CRAB/python/GetOutputServer.py

"""
Get output for server mode
"""

__revision__ = "$Id: GetOutputServer.py,v 1.32 2008/09/19 12:24:44 ewv Exp $"
__version__ = "$Revision: 1.32 $"

from GetOutput import GetOutput
from StatusServer import StatusServer
from crab_util import *
import common
import time

from ServerCommunicator import ServerCommunicator

from ProdCommon.Storage.SEAPI.SElement import SElement
from ProdCommon.Storage.SEAPI.SBinterface import SBinterface

class GetOutputServer( GetOutput, StatusServer ):

    def __init__(self, *args):

        GetOutput.__init__(self, *args)

        # init client server params...
        CliServerParams(self)

        if self.storage_path[0] != '/':
            self.storage_path = '/'+self.storage_path

        return


    def getOutput(self):

        # get updated status from server #inherited from StatusServer
        self.resynchClientSide()

        # understand whether the required output are available
        self.checkBeforeGet()

        # retrive files
        filesAndJodId = { }

        filesAndJodId.update( self.retrieveFiles(self.list_id) )
        common.logger.debug(5, "Files to be organized and notified " + str(filesAndJodId))

        self.organizeOutput()

        self.notifyRetrievalToServer(filesAndJodId)
        return

    def retrieveFiles(self, filesToRetrieve):
        """
        Real get output from server storage
        """

        self.taskuuid = str(common._db.queryTask('name'))
        common.logger.debug(3, "Task name: " + self.taskuuid)

        # create the list with the actual filenames
        remotedir = os.path.join(self.storage_path, self.taskuuid)

        # list of file to retrieve
        osbTemplate = remotedir + '/out_files_%s.tgz'
        osbFiles = [osbTemplate % str(jid) for jid in filesToRetrieve]
        if self.cfg_params['CRAB.scheduler'].lower() in ["condor_g", "glidein"]:
            osbTemplate = remotedir + '/CMSSW_%s.stdout'
            osbFiles.extend([osbTemplate % str(jid) for jid in filesToRetrieve])
            osbTemplate = remotedir + '/CMSSW_%s.stderr'
            osbFiles.extend([osbTemplate % str(jid) for jid in filesToRetrieve])
        common.logger.debug(3, "List of OSB files: " +str(osbFiles) )

        copyHere = self.outDir
        destTemplate = copyHere+'/out_files_%s.tgz'
        destFiles = [ destTemplate % str(jid) for jid in filesToRetrieve ]
        if self.cfg_params['CRAB.scheduler'].lower() in ["condor_g","glidein"]:
            destTemplate = remotedir + '/CMSSW_%s.stdout'
            destFiles.extend([destTemplate % str(jid) for jid in filesToRetrieve])
            osbTemplate = remotedir + '/CMSSW_%s.stderr'
            destFiles.extend([destTemplate % str(jid) for jid in filesToRetrieve])

        common.logger.message("Starting retrieving output from server "+str(self.storage_name)+"...")

        try:
            seEl = SElement(self.storage_name, self.storage_proto, self.storage_port)
        except Exception, ex:
            common.logger.debug(1, str(ex))
            msg = "ERROR : Unable to create SE source interface \n"
            raise CrabException(msg)
        try:
            loc = SElement("localhost", "local")
        except Exception, ex:
            common.logger.debug(1, str(ex))
            msg = "ERROR : Unable to create destination  interface \n"
            raise CrabException(msg)

        ## copy ISB ##
        sbi = SBinterface( seEl, loc )

        # retrieve them from SE
        filesAndJodId = {}
        for i in xrange(len(filesToRetrieve)):
            source = osbFiles[i]
            dest = destFiles[i]
            common.logger.debug(1, "retrieving "+ str(source) +" to "+ str(dest) )
            try:
                sbi.copy( source, dest)
                filesAndJodId[ filesToRetrieve[i] ] = dest
            except Exception, ex:
                msg = "WARNING: Unable to retrieve output file %s \n" % osbFiles[i]
                msg += str(ex)
                common.logger.debug(1,msg)
                continue

        return filesAndJodId

    def notifyRetrievalToServer(self, fileAndJobList):
        retrievedFilesJodId = []

        for jid in fileAndJobList:
            if not os.path.exists(fileAndJobList[jid]):
                # it means the file has been untarred
                retrievedFilesJodId.append(jid)

        common.logger.debug(5, "List of retrieved files notified to server: %s"%str(retrievedFilesJodId) )

        # notify to the server that output have been retrieved successfully. proxy from StatusServer
        if len(retrievedFilesJodId) > 0:
            csCommunicator = ServerCommunicator(self.server_name, self.server_port, self.cfg_params)
            try:
                csCommunicator.outputRetrieved(self.taskuuid, retrievedFilesJodId)
            except Exception, e:
                pass
        return

Revision:	1.33
Committed:	Fri Sep 19 12:31:21 2008 UTC (16 years, 7 months ago) by ewv
Content type:	text/x-python
Branch:	MAIN
CVS Tags:	CRAB_2_4_0_pre1
Changes since 1.32:	+2 -3 lines
Log Message:	n/c
#	User	Rev	Content
1	ewv	1.32	"""
2			Get output for server mode
3			"""
4
5	ewv	1.33	__revision__ = "$Id: GetOutputServer.py,v 1.32 2008/09/19 12:24:44 ewv Exp $"
6			__version__ = "$Revision: 1.32 $"
7	ewv	1.32
8	spiga	1.16	from GetOutput import GetOutput
9			from StatusServer import StatusServer
10	spiga	1.1	from crab_util import *
11			import common
12			import time
13
14	farinafa	1.14	from ServerCommunicator import ServerCommunicator
15
16	spiga	1.19	from ProdCommon.Storage.SEAPI.SElement import SElement
17			from ProdCommon.Storage.SEAPI.SBinterface import SBinterface
18	farinafa	1.14
19	spiga	1.18	class GetOutputServer( GetOutput, StatusServer ):
20	farinafa	1.14
21			def __init__(self, *args):
22	ewv	1.32
23			GetOutput.__init__(self, *args)
24	spiga	1.21
25	spiga	1.22	# init client server params...
26	ewv	1.32	CliServerParams(self)
27	farinafa	1.14
28	ewv	1.32	if self.storage_path[0] != '/':
29	farinafa	1.14	self.storage_path = '/'+self.storage_path
30	spiga	1.16
31	spiga	1.1	return
32	farinafa	1.14
33	ewv	1.32
34			def getOutput(self):
35	spiga	1.1
36	farinafa	1.14	# get updated status from server #inherited from StatusServer
37			self.resynchClientSide()
38	mcinquil	1.2
39	farinafa	1.14	# understand whether the required output are available
40			self.checkBeforeGet()
41	spiga	1.18
42			# retrive files
43	farinafa	1.29	filesAndJodId = { }
44	spiga	1.18
45	farinafa	1.29	filesAndJodId.update( self.retrieveFiles(self.list_id) )
46	ewv	1.32	common.logger.debug(5, "Files to be organized and notified " + str(filesAndJodId))
47	spiga	1.18
48	ewv	1.32	self.organizeOutput()
49	farinafa	1.29
50			self.notifyRetrievalToServer(filesAndJodId)
51	spiga	1.18	return
52
53	ewv	1.32	def retrieveFiles(self, filesToRetrieve):
54	spiga	1.18	"""
55			Real get output from server storage
56			"""
57	spiga	1.19
58			self.taskuuid = str(common._db.queryTask('name'))
59			common.logger.debug(3, "Task name: " + self.taskuuid)
60	farinafa	1.14
61	ewv	1.32	# create the list with the actual filenames
62	spiga	1.19	remotedir = os.path.join(self.storage_path, self.taskuuid)
63	ewv	1.32
64	spiga	1.19	# list of file to retrieve
65	ewv	1.32	osbTemplate = remotedir + '/out_files_%s.tgz'
66			osbFiles = [osbTemplate % str(jid) for jid in filesToRetrieve]
67			if self.cfg_params['CRAB.scheduler'].lower() in ["condor_g", "glidein"]:
68			osbTemplate = remotedir + '/CMSSW_%s.stdout'
69			osbFiles.extend([osbTemplate % str(jid) for jid in filesToRetrieve])
70			osbTemplate = remotedir + '/CMSSW_%s.stderr'
71			osbFiles.extend([osbTemplate % str(jid) for jid in filesToRetrieve])
72	spiga	1.19	common.logger.debug(3, "List of OSB files: " +str(osbFiles) )
73	ewv	1.32
74			copyHere = self.outDir
75			destTemplate = copyHere+'/out_files_%s.tgz'
76			destFiles = [ destTemplate % str(jid) for jid in filesToRetrieve ]
77			if self.cfg_params['CRAB.scheduler'].lower() in ["condor_g","glidein"]:
78			destTemplate = remotedir + '/CMSSW_%s.stdout'
79			destFiles.extend([destTemplate % str(jid) for jid in filesToRetrieve])
80			osbTemplate = remotedir + '/CMSSW_%s.stderr'
81			destFiles.extend([destTemplate % str(jid) for jid in filesToRetrieve])
82	farinafa	1.14
83	spiga	1.19	common.logger.message("Starting retrieving output from server "+str(self.storage_name)+"...")
84
85	ewv	1.32	try:
86	spiga	1.19	seEl = SElement(self.storage_name, self.storage_proto, self.storage_port)
87			except Exception, ex:
88			common.logger.debug(1, str(ex))
89			msg = "ERROR : Unable to create SE source interface \n"
90			raise CrabException(msg)
91	ewv	1.32	try:
92	spiga	1.19	loc = SElement("localhost", "local")
93			except Exception, ex:
94			common.logger.debug(1, str(ex))
95			msg = "ERROR : Unable to create destination interface \n"
96			raise CrabException(msg)
97
98			## copy ISB ##
99			sbi = SBinterface( seEl, loc )
100
101	ewv	1.32	# retrieve them from SE
102	farinafa	1.29	filesAndJodId = {}
103	ewv	1.32	for i in xrange(len(filesToRetrieve)):
104			source = osbFiles[i]
105	spiga	1.19	dest = destFiles[i]
106			common.logger.debug(1, "retrieving "+ str(source) +" to "+ str(dest) )
107	farinafa	1.15	try:
108	spiga	1.22	sbi.copy( source, dest)
109	farinafa	1.29	filesAndJodId[ filesToRetrieve[i] ] = dest
110	spiga	1.19	except Exception, ex:
111	ewv	1.32	msg = "WARNING: Unable to retrieve output file %s \n" % osbFiles[i]
112	spiga	1.20	msg += str(ex)
113			common.logger.debug(1,msg)
114	farinafa	1.15	continue
115	farinafa	1.29
116			return filesAndJodId
117
118			def notifyRetrievalToServer(self, fileAndJobList):
119	ewv	1.32	retrievedFilesJodId = []
120	farinafa	1.29
121			for jid in fileAndJobList:
122			if not os.path.exists(fileAndJobList[jid]):
123			# it means the file has been untarred
124			retrievedFilesJodId.append(jid)
125
126			common.logger.debug(5, "List of retrieved files notified to server: %s"%str(retrievedFilesJodId) )
127	farinafa	1.14
128			# notify to the server that output have been retrieved successfully. proxy from StatusServer
129	farinafa	1.23	if len(retrievedFilesJodId) > 0:
130	spiga	1.22	csCommunicator = ServerCommunicator(self.server_name, self.server_port, self.cfg_params)
131	farinafa	1.28	try:
132			csCommunicator.outputRetrieved(self.taskuuid, retrievedFilesJodId)
133			except Exception, e:
134			pass
135	farinafa	1.14	return
136	spiga	1.1