CRAB/python/SchedulerLocal.py

from Scheduler import Scheduler
from crab_exceptions import *
from crab_logger import Logger
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)

        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('EDG.se_white_list'):
            cfg_params['EDG.se_white_list']=localDomainName
            common.logger.message("Your domain name is "+str(localDomainName)+": only local dataset will be considered")
        else:
            common.logger.message("Your se_white_list is set to "+str(cfg_params['EDG.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 Env_uniqueId(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.Env_uniqueId()
            except : 
                raise CrabException('environment_unique_identifier not set')
        print self.environment_unique_identifier  
        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 += "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()
            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 += 'echo "endpoint = $endpoint"\n'
            ### Needed i.e. for caf
            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'
            cmscp_args = '--destination $endpoint --inputFileList $file_list \
                          --middleware $middleware --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 += 'if [ -f .SEinteraction.log ] ;then\n'
                txt += '########### details of SE interaction\n'
                txt += '    cat .SEinteraction.log\n'
                txt += 'else\n'
                txt += '    echo ".SEinteraction.log file not found"\n'
                txt += 'fi\n'
                txt += '##################################### \n'
            txt += 'if [ -f cmscpReport.sh ] ;then\n'
            txt += '    cat cmscpReport.sh\n'
            txt += '    source cmscpReport.sh\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 += '    ########### 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 += '    ##################################### \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

    def wsExitFunc_comm(self):
        """
        """
        txt = ''
        txt += '    if [ $PYTHONPATH ]; then \n'
        txt += '       if [ ! -s $RUNTIME_AREA/fillCrabFjr.py ]; then \n'
        txt += '           echo "WARNING: it is not possible to create crab_fjr.xml to final report" \n'
        txt += '       else \n'
        txt += '           python $RUNTIME_AREA/fillCrabFjr.py $RUNTIME_AREA/crab_fjr_$NJob.xml --errorcode $job_exit_code $executable_exit_status \n'
        txt += '       fi\n'
        txt += '    fi\n'
        txt += '    cd $RUNTIME_AREA  \n'   
        txt += '    for file in $filesToCheck ; do\n'
        txt += '        if [ -e $file ]; then\n'
        txt += '            echo "tarring file $file in  $out_files"\n'
        txt += '        else\n'
        txt += '            echo "WARNING: output file $file not found!"\n'
        txt += '        fi\n'
        txt += '    done\n'
        txt += '    export TIME_WRAP_END=`date +%s`\n'
        txt += '    let "TIME_WRAP = TIME_WRAP_END - TIME_WRAP_END_INI" \n'
        txt += '    if [ $PYTHONPATH ]; then \n'
        txt += '       if [ ! -s $RUNTIME_AREA/fillCrabFjr.py ]; then \n'
        txt += '           echo "WARNING: it is not possible to create crab_fjr.xml to final report" \n'
        txt += '       else \n'
        # call timing FJR filling
        txt += '           python $RUNTIME_AREA/fillCrabFjr.py $RUNTIME_AREA/crab_fjr_$NJob.xml --timing $TIME_WRAP $TIME_EXE $TIME_STAGEOUT \n'
        txt += '           echo "CrabWrapperTime=$TIME_WRAP" >> $RUNTIME_AREA/$repo \n' 
        txt += '           if [ $TIME_STAGEOUT -lt 0 ]; then \n'
        txt += '               export TIME_STAGEOUT=NULL \n'
        txt += '           fi\n'
        txt += '           echo "CrabStageoutTime=$TIME_STAGEOUT" >> $RUNTIME_AREA/$repo \n'
        txt += '       fi\n'
        txt += '    fi\n'
        txt += '    dumpStatus $RUNTIME_AREA/$repo \n\n'
        txt += '    echo "JOB_EXIT_STATUS = $job_exit_code"\n'
        txt += '    echo "JobExitCode=$job_exit_code" >> $RUNTIME_AREA/$repo\n'
        txt += '    dumpStatus $RUNTIME_AREA/$repo\n'

        return txt

Revision:	1.47
Committed:	Sat Mar 7 16:40:03 2009 UTC (16 years, 1 month ago) by spiga
Content type:	text/x-python
Branch:	MAIN
Changes since 1.46:	+11 -6 lines
Log Message:	adapt to task name changes.
#	User	Rev	Content
1	slacapra	1.1	from Scheduler import Scheduler
2			from crab_exceptions import *
3			from crab_logger import Logger
4	spiga	1.47	from crab_util import getLocalDomain
5	slacapra	1.1	import common
6	spiga	1.27	from PhEDExDatasvcInfo import PhEDExDatasvcInfo
7	slacapra	1.1
8			import os,string
9
10	slacapra	1.2	# Base class for all local scheduler
11	slacapra	1.1
12			class SchedulerLocal(Scheduler) :
13
14			def configure(self, cfg_params):
15	spiga	1.47	self.environment_unique_identifier = None
16	spiga	1.27	self.cfg_params = cfg_params
17	spiga	1.46	Scheduler.configure(self,cfg_params)
18	slacapra	1.1	self.jobtypeName = cfg_params['CRAB.jobtype']
19
20			name=string.upper(self.name())
21			self.queue = cfg_params.get(name+'.queue',None)
22
23			self.res = cfg_params.get(name+'.resource',None)
24
25			if (cfg_params.has_key(self.name()+'.env_id')): self.environment_unique_identifier = cfg_params[self.name()+'.env_id']
26			## is this ok?
27	spiga	1.25	localDomainName = getLocalDomain(self)
28	slacapra	1.16	if not cfg_params.has_key('EDG.se_white_list'):
29			cfg_params['EDG.se_white_list']=localDomainName
30			common.logger.message("Your domain name is "+str(localDomainName)+": only local dataset will be considered")
31			else:
32			common.logger.message("Your se_white_list is set to "+str(cfg_params['EDG.se_white_list'])+": only local dataset will be considered")
33	slacapra	1.1	return
34
35			def userName(self):
36			""" return the user name """
37	slacapra	1.5	import pwd,getpass
38			tmp=pwd.getpwnam(getpass.getuser())[4]
39			return "/CN="+tmp.strip()
40	slacapra	1.1
41	spiga	1.47	def Env_uniqueId(self):
42			return
43
44	slacapra	1.1	def wsSetupEnvironment(self):
45			"""
46			Returns part of a job script which does scheduler-specific work.
47			"""
48	spiga	1.47	taskId = common._db.queryTask('name')
49	slacapra	1.1	if not self.environment_unique_identifier:
50	spiga	1.47	try :
51			self.environment_unique_identifier = self.Env_uniqueId()
52			except :
53			raise CrabException('environment_unique_identifier not set')
54			print self.environment_unique_identifier
55	slacapra	1.12	index = int(common._db.nJobs())
56			job = common.job_list[index-1]
57			jbt = job.type()
58	ewv	1.20	# start with wrapper timing
59	spiga	1.18	txt = 'export TIME_WRAP_INI=`date +%s` \n'
60	farinafa	1.22	txt += 'export TIME_STAGEOUT=-2 \n\n'
61	spiga	1.17
62	spiga	1.19	txt += '# '+self.name()+' specific stuff\n'
63	slacapra	1.1	txt += '# strip arguments\n'
64			txt += 'echo "strip arguments"\n'
65			txt += 'args=("$@")\n'
66			txt += 'nargs=$#\n'
67			txt += 'shift $nargs\n'
68			txt += "# job number (first parameter for job wrapper)\n"
69	ewv	1.9	txt += "NJob=${args[0]}; export NJob\n"
70	slacapra	1.1
71	slacapra	1.12	txt += "out_files=out_files_${NJob}; export out_files\n"
72			txt += "echo $out_files\n"
73			txt += jbt.outList()
74
75	slacapra	1.6	txt += 'SyncGridJobId=`echo '+self.environment_unique_identifier+'`\n'
76			txt += 'MonitorJobID=`echo ${NJob}_${SyncGridJobId}`\n'
77	spiga	1.38	txt += 'MonitorID=`echo ' + taskId + '`\n'
78	slacapra	1.1
79			txt += 'echo "MonitorJobID=`echo $MonitorJobID`" \| tee -a $RUNTIME_AREA/$repo \n'
80			txt += 'echo "SyncGridJobId=`echo $SyncGridJobId`" \| tee -a $RUNTIME_AREA/$repo \n'
81			txt += 'echo "MonitorID=`echo $MonitorID`" \| tee -a $RUNTIME_AREA/$repo\n'
82	slacapra	1.7	txt += 'echo "SyncCE='+self.name()+'.`hostname -d`" \| tee -a $RUNTIME_AREA/$repo \n'
83	spiga	1.27
84	fanzago	1.23	txt += 'middleware='+self.name().upper()+' \n'
85	slacapra	1.1
86			txt += 'dumpStatus $RUNTIME_AREA/$repo \n'
87
88	slacapra	1.12	txt += 'InputSandBox=${args[3]}\n'
89
90	slacapra	1.1	txt += '\n\n'
91
92			return txt
93
94	spiga	1.27	def wsCopyOutput_comm(self, pool=None):
95	slacapra	1.3	"""
96			Write a CopyResults part of a job script, e.g.
97			to copy produced output into a storage element.
98			"""
99	spiga	1.37	index = int(common._db.nJobs())
100			job = common.job_list[index-1]
101			jbt = job.type()
102	slacapra	1.8	txt = '\n'
103	spiga	1.27	if int(self.copy_data) == 1:
104	slacapra	1.3
105	spiga	1.27	stageout = PhEDExDatasvcInfo(self.cfg_params)
106			endpoint, lfn, SE, SE_PATH, user = stageout.getEndpoint()
107	spiga	1.37	if self.check_RemoteDir == 1 :
108			self.checkRemoteDir(endpoint,jbt.outList('list') )
109	slacapra	1.3
110	spiga	1.27	txt += '#\n'
111			txt += '# COPY OUTPUT FILE TO '+SE_PATH+ '\n'
112			txt += '#\n\n'
113
114	fanzago	1.35	txt += 'export SE='+SE+'\n'
115	spiga	1.27	txt += 'echo "SE = $SE"\n'
116			txt += 'export SE_PATH='+SE_PATH+'\n'
117			txt += 'echo "SE_PATH = $SE_PATH"\n'
118			txt += 'export LFNBaseName='+lfn+'\n'
119	spiga	1.33	txt += 'echo "LFNBaseName = $LFNBaseName"\n'
120	spiga	1.27	txt += 'export USER='+user+'\n'
121	spiga	1.33	txt += 'echo "USER = $USER"\n'
122	spiga	1.27	txt += 'export endpoint='+endpoint+'\n'
123	spiga	1.33	txt += 'echo "endpoint = $endpoint"\n'
124	fanzago	1.36	### Needed i.e. for caf
125			if (pool) and (pool != 'None'):
126	spiga	1.27	txt += 'export STAGE_SVCCLASS='+str(pool)+'\n'
127
128			txt += 'echo ">>> Copy output files from WN = `hostname` to $SE_PATH :"\n'
129			txt += 'export TIME_STAGEOUT_INI=`date +%s` \n'
130			txt += 'copy_exit_status=0\n'
131	spiga	1.46	cmscp_args = '--destination $endpoint --inputFileList $file_list \
132			--middleware $middleware --lfn $LFNBaseName %s %s '%(self.loc_stage_out, self.debugWrap)
133			txt += 'echo "python cmscp.py %s "\n'%cmscp_args
134			txt += 'python cmscp.py %s \n'%cmscp_args
135	spiga	1.40	if self.debug_wrapper==1:
136	fanzago	1.43	txt += 'if [ -f .SEinteraction.log ] ;then\n'
137	spiga	1.27	txt += '########### details of SE interaction\n'
138	spiga	1.39	txt += ' cat .SEinteraction.log\n'
139			txt += 'else\n'
140			txt += ' echo ".SEinteraction.log file not found"\n'
141	spiga	1.40	txt += 'fi\n'
142	spiga	1.45	txt += '##################################### \n'
143	spiga	1.40	txt += 'if [ -f cmscpReport.sh ] ;then\n'
144	fanzago	1.43	txt += ' cat cmscpReport.sh\n'
145	spiga	1.40	txt += ' source cmscpReport.sh\n'
146			txt += 'else\n'
147			txt += ' echo "cmscpReport.sh file not found"\n'
148			txt += ' StageOutExitStatus=60307\n'
149			txt += 'fi\n'
150	spiga	1.27	txt += 'if [ $StageOutExitStatus -ne 0 ]; then\n'
151			txt += ' echo "Problem copying file to $SE $SE_PATH"\n'
152			txt += ' copy_exit_status=$StageOutExitStatus \n'
153	spiga	1.40	if not self.debug_wrapper==1:
154	spiga	1.32	txt += ' ########### details of SE interaction\n'
155	spiga	1.40	txt += ' if [ -f .SEinteraction.log ] ;then\n'
156			txt += ' cat .SEinteraction.log\n'
157			txt += ' else\n'
158			txt += ' echo ".SEinteraction.log file not found"\n'
159			txt += ' fi\n'
160	spiga	1.45	txt += ' ##################################### \n'
161	spiga	1.27	txt += ' job_exit_code=$StageOutExitStatus\n'
162			txt += 'fi\n'
163			txt += 'export TIME_STAGEOUT_END=`date +%s` \n'
164			txt += 'let "TIME_STAGEOUT = TIME_STAGEOUT_END - TIME_STAGEOUT_INI" \n'
165			else:
166			txt += 'export TIME_STAGEOUT=-1 \n'
167	slacapra	1.3	return txt
168	spiga	1.21
169			def wsExitFunc_comm(self):
170			"""
171			"""
172			txt = ''
173			txt += ' if [ $PYTHONPATH ]; then \n'
174			txt += ' if [ ! -s $RUNTIME_AREA/fillCrabFjr.py ]; then \n'
175			txt += ' echo "WARNING: it is not possible to create crab_fjr.xml to final report" \n'
176			txt += ' else \n'
177			txt += ' python $RUNTIME_AREA/fillCrabFjr.py $RUNTIME_AREA/crab_fjr_$NJob.xml --errorcode $job_exit_code $executable_exit_status \n'
178			txt += ' fi\n'
179			txt += ' fi\n'
180			txt += ' cd $RUNTIME_AREA \n'
181			txt += ' for file in $filesToCheck ; do\n'
182			txt += ' if [ -e $file ]; then\n'
183			txt += ' echo "tarring file $file in $out_files"\n'
184			txt += ' else\n'
185			txt += ' echo "WARNING: output file $file not found!"\n'
186			txt += ' fi\n'
187			txt += ' done\n'
188			txt += ' export TIME_WRAP_END=`date +%s`\n'
189			txt += ' let "TIME_WRAP = TIME_WRAP_END - TIME_WRAP_END_INI" \n'
190			txt += ' if [ $PYTHONPATH ]; then \n'
191			txt += ' if [ ! -s $RUNTIME_AREA/fillCrabFjr.py ]; then \n'
192			txt += ' echo "WARNING: it is not possible to create crab_fjr.xml to final report" \n'
193			txt += ' else \n'
194			# call timing FJR filling
195			txt += ' python $RUNTIME_AREA/fillCrabFjr.py $RUNTIME_AREA/crab_fjr_$NJob.xml --timing $TIME_WRAP $TIME_EXE $TIME_STAGEOUT \n'
196			txt += ' echo "CrabWrapperTime=$TIME_WRAP" >> $RUNTIME_AREA/$repo \n'
197			txt += ' if [ $TIME_STAGEOUT -lt 0 ]; then \n'
198			txt += ' export TIME_STAGEOUT=NULL \n'
199			txt += ' fi\n'
200			txt += ' echo "CrabStageoutTime=$TIME_STAGEOUT" >> $RUNTIME_AREA/$repo \n'
201			txt += ' fi\n'
202			txt += ' fi\n'
203			txt += ' dumpStatus $RUNTIME_AREA/$repo \n\n'
204			txt += ' echo "JOB_EXIT_STATUS = $job_exit_code"\n'
205			txt += ' echo "JobExitCode=$job_exit_code" >> $RUNTIME_AREA/$repo\n'
206			txt += ' dumpStatus $RUNTIME_AREA/$repo\n'
207
208			return txt
209