CRAB/python/SchedulerCondor.py

"""
Implements the vanilla (local) Condor scheduler
"""

__revision__ = "$Id: SchedulerCondor.py,v 1.33 2012/06/14 15:47:28 belforte Exp $"
__version__ = "$Revision: 1.33 $"

from SchedulerLocal  import SchedulerLocal
from crab_exceptions import CrabException

import common
import os
import socket

# FUTURE: for python 2.4 & 2.6
try:
    from hashlib import sha1
except:
    from sha import sha as sha1

class SchedulerCondor(SchedulerLocal) :
    """
    Class to implement the vanilla (local) Condor scheduler
     Naming convention:  Methods starting with 'ws' provide
     the corresponding part of the job script
     ('ws' stands for 'write script').
    """

    def __init__(self):
        SchedulerLocal.__init__(self,"CONDOR")
        self.datasetPath   = None
        self.selectNoInput = None
        self.OSBsize = None

        self.environment_unique_identifier = None
        return


    def configure(self, cfg_params):
        """
        Configure the scheduler with the config settings from the user
        """

        SchedulerLocal.configure(self, cfg_params)

        self.proxyValid=0
        self.dontCheckProxy=int(cfg_params.get("GRID.dont_check_proxy",0))
        self.space_token = cfg_params.get("USER.space_token",None)
        try:
            self.proxyServer = Downloader("http://cmsdoc.cern.ch/cms/LCG/crab/config/").config("myproxy_server.conf")
            self.proxyServer = self.proxyServer.strip()
            if self.proxyServer is None:
                raise CrabException("myproxy_server.conf retrieved but empty")
        except Exception, e:
            common.logger.info("Problem setting myproxy server endpoint: using myproxy.cern.ch")
            common.logger.debug(e)
            self.proxyServer= 'myproxy.cern.ch'
        self.group = cfg_params.get("GRID.group", None)
        self.role = cfg_params.get("GRID.role", None)
        self.VO = cfg_params.get('GRID.virtual_organization','cms')

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

        self.checkProxy()

        return

    def envUniqueID(self):
        taskHash = sha1(common._db.queryTask('name')).hexdigest()
        id = "https://" + socket.gethostname() + '/' + taskHash + "/${NJob}"
        return id

    def sched_parameter(self, i, task):
        """
        Return scheduler-specific parameters
        """
        req = ''
        if self.EDG_addJdlParam:
            if self.EDG_addJdlParam[-1] == '':
                self.EDG_addJdlParam = self.EDG_addJdlParam[:-1]
            for p in self.EDG_addJdlParam:
                req += p.strip()+';\n'

        return req


    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')
        tmpDir = os.path.join(common.work_space.shareDir(),'.condor_temp')
        jobDir = common.work_space.jobDir()
        params = {'tmpDir':tmpDir,
                  'jobDir':jobDir}
        return params


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

        return [True]


    def decodeLogInfo(self, fileName):
        """
        Parse logging info file and return main info
        """

        import CondorGLoggingInfo
        loggingInfo = CondorGLoggingInfo.CondorGLoggingInfo()
        reason = loggingInfo.decodeReason(fileName)
        return reason


    def wsCopyOutput(self):
        """
        Write a CopyResults part of a job script, e.g.
        to copy produced output into a storage element.
        """
        txt = self.wsCopyOutput_comm()
        return txt


    def wsExitFunc(self):
        """
        Returns the part of the job script which runs prior to exit
        """

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

        txt += 'func_exit() { \n'
        txt += self.wsExitFunc_common()

        txt += '    cp  ${out_files}.tgz $_CONDOR_SCRATCH_DIR/\n'
        txt += '    cp  crab_fjr_$NJob.xml $_CONDOR_SCRATCH_DIR/\n'

        txt += '    exit $job_exit_code\n'
        txt += '}\n'

        return txt

    def wsInitialEnvironment(self):
        """
        Returns part of a job script which does scheduler-specific work.
        """

        txt  = '\n# Written by SchedulerCondor::wsInitialEnvironment\n'
        txt += 'echo "Beginning environment"\n'
        txt += 'printenv | sort\n'

        txt += 'middleware='+self.name()+' \n'
        txt += 'if [ -e /opt/d-cache/srm/bin ]; then\n'
        txt += '  export PATH=${PATH}:/opt/d-cache/srm/bin\n'
        txt += 'fi\n'

        txt += """
if [ $_CONDOR_SCRATCH_DIR ] && [ -d $_CONDOR_SCRATCH_DIR ]; then
    echo "cd to Condor scratch directory: $_CONDOR_SCRATCH_DIR"
    if [ -e ../default.tgz ] ;then
      echo "Found ISB in parent directory (Local Condor)"
      cp ../default.tgz $_CONDOR_SCRATCH_DIR
    fi
    cd $_CONDOR_SCRATCH_DIR
fi
"""

        return txt


    def sched_fix_parameter(self):
        """
        Returns string with requirements and scheduler-specific parameters
        """

        if self.EDG_requirements:
            req = self.EDG_requirements
            taskReq = {'commonRequirements':req}
            common._db.updateTask_(taskReq)
Revision:	1.34
Committed:	Thu Jan 17 14:50:43 2013 UTC (12 years, 3 months ago) by belforte
Content type:	text/x-python
Branch:	MAIN
CVS Tags:	CRAB_2_9_1, CRAB_2_9_1_pre2, CRAB_2_9_1_pre1, CRAB_2_9_0, CRAB_2_9_0_pre2, CRAB_2_9_0_pre1, CRAB_2_8_8, CRAB_2_8_8_pre1, CRAB_2_8_7_patch3, CRAB_2_8_7_patch2, CRAB_2_8_7_patch1, CRAB_2_8_7, CRAB_2_8_7_pre2, CRAB_2_8_7_pre1, CRAB_2_8_6, CRAB_2_8_6_pre1, CRAB_2_8_5_patch3, CRAB_2_8_5_patch2, CRAB_2_8_5_patch1, CRAB_2_8_5, CRAB_2_8_5_pre5, CRAB_2_8_5_pre4, CRAB_2_8_5_pre3, HEAD
Changes since 1.33:	+2 -3 lines
Log Message:	move OSB sixe check to Scheduler.py https://savannah.cern.ch/bugs/index.php?95466
#	User	Rev	Content
1	ewv	1.10	"""
2			Implements the vanilla (local) Condor scheduler
3			"""
4
5	belforte	1.34	__revision__ = "$Id: SchedulerCondor.py,v 1.33 2012/06/14 15:47:28 belforte Exp $"
6			__version__ = "$Revision: 1.33 $"
7	ewv	1.10
8			from SchedulerLocal import SchedulerLocal
9			from crab_exceptions import CrabException
10	ewv	1.1
11			import common
12			import os
13	ewv	1.17	import socket
14	ewv	1.30
15			# FUTURE: for python 2.4 & 2.6
16	ewv	1.29	try:
17			from hashlib import sha1
18			except:
19			from sha import sha as sha1
20	ewv	1.1
21			class SchedulerCondor(SchedulerLocal) :
22	ewv	1.10	"""
23			Class to implement the vanilla (local) Condor scheduler
24	ewv	1.14	Naming convention: Methods starting with 'ws' provide
25			the corresponding part of the job script
26			('ws' stands for 'write script').
27	ewv	1.10	"""
28
29			def __init__(self):
30			SchedulerLocal.__init__(self,"CONDOR")
31			self.datasetPath = None
32			self.selectNoInput = None
33	belforte	1.33	self.OSBsize = None
34	ewv	1.19
35	ewv	1.10	self.environment_unique_identifier = None
36			return
37
38
39			def configure(self, cfg_params):
40			"""
41			Configure the scheduler with the config settings from the user
42			"""
43
44			SchedulerLocal.configure(self, cfg_params)
45
46	ewv	1.32	self.proxyValid=0
47			self.dontCheckProxy=int(cfg_params.get("GRID.dont_check_proxy",0))
48			self.space_token = cfg_params.get("USER.space_token",None)
49			try:
50			self.proxyServer = Downloader("http://cmsdoc.cern.ch/cms/LCG/crab/config/").config("myproxy_server.conf")
51			self.proxyServer = self.proxyServer.strip()
52			if self.proxyServer is None:
53			raise CrabException("myproxy_server.conf retrieved but empty")
54			except Exception, e:
55			common.logger.info("Problem setting myproxy server endpoint: using myproxy.cern.ch")
56			common.logger.debug(e)
57			self.proxyServer= 'myproxy.cern.ch'
58			self.group = cfg_params.get("GRID.group", None)
59			self.role = cfg_params.get("GRID.role", None)
60			self.VO = cfg_params.get('GRID.virtual_organization','cms')
61
62	ewv	1.10	try:
63			tmp = cfg_params['CMSSW.datasetpath']
64			if tmp.lower() == 'none':
65			self.datasetPath = None
66			self.selectNoInput = 1
67			else:
68			self.datasetPath = tmp
69			self.selectNoInput = 0
70			except KeyError:
71			msg = "Error: datasetpath not defined "
72			raise CrabException(msg)
73
74	ewv	1.32	self.checkProxy()
75
76	ewv	1.10	return
77	ewv	1.1
78	ewv	1.23	def envUniqueID(self):
79	ewv	1.29	taskHash = sha1(common._db.queryTask('name')).hexdigest()
80	spiga	1.22	id = "https://" + socket.gethostname() + '/' + taskHash + "/${NJob}"
81			return id
82	ewv	1.7
83	ewv	1.10	def sched_parameter(self, i, task):
84			"""
85			Return scheduler-specific parameters
86			"""
87	ewv	1.27	req = ''
88			if self.EDG_addJdlParam:
89			if self.EDG_addJdlParam[-1] == '':
90			self.EDG_addJdlParam = self.EDG_addJdlParam[:-1]
91			for p in self.EDG_addJdlParam:
92			req += p.strip()+';\n'
93	ewv	1.1
94	ewv	1.27	return req
95	ewv	1.1
96
97	ewv	1.10	def realSchedParams(self, cfg_params):
98			"""
99			Return dictionary with specific parameters, to use with real scheduler
100			"""
101	ewv	1.1
102	ewv	1.10	tmpDir = os.path.join(common.work_space.shareDir(),'.condor_temp')
103	ewv	1.13	tmpDir = os.path.join(common.work_space.shareDir(),'.condor_temp')
104			jobDir = common.work_space.jobDir()
105			params = {'tmpDir':tmpDir,
106			'jobDir':jobDir}
107	ewv	1.10	return params
108	ewv	1.3
109	ewv	1.2
110	ewv	1.10	def listMatch(self, seList, full):
111			"""
112			Check the compatibility of available resources
113			"""
114	ewv	1.2
115	ewv	1.16	return [True]
116	ewv	1.3
117	ewv	1.7
118	ewv	1.10	def decodeLogInfo(self, fileName):
119			"""
120			Parse logging info file and return main info
121			"""
122	ewv	1.7
123	ewv	1.10	import CondorGLoggingInfo
124			loggingInfo = CondorGLoggingInfo.CondorGLoggingInfo()
125			reason = loggingInfo.decodeReason(fileName)
126			return reason
127	ewv	1.7
128	ewv	1.3
129	ewv	1.14	def wsCopyOutput(self):
130			"""
131			Write a CopyResults part of a job script, e.g.
132			to copy produced output into a storage element.
133			"""
134			txt = self.wsCopyOutput_comm()
135			return txt
136
137
138	ewv	1.10	def wsExitFunc(self):
139			"""
140			Returns the part of the job script which runs prior to exit
141			"""
142	ewv	1.3
143	ewv	1.10	txt = '\n'
144			txt += '#\n'
145			txt += '# EXECUTE THIS FUNCTION BEFORE EXIT \n'
146			txt += '#\n\n'
147	ewv	1.3
148	ewv	1.10	txt += 'func_exit() { \n'
149			txt += self.wsExitFunc_common()
150	spiga	1.8
151	ewv	1.12	txt += ' cp ${out_files}.tgz $_CONDOR_SCRATCH_DIR/\n'
152			txt += ' cp crab_fjr_$NJob.xml $_CONDOR_SCRATCH_DIR/\n'
153	ewv	1.3
154	ewv	1.10	txt += ' exit $job_exit_code\n'
155			txt += '}\n'
156	ewv	1.3
157	ewv	1.10	return txt
158	ewv	1.5
159	ewv	1.10	def wsInitialEnvironment(self):
160			"""
161			Returns part of a job script which does scheduler-specific work.
162			"""
163	ewv	1.5
164	ewv	1.10	txt = '\n# Written by SchedulerCondor::wsInitialEnvironment\n'
165			txt += 'echo "Beginning environment"\n'
166			txt += 'printenv \| sort\n'
167	ewv	1.5
168	ewv	1.10	txt += 'middleware='+self.name()+' \n'
169	ewv	1.14	txt += 'if [ -e /opt/d-cache/srm/bin ]; then\n'
170			txt += ' export PATH=${PATH}:/opt/d-cache/srm/bin\n'
171			txt += 'fi\n'
172
173	ewv	1.10	txt += """
174	ewv	1.6	if [ $_CONDOR_SCRATCH_DIR ] && [ -d $_CONDOR_SCRATCH_DIR ]; then
175	ewv	1.12	echo "cd to Condor scratch directory: $_CONDOR_SCRATCH_DIR"
176	ewv	1.6	if [ -e ../default.tgz ] ;then
177			echo "Found ISB in parent directory (Local Condor)"
178			cp ../default.tgz $_CONDOR_SCRATCH_DIR
179			fi
180			cd $_CONDOR_SCRATCH_DIR
181			fi
182			"""
183	ewv	1.5
184	ewv	1.10	return txt
185	ewv	1.27
186
187			def sched_fix_parameter(self):
188			"""
189			Returns string with requirements and scheduler-specific parameters
190			"""
191
192			if self.EDG_requirements:
193			req = self.EDG_requirements
194			taskReq = {'commonRequirements':req}
195			common._db.updateTask_(taskReq)