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
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.16 2008/05/03 17:14:17 spiga Exp $"
__version__ = "$Revision: 1.16 $"

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.checkProxy()
        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, onlyOSG=True):
      """
      Check the compatibility of available resources
      """
      seDest = self.blackWhiteListParser.cleanForBlackWhiteList(seList,"list")
      scram = Scram.Scram(None)

      versionCMSSW = scram.getSWVersion()
      arch = scram.getArch()
      availCEs = getJobManagerList(seDest,versionCMSSW,arch,onlyOSG=onlyOSG)
      uniqCEs = []
      for ce in availCEs:
        if ce not in uniqCEs:
          uniqCEs.append(ce)

      ceDest = self.ceBlackWhiteListParser.cleanForBlackWhiteList(uniqCEs,"list")
      return ceDest

    def userName(self):
      """ return the user name """
      self.checkProxy()
      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 OSG 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.17
Committed:	Mon May 5 18:46:02 2008 UTC (16 years, 11 months ago) by ewv
Content type:	text/x-python
Branch:	MAIN
CVS Tags:	PRODCOMMON_0_10_7_testCS2, CRAB_2_2_1_pre3, CRAB_2_2_1_pre2, CRAB_2_2_1_pre1, CRAB_2_2_0, CRAB_2_2_0_pre21, CRAB_2_2_0_pre19, CRAB_2_2_0_pre18, CRAB_2_2_0_pre17
Changes since 1.16:	+12 -4 lines
Log Message:	realScheduler for CondorG/Glidein temp dir
#	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.1	import time
8			import common
9			import popen2
10			import os
11			from BlackWhiteListParser import BlackWhiteListParser
12	ewv	1.7	from BlackWhiteListParser import CEBlackWhiteListParser
13			import Scram
14	ewv	1.1	import CondorGLoggingInfo
15
16			# This class was originally SchedulerCondor_g. For a history of this code, see that file.
17
18	ewv	1.17	__revision__ = "$Id: SchedulerCondorCommon.py,v 1.16 2008/05/03 17:14:17 spiga Exp $"
19			__version__ = "$Revision: 1.16 $"
20	ewv	1.1
21			class SchedulerCondorCommon(SchedulerGrid):
22			def __init__(self,name):
23			SchedulerGrid.__init__(self,name)
24
25			# check for locally running condor scheduler
26			cmd = 'ps xau \| grep -i condor_schedd \| grep -v grep'
27			cmd_out = runCommand(cmd)
28			if cmd_out == None:
29			msg = '[Condor-G Scheduler]: condor_schedd is not running on this machine.\n'
30			msg += '[Condor-G Scheduler]: Please use another machine with installed condor and running condor_schedd or change the Scheduler in your crab.cfg.'
31			common.logger.debug(2,msg)
32			raise CrabException(msg)
33
34			self.checkExecutableInPath('condor_q')
35			self.checkExecutableInPath('condor_submit')
36			self.checkExecutableInPath('condor_version')
37
38			# get version number
39			cmd = 'condor_version'
40			cmd_out = runCommand(cmd)
41			if cmd != None :
42			tmp = cmd_out.find('CondorVersion') + 15
43			version = cmd_out[tmp:tmp+6].split('.')
44			version_master = int(version[0])
45			version_major = int(version[1])
46			version_minor = int(version[2])
47			else :
48			msg = '[Condor-G Scheduler]: condor_version was not able to determine the installed condor version.\n'
49			msg += '[Condor-G Scheduler]: Please use another machine with properly installed condor or change the Scheduler in your crab.cfg.'
50			common.logger.debug(2,msg)
51			raise CrabException(msg)
52
53			self.checkExecutableInPath('condor_config_val')
54			self.checkCondorVariablePointsToFile('GRIDMANAGER')
55			self.checkCondorVariablePointsToFile('GT2_GAHP',alternate_name='GAHP')
56			self.checkCondorVariablePointsToFile('GRID_MONITOR')
57			self.checkCondorVariableIsTrue('ENABLE_GRID_MONITOR')
58
59			max_submit = self.queryCondorVariable('GRIDMANAGER_MAX_SUBMITTED_JOBS_PER_RESOURCE',100).strip()
60			max_pending = self.queryCondorVariable('GRIDMANAGER_MAX_PENDING_SUBMITS_PER_RESOURCE','Unlimited').strip()
61
62			msg = '[Condor-G Scheduler]\n'
63			msg += 'Maximal number of jobs submitted to the grid : GRIDMANAGER_MAX_SUBMITTED_JOBS_PER_RESOURCE = '+max_submit+'\n'
64			msg += 'Maximal number of parallel submits to the grid : GRIDMANAGER_MAX_PENDING_SUBMITS_PER_RESOURCE = '+max_pending+'\n'
65			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'
66			common.logger.debug(2,msg)
67
68			# create hash
69			self.hash = makeCksum(common.work_space.cfgFileName())
70
71			return
72
73			def checkExecutableInPath(self, name):
74			# check if executable is in PATH
75			cmd = 'which '+name
76			cmd_out = runCommand(cmd)
77			if cmd_out == None:
78			msg = '[Condor-G Scheduler]: '+name+' is not in the $PATH on this machine.\n'
79			msg += '[Condor-G Scheduler]: Please use another machine with installed condor or change the Scheduler in your crab.cfg.'
80			common.logger.debug(2,msg)
81			raise CrabException(msg)
82
83			def checkCondorVariablePointsToFile(self, name, alternate_name=None):
84			## check for condor variable
85			cmd = 'condor_config_val '+name
86			cmd_out = runCommand(cmd)
87			if alternate_name and not cmd_out:
88			cmd = 'condor_config_val '+alternate_name
89			cmd_out = runCommand(cmd)
90			if cmd_out:
91			cmd_out = string.strip(cmd_out)
92			if not cmd_out or not os.path.isfile(cmd_out) :
93			msg = '[Condor-G Scheduler]: the variable '+name+' is not properly set for the condor installation on this machine.\n'
94			msg += '[Condor-G Scheduler]: Please ask the administrator of the local condor installation to set the variable '+name+' properly, ' + \
95			'use another machine with properly installed condor or change the Scheduler in your crab.cfg.'
96			common.logger.debug(2,msg)
97			raise CrabException(msg)
98
99			def checkCondorVariableIsTrue(self, name):
100			## check for condor variable
101			cmd = 'condor_config_val '+name
102			cmd_out = runCommand(cmd)
103			if cmd_out == 'TRUE' :
104			msg = '[Condor-G Scheduler]: the variable '+name+' is not set to true for the condor installation on this machine.\n'
105			msg += '[Condor-G Scheduler]: Please ask the administrator of the local condor installation to set the variable '+name+' to true,',
106			'use another machine with properly installed condor or change the Scheduler in your crab.cfg.'
107			common.logger.debug(2,msg)
108			raise CrabException(msg)
109
110			def queryCondorVariable(self, name, default):
111			## check for condor variable
112			cmd = 'condor_config_val '+name
113			out = popen2.Popen3(cmd,1)
114			exit_code = out.wait()
115			cmd_out = out.fromchild.readline().strip()
116			if exit_code != 0 :
117			cmd_out = str(default)
118
119			return cmd_out
120
121			def configure(self, cfg_params):
122			SchedulerGrid.configure(self,cfg_params)
123
124			# init BlackWhiteListParser
125	ewv	1.7	self.ceBlackWhiteListParser = CEBlackWhiteListParser(cfg_params)
126	ewv	1.1
127			try:
128			self.GLOBUS_RSL = cfg_params['CONDORG.globus_rsl']
129			except KeyError:
130			self.GLOBUS_RSL = ''
131
132			# Provide an override for the batchsystem that condor_g specifies as a grid resource.
133			# this is to handle the case where the site supports several batchsystem but bdii
134			# only allows a site to public one.
135			try:
136			self.batchsystem = cfg_params['CONDORG.batchsystem']
137			msg = '[Condor-G Scheduler]: batchsystem overide specified in your crab.cfg'
138			common.logger.debug(2,msg)
139			except KeyError:
140			self.batchsystem = ''
141
142			try:
143			self.UseGT4 = cfg_params['USER.use_gt_4'];
144			except KeyError:
145			self.UseGT4 = 0;
146
147			# added here because checklistmatch is not used
148			self.checkProxy()
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.14	def listMatch(self, seList, onlyOSG=True):
193	ewv	1.1	"""
194	ewv	1.14	Check the compatibility of available resources
195	ewv	1.1	"""
196	ewv	1.7	seDest = self.blackWhiteListParser.cleanForBlackWhiteList(seList,"list")
197			scram = Scram.Scram(None)
198
199			versionCMSSW = scram.getSWVersion()
200			arch = scram.getArch()
201	ewv	1.11	availCEs = getJobManagerList(seDest,versionCMSSW,arch,onlyOSG=onlyOSG)
202	ewv	1.7	uniqCEs = []
203			for ce in availCEs:
204			if ce not in uniqCEs:
205			uniqCEs.append(ce)
206
207			ceDest = self.ceBlackWhiteListParser.cleanForBlackWhiteList(uniqCEs,"list")
208	ewv	1.14	return ceDest
209
210			def userName(self):
211			""" return the user name """
212			self.checkProxy()
213			return runCommand("voms-proxy-info -identity")
214
215			def ce_list(self):
216			"""
217			Returns string with requirement CE related, dummy for now
218			"""
219			req = ''
220
221			return req,self.EDG_ce_white_list,self.EDG_ce_black_list
222
223			def seListToCElist(self, seList, onlyOSG=True):
224			ceDest = self.listMatch(seList, onlyOSG)
225	ewv	1.7
226			if (not ceDest):
227			msg = 'No OSG sites found hosting the data or all sites blocked by CE/SE white/blacklisting'
228			print msg
229			raise CrabException(msg)
230
231			return ceDest
232	ewv	1.15
233	spiga	1.16
234			def wsExitFunc(self):
235			"""
236			"""
237			txt = '\n'
238
239			txt += '#\n'
240			txt += '# EXECUTE THIS FUNCTION BEFORE EXIT \n'
241			txt += '#\n\n'
242
243			txt += 'func_exit() { \n'
244			txt += self.wsExitFunc_common()
245			txt += ' echo "JOB_EXIT_STATUS = $job_exit_code"\n'
246			txt += ' echo "JobExitCode=$job_exit_code" >> $RUNTIME_AREA/$repo\n'
247			txt += ' dumpStatus $RUNTIME_AREA/$repo\n'
248			txt += ' tar zcvf ${out_files}.tgz ${final_list}\n'
249			txt += ' exit $job_exit_code\n'
250			txt += '}\n'
251			return txt