CRAB/python/GetOutputServer.py

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

import traceback
from xml.dom import minidom
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 ]
        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 ]

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