CRAB/python/SchedulerLocal.py

from Scheduler import Scheduler
from crab_exceptions import *
from crab_util import getLocalDomain
import common
from PhEDExDatasvcInfo import PhEDExDatasvcInfo

import os,string

# Base class for all local scheduler

class SchedulerLocal(Scheduler) :

    def configure(self, cfg_params):
        self.environment_unique_identifier = None
        self.cfg_params = cfg_params
        Scheduler.configure(self,cfg_params)
        self.jobtypeName = cfg_params['CRAB.jobtype']

        name=string.upper(self.name())
        self.queue = cfg_params.get(name+'.queue',None)

        self.res = cfg_params.get(name+'.resource',None)

        # minimal padding time for jobs. For local schedulers is disabled.
        # Added for alignment purpose only (and for test) with Grid schedulers
        self.minimal_job_duration = 0

        if (cfg_params.has_key(self.name()+'.env_id')): self.environment_unique_identifier = cfg_params[self.name()+'.env_id']
        ## is this ok?
        localDomainName = getLocalDomain(self)
        if not cfg_params.has_key('GRID.se_white_list'):
            cfg_params['GRID.se_white_list']=localDomainName
            common.logger.info("Your domain name is "+str(localDomainName)+": only local dataset will be considered")
        else:
            common.logger.info("Your se_white_list is set to "+str(cfg_params['GRID.se_white_list'])+": only local dataset will be considered")
        return

    def userName(self):
        """ return the user name """
        import pwd,getpass
        tmp=pwd.getpwnam(getpass.getuser())[4]
        return "/CN="+tmp.strip()

    def envUniqueID(self):
        return

    def wsSetupEnvironment(self):
        """
        Returns part of a job script which does scheduler-specific work.
        """
        taskId = common._db.queryTask('name')
        if not self.environment_unique_identifier:
            try :
                self.environment_unique_identifier = self.envUniqueID()
            except :
                raise CrabException('environment_unique_identifier not set')
        index = int(common._db.nJobs())
        job = common.job_list[index-1]
        jbt = job.type()
        # start with wrapper timing
        txt  = 'export TIME_WRAP_INI=`date +%s` \n'
        txt += 'export TIME_STAGEOUT=-2 \n\n'

        txt += '# '+self.name()+' specific stuff\n'
        txt += '# strip arguments\n'
        txt += 'echo "strip arguments"\n'
        txt += 'args=("$@")\n'
        txt += 'nargs=$#\n'
        txt += 'shift $nargs\n'
        txt += "# job number (first parameter for job wrapper)\n"
        txt += "NJob=${args[0]}; export NJob\n"
        txt += "NResub=${args[1]}; export NResub\n"
        txt += "NRand=`getRandSeed`; export NRand\n"
        txt += 'OutUniqueID=_$NRand\n'
        txt += 'OutUniqueID=_$NResub$OutUniqueID\n'
        txt += 'OutUniqueID=$NJob$OutUniqueID; export OutUniqueID\n'

        txt += "out_files=out_files_${NJob}; export out_files\n"
        txt += "echo $out_files\n"
        txt += jbt.outList()

        txt += 'SyncGridJobId=`echo '+self.environment_unique_identifier+'`\n'
        txt += 'MonitorJobID=`echo ${NJob}_${SyncGridJobId}`\n'
        txt += 'MonitorID=`echo ' + taskId + '`\n'

        txt += 'echo "MonitorJobID=`echo $MonitorJobID`" | tee -a $RUNTIME_AREA/$repo \n'
        txt += 'echo "SyncGridJobId=`echo $SyncGridJobId`" | tee -a $RUNTIME_AREA/$repo \n'
        txt += 'echo "MonitorID=`echo $MonitorID`" | tee -a $RUNTIME_AREA/$repo\n'
        txt += 'echo "SyncCE='+self.name()+'.`hostname -d`" | tee -a $RUNTIME_AREA/$repo \n'

        txt += 'middleware='+self.name().upper()+' \n'

        txt += 'dumpStatus $RUNTIME_AREA/$repo \n'

        txt += 'InputSandBox=${args[3]}\n'

        txt += '\n\n'

        return txt

    def wsCopyOutput_comm(self, pool=None):
        """
        Write a CopyResults part of a job script, e.g.
        to copy produced output into a storage element.
        """
        index = int(common._db.nJobs())
        job = common.job_list[index-1]
        jbt = job.type()
        txt = '\n'
        if int(self.copy_data) == 1:

            stageout = PhEDExDatasvcInfo(self.cfg_params)
            endpoint, lfn, SE, SE_PATH, user = stageout.getEndpoint()
            print "endpoint = ", endpoint

            ########################################################
            ##################### FEDE FOR CAF #####################
            cmscp_args = '' 
            if  common.scheduler.name().upper() == 'CAF':
                    if endpoint.find('root:') != -1:
                        SE_PATH = '/' + endpoint.split('//')[2]
                        caf_protocol = 'xrootd'
                        cmscp_args += '--protocol %s '%caf_protocol
                    elif endpoint.find('rfio:') != -1:
                        if endpoint.find('path=') != -1:
                            SE_PATH = endpoint.split('path=')[1]
                        else:
                            SE_PATH = endpoint
                        caf_protocol = 'rfio'
                        cmscp_args += '--protocol %s '%caf_protocol
            ########################################################            
            ########################################################            

            if self.check_RemoteDir == 1 :
                self.checkRemoteDir(endpoint,jbt.outList('list') )

            txt += '#\n'
            txt += '# COPY OUTPUT FILE TO '+SE_PATH+ '\n'
            txt += '#\n\n'

            txt += 'export SE='+SE+'\n'
            txt += 'echo "SE = $SE"\n'
            txt += 'export SE_PATH='+SE_PATH+'\n'
            txt += 'echo "SE_PATH = $SE_PATH"\n'
            txt += 'export LFNBaseName='+lfn+'\n'
            txt += 'echo "LFNBaseName = $LFNBaseName"\n'
            txt += 'export USER='+user+'\n'
            txt += 'echo "USER = $USER"\n'
            #txt += 'export endpoint='+"'"+endpoint+"'"+'\n'
            txt += 'export endpoint='+endpoint+'\n'
            txt += 'echo "endpoint = $endpoint"\n'
            
            if (pool) and (pool != 'None'):
                txt += 'export STAGE_SVCCLASS='+str(pool)+'\n'

            txt += 'echo ">>> Copy output files from WN = `hostname` to $SE_PATH :"\n'
            txt += 'export TIME_STAGEOUT_INI=`date +%s` \n'
            txt += 'copy_exit_status=0\n'
            #### FEDE added += ######### 
            cmscp_args += ' --destination $endpoint --inputFileList $file_list'
            #######################################################
            #######################################################
            cmscp_args +=' --middleware $middleware --se_name $SE --for_lfn $LFNBaseName %s %s '%(self.loc_stage_out,self.debugWrap)
            txt += 'echo "python cmscp.py %s "\n'%cmscp_args
            txt += 'python cmscp.py %s \n'%cmscp_args
            if self.debug_wrapper==1:
                txt += 'echo "########### details of SE interaction"\n'
                txt += 'if [ -f .SEinteraction.log ] ;then\n'
                txt += '    cat .SEinteraction.log\n'
                txt += 'else\n'
                txt += '    echo ".SEinteraction.log file not found"\n'
                txt += 'fi\n'
                txt += 'echo "#####################################"\n'
            txt += 'if [ -f $RUNTIME_AREA/resultCopyFile ] ;then\n'
            txt += '    cat $RUNTIME_AREA/resultCopyFile\n'
            txt += '    pwd\n'
            txt += 'else\n'
            ### to avoid some 70500 error ....
            txt += '    echo "ERROR ==> $RUNTIME_AREA/resultCopyFile file not found. Problem during the stageout"\n'
            txt += '    echo "RUNTIME_AREA content: "\n'
            txt += '    ls $RUNTIME_AREA \n'
            txt += '    job_exit_code=60318\n'
            txt += '    func_exit \n'
            txt += 'fi\n'
            ##########################################

            txt += 'if [ -f ${RUNTIME_AREA}/cmscpReport.sh ] ;then\n'
            txt += '    echo "-------- cat ${RUNTIME_AREA}/cmscpReport.sh "\n'
            txt += '    cat ${RUNTIME_AREA}/cmscpReport.sh\n'
            txt += '    echo "-------- end of ${RUNTIME_AREA}/cmscpReport.sh "\n'
            txt += '    source ${RUNTIME_AREA}/cmscpReport.sh\n'
            txt += '    source_result=$? \n'
            txt += '    if [ $source_result -ne 0 ]; then\n'
            txt += '        echo "problem with the source of cmscpReport.sh file"\n'
            txt += '        StageOutExitStatus=60307\n'
            txt += '    fi\n'
            txt += 'else\n'
            txt += '    echo "cmscpReport.sh file not found"\n'
            txt += '    StageOutExitStatus=60307\n'
            txt += 'fi\n'
            txt += 'if [ $StageOutExitStatus -ne 0 ]; then\n'
            txt += '    echo "Problem copying file to $SE $SE_PATH"\n'
            txt += '    copy_exit_status=$StageOutExitStatus \n'
            if not self.debug_wrapper==1:
                txt += 'if [ -f .SEinteraction.log ] ;then\n'
                txt += '    echo "########## contents of SE interaction"\n'
                txt += '    cat .SEinteraction.log\n'
                txt += '    echo "#####################################"\n'
                txt += 'else\n'
                txt += '    echo ".SEinteraction.log file not found"\n'
                txt += 'fi\n'
            txt += '    job_exit_code=$StageOutExitStatus\n'
            txt += 'fi\n'
            txt += 'export TIME_STAGEOUT_END=`date +%s` \n'
            txt += 'let "TIME_STAGEOUT = TIME_STAGEOUT_END - TIME_STAGEOUT_INI" \n'
        else:
            txt += 'export TIME_STAGEOUT=-1 \n'
        return txt
Revision:	1.69
Committed:	Fri Sep 7 20:27:40 2012 UTC (12 years, 7 months ago) by belforte
Content type:	text/x-python
Branch:	MAIN
CVS Tags:	CRAB_2_9_1, CRAB_2_9_1_pre2, CRAB_2_9_1_pre1, CRAB_2_9_0, CRAB_2_9_0_pre2, CRAB_2_9_0_pre1, CRAB_2_8_8, CRAB_2_8_8_pre1, CRAB_2_8_7_patch3, CRAB_2_8_7_patch2, CRAB_2_8_7_patch1, CRAB_2_8_7, CRAB_2_8_7_pre2, CRAB_2_8_7_pre1, CRAB_2_8_6, CRAB_2_8_6_pre1, CRAB_2_8_5_patch3, CRAB_2_8_5_patch2, CRAB_2_8_5_patch1, CRAB_2_8_5, CRAB_2_8_5_pre5, CRAB_2_8_5_pre4, CRAB_2_8_5_pre3, CRAB_2_8_4_patch3, CRAB_2_8_5_pre2, CRAB_2_8_4_patch2, CRAB_2_8_5_pre1, CRAB_2_8_4_patch1, CRAB_2_8_4, CRAB_2_8_4_pre5, CRAB_2_8_4_pre4, CRAB_2_8_4_pre3, CRAB_2_8_4_pre2, CRAB_2_8_4_pre1, CRAB_2_8_3, CRAB_2_8_3_pre4, CRAB_2_8_3_pre3, CRAB_2_8_3_pre2, CRAB_2_8_3_pre1, HEAD
Changes since 1.68:	+5 -3 lines
Log Message:	use full path when sourcing cmscpReport.sh, fixes https://savannah.cern.ch/bugs/?97065
#	User	Rev	Content
1	slacapra	1.1	from Scheduler import Scheduler
2			from crab_exceptions import *
3	spiga	1.47	from crab_util import getLocalDomain
4	slacapra	1.1	import common
5	spiga	1.27	from PhEDExDatasvcInfo import PhEDExDatasvcInfo
6	slacapra	1.1
7			import os,string
8
9	slacapra	1.2	# Base class for all local scheduler
10	slacapra	1.1
11			class SchedulerLocal(Scheduler) :
12
13			def configure(self, cfg_params):
14	spiga	1.47	self.environment_unique_identifier = None
15	spiga	1.27	self.cfg_params = cfg_params
16	spiga	1.46	Scheduler.configure(self,cfg_params)
17	slacapra	1.1	self.jobtypeName = cfg_params['CRAB.jobtype']
18
19			name=string.upper(self.name())
20			self.queue = cfg_params.get(name+'.queue',None)
21
22			self.res = cfg_params.get(name+'.resource',None)
23
24	ewv	1.58	# minimal padding time for jobs. For local schedulers is disabled.
25			# Added for alignment purpose only (and for test) with Grid schedulers
26	farinafa	1.55	self.minimal_job_duration = 0
27	farinafa	1.53
28	slacapra	1.1	if (cfg_params.has_key(self.name()+'.env_id')): self.environment_unique_identifier = cfg_params[self.name()+'.env_id']
29			## is this ok?
30	spiga	1.25	localDomainName = getLocalDomain(self)
31	spiga	1.52	if not cfg_params.has_key('GRID.se_white_list'):
32			cfg_params['GRID.se_white_list']=localDomainName
33	spiga	1.51	common.logger.info("Your domain name is "+str(localDomainName)+": only local dataset will be considered")
34	slacapra	1.16	else:
35	spiga	1.52	common.logger.info("Your se_white_list is set to "+str(cfg_params['GRID.se_white_list'])+": only local dataset will be considered")
36	slacapra	1.1	return
37
38			def userName(self):
39			""" return the user name """
40	slacapra	1.5	import pwd,getpass
41			tmp=pwd.getpwnam(getpass.getuser())[4]
42			return "/CN="+tmp.strip()
43	slacapra	1.1
44	ewv	1.50	def envUniqueID(self):
45	spiga	1.47	return
46
47	slacapra	1.1	def wsSetupEnvironment(self):
48			"""
49			Returns part of a job script which does scheduler-specific work.
50			"""
51	spiga	1.47	taskId = common._db.queryTask('name')
52	slacapra	1.1	if not self.environment_unique_identifier:
53	spiga	1.47	try :
54	ewv	1.50	self.environment_unique_identifier = self.envUniqueID()
55			except :
56	spiga	1.47	raise CrabException('environment_unique_identifier not set')
57	slacapra	1.12	index = int(common._db.nJobs())
58			job = common.job_list[index-1]
59			jbt = job.type()
60	ewv	1.20	# start with wrapper timing
61	spiga	1.18	txt = 'export TIME_WRAP_INI=`date +%s` \n'
62	farinafa	1.22	txt += 'export TIME_STAGEOUT=-2 \n\n'
63	spiga	1.17
64	spiga	1.19	txt += '# '+self.name()+' specific stuff\n'
65	slacapra	1.1	txt += '# strip arguments\n'
66			txt += 'echo "strip arguments"\n'
67			txt += 'args=("$@")\n'
68			txt += 'nargs=$#\n'
69			txt += 'shift $nargs\n'
70			txt += "# job number (first parameter for job wrapper)\n"
71	ewv	1.9	txt += "NJob=${args[0]}; export NJob\n"
72	farinafa	1.57	txt += "NResub=${args[1]}; export NResub\n"
73	farinafa	1.60	txt += "NRand=`getRandSeed`; export NRand\n"
74			txt += 'OutUniqueID=_$NRand\n'
75			txt += 'OutUniqueID=_$NResub$OutUniqueID\n'
76	farinafa	1.57	txt += 'OutUniqueID=$NJob$OutUniqueID; export OutUniqueID\n'
77	slacapra	1.1
78	slacapra	1.12	txt += "out_files=out_files_${NJob}; export out_files\n"
79			txt += "echo $out_files\n"
80			txt += jbt.outList()
81
82	slacapra	1.6	txt += 'SyncGridJobId=`echo '+self.environment_unique_identifier+'`\n'
83			txt += 'MonitorJobID=`echo ${NJob}_${SyncGridJobId}`\n'
84	spiga	1.38	txt += 'MonitorID=`echo ' + taskId + '`\n'
85	slacapra	1.1
86			txt += 'echo "MonitorJobID=`echo $MonitorJobID`" \| tee -a $RUNTIME_AREA/$repo \n'
87			txt += 'echo "SyncGridJobId=`echo $SyncGridJobId`" \| tee -a $RUNTIME_AREA/$repo \n'
88			txt += 'echo "MonitorID=`echo $MonitorID`" \| tee -a $RUNTIME_AREA/$repo\n'
89	slacapra	1.7	txt += 'echo "SyncCE='+self.name()+'.`hostname -d`" \| tee -a $RUNTIME_AREA/$repo \n'
90	spiga	1.27
91	fanzago	1.23	txt += 'middleware='+self.name().upper()+' \n'
92	slacapra	1.1
93			txt += 'dumpStatus $RUNTIME_AREA/$repo \n'
94
95	slacapra	1.12	txt += 'InputSandBox=${args[3]}\n'
96
97	slacapra	1.1	txt += '\n\n'
98
99			return txt
100
101	spiga	1.27	def wsCopyOutput_comm(self, pool=None):
102	slacapra	1.3	"""
103			Write a CopyResults part of a job script, e.g.
104			to copy produced output into a storage element.
105			"""
106	spiga	1.37	index = int(common._db.nJobs())
107			job = common.job_list[index-1]
108			jbt = job.type()
109	slacapra	1.8	txt = '\n'
110	spiga	1.27	if int(self.copy_data) == 1:
111	slacapra	1.3
112	spiga	1.27	stageout = PhEDExDatasvcInfo(self.cfg_params)
113			endpoint, lfn, SE, SE_PATH, user = stageout.getEndpoint()
114	fanzago	1.65	print "endpoint = ", endpoint
115
116			########################################################
117			##################### FEDE FOR CAF #####################
118			cmscp_args = ''
119			if common.scheduler.name().upper() == 'CAF':
120			if endpoint.find('root:') != -1:
121			SE_PATH = '/' + endpoint.split('//')[2]
122			caf_protocol = 'xrootd'
123			cmscp_args += '--protocol %s '%caf_protocol
124			elif endpoint.find('rfio:') != -1:
125	fanzago	1.66	if endpoint.find('path=') != -1:
126			SE_PATH = endpoint.split('path=')[1]
127			else:
128			SE_PATH = endpoint
129	fanzago	1.65	caf_protocol = 'rfio'
130			cmscp_args += '--protocol %s '%caf_protocol
131			########################################################
132			########################################################
133
134	ewv	1.50	if self.check_RemoteDir == 1 :
135	spiga	1.37	self.checkRemoteDir(endpoint,jbt.outList('list') )
136	slacapra	1.3
137	spiga	1.27	txt += '#\n'
138			txt += '# COPY OUTPUT FILE TO '+SE_PATH+ '\n'
139			txt += '#\n\n'
140
141	fanzago	1.35	txt += 'export SE='+SE+'\n'
142	spiga	1.27	txt += 'echo "SE = $SE"\n'
143			txt += 'export SE_PATH='+SE_PATH+'\n'
144			txt += 'echo "SE_PATH = $SE_PATH"\n'
145			txt += 'export LFNBaseName='+lfn+'\n'
146	spiga	1.33	txt += 'echo "LFNBaseName = $LFNBaseName"\n'
147	spiga	1.27	txt += 'export USER='+user+'\n'
148	spiga	1.33	txt += 'echo "USER = $USER"\n'
149	fanzago	1.67	#txt += 'export endpoint='+"'"+endpoint+"'"+'\n'
150			txt += 'export endpoint='+endpoint+'\n'
151	spiga	1.33	txt += 'echo "endpoint = $endpoint"\n'
152	fanzago	1.65
153	fanzago	1.36	if (pool) and (pool != 'None'):
154	spiga	1.27	txt += 'export STAGE_SVCCLASS='+str(pool)+'\n'
155	ewv	1.50
156	spiga	1.27	txt += 'echo ">>> Copy output files from WN = `hostname` to $SE_PATH :"\n'
157			txt += 'export TIME_STAGEOUT_INI=`date +%s` \n'
158			txt += 'copy_exit_status=0\n'
159	fanzago	1.65	#### FEDE added += #########
160			cmscp_args += ' --destination $endpoint --inputFileList $file_list'
161			#######################################################
162			#######################################################
163	fanzago	1.59	cmscp_args +=' --middleware $middleware --se_name $SE --for_lfn $LFNBaseName %s %s '%(self.loc_stage_out,self.debugWrap)
164	spiga	1.46	txt += 'echo "python cmscp.py %s "\n'%cmscp_args
165			txt += 'python cmscp.py %s \n'%cmscp_args
166	ewv	1.50	if self.debug_wrapper==1:
167	fanzago	1.62	txt += 'echo "########### details of SE interaction"\n'
168	fanzago	1.43	txt += 'if [ -f .SEinteraction.log ] ;then\n'
169	spiga	1.39	txt += ' cat .SEinteraction.log\n'
170			txt += 'else\n'
171			txt += ' echo ".SEinteraction.log file not found"\n'
172	spiga	1.40	txt += 'fi\n'
173	fanzago	1.62	txt += 'echo "#####################################"\n'
174	spiga	1.63	txt += 'if [ -f $RUNTIME_AREA/resultCopyFile ] ;then\n'
175			txt += ' cat $RUNTIME_AREA/resultCopyFile\n'
176	fanzago	1.62	txt += ' pwd\n'
177			txt += 'else\n'
178	fanzago	1.65	### to avoid some 70500 error ....
179	spiga	1.63	txt += ' echo "ERROR ==> $RUNTIME_AREA/resultCopyFile file not found. Problem during the stageout"\n'
180	fanzago	1.64	txt += ' echo "RUNTIME_AREA content: "\n'
181			txt += ' ls $RUNTIME_AREA \n'
182	fanzago	1.62	txt += ' job_exit_code=60318\n'
183			txt += ' func_exit \n'
184			txt += 'fi\n'
185			##########################################
186
187	belforte	1.69	txt += 'if [ -f ${RUNTIME_AREA}/cmscpReport.sh ] ;then\n'
188			txt += ' echo "-------- cat ${RUNTIME_AREA}/cmscpReport.sh "\n'
189			txt += ' cat ${RUNTIME_AREA}/cmscpReport.sh\n'
190			txt += ' echo "-------- end of ${RUNTIME_AREA}/cmscpReport.sh "\n'
191			txt += ' source ${RUNTIME_AREA}/cmscpReport.sh\n'
192	fanzago	1.61	txt += ' source_result=$? \n'
193			txt += ' if [ $source_result -ne 0 ]; then\n'
194			txt += ' echo "problem with the source of cmscpReport.sh file"\n'
195			txt += ' StageOutExitStatus=60307\n'
196			txt += ' fi\n'
197	spiga	1.40	txt += 'else\n'
198	ewv	1.50	txt += ' echo "cmscpReport.sh file not found"\n'
199	spiga	1.40	txt += ' StageOutExitStatus=60307\n'
200			txt += 'fi\n'
201	spiga	1.27	txt += 'if [ $StageOutExitStatus -ne 0 ]; then\n'
202			txt += ' echo "Problem copying file to $SE $SE_PATH"\n'
203			txt += ' copy_exit_status=$StageOutExitStatus \n'
204	ewv	1.50	if not self.debug_wrapper==1:
205	fanzago	1.62	txt += 'if [ -f .SEinteraction.log ] ;then\n'
206			txt += ' echo "########## contents of SE interaction"\n'
207			txt += ' cat .SEinteraction.log\n'
208			txt += ' echo "#####################################"\n'
209			txt += 'else\n'
210			txt += ' echo ".SEinteraction.log file not found"\n'
211			txt += 'fi\n'
212	spiga	1.27	txt += ' job_exit_code=$StageOutExitStatus\n'
213			txt += 'fi\n'
214			txt += 'export TIME_STAGEOUT_END=`date +%s` \n'
215			txt += 'let "TIME_STAGEOUT = TIME_STAGEOUT_END - TIME_STAGEOUT_INI" \n'
216			else:
217			txt += 'export TIME_STAGEOUT=-1 \n'
218	slacapra	1.3	return txt