CRAB/python/SchedulerLocal.py

from Scheduler import Scheduler
from crab_exceptions import *
from crab_logger import Logger
import common

import os,string

# Base class for all local scheduler

class SchedulerLocal(Scheduler) :

    def configure(self, 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)

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

        if (cfg_params.has_key(self.name()+'.env_id')): self.environment_unique_identifier = cfg_params[self.name()+'.env_id']

        self._taskId = common.taskDB.dict('taskId')

        self.return_data = int(cfg_params.get('USER.return_data',0))

        self.copy_data = int(cfg_params.get("USER.copy_data",0))
        if self.copy_data == 1:
            self._copyCommand = cfg_params.get('USER.copyCommand','rfcp')
            self.SE_path= cfg_params.get('USER.storage_path',None)
            self.SE_path+='/'
            if not self.SE_path:
                if os.environ.has_key('CASTOR_HOME'):
                    self.SE_path=os.environ['CASTOR_HOME']
                else:
                    msg='No USER.storage_path has been provided: cannot copy_output'
                    raise CrabException(msg)
                pass
            pass

        if ( self.return_data == 0 and self.copy_data == 0 ):
           msg = 'Error: return_data = 0 and copy_data = 0 ==> your exe output will be lost\n'
           msg = msg + 'Please modify return_data and copy_data value in your crab.cfg file\n'
           raise CrabException(msg)

        if ( self.return_data == 1 and self.copy_data == 1 ):
           msg = 'Error: return_data and copy_data cannot be set both to 1\n'
           msg = msg + 'Please modify return_data or copy_data value in your crab.cfg file\n'
           raise CrabException(msg)

        ## Get local domain name
        import socket
        tmp=socket.gethostname()
        dot=string.find(tmp,'.')
        if (dot==-1):
            msg='Unkown domain name. Cannot use local scheduler'
            raise CrabException(msg)
        localDomainName = string.split(tmp,'.',1)[-1]
        ## is this ok?
        cfg_params['EDG.se_white_list']=localDomainName
        common.logger.message("Your domain name is "+str(localDomainName)+": only local dataset will be considered")

        return

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

    def wsSetupEnvironment(self):
        """
        Returns part of a job script which does scheduler-specific work.
        """
        if not self.environment_unique_identifier:
            raise CrabException('environment_unique_identifier not set')

        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]}\n"

        txt += 'MonitorJobID=`echo ${NJob}_$'+self.environment_unique_identifier+'`\n'
        txt += 'SyncGridJobId=`echo $'+self.environment_unique_identifier+'`\n'
        txt += 'MonitorID=`echo ' + self._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 += 'middleware='+self.name()+' \n'

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

        txt += '\n\n'

        return txt

    def wsCopyOutput(self):
        """
        Write a CopyResults part of a job script, e.g.
        to copy produced output into a storage element.
        """
        if not self.copy_data: return

        txt = '\n'

        txt += '#\n'
        txt += '# COPY OUTPUT FILE TO '+self.SE_path
        txt += '#\n\n'

        txt += 'export SE_PATH='+self.SE_path+'\n'

        txt += 'export CP_CMD='+self._copyCommand+'\n'

        txt += 'echo ">>> Copy output files from WN = `hostname` to PATH = $SE_PATH using $CP_CMD :"\n'

        txt += 'if [ $output_exit_status -eq 60302 ]; then\n'
        txt += '    echo "--> No output file to copy to $SE"\n'
        txt += '    copy_exit_status=$output_exit_status\n'
        txt += '    echo "COPY_EXIT_STATUS = $copy_exit_status"\n'
        txt += 'else\n'
        txt += '    for out_file in $file_list ; do\n'
        txt += '        echo "Trying to copy output file to $SE_PATH"\n'
        txt += '        $CP_CMD $SOFTWARE_DIR/$out_file ${SE_PATH}/$out_file\n'
        txt += '        copy_exit_status=$?\n'
        txt += '        echo "COPY_EXIT_STATUS = $copy_exit_status"\n'
        txt += '        echo "STAGE_OUT = $copy_exit_status"\n'
        txt += '        if [ $copy_exit_status -ne 0 ]; then\n'
        txt += '            echo "Problem copying $out_file to $SE $SE_PATH"\n'
        txt += '            echo "StageOutExitStatus = $copy_exit_status " | tee -a $RUNTIME_AREA/$repo\n'
        txt += '            copy_exit_status=60307\n'
        txt += '        else\n'
        txt += '            echo "StageOutSE = $SE" | tee -a $RUNTIME_AREA/$repo\n'
        txt += '            echo "StageOutCatalog = " | tee -a $RUNTIME_AREA/$repo\n'
        txt += '            echo "output copied into $SE/$SE_PATH directory"\n'
        txt += '            echo "StageOutExitStatus = 0" | tee -a $RUNTIME_AREA/$repo\n'
        txt += '        fi\n'
        txt += '    done\n'
        txt += 'fi\n'
        txt += 'exit_status=$copy_exit_status\n'

        return txt

    def createXMLSchScript(self, nj, argsList):

        """
        Create a XML-file for BOSS4.
        """

        """
        INDY
        [begin] FIX-ME:
        I would pass jobType instead of job
        """
        index = nj - 1
        job = common.job_list[index]
        jbt = job.type()
        inp_sandbox = jbt.inputSandbox(index)
        #out_sandbox = jbt.outputSandbox(index)
        """
        [end] FIX-ME
        """


        title = '<?xml version="1.0" encoding="UTF-8" standalone="no"?>\n'
        jt_string = ''

        xml_fname = str(self.jobtypeName)+'.xml'
        xml = open(common.work_space.shareDir()+'/'+xml_fname, 'a')

        #TaskName
        dir = string.split(common.work_space.topDir(), '/')
        taskName = dir[len(dir)-2]

        to_write = ''

        req=' '
        req = req + jbt.getRequirements()

        #TaskName
        dir = string.split(common.work_space.topDir(), '/')
        taskName = dir[len(dir)-2]

        xml.write(str(title))

        #First check the X509_USER_PROXY. In not there use the default
        xml.write('<task name="' +str(taskName)+ '" sub_path="' +common.work_space.pathForTgz() + 'share/.boss_cache"' + '>\n')

        xml.write(jt_string)

        if (to_write != ''):
            xml.write('<extraTags\n')
            xml.write(to_write)
            xml.write('/>\n')
            pass

        xml.write('<iterator>\n')
        xml.write('\t<iteratorRule name="ITR1">\n')
        xml.write('\t\t<ruleElement> 1:'+ str(nj) + ' </ruleElement>\n')
        xml.write('\t</iteratorRule>\n')
        xml.write('\t<iteratorRule name="ITR2">\n')
        for arg in argsList:
            xml.write('\t\t<ruleElement> <![CDATA[\n'+ arg + '\n\t\t]]> </ruleElement>\n')
            pass
        xml.write('\t</iteratorRule>\n')
        #print jobList
        xml.write('\t<iteratorRule name="ITR3">\n')
        xml.write('\t\t<ruleElement> 1:'+ str(nj) + ':1:6 </ruleElement>\n')
        xml.write('\t</iteratorRule>\n')

        xml.write('<chain name="' +str(taskName)+'__ITR1_" scheduler="'+str(self.name())+'">\n')
       # xml.write('<chain scheduler="'+str(self.schedulerName)+'">\n')
        xml.write(jt_string)

        #executable

        script = job.scriptFilename()
        xml.write('<program>\n')
        xml.write('<exec> ' + os.path.basename(script) +' </exec>\n')
        xml.write(jt_string)

        xml.write('<args> <![CDATA[\n _ITR2_ \n]]> </args>\n')
        xml.write('<program_types> crabjob </program_types>\n')
        inp_box = common.work_space.pathForTgz() + 'job/' + jbt.scriptName + ','

        if inp_sandbox != None:
            for fl in inp_sandbox:
                inp_box = inp_box + '' + fl + ','
                pass
            pass

        if inp_box[-1] == ',' : inp_box = inp_box[:-1]
        inp_box = '<infiles> <![CDATA[\n' + inp_box + '\n]]> </infiles>\n'
        xml.write(inp_box)

        base = jbt.name()
        stdout = base + '__ITR3_.stdout'
        stderr = base + '__ITR3_.stderr'

        xml.write('<stderr> ' + stderr + '</stderr>\n')
        xml.write('<stdout> ' + stdout + '</stdout>\n')


        out_box = stdout + ',' + \
                  stderr + ',.BrokerInfo,'

        # Stuff to be returned _always_ via sandbox
        for fl in jbt.output_file_sandbox:
            out_box = out_box + '' + jbt.numberFile_(fl, '_ITR1_') + ','
            pass
        pass

        # via sandbox iif required return_data
        if int(self.return_data) == 1:
            for fl in jbt.output_file:
                out_box = out_box + '' + jbt.numberFile_(fl, '_ITR1_') + ','
                pass
            pass

        if out_box[-1] == ',' : out_box = out_box[:-1]
        out_box = '<outfiles> <![CDATA[\n' + out_box + '\n]]></outfiles>\n'
        xml.write(out_box)

        xml.write('<BossAttr> crabjob.INTERNAL_ID=_ITR1_ </BossAttr>\n')

        xml.write('</program>\n')
        xml.write('</chain>\n')

        xml.write('</iterator>\n')
        xml.write('</task>\n')

        xml.close()


        return
Revision:	1.3
Committed:	Mon Jan 21 18:39:37 2008 UTC (17 years, 3 months ago) by slacapra
Content type:	text/x-python
Branch:	MAIN
Changes since 1.2:	+71 -1 lines
Log Message:	add support for copy output to user defined location and copy command
#	User	Rev	Content
1	slacapra	1.1	from Scheduler import Scheduler
2			from crab_exceptions import *
3			from crab_logger import Logger
4			import common
5
6			import os,string
7
8	slacapra	1.2	# Base class for all local scheduler
9	slacapra	1.1
10			class SchedulerLocal(Scheduler) :
11
12			def configure(self, cfg_params):
13	slacapra	1.2	Scheduler.configure(self,cfg_params)
14	slacapra	1.1
15			self.jobtypeName = cfg_params['CRAB.jobtype']
16
17			name=string.upper(self.name())
18			self.queue = cfg_params.get(name+'.queue',None)
19
20			self.res = cfg_params.get(name+'.resource',None)
21
22			self.user = cfg_params.get(name+'.user',None)
23
24			if (cfg_params.has_key(self.name()+'.env_id')): self.environment_unique_identifier = cfg_params[self.name()+'.env_id']
25
26			self._taskId = common.taskDB.dict('taskId')
27
28	slacapra	1.3	self.return_data = int(cfg_params.get('USER.return_data',0))
29
30			self.copy_data = int(cfg_params.get("USER.copy_data",0))
31			if self.copy_data == 1:
32			self._copyCommand = cfg_params.get('USER.copyCommand','rfcp')
33			self.SE_path= cfg_params.get('USER.storage_path',None)
34			self.SE_path+='/'
35			if not self.SE_path:
36			if os.environ.has_key('CASTOR_HOME'):
37			self.SE_path=os.environ['CASTOR_HOME']
38			else:
39			msg='No USER.storage_path has been provided: cannot copy_output'
40			raise CrabException(msg)
41			pass
42			pass
43
44			if ( self.return_data == 0 and self.copy_data == 0 ):
45			msg = 'Error: return_data = 0 and copy_data = 0 ==> your exe output will be lost\n'
46			msg = msg + 'Please modify return_data and copy_data value in your crab.cfg file\n'
47			raise CrabException(msg)
48
49			if ( self.return_data == 1 and self.copy_data == 1 ):
50			msg = 'Error: return_data and copy_data cannot be set both to 1\n'
51			msg = msg + 'Please modify return_data or copy_data value in your crab.cfg file\n'
52			raise CrabException(msg)
53	slacapra	1.1
54			## Get local domain name
55			import socket
56			tmp=socket.gethostname()
57			dot=string.find(tmp,'.')
58			if (dot==-1):
59			msg='Unkown domain name. Cannot use local scheduler'
60			raise CrabException(msg)
61			localDomainName = string.split(tmp,'.',1)[-1]
62			## is this ok?
63			cfg_params['EDG.se_white_list']=localDomainName
64			common.logger.message("Your domain name is "+str(localDomainName)+": only local dataset will be considered")
65
66			return
67
68			def userName(self):
69			""" return the user name """
70			if self.user:
71			return self.user
72			else:
73			import pwd,getpass
74			tmp=pwd.getpwnam(getpass.getuser())[4]
75			return tmp.strip()
76
77			def wsSetupEnvironment(self):
78			"""
79			Returns part of a job script which does scheduler-specific work.
80			"""
81			if not self.environment_unique_identifier:
82			raise CrabException('environment_unique_identifier not set')
83
84			txt = '# '+self.name()+' specific stuff\n'
85			txt += '# strip arguments\n'
86			txt += 'echo "strip arguments"\n'
87			txt += 'args=("$@")\n'
88			txt += 'nargs=$#\n'
89			txt += 'shift $nargs\n'
90			txt += "# job number (first parameter for job wrapper)\n"
91			txt += "NJob=${args[0]}\n"
92
93			txt += 'MonitorJobID=`echo ${NJob}_$'+self.environment_unique_identifier+'`\n'
94			txt += 'SyncGridJobId=`echo $'+self.environment_unique_identifier+'`\n'
95			txt += 'MonitorID=`echo ' + self._taskId + '`\n'
96
97			txt += 'echo "MonitorJobID=`echo $MonitorJobID`" \| tee -a $RUNTIME_AREA/$repo \n'
98			txt += 'echo "SyncGridJobId=`echo $SyncGridJobId`" \| tee -a $RUNTIME_AREA/$repo \n'
99			txt += 'echo "MonitorID=`echo $MonitorID`" \| tee -a $RUNTIME_AREA/$repo\n'
100
101			txt += 'middleware='+self.name()+' \n'
102
103			txt += 'dumpStatus $RUNTIME_AREA/$repo \n'
104
105			txt += '\n\n'
106
107			return txt
108
109	slacapra	1.3	def wsCopyOutput(self):
110			"""
111			Write a CopyResults part of a job script, e.g.
112			to copy produced output into a storage element.
113			"""
114			if not self.copy_data: return
115
116			txt = '\n'
117
118			txt += '#\n'
119			txt += '# COPY OUTPUT FILE TO '+self.SE_path
120			txt += '#\n\n'
121
122			txt += 'export SE_PATH='+self.SE_path+'\n'
123
124			txt += 'export CP_CMD='+self._copyCommand+'\n'
125
126			txt += 'echo ">>> Copy output files from WN = `hostname` to PATH = $SE_PATH using $CP_CMD :"\n'
127
128			txt += 'if [ $output_exit_status -eq 60302 ]; then\n'
129			txt += ' echo "--> No output file to copy to $SE"\n'
130			txt += ' copy_exit_status=$output_exit_status\n'
131			txt += ' echo "COPY_EXIT_STATUS = $copy_exit_status"\n'
132			txt += 'else\n'
133			txt += ' for out_file in $file_list ; do\n'
134			txt += ' echo "Trying to copy output file to $SE_PATH"\n'
135			txt += ' $CP_CMD $SOFTWARE_DIR/$out_file ${SE_PATH}/$out_file\n'
136			txt += ' copy_exit_status=$?\n'
137			txt += ' echo "COPY_EXIT_STATUS = $copy_exit_status"\n'
138			txt += ' echo "STAGE_OUT = $copy_exit_status"\n'
139			txt += ' if [ $copy_exit_status -ne 0 ]; then\n'
140			txt += ' echo "Problem copying $out_file to $SE $SE_PATH"\n'
141			txt += ' echo "StageOutExitStatus = $copy_exit_status " \| tee -a $RUNTIME_AREA/$repo\n'
142			txt += ' copy_exit_status=60307\n'
143			txt += ' else\n'
144			txt += ' echo "StageOutSE = $SE" \| tee -a $RUNTIME_AREA/$repo\n'
145			txt += ' echo "StageOutCatalog = " \| tee -a $RUNTIME_AREA/$repo\n'
146			txt += ' echo "output copied into $SE/$SE_PATH directory"\n'
147			txt += ' echo "StageOutExitStatus = 0" \| tee -a $RUNTIME_AREA/$repo\n'
148			txt += ' fi\n'
149			txt += ' done\n'
150			txt += 'fi\n'
151			txt += 'exit_status=$copy_exit_status\n'
152
153			return txt
154
155	slacapra	1.1	def createXMLSchScript(self, nj, argsList):
156
157			"""
158			Create a XML-file for BOSS4.
159			"""
160
161			"""
162			INDY
163			[begin] FIX-ME:
164			I would pass jobType instead of job
165			"""
166			index = nj - 1
167			job = common.job_list[index]
168			jbt = job.type()
169			inp_sandbox = jbt.inputSandbox(index)
170			#out_sandbox = jbt.outputSandbox(index)
171			"""
172			[end] FIX-ME
173			"""
174
175
176			title = '<?xml version="1.0" encoding="UTF-8" standalone="no"?>\n'
177			jt_string = ''
178
179			xml_fname = str(self.jobtypeName)+'.xml'
180			xml = open(common.work_space.shareDir()+'/'+xml_fname, 'a')
181
182			#TaskName
183			dir = string.split(common.work_space.topDir(), '/')
184			taskName = dir[len(dir)-2]
185
186			to_write = ''
187
188			req=' '
189			req = req + jbt.getRequirements()
190
191			#TaskName
192			dir = string.split(common.work_space.topDir(), '/')
193			taskName = dir[len(dir)-2]
194
195			xml.write(str(title))
196
197			#First check the X509_USER_PROXY. In not there use the default
198			xml.write('<task name="' +str(taskName)+ '" sub_path="' +common.work_space.pathForTgz() + 'share/.boss_cache"' + '>\n')
199
200			xml.write(jt_string)
201
202			if (to_write != ''):
203			xml.write('<extraTags\n')
204			xml.write(to_write)
205			xml.write('/>\n')
206			pass
207
208			xml.write('<iterator>\n')
209			xml.write('\t<iteratorRule name="ITR1">\n')
210			xml.write('\t\t<ruleElement> 1:'+ str(nj) + ' </ruleElement>\n')
211			xml.write('\t</iteratorRule>\n')
212			xml.write('\t<iteratorRule name="ITR2">\n')
213			for arg in argsList:
214			xml.write('\t\t<ruleElement> <![CDATA[\n'+ arg + '\n\t\t]]> </ruleElement>\n')
215			pass
216			xml.write('\t</iteratorRule>\n')
217			#print jobList
218			xml.write('\t<iteratorRule name="ITR3">\n')
219			xml.write('\t\t<ruleElement> 1:'+ str(nj) + ':1:6 </ruleElement>\n')
220			xml.write('\t</iteratorRule>\n')
221
222			xml.write('<chain name="' +str(taskName)+'__ITR1_" scheduler="'+str(self.name())+'">\n')
223			# xml.write('<chain scheduler="'+str(self.schedulerName)+'">\n')
224			xml.write(jt_string)
225
226			#executable
227
228			script = job.scriptFilename()
229			xml.write('<program>\n')
230			xml.write('<exec> ' + os.path.basename(script) +' </exec>\n')
231			xml.write(jt_string)
232
233			xml.write('<args> <![CDATA[\n _ITR2_ \n]]> </args>\n')
234			xml.write('<program_types> crabjob </program_types>\n')
235			inp_box = common.work_space.pathForTgz() + 'job/' + jbt.scriptName + ','
236
237			if inp_sandbox != None:
238			for fl in inp_sandbox:
239			inp_box = inp_box + '' + fl + ','
240			pass
241			pass
242
243			if inp_box[-1] == ',' : inp_box = inp_box[:-1]
244			inp_box = '<infiles> <![CDATA[\n' + inp_box + '\n]]> </infiles>\n'
245			xml.write(inp_box)
246
247			base = jbt.name()
248			stdout = base + '__ITR3_.stdout'
249			stderr = base + '__ITR3_.stderr'
250
251			xml.write('<stderr> ' + stderr + '</stderr>\n')
252			xml.write('<stdout> ' + stdout + '</stdout>\n')
253
254
255			out_box = stdout + ',' + \
256			stderr + ',.BrokerInfo,'
257
258			# Stuff to be returned _always_ via sandbox
259			for fl in jbt.output_file_sandbox:
260			out_box = out_box + '' + jbt.numberFile_(fl, '_ITR1_') + ','
261			pass
262			pass
263
264			# via sandbox iif required return_data
265			if int(self.return_data) == 1:
266			for fl in jbt.output_file:
267			out_box = out_box + '' + jbt.numberFile_(fl, '_ITR1_') + ','
268			pass
269			pass
270
271			if out_box[-1] == ',' : out_box = out_box[:-1]
272			out_box = '<outfiles> <![CDATA[\n' + out_box + '\n]]></outfiles>\n'
273			xml.write(out_box)
274
275			xml.write('<BossAttr> crabjob.INTERNAL_ID=_ITR1_ </BossAttr>\n')
276
277			xml.write('</program>\n')
278			xml.write('</chain>\n')
279
280			xml.write('</iterator>\n')
281			xml.write('</task>\n')
282
283			xml.close()
284
285
286			return