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.17 2008/05/05 18:46:02 ewv Exp $"
__version__ = "$Revision: 1.17 $"

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, 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 """
      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.18
Committed:	Thu May 29 22:06:25 2008 UTC (16 years, 11 months ago) by spiga
Content type:	text/x-python
Branch:	MAIN
CVS Tags:	CRAB_2_2_2_pre4, CRAB_2_2_2_pre3, CRAB_2_2_2_pre2, CRAB_2_2_2_pre1, CRAB_2_2_1, CRAB_2_2_1_pre6, CRAB_2_2_1_pre5, CRAB_2_2_1_pre4
Changes since 1.17:	+2 -4 lines
Log Message:	others adjustments on check proxy. Now shoud be ok.
#	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	spiga	1.18	__revision__ = "$Id: SchedulerCondorCommon.py,v 1.17 2008/05/05 18:46:02 ewv Exp $"
19			__version__ = "$Revision: 1.17 $"
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.environment_unique_identifier = 'GLOBUS_GRAM_JOB_CONTACT'
149	ewv	1.4	self.datasetPath = ''
150	ewv	1.1
151			try:
152			tmp = cfg_params['CMSSW.datasetpath']
153			if string.lower(tmp)=='none':
154			self.datasetPath = None
155			self.selectNoInput = 1
156			else:
157			self.datasetPath = tmp
158			self.selectNoInput = 0
159			except KeyError:
160			msg = "Error: datasetpath not defined "
161			raise CrabException(msg)
162
163			return
164
165	ewv	1.17	def realSchedParams(self,cfg_params):
166			"""
167			Return dictionary with specific parameters, to use
168			with real scheduler
169			"""
170
171			tmpDir = os.path.join(common.work_space.shareDir(),'.condor_temp')
172			params = {'tmpDir':tmpDir}
173			return params
174
175	ewv	1.1	def sched_parameter(self,i,task):
176			"""
177			Returns scheduler-specific parameters
178			"""
179			jobParams = ''
180
181			return jobParams
182
183			def decodeLogInfo(self, file):
184			"""
185			Parse logging info file and return main info
186			"""
187			loggingInfo = CondorGLoggingInfo.CondorGLoggingInfo()
188			reason = loggingInfo.decodeReason(file)
189			return reason
190
191	ewv	1.14	def listMatch(self, seList, onlyOSG=True):
192	ewv	1.1	"""
193	ewv	1.14	Check the compatibility of available resources
194	ewv	1.1	"""
195	ewv	1.7	seDest = self.blackWhiteListParser.cleanForBlackWhiteList(seList,"list")
196			scram = Scram.Scram(None)
197
198			versionCMSSW = scram.getSWVersion()
199			arch = scram.getArch()
200	ewv	1.11	availCEs = getJobManagerList(seDest,versionCMSSW,arch,onlyOSG=onlyOSG)
201	ewv	1.7	uniqCEs = []
202			for ce in availCEs:
203			if ce not in uniqCEs:
204			uniqCEs.append(ce)
205
206			ceDest = self.ceBlackWhiteListParser.cleanForBlackWhiteList(uniqCEs,"list")
207	ewv	1.14	return ceDest
208
209			def userName(self):
210			""" return the user name """
211			return runCommand("voms-proxy-info -identity")
212
213			def ce_list(self):
214			"""
215			Returns string with requirement CE related, dummy for now
216			"""
217			req = ''
218
219			return req,self.EDG_ce_white_list,self.EDG_ce_black_list
220
221			def seListToCElist(self, seList, onlyOSG=True):
222			ceDest = self.listMatch(seList, onlyOSG)
223	ewv	1.7
224			if (not ceDest):
225			msg = 'No OSG sites found hosting the data or all sites blocked by CE/SE white/blacklisting'
226			print msg
227			raise CrabException(msg)
228
229			return ceDest
230	ewv	1.15
231	spiga	1.16
232			def wsExitFunc(self):
233			"""
234			"""
235			txt = '\n'
236
237			txt += '#\n'
238			txt += '# EXECUTE THIS FUNCTION BEFORE EXIT \n'
239			txt += '#\n\n'
240
241			txt += 'func_exit() { \n'
242			txt += self.wsExitFunc_common()
243			txt += ' echo "JOB_EXIT_STATUS = $job_exit_code"\n'
244			txt += ' echo "JobExitCode=$job_exit_code" >> $RUNTIME_AREA/$repo\n'
245			txt += ' dumpStatus $RUNTIME_AREA/$repo\n'
246			txt += ' tar zcvf ${out_files}.tgz ${final_list}\n'
247			txt += ' exit $job_exit_code\n'
248			txt += '}\n'
249			return txt