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 *
import time
import common
import popen2
import os
from BlackWhiteListParser import BlackWhiteListParser
import CondorGLoggingInfo

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

import pdb # FIXME: Use while debugging

__revision__ = "$Id: SchedulerCondorCommon.py,v 1.5 2008/04/22 20:26:45 ewv Exp $"
__version__ = "$Revision: 1.5 $"

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.blackWhiteListParser = BlackWhiteListParser(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 sched_parameter(self,i,task):
      """
      Returns scheduler-specific parameters
      """
      jobParams = ''
      globusRSL = self.GLOBUS_RSL
      if (self.EDG_clock_time):
        globusRSL += '(maxWalltime='+self.EDG_clock_time+')'
      if (globusRSL != ''):
        jobParams +=  'globusrsl = ' + globusRSL + '; '

      return jobParams

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

    def wsSetupEnvironment(self):
        """
        Returns part of a job script which does scheduler-specific work.
        """
        txt = SchedulerGrid.wsSetupEnvironment(self)

        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 wsSetupEnvironment(self):
      #"""
      #Returns part of a job script which does scheduler-specific work.
      #"""
      #txt = SchedulerGrid.wsSetupEnvironment(self)

      #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 + '/'
          #txt += 'export SE_PATH='+self.SE_PATH+'\n'
          #txt += 'echo "SE_PATH = $SE_PATH"\n'

      #txt += 'export VO='+self.VO+'\n'
      #txt += 'CE=${args[3]}\n'
      #txt += 'echo "CE = $CE"\n'
      #return txt

    #def wsCopyInput(self):
        #"""
        #Copy input data from SE to WN
        #"""
        #txt = '\n'
        #return txt

    def listMatch(self, nj):
        """
        Check the compatibility of available resources
        """
        #self.checkProxy()
        return [""]

    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
Revision:	1.6
Committed:	Wed Apr 23 18:20:01 2008 UTC (17 years ago) by ewv
Content type:	text/x-python
Branch:	MAIN
Changes since 1.5:	+49 -25 lines
Log Message:	Fix iterator bug in osg_bdii, remove Condor specific WN script
#	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			from osg_bdii import *
7			import time
8			import common
9			import popen2
10			import os
11			from BlackWhiteListParser import BlackWhiteListParser
12			import CondorGLoggingInfo
13
14			# This class was originally SchedulerCondor_g. For a history of this code, see that file.
15
16	ewv	1.5	import pdb # FIXME: Use while debugging
17	ewv	1.1
18	ewv	1.6	__revision__ = "$Id: SchedulerCondorCommon.py,v 1.5 2008/04/22 20:26:45 ewv Exp $"
19			__version__ = "$Revision: 1.5 $"
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			#self.blackWhiteListParser = BlackWhiteListParser(cfg_params)
126
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			def sched_parameter(self,i,task):
167			"""
168			Returns scheduler-specific parameters
169			"""
170			jobParams = ''
171			globusRSL = self.GLOBUS_RSL
172			if (self.EDG_clock_time):
173			globusRSL += '(maxWalltime='+self.EDG_clock_time+')'
174			if (globusRSL != ''):
175			jobParams += 'globusrsl = ' + globusRSL + '; '
176
177			return jobParams
178
179			def decodeLogInfo(self, file):
180			"""
181			Parse logging info file and return main info
182			"""
183			loggingInfo = CondorGLoggingInfo.CondorGLoggingInfo()
184			reason = loggingInfo.decodeReason(file)
185			return reason
186
187			def wsSetupEnvironment(self):
188	ewv	1.6	"""
189			Returns part of a job script which does scheduler-specific work.
190			"""
191			txt = SchedulerGrid.wsSetupEnvironment(self)
192	ewv	1.1
193	ewv	1.6	txt += 'export VO='+self.VO+'\n'
194			txt += 'if [ $middleware == LCG ]; then\n'
195			txt += ' CloseCEs=`glite-brokerinfo getCE`\n'
196			txt += ' echo "CloseCEs = $CloseCEs"\n'
197			txt += ' CE=`echo $CloseCEs \| sed -e "s/:.*//"`\n'
198			txt += ' echo "CE = $CE"\n'
199			txt += 'elif [ $middleware == OSG ]; then \n'
200			txt += ' if [ $OSG_JOB_CONTACT ]; then \n'
201			txt += ' CE=`echo $OSG_JOB_CONTACT \| /usr/bin/awk -F\/ \'{print $1}\'` \n'
202			txt += ' else \n'
203			txt += ' echo "ERROR ==> OSG mode in setting CE name from OSG_JOB_CONTACT" \n'
204			txt += ' job_exit_code=10099\n'
205			txt += ' func_exit\n'
206			txt += ' fi \n'
207			txt += 'fi \n'
208	ewv	1.1
209			return txt
210
211	ewv	1.6	#def wsSetupEnvironment(self):
212			#"""
213			#Returns part of a job script which does scheduler-specific work.
214			#"""
215			#txt = SchedulerGrid.wsSetupEnvironment(self)
216
217			#if int(self.copy_data) == 1:
218			#if self.SE:
219			#txt += 'export SE='+self.SE+'\n'
220			#txt += 'echo "SE = $SE"\n'
221			#if self.SE_PATH:
222			#if self.SE_PATH[-1] != '/':
223			#self.SE_PATH = self.SE_PATH + '/'
224			#txt += 'export SE_PATH='+self.SE_PATH+'\n'
225			#txt += 'echo "SE_PATH = $SE_PATH"\n'
226
227			#txt += 'export VO='+self.VO+'\n'
228			#txt += 'CE=${args[3]}\n'
229			#txt += 'echo "CE = $CE"\n'
230			#return txt
231
232			#def wsCopyInput(self):
233			#"""
234			#Copy input data from SE to WN
235			#"""
236			#txt = '\n'
237			#return txt
238
239	ewv	1.1	def listMatch(self, nj):
240			"""
241			Check the compatibility of available resources
242			"""
243			#self.checkProxy()
244	slacapra	1.2	return [""]
245	ewv	1.1
246			def userName(self):
247			""" return the user name """
248			self.checkProxy()
249			return runCommand("voms-proxy-info -identity")
250
251			def ce_list(self):
252			"""
253			Returns string with requirement CE related, dummy for now
254			"""
255			req = ''
256
257			return req,self.EDG_ce_white_list,self.EDG_ce_black_list