CRAB/python/SchedulerCondorCommon.py

from SchedulerGrid import SchedulerGrid
from JobList import JobList
from crab_logger import Logger
from crab_exceptions import *
from crab_util import *
from osg_bdii import getJobManagerList
from osg_bdii import listAllSEs, listAllCEs
import time
import common
import popen2
import os
from BlackWhiteListParser import BlackWhiteListParser
from BlackWhiteListParser import CEBlackWhiteListParser
import Scram
import CondorGLoggingInfo

# This class was originally SchedulerCondor_g. For a history of this code, see that file.

__revision__ = "$Id: SchedulerCondorCommon.py,v 1.19 2008/06/06 18:02:08 ewv Exp $"
__version__ = "$Revision: 1.19 $"

class SchedulerCondorCommon(SchedulerGrid):
    def __init__(self,name):
        SchedulerGrid.__init__(self,name)

        # check for locally running condor scheduler
        cmd = 'ps xau | grep -i condor_schedd | grep -v grep'
        cmd_out = runCommand(cmd)
        if cmd_out == None:
            msg  = '[Condor-G Scheduler]: condor_schedd is not running on this machine.\n'
            msg += '[Condor-G Scheduler]: Please use another machine with installed condor and running condor_schedd or change the Scheduler in your crab.cfg.'
            common.logger.debug(2,msg)
            raise CrabException(msg)

        self.checkExecutableInPath('condor_q')
        self.checkExecutableInPath('condor_submit')
        self.checkExecutableInPath('condor_version')

        # get version number
        cmd = 'condor_version'
        cmd_out = runCommand(cmd)
        if cmd != None :
            tmp = cmd_out.find('CondorVersion') + 15
            version = cmd_out[tmp:tmp+6].split('.')
            version_master = int(version[0])
            version_major  = int(version[1])
            version_minor  = int(version[2])
        else :
            msg  = '[Condor-G Scheduler]: condor_version was not able to determine the installed condor version.\n'
            msg += '[Condor-G Scheduler]: Please use another machine with properly installed condor or change the Scheduler in your crab.cfg.'
            common.logger.debug(2,msg)
            raise CrabException(msg)

        self.checkExecutableInPath('condor_config_val')
        self.checkCondorVariablePointsToFile('GRIDMANAGER')
        self.checkCondorVariablePointsToFile('GT2_GAHP',alternate_name='GAHP')
        self.checkCondorVariablePointsToFile('GRID_MONITOR')
        self.checkCondorVariableIsTrue('ENABLE_GRID_MONITOR')

        max_submit = self.queryCondorVariable('GRIDMANAGER_MAX_SUBMITTED_JOBS_PER_RESOURCE',100).strip()
        max_pending = self.queryCondorVariable('GRIDMANAGER_MAX_PENDING_SUBMITS_PER_RESOURCE','Unlimited').strip()

        msg  = '[Condor-G Scheduler]\n'
        msg += 'Maximal number of jobs submitted to the grid   : GRIDMANAGER_MAX_SUBMITTED_JOBS_PER_RESOURCE  = '+max_submit+'\n'
        msg += 'Maximal number of parallel submits to the grid : GRIDMANAGER_MAX_PENDING_SUBMITS_PER_RESOURCE = '+max_pending+'\n'
        msg += 'Ask the administrator of your local condor installation to increase these variables to enable more jobs to be executed on the grid in parallel.\n'
        common.logger.debug(2,msg)

        # create hash
        self.hash = makeCksum(common.work_space.cfgFileName())

        return

    def checkExecutableInPath(self, name):
        # check if executable is in PATH
        cmd = 'which '+name
        cmd_out = runCommand(cmd)
        if cmd_out == None:
            msg  = '[Condor-G Scheduler]: '+name+' is not in the $PATH on this machine.\n'
            msg += '[Condor-G Scheduler]: Please use another machine with installed condor or change the Scheduler in your crab.cfg.'
            common.logger.debug(2,msg)
            raise CrabException(msg)

    def checkCondorVariablePointsToFile(self, name, alternate_name=None):
        ## check for condor variable
        cmd = 'condor_config_val '+name
        cmd_out = runCommand(cmd)
        if alternate_name and not cmd_out:
            cmd = 'condor_config_val '+alternate_name
            cmd_out = runCommand(cmd)
        if cmd_out:
            cmd_out = string.strip(cmd_out)
        if not cmd_out or not os.path.isfile(cmd_out) :
            msg  = '[Condor-G Scheduler]: the variable '+name+' is not properly set for the condor installation on this machine.\n'
            msg += '[Condor-G Scheduler]: Please ask the administrator of the local condor installation to set the variable '+name+' properly, ' + \
            'use another machine with properly installed condor or change the Scheduler in your crab.cfg.'
            common.logger.debug(2,msg)
            raise CrabException(msg)

    def checkCondorVariableIsTrue(self, name):
        ## check for condor variable
        cmd = 'condor_config_val '+name
        cmd_out = runCommand(cmd)
        if cmd_out == 'TRUE' :
            msg  = '[Condor-G Scheduler]: the variable '+name+' is not set to true for the condor installation on this machine.\n'
            msg += '[Condor-G Scheduler]: Please ask the administrator of the local condor installation to set the variable '+name+' to true,',
            'use another machine with properly installed condor or change the Scheduler in your crab.cfg.'
            common.logger.debug(2,msg)
            raise CrabException(msg)

    def queryCondorVariable(self, name, default):
        ## check for condor variable
        cmd = 'condor_config_val '+name
        out = popen2.Popen3(cmd,1)
        exit_code = out.wait()
        cmd_out = out.fromchild.readline().strip()
        if exit_code != 0 :
            cmd_out = str(default)

        return cmd_out

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

        # init BlackWhiteListParser
        self.ceBlackWhiteListParser = CEBlackWhiteListParser(cfg_params)

        try:
          self.GLOBUS_RSL = cfg_params['CONDORG.globus_rsl']
        except KeyError:
          self.GLOBUS_RSL = ''

        # Provide an override for the batchsystem that condor_g specifies as a grid resource.
        # this is to handle the case where the site supports several batchsystem but bdii
        # only allows a site to public one.
        try:
          self.batchsystem = cfg_params['CONDORG.batchsystem']
          msg = '[Condor-G Scheduler]: batchsystem overide specified in your crab.cfg'
          common.logger.debug(2,msg)
        except KeyError:
          self.batchsystem = ''

        try:
            self.UseGT4 = cfg_params['USER.use_gt_4'];
        except KeyError:
            self.UseGT4 = 0;

        # added here because checklistmatch is not used
        self.environment_unique_identifier = 'GLOBUS_GRAM_JOB_CONTACT'
        self.datasetPath = ''

        try:
            tmp =  cfg_params['CMSSW.datasetpath']
            if string.lower(tmp)=='none':
                self.datasetPath = None
                self.selectNoInput = 1
            else:
                self.datasetPath = tmp
                self.selectNoInput = 0
        except KeyError:
            msg = "Error: datasetpath not defined "
            raise CrabException(msg)

        return

    def realSchedParams(self,cfg_params):
      """
      Return dictionary with specific parameters, to use
      with real scheduler
      """

      tmpDir = os.path.join(common.work_space.shareDir(),'.condor_temp')
      params = {'tmpDir':tmpDir}
      return  params

    def sched_parameter(self,i,task):
      """
      Returns scheduler-specific parameters
      """
      jobParams = ''

      return jobParams

    def decodeLogInfo(self, file):
        """
        Parse logging info file and return main info
        """
        loggingInfo = CondorGLoggingInfo.CondorGLoggingInfo()
        reason = loggingInfo.decodeReason(file)
        return reason

    def listMatch(self, seList, full, onlyOSG=True):
      """
      Check the compatibility of available resources
      """

      # May have problems with onlyOSG being false, probably due to lengths of lists and command line.
      # Either re-write osg_bdii.py with a proper ldap library or break the queries apart

      scram = Scram.Scram(None)
      versionCMSSW = scram.getSWVersion()
      arch = scram.getArch()

      if self.selectNoInput:
        availCEs = listAllCEs(versionCMSSW, arch, onlyOSG=onlyOSG)
      else:
        seDest = self.blackWhiteListParser.cleanForBlackWhiteList(seList,"list")
        availCEs = getJobManagerList(seDest,versionCMSSW,arch,onlyOSG=onlyOSG)
      #print "\n\navailCEs =",availCEs
      uniqCEs = []
      for ce in availCEs:
        if ce not in uniqCEs:
          uniqCEs.append(ce)

      ceDest = self.ceBlackWhiteListParser.cleanForBlackWhiteList(uniqCEs,"list")
      #print "\n\nceDest =",ceDest
      uniqCEs = []
      return ceDest

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

    def ce_list(self):
      """
      Returns string with requirement CE related, dummy for now
      """
      req = ''

      return req,self.EDG_ce_white_list,self.EDG_ce_black_list

    def seListToCElist(self, seList, onlyOSG=True):
      ceDest = self.listMatch(seList, onlyOSG)

      if (not ceDest):
        msg = 'No sites found hosting the data or all sites blocked by CE/SE white/blacklisting'
        print msg
        raise CrabException(msg)

      return ceDest


    def wsExitFunc(self):
        """
        """
        txt = '\n'

        txt += '#\n'
        txt += '# EXECUTE THIS FUNCTION BEFORE EXIT \n'
        txt += '#\n\n'

        txt += 'func_exit() { \n'
        txt += self.wsExitFunc_common()
        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'
        txt += '    tar zcvf ${out_files}.tgz  ${final_list}\n'
        txt += '    exit $job_exit_code\n'
        txt += '}\n'
        return txt
Revision:	1.20
Committed:	Fri Jun 6 20:32:01 2008 UTC (16 years, 10 months ago) by ewv
Content type:	text/x-python
Branch:	MAIN
CVS Tags:	CRAB_2_3_1, CRAB_2_3_1_pre6, CRAB_2_3_1_pre5, CRAB_2_3_1_pre4, CRAB_2_3_1_pre3, CRAB_2_3_1_pre2, CRAB_2_3_1_pre1, CRAB_2_3_0, CRAB_2_3_0_pre6, CRAB_2_3_0_pre1, CRAB_2_2_2_pre5
Branch point for:	CRAB_2_3_0_br
Changes since 1.19:	+15 -23 lines
Log Message:	Fix pythia on OSG. Still a little flakey
#	User	Rev	Content
1	ewv	1.1	from SchedulerGrid import SchedulerGrid
2			from JobList import JobList
3			from crab_logger import Logger
4			from crab_exceptions import *
5			from crab_util import *
6	ewv	1.7	from osg_bdii import getJobManagerList
7	ewv	1.19	from osg_bdii import listAllSEs, listAllCEs
8	ewv	1.1	import time
9			import common
10			import popen2
11			import os
12			from BlackWhiteListParser import BlackWhiteListParser
13	ewv	1.7	from BlackWhiteListParser import CEBlackWhiteListParser
14			import Scram
15	ewv	1.1	import CondorGLoggingInfo
16
17			# This class was originally SchedulerCondor_g. For a history of this code, see that file.
18
19	ewv	1.20	__revision__ = "$Id: SchedulerCondorCommon.py,v 1.19 2008/06/06 18:02:08 ewv Exp $"
20			__version__ = "$Revision: 1.19 $"
21	ewv	1.1
22			class SchedulerCondorCommon(SchedulerGrid):
23			def __init__(self,name):
24			SchedulerGrid.__init__(self,name)
25
26			# check for locally running condor scheduler
27			cmd = 'ps xau \| grep -i condor_schedd \| grep -v grep'
28			cmd_out = runCommand(cmd)
29			if cmd_out == None:
30			msg = '[Condor-G Scheduler]: condor_schedd is not running on this machine.\n'
31			msg += '[Condor-G Scheduler]: Please use another machine with installed condor and running condor_schedd or change the Scheduler in your crab.cfg.'
32			common.logger.debug(2,msg)
33			raise CrabException(msg)
34
35			self.checkExecutableInPath('condor_q')
36			self.checkExecutableInPath('condor_submit')
37			self.checkExecutableInPath('condor_version')
38
39			# get version number
40			cmd = 'condor_version'
41			cmd_out = runCommand(cmd)
42			if cmd != None :
43			tmp = cmd_out.find('CondorVersion') + 15
44			version = cmd_out[tmp:tmp+6].split('.')
45			version_master = int(version[0])
46			version_major = int(version[1])
47			version_minor = int(version[2])
48			else :
49			msg = '[Condor-G Scheduler]: condor_version was not able to determine the installed condor version.\n'
50			msg += '[Condor-G Scheduler]: Please use another machine with properly installed condor or change the Scheduler in your crab.cfg.'
51			common.logger.debug(2,msg)
52			raise CrabException(msg)
53
54			self.checkExecutableInPath('condor_config_val')
55			self.checkCondorVariablePointsToFile('GRIDMANAGER')
56			self.checkCondorVariablePointsToFile('GT2_GAHP',alternate_name='GAHP')
57			self.checkCondorVariablePointsToFile('GRID_MONITOR')
58			self.checkCondorVariableIsTrue('ENABLE_GRID_MONITOR')
59
60			max_submit = self.queryCondorVariable('GRIDMANAGER_MAX_SUBMITTED_JOBS_PER_RESOURCE',100).strip()
61			max_pending = self.queryCondorVariable('GRIDMANAGER_MAX_PENDING_SUBMITS_PER_RESOURCE','Unlimited').strip()
62
63			msg = '[Condor-G Scheduler]\n'
64			msg += 'Maximal number of jobs submitted to the grid : GRIDMANAGER_MAX_SUBMITTED_JOBS_PER_RESOURCE = '+max_submit+'\n'
65			msg += 'Maximal number of parallel submits to the grid : GRIDMANAGER_MAX_PENDING_SUBMITS_PER_RESOURCE = '+max_pending+'\n'
66			msg += 'Ask the administrator of your local condor installation to increase these variables to enable more jobs to be executed on the grid in parallel.\n'
67			common.logger.debug(2,msg)
68
69			# create hash
70			self.hash = makeCksum(common.work_space.cfgFileName())
71
72			return
73
74			def checkExecutableInPath(self, name):
75			# check if executable is in PATH
76			cmd = 'which '+name
77			cmd_out = runCommand(cmd)
78			if cmd_out == None:
79			msg = '[Condor-G Scheduler]: '+name+' is not in the $PATH on this machine.\n'
80			msg += '[Condor-G Scheduler]: Please use another machine with installed condor or change the Scheduler in your crab.cfg.'
81			common.logger.debug(2,msg)
82			raise CrabException(msg)
83
84			def checkCondorVariablePointsToFile(self, name, alternate_name=None):
85			## check for condor variable
86			cmd = 'condor_config_val '+name
87			cmd_out = runCommand(cmd)
88			if alternate_name and not cmd_out:
89			cmd = 'condor_config_val '+alternate_name
90			cmd_out = runCommand(cmd)
91			if cmd_out:
92			cmd_out = string.strip(cmd_out)
93			if not cmd_out or not os.path.isfile(cmd_out) :
94			msg = '[Condor-G Scheduler]: the variable '+name+' is not properly set for the condor installation on this machine.\n'
95			msg += '[Condor-G Scheduler]: Please ask the administrator of the local condor installation to set the variable '+name+' properly, ' + \
96			'use another machine with properly installed condor or change the Scheduler in your crab.cfg.'
97			common.logger.debug(2,msg)
98			raise CrabException(msg)
99
100			def checkCondorVariableIsTrue(self, name):
101			## check for condor variable
102			cmd = 'condor_config_val '+name
103			cmd_out = runCommand(cmd)
104			if cmd_out == 'TRUE' :
105			msg = '[Condor-G Scheduler]: the variable '+name+' is not set to true for the condor installation on this machine.\n'
106			msg += '[Condor-G Scheduler]: Please ask the administrator of the local condor installation to set the variable '+name+' to true,',
107			'use another machine with properly installed condor or change the Scheduler in your crab.cfg.'
108			common.logger.debug(2,msg)
109			raise CrabException(msg)
110
111			def queryCondorVariable(self, name, default):
112			## check for condor variable
113			cmd = 'condor_config_val '+name
114			out = popen2.Popen3(cmd,1)
115			exit_code = out.wait()
116			cmd_out = out.fromchild.readline().strip()
117			if exit_code != 0 :
118			cmd_out = str(default)
119
120			return cmd_out
121
122			def configure(self, cfg_params):
123			SchedulerGrid.configure(self,cfg_params)
124
125			# init BlackWhiteListParser
126	ewv	1.7	self.ceBlackWhiteListParser = CEBlackWhiteListParser(cfg_params)
127	ewv	1.1
128			try:
129			self.GLOBUS_RSL = cfg_params['CONDORG.globus_rsl']
130			except KeyError:
131			self.GLOBUS_RSL = ''
132
133			# Provide an override for the batchsystem that condor_g specifies as a grid resource.
134			# this is to handle the case where the site supports several batchsystem but bdii
135			# only allows a site to public one.
136			try:
137			self.batchsystem = cfg_params['CONDORG.batchsystem']
138			msg = '[Condor-G Scheduler]: batchsystem overide specified in your crab.cfg'
139			common.logger.debug(2,msg)
140			except KeyError:
141			self.batchsystem = ''
142
143			try:
144			self.UseGT4 = cfg_params['USER.use_gt_4'];
145			except KeyError:
146			self.UseGT4 = 0;
147
148			# added here because checklistmatch is not used
149			self.environment_unique_identifier = 'GLOBUS_GRAM_JOB_CONTACT'
150	ewv	1.4	self.datasetPath = ''
151	ewv	1.1
152			try:
153			tmp = cfg_params['CMSSW.datasetpath']
154			if string.lower(tmp)=='none':
155			self.datasetPath = None
156			self.selectNoInput = 1
157			else:
158			self.datasetPath = tmp
159			self.selectNoInput = 0
160			except KeyError:
161			msg = "Error: datasetpath not defined "
162			raise CrabException(msg)
163
164			return
165
166	ewv	1.17	def realSchedParams(self,cfg_params):
167			"""
168			Return dictionary with specific parameters, to use
169			with real scheduler
170			"""
171
172			tmpDir = os.path.join(common.work_space.shareDir(),'.condor_temp')
173			params = {'tmpDir':tmpDir}
174			return params
175
176	ewv	1.1	def sched_parameter(self,i,task):
177			"""
178			Returns scheduler-specific parameters
179			"""
180			jobParams = ''
181
182			return jobParams
183
184			def decodeLogInfo(self, file):
185			"""
186			Parse logging info file and return main info
187			"""
188			loggingInfo = CondorGLoggingInfo.CondorGLoggingInfo()
189			reason = loggingInfo.decodeReason(file)
190			return reason
191
192	ewv	1.20	def listMatch(self, seList, full, onlyOSG=True):
193	ewv	1.1	"""
194	ewv	1.14	Check the compatibility of available resources
195	ewv	1.1	"""
196	ewv	1.20
197			# May have problems with onlyOSG being false, probably due to lengths of lists and command line.
198			# Either re-write osg_bdii.py with a proper ldap library or break the queries apart
199
200			scram = Scram.Scram(None)
201			versionCMSSW = scram.getSWVersion()
202			arch = scram.getArch()
203
204	ewv	1.19	if self.selectNoInput:
205	ewv	1.20	availCEs = listAllCEs(versionCMSSW, arch, onlyOSG=onlyOSG)
206	ewv	1.19	else:
207			seDest = self.blackWhiteListParser.cleanForBlackWhiteList(seList,"list")
208			availCEs = getJobManagerList(seDest,versionCMSSW,arch,onlyOSG=onlyOSG)
209	ewv	1.20	#print "\n\navailCEs =",availCEs
210	ewv	1.7	uniqCEs = []
211			for ce in availCEs:
212			if ce not in uniqCEs:
213			uniqCEs.append(ce)
214
215			ceDest = self.ceBlackWhiteListParser.cleanForBlackWhiteList(uniqCEs,"list")
216	ewv	1.20	#print "\n\nceDest =",ceDest
217	ewv	1.19	uniqCEs = []
218	ewv	1.14	return ceDest
219
220			def userName(self):
221			""" return the user name """
222			return runCommand("voms-proxy-info -identity")
223
224			def ce_list(self):
225			"""
226			Returns string with requirement CE related, dummy for now
227			"""
228			req = ''
229
230			return req,self.EDG_ce_white_list,self.EDG_ce_black_list
231
232			def seListToCElist(self, seList, onlyOSG=True):
233			ceDest = self.listMatch(seList, onlyOSG)
234	ewv	1.7
235			if (not ceDest):
236	ewv	1.20	msg = 'No sites found hosting the data or all sites blocked by CE/SE white/blacklisting'
237	ewv	1.7	print msg
238			raise CrabException(msg)
239
240			return ceDest
241	ewv	1.15
242	spiga	1.16
243			def wsExitFunc(self):
244			"""
245			"""
246			txt = '\n'
247
248			txt += '#\n'
249			txt += '# EXECUTE THIS FUNCTION BEFORE EXIT \n'
250			txt += '#\n\n'
251
252			txt += 'func_exit() { \n'
253			txt += self.wsExitFunc_common()
254			txt += ' echo "JOB_EXIT_STATUS = $job_exit_code"\n'
255			txt += ' echo "JobExitCode=$job_exit_code" >> $RUNTIME_AREA/$repo\n'
256			txt += ' dumpStatus $RUNTIME_AREA/$repo\n'
257			txt += ' tar zcvf ${out_files}.tgz ${final_list}\n'
258			txt += ' exit $job_exit_code\n'
259			txt += '}\n'
260			return txt