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 ServerConfig import *


class GetOutputServer( GetOutput, StatusServer ):

    def __init__(self, *args):
        self.cfg_params = args[0]
        self.jobs = args[1]
 
        self.server_name = None
        self.server_port = None

        self.storage_name = None
        self.storage_path = None
        self.storage_proto = None
        self.storage_port = None

        self.srvCfg = {}

        try:
            self.srvCfg = ServerConfig(self.cfg_params['CRAB.server_name']).config()
            self.server_name = str(self.srvCfg['serverName'])
            self.server_port = int(self.srvCfg['serverPort'])
            self.storage_name = str(self.srvCfg['storageName'])
            self.storage_path = str(self.srvCfg['storagePath'])
            self.storage_proto = str(self.srvCfg['storageProtocol'])
            self.storage_port = str(self.srvCfg['storagePort'])
        except KeyError:
            msg = 'No server selected or port specified.'
            msg = msg + 'Please specify a server in the crab cfg file'
            raise CrabException(msg)

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

        self.outDir = common.work_space.resDir()
        self.logDir = common.work_space.resDir()
        self.return_data = self.cfg_params.get('USER.return_data',0)

        self.possible_status = {
                         'UN': 'Unknown',
                         'SU': 'Submitted',
                         'SW': 'Waiting',
                         'SS': 'Scheduled',
                         'R': 'Running',
                         'SD': 'Done',
                         'SK': 'Killed',
                         'SA': 'Aborted',
                         'SE': 'Cleared',
                         'E': 'Cleared'
                         }
        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
        self.retrieveFiles(self.list_id) 

        self.organizeOutput()    

        return

    def retrieveFiles(self,filesToRetrieve):
        """
        Real get output from server storage
        """
        common.scheduler.checkProxy()

        # create the list with the actual filenames 
        taskuuid = str(common._db.queryTask('name'))
        remotedir = os.path.join(self.storage_path, taskuuid)

        osbTemplate = self.storage_proto + '://'+ self.storage_name +\
            ':' + self.storage_port + remotedir + '/out_files_%s.tgz'  
        osbFiles = [ osbTemplate%str(jid) for jid in filesToRetrieve ]

        copyHere = common.work_space.resDir()
        destTemplate = 'file://'+copyHere+'/out_files_%s.tgz'  
        destFiles = [ destTemplate%str(jid) for jid in filesToRetrieve ]

        # retrieve them from SE #TODO replace this with SE-API
        for i in xrange(len(filesToRetrieve)):
            try:
                cmd = 'lcg-cp --vo cms %s %s'%(osbFiles[i], destFiles[i])
                out = os.system(cmd +' >& /dev/null')
                common.logger.debug(5, cmd)
                if out != 0:
                    print "Unable to retrieve output file %s"%osbFiles[i]
                    del filesToRetrieve[i]
                    continue

                ##    
                ## TODO check if sizes are ok/the transfer was safe. Simpler with the API
                ## 
 
                ## clean-up SE
               # cmd = 'lcg-del --vo cms %s'%osbFiles[i]
               # out = os.system(cmd +' >& /dev/null')
               # common.logger.debug(5, cmd)
               # if out != 0:
               #     print "Unable to clean up SE for output file %s. The deletion will be performed by the server"%osbFiles[i]
               #     continue

            except Exception, e:
                print "Unable to retrieve output file %s"%osbFiles[i] 
                del filesToRetrieve[i]
                continue

        # notify to the server that output have been retrieved successfully. proxy from StatusServer
        if len(filesToRetrieve) > 0:
            common.logger.debug(5, "List of retrieved files notified to server: %s"%str(filesToRetrieve) ) 
            csCommunicator = ServerCommunicator(self.server_name, self.server_port, self.cfg_params, self.proxy)
            csCommunicator.outputRetrieved(taskuuid, filesToRetrieve)

        return


Revision:	1.18
Committed:	Tue Apr 15 14:20:18 2008 UTC (17 years ago) by spiga
Content type:	text/x-python
Branch:	MAIN
CVS Tags:	CRAB_2_2_0_pre7
Changes since 1.17:	+36 -28 lines
Log Message:	many improvement on getOutput. Both on client and standalone classes
#	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			from ServerConfig import *
11
12
13	spiga	1.18	class GetOutputServer( GetOutput, StatusServer ):
14	farinafa	1.14
15			def __init__(self, *args):
16			self.cfg_params = args[0]
17			self.jobs = args[1]
18
19			self.server_name = None
20			self.server_port = None
21
22			self.storage_name = None
23			self.storage_path = None
24			self.storage_proto = None
25			self.storage_port = None
26	mcinquil	1.2
27	farinafa	1.14	self.srvCfg = {}
28
29			try:
30			self.srvCfg = ServerConfig(self.cfg_params['CRAB.server_name']).config()
31			self.server_name = str(self.srvCfg['serverName'])
32			self.server_port = int(self.srvCfg['serverPort'])
33			self.storage_name = str(self.srvCfg['storageName'])
34			self.storage_path = str(self.srvCfg['storagePath'])
35			self.storage_proto = str(self.srvCfg['storageProtocol'])
36			self.storage_port = str(self.srvCfg['storagePort'])
37	spiga	1.3	except KeyError:
38	farinafa	1.14	msg = 'No server selected or port specified.'
39			msg = msg + 'Please specify a server in the crab cfg file'
40			raise CrabException(msg)
41
42			if self.storage_path[0]!='/':
43			self.storage_path = '/'+self.storage_path
44	spiga	1.16
45	slacapra	1.17	self.outDir = common.work_space.resDir()
46			self.logDir = common.work_space.resDir()
47	spiga	1.16	self.return_data = self.cfg_params.get('USER.return_data',0)
48
49	spiga	1.18	self.possible_status = {
50			'UN': 'Unknown',
51			'SU': 'Submitted',
52			'SW': 'Waiting',
53			'SS': 'Scheduled',
54			'R': 'Running',
55			'SD': 'Done',
56			'SK': 'Killed',
57			'SA': 'Aborted',
58			'SE': 'Cleared',
59			'E': 'Cleared'
60			}
61	spiga	1.1	return
62	farinafa	1.14
63	spiga	1.18
64			def getOutput(self):
65	spiga	1.1
66	farinafa	1.14	# get updated status from server #inherited from StatusServer
67			self.resynchClientSide()
68	mcinquil	1.2
69	farinafa	1.14	# understand whether the required output are available
70			self.checkBeforeGet()
71	spiga	1.18
72			# retrive files
73			self.retrieveFiles(self.list_id)
74
75			self.organizeOutput()
76
77			return
78
79			def retrieveFiles(self,filesToRetrieve):
80			"""
81			Real get output from server storage
82			"""
83			common.scheduler.checkProxy()
84	farinafa	1.14
85			# create the list with the actual filenames
86			taskuuid = str(common._db.queryTask('name'))
87			remotedir = os.path.join(self.storage_path, taskuuid)
88
89			osbTemplate = self.storage_proto + '://'+ self.storage_name +\
90	spiga	1.18	':' + self.storage_port + remotedir + '/out_files_%s.tgz'
91	farinafa	1.14	osbFiles = [ osbTemplate%str(jid) for jid in filesToRetrieve ]
92
93	slacapra	1.17	copyHere = common.work_space.resDir()
94	spiga	1.18	destTemplate = 'file://'+copyHere+'/out_files_%s.tgz'
95	farinafa	1.14	destFiles = [ destTemplate%str(jid) for jid in filesToRetrieve ]
96
97			# retrieve them from SE #TODO replace this with SE-API
98			for i in xrange(len(filesToRetrieve)):
99	farinafa	1.15	try:
100			cmd = 'lcg-cp --vo cms %s %s'%(osbFiles[i], destFiles[i])
101			out = os.system(cmd +' >& /dev/null')
102			common.logger.debug(5, cmd)
103			if out != 0:
104			print "Unable to retrieve output file %s"%osbFiles[i]
105			del filesToRetrieve[i]
106			continue
107
108			##
109			## TODO check if sizes are ok/the transfer was safe. Simpler with the API
110			##
111
112			## clean-up SE
113	spiga	1.18	# cmd = 'lcg-del --vo cms %s'%osbFiles[i]
114			# out = os.system(cmd +' >& /dev/null')
115			# common.logger.debug(5, cmd)
116			# if out != 0:
117			# print "Unable to clean up SE for output file %s. The deletion will be performed by the server"%osbFiles[i]
118			# continue
119	farinafa	1.15
120			except Exception, e:
121			print "Unable to retrieve output file %s"%osbFiles[i]
122			del filesToRetrieve[i]
123			continue
124	farinafa	1.14
125			# notify to the server that output have been retrieved successfully. proxy from StatusServer
126	farinafa	1.15	if len(filesToRetrieve) > 0:
127			common.logger.debug(5, "List of retrieved files notified to server: %s"%str(filesToRetrieve) )
128			csCommunicator = ServerCommunicator(self.server_name, self.server_port, self.cfg_params, self.proxy)
129			csCommunicator.outputRetrieved(taskuuid, filesToRetrieve)
130	spiga	1.1
131	farinafa	1.14	return
132	spiga	1.1
133