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.

import pdb # FIXME: Use while debugging

__revision__ = "$Id: SchedulerCondorCommon.py,v 1.14 2008/05/02 19:33:17 ewv Exp $"
__version__ = "$Revision: 1.14 $"

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 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

Revision:	1.15
Committed:	Fri May 2 19:55:39 2008 UTC (17 years ago) by ewv
Content type:	text/x-python
Branch:	MAIN
Changes since 1.14:	+3 -4 lines
Log Message:	Patch from Burt, restore onlyOSG function
#	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.5	import pdb # FIXME: Use while debugging
19	ewv	1.1
20	ewv	1.15	__revision__ = "$Id: SchedulerCondorCommon.py,v 1.14 2008/05/02 19:33:17 ewv Exp $"
21			__version__ = "$Revision: 1.14 $"
22	ewv	1.1
23			class SchedulerCondorCommon(SchedulerGrid):
24			def __init__(self,name):
25			SchedulerGrid.__init__(self,name)
26
27			# check for locally running condor scheduler
28			cmd = 'ps xau \| grep -i condor_schedd \| grep -v grep'
29			cmd_out = runCommand(cmd)
30			if cmd_out == None:
31			msg = '[Condor-G Scheduler]: condor_schedd is not running on this machine.\n'
32			msg += '[Condor-G Scheduler]: Please use another machine with installed condor and running condor_schedd or change the Scheduler in your crab.cfg.'
33			common.logger.debug(2,msg)
34			raise CrabException(msg)
35
36			self.checkExecutableInPath('condor_q')
37			self.checkExecutableInPath('condor_submit')
38			self.checkExecutableInPath('condor_version')
39
40			# get version number
41			cmd = 'condor_version'
42			cmd_out = runCommand(cmd)
43			if cmd != None :
44			tmp = cmd_out.find('CondorVersion') + 15
45			version = cmd_out[tmp:tmp+6].split('.')
46			version_master = int(version[0])
47			version_major = int(version[1])
48			version_minor = int(version[2])
49			else :
50			msg = '[Condor-G Scheduler]: condor_version was not able to determine the installed condor version.\n'
51			msg += '[Condor-G Scheduler]: Please use another machine with properly installed condor or change the Scheduler in your crab.cfg.'
52			common.logger.debug(2,msg)
53			raise CrabException(msg)
54
55			self.checkExecutableInPath('condor_config_val')
56			self.checkCondorVariablePointsToFile('GRIDMANAGER')
57			self.checkCondorVariablePointsToFile('GT2_GAHP',alternate_name='GAHP')
58			self.checkCondorVariablePointsToFile('GRID_MONITOR')
59			self.checkCondorVariableIsTrue('ENABLE_GRID_MONITOR')
60
61			max_submit = self.queryCondorVariable('GRIDMANAGER_MAX_SUBMITTED_JOBS_PER_RESOURCE',100).strip()
62			max_pending = self.queryCondorVariable('GRIDMANAGER_MAX_PENDING_SUBMITS_PER_RESOURCE','Unlimited').strip()
63
64			msg = '[Condor-G Scheduler]\n'
65			msg += 'Maximal number of jobs submitted to the grid : GRIDMANAGER_MAX_SUBMITTED_JOBS_PER_RESOURCE = '+max_submit+'\n'
66			msg += 'Maximal number of parallel submits to the grid : GRIDMANAGER_MAX_PENDING_SUBMITS_PER_RESOURCE = '+max_pending+'\n'
67			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'
68			common.logger.debug(2,msg)
69
70			# create hash
71			self.hash = makeCksum(common.work_space.cfgFileName())
72
73			return
74
75			def checkExecutableInPath(self, name):
76			# check if executable is in PATH
77			cmd = 'which '+name
78			cmd_out = runCommand(cmd)
79			if cmd_out == None:
80			msg = '[Condor-G Scheduler]: '+name+' is not in the $PATH on this machine.\n'
81			msg += '[Condor-G Scheduler]: Please use another machine with installed condor or change the Scheduler in your crab.cfg.'
82			common.logger.debug(2,msg)
83			raise CrabException(msg)
84
85			def checkCondorVariablePointsToFile(self, name, alternate_name=None):
86			## check for condor variable
87			cmd = 'condor_config_val '+name
88			cmd_out = runCommand(cmd)
89			if alternate_name and not cmd_out:
90			cmd = 'condor_config_val '+alternate_name
91			cmd_out = runCommand(cmd)
92			if cmd_out:
93			cmd_out = string.strip(cmd_out)
94			if not cmd_out or not os.path.isfile(cmd_out) :
95			msg = '[Condor-G Scheduler]: the variable '+name+' is not properly set for the condor installation on this machine.\n'
96			msg += '[Condor-G Scheduler]: Please ask the administrator of the local condor installation to set the variable '+name+' properly, ' + \
97			'use another machine with properly installed condor or change the Scheduler in your crab.cfg.'
98			common.logger.debug(2,msg)
99			raise CrabException(msg)
100
101			def checkCondorVariableIsTrue(self, name):
102			## check for condor variable
103			cmd = 'condor_config_val '+name
104			cmd_out = runCommand(cmd)
105			if cmd_out == 'TRUE' :
106			msg = '[Condor-G Scheduler]: the variable '+name+' is not set to true for the condor installation on this machine.\n'
107			msg += '[Condor-G Scheduler]: Please ask the administrator of the local condor installation to set the variable '+name+' to true,',
108			'use another machine with properly installed condor or change the Scheduler in your crab.cfg.'
109			common.logger.debug(2,msg)
110			raise CrabException(msg)
111
112			def queryCondorVariable(self, name, default):
113			## check for condor variable
114			cmd = 'condor_config_val '+name
115			out = popen2.Popen3(cmd,1)
116			exit_code = out.wait()
117			cmd_out = out.fromchild.readline().strip()
118			if exit_code != 0 :
119			cmd_out = str(default)
120
121			return cmd_out
122
123			def configure(self, cfg_params):
124			SchedulerGrid.configure(self,cfg_params)
125
126			# init BlackWhiteListParser
127	ewv	1.7	self.ceBlackWhiteListParser = CEBlackWhiteListParser(cfg_params)
128	ewv	1.1
129			try:
130			self.GLOBUS_RSL = cfg_params['CONDORG.globus_rsl']
131			except KeyError:
132			self.GLOBUS_RSL = ''
133
134			# Provide an override for the batchsystem that condor_g specifies as a grid resource.
135			# this is to handle the case where the site supports several batchsystem but bdii
136			# only allows a site to public one.
137			try:
138			self.batchsystem = cfg_params['CONDORG.batchsystem']
139			msg = '[Condor-G Scheduler]: batchsystem overide specified in your crab.cfg'
140			common.logger.debug(2,msg)
141			except KeyError:
142			self.batchsystem = ''
143
144			try:
145			self.UseGT4 = cfg_params['USER.use_gt_4'];
146			except KeyError:
147			self.UseGT4 = 0;
148
149			# added here because checklistmatch is not used
150			self.checkProxy()
151			self.environment_unique_identifier = 'GLOBUS_GRAM_JOB_CONTACT'
152	ewv	1.4	self.datasetPath = ''
153	ewv	1.1
154			try:
155			tmp = cfg_params['CMSSW.datasetpath']
156			if string.lower(tmp)=='none':
157			self.datasetPath = None
158			self.selectNoInput = 1
159			else:
160			self.datasetPath = tmp
161			self.selectNoInput = 0
162			except KeyError:
163			msg = "Error: datasetpath not defined "
164			raise CrabException(msg)
165
166			return
167
168			def sched_parameter(self,i,task):
169			"""
170			Returns scheduler-specific parameters
171			"""
172			jobParams = ''
173
174			return jobParams
175
176			def decodeLogInfo(self, file):
177			"""
178			Parse logging info file and return main info
179			"""
180			loggingInfo = CondorGLoggingInfo.CondorGLoggingInfo()
181			reason = loggingInfo.decodeReason(file)
182			return reason
183
184	ewv	1.14	def listMatch(self, seList, onlyOSG=True):
185	ewv	1.1	"""
186	ewv	1.14	Check the compatibility of available resources
187	ewv	1.1	"""
188	ewv	1.7	seDest = self.blackWhiteListParser.cleanForBlackWhiteList(seList,"list")
189			scram = Scram.Scram(None)
190
191			versionCMSSW = scram.getSWVersion()
192			arch = scram.getArch()
193	ewv	1.11	availCEs = getJobManagerList(seDest,versionCMSSW,arch,onlyOSG=onlyOSG)
194	ewv	1.7	uniqCEs = []
195			for ce in availCEs:
196			if ce not in uniqCEs:
197			uniqCEs.append(ce)
198
199			ceDest = self.ceBlackWhiteListParser.cleanForBlackWhiteList(uniqCEs,"list")
200	ewv	1.14	return ceDest
201
202			def userName(self):
203			""" return the user name """
204			self.checkProxy()
205			return runCommand("voms-proxy-info -identity")
206
207			def ce_list(self):
208			"""
209			Returns string with requirement CE related, dummy for now
210			"""
211			req = ''
212
213			return req,self.EDG_ce_white_list,self.EDG_ce_black_list
214
215			def seListToCElist(self, seList, onlyOSG=True):
216			ceDest = self.listMatch(seList, onlyOSG)
217	ewv	1.7
218			if (not ceDest):
219			msg = 'No OSG sites found hosting the data or all sites blocked by CE/SE white/blacklisting'
220			print msg
221			raise CrabException(msg)
222
223			return ceDest
224	ewv	1.15