CRAB/python/SchedulerGrid.py

from Scheduler import Scheduler
from crab_logger import Logger
from crab_exceptions import *
from crab_util import *
from BlackWhiteListParser import BlackWhiteListParser
import common
from LFNBaseName import *
from JobList import JobList

import os, sys, time

#
# Base class for all grid scheduler
#

class SchedulerGrid(Scheduler):

    def __init__(self, name):
        Scheduler.__init__(self,name)
        self.states = [ "Acl", "cancelReason", "cancelling","ce_node","children", \
                      "children_hist","children_num","children_states","condorId","condor_jdl", \
                      "cpuTime","destination", "done_code","exit_code","expectFrom", \
                      "expectUpdate","globusId","jdl","jobId","jobtype", \
                      "lastUpdateTime","localId","location", "matched_jdl","network_server", \
                      "owner","parent_job", "reason","resubmitted","rsl","seed",\
                      "stateEnterTime","stateEnterTimes","subjob_failed", \
                      "user tags" , "status" , "status_code","hierarchy"]
        return

    def configure(self, cfg_params):
        Scheduler.configure(self,cfg_params)

        # init BlackWhiteListParser
        self.blackWhiteListParser = BlackWhiteListParser(cfg_params)

        self.proxyValid=0
        self.dontCheckProxy=int(cfg_params.get("EDG.dont_check_proxy",0))

#        self.rb_param_file=None
#        if (cfg_params.has_key('EDG.rb')):
#            self.rb_param_file=self.rb_configure(cfg_params.get("EDG.rb"))

        self.proxyServer = cfg_params.get("EDG.proxy_server",'myproxy.cern.ch')
        common.logger.debug(5,'Setting myproxy server to '+self.proxyServer)

        self.group = cfg_params.get("EDG.group", None)

        self.role = cfg_params.get("EDG.role", None)

        self.EDG_ce_black_list = cfg_params.get('EDG.ce_black_list',None)
        if (self.EDG_ce_black_list): self.EDG_ce_black_list = string.split(self.EDG_ce_black_list,',')

        self.EDG_ce_white_list = cfg_params.get('EDG.ce_white_list',None)
        if (self.EDG_ce_white_list): self.EDG_ce_white_list = string.split(self.EDG_ce_white_list,',')

        self.VO = cfg_params.get('EDG.virtual_organization','cms')

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

        self.copy_data = cfg_params.get("USER.copy_data",0)
        if int(self.copy_data) == 1:
            self.SE = cfg_params.get('USER.storage_element',None)
            self.SE_PATH = cfg_params.get('USER.storage_path',None)
            self.srm_ver  = cfg_params.get('USER.srm_version',0) ## DS for srmv2
            if not self.SE or not self.SE_PATH:
                msg = "Error. The [USER] section does not have 'storage_element'"
                msg = msg + " and/or 'storage_path' entries, necessary to copy the output"
                common.logger.message(msg)
                raise CrabException(msg)
            if self.srm_ver != 1 and self.SE_PATH.find("managerv1") != -1:
                msg = "Warning: The storage_path and srm_version seem to be incompatible."
                msg = msg + " Set srm_version = 1 if your path contains manager1"
                common.logger.message(msg)

        if ( int(self.return_data) == 0 and int(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 ( int(self.return_data) == 1 and int(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)

        self.publish_data = cfg_params.get("USER.publish_data",0)
        self.usenamespace = cfg_params.get("USER.usenamespace",0)
        if int(self.publish_data) == 1 or int(self.usenamespace) == 1:
            self.publish_data_name = cfg_params.get('USER.publish_data_name',None)
            if not self.publish_data_name and int(self.publish_data) == 1:
                msg = "Error. The [USER] section does not have 'publish_data_name'"
                raise CrabException(msg)
            if not self.publish_data_name and int(self.usenamespace) == 1:
               self.publish_data_name = "DefaultDataset"

            ## SL I don't like a direct call to voms-proxy here

            import urllib
            try:
                userdn = runCommand("voms-proxy-info -identity")
                self.userdn = string.strip(userdn)
            except:
                msg = "Error. Problem with voms-proxy-info -identity command"
                raise CrabException(msg)
            try:
                sitedburl="https://cmsweb.cern.ch/sitedb/sitedb/json/index/dnUserName"
                params = urllib.urlencode({'dn': self.userdn })
                f = urllib.urlopen(sitedburl,params)
                udata = f.read()
                userinfo= eval(udata)
                self.hnUserName = userinfo['user']
            except:
                msg = "Error. Problem extracting user name from %s"%sitedburl
                raise CrabException(msg)

            if not self.hnUserName:
                msg = "Error. There is no user name associated to DN %s in %s.You need to register in SiteDB"%(userdn,sitedburl)
                raise CrabException(msg)

        if ( int(self.copy_data) == 0 and int(self.publish_data) == 1 ):
           msg = 'Warning: publish_data = 1 must be used with copy_data = 1\n'
           msg = msg + 'Please modify copy_data value in your crab.cfg file\n'
           common.logger.message(msg)
           raise CrabException(msg)

        self.EDG_requirements = cfg_params.get('EDG.requirements',None)

        self.EDG_addJdlParam = cfg_params.get('EDG.additional_jdl_parameters',None)
        if (self.EDG_addJdlParam): self.EDG_addJdlParam = string.split(self.EDG_addJdlParam,';')

        self.EDG_retry_count = cfg_params.get('EDG.retry_count',0)

        self.EDG_shallow_retry_count= cfg_params.get('EDG.shallow_retry_count',-1)

        self.EDG_clock_time = cfg_params.get('EDG.max_wall_clock_time',None)

        # Default minimum CPU time to >= 130 minutes
        self.EDG_cpu_time = cfg_params.get('EDG.max_cpu_time', '130')

        # Add EDG_WL_LOCATION to the python path

        if not os.environ.has_key('EDG_WL_LOCATION'):
            msg = "Error: the EDG_WL_LOCATION variable is not set."
            raise CrabException(msg)
        path = os.environ['EDG_WL_LOCATION']

        libPath=os.path.join(path, "lib")
        sys.path.append(libPath)
        libPath=os.path.join(path, "lib", "python")
        sys.path.append(libPath)

        self._taskId=str("_".join(common._db.queryTask('name').split('_')[:-1]))
        self.jobtypeName = cfg_params.get('CRAB.jobtype','')

        self.schedulerName = cfg_params.get('CRAB.scheduler','')

        return

    def rb_configure(self, RB):
        """
        Return a requirement to be add to Jdl to select a specific RB/WMS:
        return None if RB=None
        To be re-implemented in concrete scheduler
        """
        return None

    def sched_fix_parameter(self):
        return

    def listMatch(self, dest):
        ces=Scheduler.listMatch(self,dest)
        sites=[]
        for ce in ces:
            site=ce.split(":")[0]
            if site not in sites:
                sites.append(site)
            pass

        return sites


    def wsSetupEnvironment(self):
        """
        Returns part of a job script which does scheduler-specific work.
        """
        index = int(common._db.nJobs())
        job = common.job_list[index-1]
        jbt = job.type()
        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]}; export NJob\n"

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

        txt += 'MonitorJobID=${NJob}_$'+self.environment_unique_identifier+'\n'
        txt += 'SyncGridJobId=$'+self.environment_unique_identifier+'\n'
        txt += 'MonitorID='+self._taskId+'\n'
        txt += 'echo "MonitorJobID=$MonitorJobID" > $RUNTIME_AREA/$repo \n'
        txt += 'echo "SyncGridJobId=$SyncGridJobId" >> $RUNTIME_AREA/$repo \n'
        txt += 'echo "MonitorID=$MonitorID" >> $RUNTIME_AREA/$repo\n'

        txt += 'echo ">>> GridFlavour discovery: " \n'
        txt += 'if [ $OSG_APP ]; then \n'
        txt += '    middleware=OSG \n'
        txt += '    if [ $OSG_JOB_CONTACT ]; then \n'
        txt += '        SyncCE="$OSG_JOB_CONTACT"; \n'
        txt += '        echo "SyncCE=$SyncCE" >> $RUNTIME_AREA/$repo ;\n'
        txt += '    else\n'
        txt += '        echo "not reporting SyncCE";\n'
        txt += '    fi\n';
        txt += '    echo "GridFlavour=$middleware" | tee -a $RUNTIME_AREA/$repo \n'
        txt += 'elif [ $VO_CMS_SW_DIR ]; then \n'
        txt += '    middleware=LCG \n'
        txt += '    echo "SyncCE=`glite-brokerinfo getCE`" >> $RUNTIME_AREA/$repo \n'
        txt += '    echo "GridFlavour=$middleware" | tee -a $RUNTIME_AREA/$repo \n'
        txt += 'else \n'
        txt += '    echo "ERROR ==> GridFlavour not identified" \n'
        txt += '    job_exit_code=10030 \n'
        txt += '    func_exit \n'
        txt += 'fi \n'

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

        txt += 'export VO='+self.VO+'\n'
        txt += 'if [ $middleware == LCG ]; then\n'
        txt += '    CloseCEs=`glite-brokerinfo getCE`\n'
        txt += '    echo "CloseCEs = $CloseCEs"\n'
        txt += '    CE=`echo $CloseCEs | sed -e "s/:.*//"`\n'
        txt += '    echo "CE = $CE"\n'
        txt += 'elif [ $middleware == OSG ]; then \n'
        txt += '    if [ $OSG_JOB_CONTACT ]; then \n'
        txt += '        CE=`echo $OSG_JOB_CONTACT | /usr/bin/awk -F\/ \'{print $1}\'` \n'
        txt += '    else \n'
        txt += '        echo "ERROR ==> OSG mode in setting CE name from OSG_JOB_CONTACT" \n'
        txt += '        job_exit_code=10099\n'
        txt += '        func_exit\n'
        txt += '    fi \n'
        txt += 'fi \n'

        return txt

    # def wsCopyInput(self):
    #     """
    #     Copy input data from SE to WN
    #     """
    #     txt = ''
    #     if not self.copy_input_data: return txt

    #     ## OLI_Daniele deactivate for OSG (wait for LCG UI installed on OSG)
    #     txt += 'if [ $middleware == OSG ]; then\n'
    #     txt += '   #\n'
    #     txt += '   #   Copy Input Data from SE to this WN deactivated in OSG mode\n'
    #     txt += '   #\n'
    #     txt += '   echo "Copy Input Data from SE to this WN deactivated in OSG mode"\n'
    #     txt += 'elif [ $middleware == LCG ]; then \n'
    #     txt += '   #\n'
    #     txt += '   #   Copy Input Data from SE to this WN\n'
    #     txt += '   #\n'
    #     ### changed by georgia (put a loop copying more than one input files per jobs)
    #     txt += '   for input_file in $cur_file_list \n'
    #     txt += '   do \n'
    #     txt += '      lcg-cp --vo $VO --verbose -t 1200 lfn:$input_lfn/$input_file file:`pwd`/$input_file 2>&1\n'
    #     txt += '      copy_input_exit_status=$?\n'
    #     txt += '      echo "COPY_INPUT_EXIT_STATUS = $copy_input_exit_status"\n'
    #     txt += '      if [ $copy_input_exit_status -ne 0 ]; then \n'
    #     txt += '         echo "Problems with copying to WN" \n'
    #     txt += '      else \n'
    #     txt += '         echo "input copied into WN" \n'
    #     txt += '      fi \n'
    #     txt += '   done \n'
    #     ### copy a set of PU ntuples (same for each jobs -- but accessed randomly)
    #     txt += '   for file in $cur_pu_list \n'
    #     txt += '   do \n'
    #     txt += '      lcg-cp --vo $VO --verbose -t 1200 lfn:$pu_lfn/$file file:`pwd`/$file 2>&1\n'
    #     txt += '      copy_input_pu_exit_status=$?\n'
    #     txt += '      echo "COPY_INPUT_PU_EXIT_STATUS = $copy_input_pu_exit_status"\n'
    #     txt += '      if [ $copy_input_pu_exit_status -ne 0 ]; then \n'
    #     txt += '         echo "Problems with copying pu to WN" \n'
    #     txt += '      else \n'
    #     txt += '         echo "input pu files copied into WN" \n'
    #     txt += '      fi \n'
    #     txt += '   done \n'
    #     txt += '   \n'
    #     txt += '   ### Check SCRATCH space available on WN : \n'
    #     txt += '   df -h \n'
    #     txt += 'fi \n'

    #     return txt

    def wsCopyOutput(self):
        """
        Write a CopyResults part of a job script, e.g.
        to copy produced output into a storage element.
        """
        txt = '\n'

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

        SE_PATH=''
        if int(self.copy_data) == 1:
            if self.SE:
                txt += 'export SE='+self.SE+'\n'
                txt += 'echo "SE = $SE"\n'
            if self.SE_PATH:
                if ( self.SE_PATH[-1] != '/' ) : self.SE_PATH = self.SE_PATH + '/'
                SE_PATH=self.SE_PATH
            if int(self.publish_data) == 1 or int(self.usenamespace) == 1:
                txt += '### publish_data = 1 so the SE path where to copy the output is: \n'
                #path_add = self.hnUserName + '/' + self.publish_data_name +'_${PSETHASH}/'
                path_add = PFNportion(self.publish_data_name) +'_${PSETHASH}/'
                SE_PATH = SE_PATH + path_add
            txt += 'export SE_PATH='+SE_PATH+'\n'
            txt += 'echo "SE_PATH = $SE_PATH"\n'

            txt += 'export SRM_VER='+str(self.srm_ver)+'\n' ## DS for srmVer
            txt += 'echo "SRM_VER = $SRM_VER"\n' ## DS for srmVer

            txt += 'echo ">>> Copy output files from WN = `hostname` to SE = $SE :"\n'
            txt += 'copy_exit_status=0\n'
            txt += 'for out_file in $file_list ; do\n'
            txt += '    if [ -e $SOFTWARE_DIR/$out_file ] ; then\n'
            txt += '        echo "Trying to copy output file $SOFTWARE_DIR/$out_file to $SE"\n'
            txt += '        cmscp $SOFTWARE_DIR/$out_file ${SE} ${SE_PATH} $out_file ${SRM_VER} $middleware\n'
            txt += '        if [ $cmscp_exit_status -ne 0 ]; then\n'
            txt += '            echo "Problem copying $out_file to $SE $SE_PATH"\n'
            txt += '            copy_exit_status=$cmscp_exit_status\n'
            txt += '        else\n'
            txt += '            echo "output copied into $SE/$SE_PATH directory"\n'
            txt += '        fi\n'
            txt += '    else\n'
            txt += '        copy_exit_status=60302\n'
            txt += '        echo "StageOutExitStatus = $copy_exit_status" | tee -a $RUNTIME_AREA/$repo\n'
            txt += '        echo "StageOutExitStatusReason = file to copy not found" | tee -a $RUNTIME_AREA/$repo\n'
            txt += '    fi\n'
            txt += 'done\n'
            txt += 'if [ $copy_exit_status -ne 0 ]; then\n'
            txt += '    SE=""\n'
            txt += '    SE_PATH=""\n'
            txt += '    job_exit_code=$copy_exit_status\n'
            txt += 'fi\n'
            pass
        return txt


    def wsExitFunc_common(self):
        """
        """
        txt = '' 
        txt += '    if [ $PYTHONPATH ]; then \n'
        txt += '        update_fjr\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 += '    if [ $middleware == OSG ]; then\n'
        txt += '        final_list = $filesToCheck\n'
        txt += '        if [ $WORKING_DIR]; then\n'
        txt += '            remove_working_dir\n'
        txt += '        fi\n'
        txt += '    else\n'
        txt += '        final_list = $filesToCheck" .BrokerInfo"\n'
        txt += '    fi\n'

        return txt

    def checkProxy(self):
        """
        Function to check the Globus proxy.
        """
        if (self.proxyValid): return

        ### Just return if asked to do so
        if (self.dontCheckProxy==1):
            self.proxyValid=1
            return

        minTimeLeft=10*3600 # in seconds

        minTimeLeftServer = 100 # in hours

        mustRenew = 0
        timeLeftLocal = runCommand('voms-proxy-info -timeleft 2>/dev/null')
        timeLeftServer = -999
        if not timeLeftLocal or int(timeLeftLocal) <= 0 or not isInt(timeLeftLocal):
            mustRenew = 1
        else:
            timeLeftServer = runCommand('voms-proxy-info -actimeleft 2>/dev/null | head -1')
            if not timeLeftServer or not isInt(timeLeftServer):
                mustRenew = 1
            elif int(timeLeftLocal)<minTimeLeft or int(timeLeftServer)<minTimeLeft:
                mustRenew = 1
            pass
        pass

        if mustRenew:
            common.logger.message( "No valid proxy found or remaining time of validity of already existing proxy shorter than 10 hours!\n Creating a user proxy with default length of 192h\n")
            cmd = 'voms-proxy-init -voms '+self.VO
            if self.group:
                cmd += ':/'+self.VO+'/'+self.group
            if self.role:
                cmd += '/role='+self.role
            cmd += ' -valid 192:00'
            try:
                # SL as above: damn it!
                common.logger.debug(10,cmd)
                out = os.system(cmd)
                if (out>0): raise CrabException("Unable to create a valid proxy!\n")
            except:
                msg = "Unable to create a valid proxy!\n"
                raise CrabException(msg)
            pass

        ## now I do have a voms proxy valid, and I check the myproxy server
        renewProxy = 0
        cmd = 'myproxy-info -d -s '+self.proxyServer
        cmd_out = runCommand(cmd,0,20)
        if not cmd_out:
            common.logger.message('No credential delegated to myproxy server '+self.proxyServer+' will do now')
            renewProxy = 1
        else:
            ## minimum time: 5 days
            minTime = 4 * 24 * 3600
            ## regex to extract the right information
            myproxyRE = re.compile("timeleft: (?P<hours>[\\d]*):(?P<minutes>[\\d]*):(?P<seconds>[\\d]*)")
            for row in cmd_out.split("\n"):
                g = myproxyRE.search(row)
                if g:
                    hours = g.group("hours")
                    minutes = g.group("minutes")
                    seconds = g.group("seconds")
                    timeleft = int(hours)*3600 + int(minutes)*60 + int(seconds)
                    if timeleft < minTime:
                        renewProxy = 1
                        common.logger.message('Your proxy will expire in:\n\t'+hours+' hours '+minutes+' minutes '+seconds+' seconds\n')
                        common.logger.message('Need to renew it:')
                    pass
                pass
            pass

        # if not, create one.
        if renewProxy:
            cmd = 'myproxy-init -d -n -s '+self.proxyServer
            out = os.system(cmd)
            if (out>0):
                raise CrabException("Unable to delegate the proxy to myproxyserver "+self.proxyServer+" !\n")
            pass

        # cache proxy validity
        self.proxyValid=1
        return

    def userName(self):
        """ return the user name """
        tmp=runCommand("voms-proxy-info -identity")
        return tmp.strip()

    def configOpt_(self):
        edg_ui_cfg_opt = ' '
        if self.edg_config:
            edg_ui_cfg_opt = ' -c ' + self.edg_config + ' '
        if self.edg_config_vo:
            edg_ui_cfg_opt += ' --config-vo ' + self.edg_config_vo + ' '
        return edg_ui_cfg_opt


    def tags(self):
        task=common._db.getTask()
        tags_tmp=string.split(task['jobType'],'"')
        tags=[str(tags_tmp[1]),str(tags_tmp[3])]
        return tags

Revision:	1.41
Committed:	Sat May 3 17:14:16 2008 UTC (17 years ago) by spiga
Content type:	text/x-python
Branch:	MAIN
Changes since 1.40:	+26 -2 lines
Log Message:	removed func_exit() from template. Now it is written by crab... and configured with scheduler specific needed. Restored check for OSB size if scheduler is gLite, reorganized the real scheduler (means the BLite script interacting with real scheduler) configuration parameters. Now the dictionary is prepared by different schedulers and any needed parameters can be shipped.
#	User	Rev	Content
1	gutsche	1.1	from Scheduler import Scheduler
2			from crab_logger import Logger
3			from crab_exceptions import *
4			from crab_util import *
5	slacapra	1.8	from BlackWhiteListParser import BlackWhiteListParser
6	gutsche	1.1	import common
7	fanzago	1.30	from LFNBaseName import *
8	spiga	1.28	from JobList import JobList
9	gutsche	1.1
10			import os, sys, time
11
12	slacapra	1.8	#
13			# Base class for all grid scheduler
14			#
15
16	gutsche	1.1	class SchedulerGrid(Scheduler):
17	slacapra	1.8
18			def __init__(self, name):
19			Scheduler.__init__(self,name)
20	gutsche	1.1	self.states = [ "Acl", "cancelReason", "cancelling","ce_node","children", \
21			"children_hist","children_num","children_states","condorId","condor_jdl", \
22			"cpuTime","destination", "done_code","exit_code","expectFrom", \
23			"expectUpdate","globusId","jdl","jobId","jobtype", \
24			"lastUpdateTime","localId","location", "matched_jdl","network_server", \
25			"owner","parent_job", "reason","resubmitted","rsl","seed",\
26			"stateEnterTime","stateEnterTimes","subjob_failed", \
27			"user tags" , "status" , "status_code","hierarchy"]
28			return
29	slacapra	1.12
30	slacapra	1.8	def configure(self, cfg_params):
31			Scheduler.configure(self,cfg_params)
32
33			# init BlackWhiteListParser
34			self.blackWhiteListParser = BlackWhiteListParser(cfg_params)
35
36			self.proxyValid=0
37			self.dontCheckProxy=int(cfg_params.get("EDG.dont_check_proxy",0))
38
39	spiga	1.22	# self.rb_param_file=None
40			# if (cfg_params.has_key('EDG.rb')):
41			# self.rb_param_file=self.rb_configure(cfg_params.get("EDG.rb"))
42	slacapra	1.8
43			self.proxyServer = cfg_params.get("EDG.proxy_server",'myproxy.cern.ch')
44			common.logger.debug(5,'Setting myproxy server to '+self.proxyServer)
45
46			self.group = cfg_params.get("EDG.group", None)
47
48			self.role = cfg_params.get("EDG.role", None)
49
50			self.EDG_ce_black_list = cfg_params.get('EDG.ce_black_list',None)
51			if (self.EDG_ce_black_list): self.EDG_ce_black_list = string.split(self.EDG_ce_black_list,',')
52
53			self.EDG_ce_white_list = cfg_params.get('EDG.ce_white_list',None)
54			if (self.EDG_ce_white_list): self.EDG_ce_white_list = string.split(self.EDG_ce_white_list,',')
55
56			self.VO = cfg_params.get('EDG.virtual_organization','cms')
57
58			self.return_data = cfg_params.get('USER.return_data',0)
59	gutsche	1.1
60	slacapra	1.8	self.copy_data = cfg_params.get("USER.copy_data",0)
61			if int(self.copy_data) == 1:
62			self.SE = cfg_params.get('USER.storage_element',None)
63			self.SE_PATH = cfg_params.get('USER.storage_path',None)
64	spiga	1.14	self.srm_ver = cfg_params.get('USER.srm_version',0) ## DS for srmv2
65	slacapra	1.8	if not self.SE or not self.SE_PATH:
66			msg = "Error. The [USER] section does not have 'storage_element'"
67			msg = msg + " and/or 'storage_path' entries, necessary to copy the output"
68			common.logger.message(msg)
69			raise CrabException(msg)
70	ewv	1.40	if self.srm_ver != 1 and self.SE_PATH.find("managerv1") != -1:
71			msg = "Warning: The storage_path and srm_version seem to be incompatible."
72			msg = msg + " Set srm_version = 1 if your path contains manager1"
73			common.logger.message(msg)
74	gutsche	1.1
75			if ( int(self.return_data) == 0 and int(self.copy_data) == 0 ):
76	slacapra	1.8	msg = 'Error: return_data = 0 and copy_data = 0 ==> your exe output will be lost\n'
77			msg = msg + 'Please modify return_data and copy_data value in your crab.cfg file\n'
78	gutsche	1.1	raise CrabException(msg)
79	fanzago	1.6
80	slacapra	1.8	if ( int(self.return_data) == 1 and int(self.copy_data) == 1 ):
81			msg = 'Error: return_data and copy_data cannot be set both to 1\n'
82			msg = msg + 'Please modify return_data or copy_data value in your crab.cfg file\n'
83			raise CrabException(msg)
84
85			self.publish_data = cfg_params.get("USER.publish_data",0)
86	afanfani	1.39	self.usenamespace = cfg_params.get("USER.usenamespace",0)
87			if int(self.publish_data) == 1 or int(self.usenamespace) == 1:
88	slacapra	1.8	self.publish_data_name = cfg_params.get('USER.publish_data_name',None)
89	afanfani	1.39	if not self.publish_data_name and int(self.publish_data) == 1:
90	slacapra	1.8	msg = "Error. The [USER] section does not have 'publish_data_name'"
91			raise CrabException(msg)
92	afanfani	1.39	if not self.publish_data_name and int(self.usenamespace) == 1:
93			self.publish_data_name = "DefaultDataset"
94	slacapra	1.8
95			## SL I don't like a direct call to voms-proxy here
96	fanzago	1.15
97			import urllib
98	slacapra	1.8	try:
99	fanzago	1.15	userdn = runCommand("voms-proxy-info -identity")
100			self.userdn = string.strip(userdn)
101	slacapra	1.8	except:
102			msg = "Error. Problem with voms-proxy-info -identity command"
103			raise CrabException(msg)
104	fanzago	1.15	try:
105	afanfani	1.21	sitedburl="https://cmsweb.cern.ch/sitedb/sitedb/json/index/dnUserName"
106	fanzago	1.15	params = urllib.urlencode({'dn': self.userdn })
107			f = urllib.urlopen(sitedburl,params)
108			udata = f.read()
109			userinfo= eval(udata)
110			self.hnUserName = userinfo['user']
111			except:
112			msg = "Error. Problem extracting user name from %s"%sitedburl
113			raise CrabException(msg)
114	ewv	1.24
115	fanzago	1.15	if not self.hnUserName:
116			msg = "Error. There is no user name associated to DN %s in %s.You need to register in SiteDB"%(userdn,sitedburl)
117			raise CrabException(msg)
118	slacapra	1.8
119			if ( int(self.copy_data) == 0 and int(self.publish_data) == 1 ):
120			msg = 'Warning: publish_data = 1 must be used with copy_data = 1\n'
121			msg = msg + 'Please modify copy_data value in your crab.cfg file\n'
122	gutsche	1.1	common.logger.message(msg)
123			raise CrabException(msg)
124	slacapra	1.8
125			self.EDG_requirements = cfg_params.get('EDG.requirements',None)
126
127			self.EDG_addJdlParam = cfg_params.get('EDG.additional_jdl_parameters',None)
128			if (self.EDG_addJdlParam): self.EDG_addJdlParam = string.split(self.EDG_addJdlParam,';')
129
130	spiga	1.20	self.EDG_retry_count = cfg_params.get('EDG.retry_count',0)
131	slacapra	1.8
132	spiga	1.20	self.EDG_shallow_retry_count= cfg_params.get('EDG.shallow_retry_count',-1)
133	gutsche	1.1
134	slacapra	1.8	self.EDG_clock_time = cfg_params.get('EDG.max_wall_clock_time',None)
135
136	afanfani	1.23	# Default minimum CPU time to >= 130 minutes
137			self.EDG_cpu_time = cfg_params.get('EDG.max_cpu_time', '130')
138	gutsche	1.1
139			# Add EDG_WL_LOCATION to the python path
140
141	slacapra	1.8	if not os.environ.has_key('EDG_WL_LOCATION'):
142	gutsche	1.1	msg = "Error: the EDG_WL_LOCATION variable is not set."
143	ewv	1.27	raise CrabException(msg)
144			path = os.environ['EDG_WL_LOCATION']
145	gutsche	1.1
146	ewv	1.27	libPath=os.path.join(path, "lib")
147			sys.path.append(libPath)
148			libPath=os.path.join(path, "lib", "python")
149			sys.path.append(libPath)
150	ewv	1.24
151	spiga	1.36	self._taskId=str("_".join(common._db.queryTask('name').split('_')[:-1]))
152	slacapra	1.8	self.jobtypeName = cfg_params.get('CRAB.jobtype','')
153
154			self.schedulerName = cfg_params.get('CRAB.scheduler','')
155
156	gutsche	1.1	return
157	slacapra	1.8
158			def rb_configure(self, RB):
159			"""
160	ewv	1.16	Return a requirement to be add to Jdl to select a specific RB/WMS:
161	slacapra	1.8	return None if RB=None
162			To be re-implemented in concrete scheduler
163			"""
164			return None
165	ewv	1.24
166			def sched_fix_parameter(self):
167	spiga	1.20	return
168	gutsche	1.1
169	slacapra	1.35	def listMatch(self, dest):
170			ces=Scheduler.listMatch(self,dest)
171			sites=[]
172			for ce in ces:
173			site=ce.split(":")[0]
174			if site not in sites:
175			sites.append(site)
176			pass
177
178			return sites
179
180
181	gutsche	1.1	def wsSetupEnvironment(self):
182			"""
183			Returns part of a job script which does scheduler-specific work.
184			"""
185	spiga	1.28	index = int(common._db.nJobs())
186			job = common.job_list[index-1]
187			jbt = job.type()
188	slacapra	1.8	if not self.environment_unique_identifier:
189			raise CrabException('environment_unique_identifier not set')
190
191			txt = '# '+self.name()+' specific stuff\n'
192			txt += '# strip arguments\n'
193			txt += 'echo "strip arguments"\n'
194			txt += 'args=("$@")\n'
195			txt += 'nargs=$#\n'
196			txt += 'shift $nargs\n'
197			txt += "# job number (first parameter for job wrapper)\n"
198	ewv	1.16	txt += "NJob=${args[0]}; export NJob\n"
199	slacapra	1.8
200	spiga	1.29	txt += "out_files=out_files_${NJob}; export out_files\n"
201			txt += "echo $out_files\n"
202	spiga	1.28	txt += jbt.outList()
203	spiga	1.29
204	fanzago	1.17	txt += 'MonitorJobID=${NJob}_$'+self.environment_unique_identifier+'\n'
205			txt += 'SyncGridJobId=$'+self.environment_unique_identifier+'\n'
206			txt += 'MonitorID='+self._taskId+'\n'
207			txt += 'echo "MonitorJobID=$MonitorJobID" > $RUNTIME_AREA/$repo \n'
208			txt += 'echo "SyncGridJobId=$SyncGridJobId" >> $RUNTIME_AREA/$repo \n'
209			txt += 'echo "MonitorID=$MonitorID" >> $RUNTIME_AREA/$repo\n'
210	slacapra	1.8
211	fanzago	1.17	txt += 'echo ">>> GridFlavour discovery: " \n'
212	slacapra	1.8	txt += 'if [ $OSG_APP ]; then \n'
213			txt += ' middleware=OSG \n'
214			txt += ' if [ $OSG_JOB_CONTACT ]; then \n'
215			txt += ' SyncCE="$OSG_JOB_CONTACT"; \n'
216	fanzago	1.17	txt += ' echo "SyncCE=$SyncCE" >> $RUNTIME_AREA/$repo ;\n'
217	slacapra	1.8	txt += ' else\n'
218			txt += ' echo "not reporting SyncCE";\n'
219			txt += ' fi\n';
220	fanzago	1.17	txt += ' echo "GridFlavour=$middleware" \| tee -a $RUNTIME_AREA/$repo \n'
221	slacapra	1.8	txt += 'elif [ $VO_CMS_SW_DIR ]; then \n'
222	spiga	1.3	txt += ' middleware=LCG \n'
223	fanzago	1.17	txt += ' echo "SyncCE=`glite-brokerinfo getCE`" >> $RUNTIME_AREA/$repo \n'
224			txt += ' echo "GridFlavour=$middleware" \| tee -a $RUNTIME_AREA/$repo \n'
225	gutsche	1.1	txt += 'else \n'
226	fanzago	1.17	txt += ' echo "ERROR ==> GridFlavour not identified" \n'
227			txt += ' job_exit_code=10030 \n'
228			txt += ' func_exit \n'
229	slacapra	1.8	txt += 'fi \n'
230
231			txt += 'dumpStatus $RUNTIME_AREA/$repo \n'
232	gutsche	1.1	txt += '\n\n'
233
234	ewv	1.37	txt += 'export VO='+self.VO+'\n'
235			txt += 'if [ $middleware == LCG ]; then\n'
236			txt += ' CloseCEs=`glite-brokerinfo getCE`\n'
237			txt += ' echo "CloseCEs = $CloseCEs"\n'
238			txt += ' CE=`echo $CloseCEs \| sed -e "s/:.*//"`\n'
239			txt += ' echo "CE = $CE"\n'
240			txt += 'elif [ $middleware == OSG ]; then \n'
241			txt += ' if [ $OSG_JOB_CONTACT ]; then \n'
242			txt += ' CE=`echo $OSG_JOB_CONTACT \| /usr/bin/awk -F\/ \'{print $1}\'` \n'
243			txt += ' else \n'
244			txt += ' echo "ERROR ==> OSG mode in setting CE name from OSG_JOB_CONTACT" \n'
245			txt += ' job_exit_code=10099\n'
246			txt += ' func_exit\n'
247			txt += ' fi \n'
248			txt += 'fi \n'
249
250	gutsche	1.1	return txt
251
252	slacapra	1.34	# def wsCopyInput(self):
253			# """
254			# Copy input data from SE to WN
255			# """
256			# txt = ''
257			# if not self.copy_input_data: return txt
258
259			# ## OLI_Daniele deactivate for OSG (wait for LCG UI installed on OSG)
260			# txt += 'if [ $middleware == OSG ]; then\n'
261			# txt += ' #\n'
262			# txt += ' # Copy Input Data from SE to this WN deactivated in OSG mode\n'
263			# txt += ' #\n'
264			# txt += ' echo "Copy Input Data from SE to this WN deactivated in OSG mode"\n'
265			# txt += 'elif [ $middleware == LCG ]; then \n'
266			# txt += ' #\n'
267			# txt += ' # Copy Input Data from SE to this WN\n'
268			# txt += ' #\n'
269			# ### changed by georgia (put a loop copying more than one input files per jobs)
270			# txt += ' for input_file in $cur_file_list \n'
271			# txt += ' do \n'
272			# txt += ' lcg-cp --vo $VO --verbose -t 1200 lfn:$input_lfn/$input_file file:`pwd`/$input_file 2>&1\n'
273			# txt += ' copy_input_exit_status=$?\n'
274			# txt += ' echo "COPY_INPUT_EXIT_STATUS = $copy_input_exit_status"\n'
275			# txt += ' if [ $copy_input_exit_status -ne 0 ]; then \n'
276			# txt += ' echo "Problems with copying to WN" \n'
277			# txt += ' else \n'
278			# txt += ' echo "input copied into WN" \n'
279			# txt += ' fi \n'
280			# txt += ' done \n'
281			# ### copy a set of PU ntuples (same for each jobs -- but accessed randomly)
282			# txt += ' for file in $cur_pu_list \n'
283			# txt += ' do \n'
284			# txt += ' lcg-cp --vo $VO --verbose -t 1200 lfn:$pu_lfn/$file file:`pwd`/$file 2>&1\n'
285			# txt += ' copy_input_pu_exit_status=$?\n'
286			# txt += ' echo "COPY_INPUT_PU_EXIT_STATUS = $copy_input_pu_exit_status"\n'
287			# txt += ' if [ $copy_input_pu_exit_status -ne 0 ]; then \n'
288			# txt += ' echo "Problems with copying pu to WN" \n'
289			# txt += ' else \n'
290			# txt += ' echo "input pu files copied into WN" \n'
291			# txt += ' fi \n'
292			# txt += ' done \n'
293			# txt += ' \n'
294			# txt += ' ### Check SCRATCH space available on WN : \n'
295			# txt += ' df -h \n'
296			# txt += 'fi \n'
297	slacapra	1.8
298	slacapra	1.34	# return txt
299	gutsche	1.1
300			def wsCopyOutput(self):
301			"""
302			Write a CopyResults part of a job script, e.g.
303			to copy produced output into a storage element.
304			"""
305	slacapra	1.8	txt = '\n'
306	gutsche	1.1
307	spiga	1.18	txt += '#\n'
308			txt += '# COPY OUTPUT FILE TO SE\n'
309			txt += '#\n\n'
310	gutsche	1.1
311	slacapra	1.8	SE_PATH=''
312			if int(self.copy_data) == 1:
313			if self.SE:
314			txt += 'export SE='+self.SE+'\n'
315			txt += 'echo "SE = $SE"\n'
316			if self.SE_PATH:
317			if ( self.SE_PATH[-1] != '/' ) : self.SE_PATH = self.SE_PATH + '/'
318			SE_PATH=self.SE_PATH
319	afanfani	1.39	if int(self.publish_data) == 1 or int(self.usenamespace) == 1:
320	slacapra	1.8	txt += '### publish_data = 1 so the SE path where to copy the output is: \n'
321	fanzago	1.30	#path_add = self.hnUserName + '/' + self.publish_data_name +'_${PSETHASH}/'
322			path_add = PFNportion(self.publish_data_name) +'_${PSETHASH}/'
323	slacapra	1.8	SE_PATH = SE_PATH + path_add
324			txt += 'export SE_PATH='+SE_PATH+'\n'
325			txt += 'echo "SE_PATH = $SE_PATH"\n'
326
327	spiga	1.14	txt += 'export SRM_VER='+str(self.srm_ver)+'\n' ## DS for srmVer
328			txt += 'echo "SRM_VER = $SRM_VER"\n' ## DS for srmVer
329
330	slacapra	1.8	txt += 'echo ">>> Copy output files from WN = `hostname` to SE = $SE :"\n'
331	fanzago	1.17	txt += 'copy_exit_status=0\n'
332			txt += 'for out_file in $file_list ; do\n'
333			txt += ' if [ -e $SOFTWARE_DIR/$out_file ] ; then\n'
334			txt += ' echo "Trying to copy output file $SOFTWARE_DIR/$out_file to $SE"\n'
335	spiga	1.14	txt += ' cmscp $SOFTWARE_DIR/$out_file ${SE} ${SE_PATH} $out_file ${SRM_VER} $middleware\n'
336	fanzago	1.17	txt += ' if [ $cmscp_exit_status -ne 0 ]; then\n'
337	slacapra	1.8	txt += ' echo "Problem copying $out_file to $SE $SE_PATH"\n'
338	fanzago	1.17	txt += ' copy_exit_status=$cmscp_exit_status\n'
339	slacapra	1.8	txt += ' else\n'
340			txt += ' echo "output copied into $SE/$SE_PATH directory"\n'
341			txt += ' fi\n'
342	fanzago	1.17	txt += ' else\n'
343			txt += ' copy_exit_status=60302\n'
344			txt += ' echo "StageOutExitStatus = $copy_exit_status" \| tee -a $RUNTIME_AREA/$repo\n'
345			txt += ' echo "StageOutExitStatusReason = file to copy not found" \| tee -a $RUNTIME_AREA/$repo\n'
346	slacapra	1.8	txt += ' fi\n'
347	fanzago	1.17	txt += 'done\n'
348			txt += 'if [ $copy_exit_status -ne 0 ]; then\n'
349			txt += ' SE=""\n'
350			txt += ' SE_PATH=""\n'
351			txt += ' job_exit_code=$copy_exit_status\n'
352	slacapra	1.8	txt += 'fi\n'
353			pass
354	gutsche	1.1	return txt
355
356
357	spiga	1.41	def wsExitFunc_common(self):
358			"""
359			"""
360			txt = ''
361			txt += ' if [ $PYTHONPATH ]; then \n'
362			txt += ' update_fjr\n'
363			txt += ' fi\n'
364			txt += ' cd $RUNTIME_AREA \n'
365			txt += ' for file in $filesToCheck ; do\n'
366			txt += ' if [ -e $file ]; then\n'
367			txt += ' echo "tarring file $file in $out_files"\n'
368			txt += ' else\n'
369			txt += ' echo "WARNING: output file $file not found!"\n'
370			txt += ' fi\n'
371			txt += ' done\n'
372			txt += ' if [ $middleware == OSG ]; then\n'
373			txt += ' final_list = $filesToCheck\n'
374			txt += ' if [ $WORKING_DIR]; then\n'
375			txt += ' remove_working_dir\n'
376			txt += ' fi\n'
377			txt += ' else\n'
378			txt += ' final_list = $filesToCheck" .BrokerInfo"\n'
379			txt += ' fi\n'
380
381			return txt
382
383	gutsche	1.1	def checkProxy(self):
384			"""
385			Function to check the Globus proxy.
386			"""
387			if (self.proxyValid): return
388	slacapra	1.8
389			### Just return if asked to do so
390			if (self.dontCheckProxy==1):
391			self.proxyValid=1
392			return
393
394			minTimeLeft=10*3600 # in seconds
395
396			minTimeLeftServer = 100 # in hours
397
398			mustRenew = 0
399			timeLeftLocal = runCommand('voms-proxy-info -timeleft 2>/dev/null')
400			timeLeftServer = -999
401			if not timeLeftLocal or int(timeLeftLocal) <= 0 or not isInt(timeLeftLocal):
402			mustRenew = 1
403			else:
404			timeLeftServer = runCommand('voms-proxy-info -actimeleft 2>/dev/null \| head -1')
405			if not timeLeftServer or not isInt(timeLeftServer):
406			mustRenew = 1
407	slacapra	1.33	elif int(timeLeftLocal)<minTimeLeft or int(timeLeftServer)<minTimeLeft:
408	slacapra	1.8	mustRenew = 1
409			pass
410			pass
411
412			if mustRenew:
413			common.logger.message( "No valid proxy found or remaining time of validity of already existing proxy shorter than 10 hours!\n Creating a user proxy with default length of 192h\n")
414			cmd = 'voms-proxy-init -voms '+self.VO
415			if self.group:
416			cmd += ':/'+self.VO+'/'+self.group
417			if self.role:
418			cmd += '/role='+self.role
419			cmd += ' -valid 192:00'
420	gutsche	1.1	try:
421			# SL as above: damn it!
422	slacapra	1.8	common.logger.debug(10,cmd)
423	gutsche	1.1	out = os.system(cmd)
424			if (out>0): raise CrabException("Unable to create a valid proxy!\n")
425			except:
426			msg = "Unable to create a valid proxy!\n"
427			raise CrabException(msg)
428			pass
429	slacapra	1.8
430			## now I do have a voms proxy valid, and I check the myproxy server
431			renewProxy = 0
432			cmd = 'myproxy-info -d -s '+self.proxyServer
433			cmd_out = runCommand(cmd,0,20)
434			if not cmd_out:
435			common.logger.message('No credential delegated to myproxy server '+self.proxyServer+' will do now')
436			renewProxy = 1
437			else:
438			## minimum time: 5 days
439			minTime = 4 * 24 * 3600
440			## regex to extract the right information
441			myproxyRE = re.compile("timeleft: (?P<hours>[\\d]):(?P<minutes>[\\d]):(?P<seconds>[\\d]*)")
442			for row in cmd_out.split("\n"):
443			g = myproxyRE.search(row)
444			if g:
445			hours = g.group("hours")
446			minutes = g.group("minutes")
447			seconds = g.group("seconds")
448			timeleft = int(hours)3600 + int(minutes)60 + int(seconds)
449			if timeleft < minTime:
450			renewProxy = 1
451			common.logger.message('Your proxy will expire in:\n\t'+hours+' hours '+minutes+' minutes '+seconds+' seconds\n')
452			common.logger.message('Need to renew it:')
453			pass
454			pass
455			pass
456
457			# if not, create one.
458			if renewProxy:
459			cmd = 'myproxy-init -d -n -s '+self.proxyServer
460			out = os.system(cmd)
461			if (out>0):
462			raise CrabException("Unable to delegate the proxy to myproxyserver "+self.proxyServer+" !\n")
463			pass
464
465			# cache proxy validity
466	gutsche	1.1	self.proxyValid=1
467			return
468
469	slacapra	1.8	def userName(self):
470			""" return the user name """
471			tmp=runCommand("voms-proxy-info -identity")
472			return tmp.strip()
473
474	gutsche	1.1	def configOpt_(self):
475			edg_ui_cfg_opt = ' '
476			if self.edg_config:
477	slacapra	1.8	edg_ui_cfg_opt = ' -c ' + self.edg_config + ' '
478			if self.edg_config_vo:
479			edg_ui_cfg_opt += ' --config-vo ' + self.edg_config_vo + ' '
480	gutsche	1.1	return edg_ui_cfg_opt
481	slacapra	1.8
482
483
484	slacapra	1.34	def tags(self):
485			task=common._db.getTask()
486	ewv	1.37	tags_tmp=string.split(task['jobType'],'"')
487	slacapra	1.34	tags=[str(tags_tmp[1]),str(tags_tmp[3])]
488			return tags
489	ewv	1.37