CRAB/python/SchedulerCondor.py

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

__revision__ = "$Id: SchedulerCondor.py,v 1.31 2010/05/11 13:39:50 farinafa Exp $"
__version__ = "$Revision: 1.31 $"

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.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 += '    tar zcvf ${out_files}.tgz  ${final_list}\n'
        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.32
Committed:	Mon Jun 28 18:50:38 2010 UTC (14 years, 10 months ago) by ewv
Content type:	text/x-python
Branch:	MAIN
CVS Tags:	CRAB_2_8_2_pre2, CRAB_2_8_2_pre1, CRAB_2_8_1, CRAB_2_8_0, CRAB_2_8_0_pre1, CRAB_2_7_10_pre3, CRAB_2_7_9_patch2_pre1, CRAB_2_7_10_pre2, CRAB_2_7_10_pre1, CRAB_2_7_9_patch1, CRAB_2_7_9, CRAB_2_7_9_pre5, CRAB_2_7_9_pre4, CRAB_2_7_9_pre3, CRAB_2_7_9_pre2, CRAB_2_7_8_patch2, CRAB_2_7_9_pre1, CRAB_2_7_8_patch2_pre1, CRAB_2_7_8_patch1, CRAB_2_7_8_patch1_pre1, CRAB_2_7_8, CRAB_2_7_8_pre3, CRAB_2_7_8_pre2, CRAB_2_7_8_dash3, CRAB_2_7_8_dash2, CRAB_2_7_8_dash, CRAB_2_7_7_patch1, CRAB_2_7_7_patch1_pre1, CRAB_2_7_8_pre1, CRAB_2_7_7, CRAB_2_7_7_pre2, CRAB_2_7_7_pre1, CRAB_2_7_6_patch1, CRAB_2_7_6, CRAB_2_7_6_pre1, CRAB_2_7_5_patch1, CRAB_2_7_5, CRAB_2_7_5_pre3, CRAB_2_7_5_pre2, CRAB_2_7_5_pre1, CRAB_2_7_4_patch1, CRAB_2_7_4, CRAB_2_7_4_pre6, CRAB_2_7_4_pre5, CRAB_2_7_4_pre4, CRAB_2_7_4_pre3, CRAB_2_7_4_pre2, CRAB_2_7_4_pre1
Changes since 1.31:	+20 -2 lines
Log Message:	Add proxy stuff to Condor
#	User	Rev	Content
1	ewv	1.10	"""
2			Implements the vanilla (local) Condor scheduler
3			"""
4
5	ewv	1.32	__revision__ = "$Id: SchedulerCondor.py,v 1.31 2010/05/11 13:39:50 farinafa Exp $"
6			__version__ = "$Revision: 1.31 $"
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	ewv	1.19
34	ewv	1.10	self.environment_unique_identifier = None
35			return
36
37
38			def configure(self, cfg_params):
39			"""
40			Configure the scheduler with the config settings from the user
41			"""
42
43			SchedulerLocal.configure(self, cfg_params)
44
45	ewv	1.32	self.proxyValid=0
46			self.dontCheckProxy=int(cfg_params.get("GRID.dont_check_proxy",0))
47			self.space_token = cfg_params.get("USER.space_token",None)
48			try:
49			self.proxyServer = Downloader("http://cmsdoc.cern.ch/cms/LCG/crab/config/").config("myproxy_server.conf")
50			self.proxyServer = self.proxyServer.strip()
51			if self.proxyServer is None:
52			raise CrabException("myproxy_server.conf retrieved but empty")
53			except Exception, e:
54			common.logger.info("Problem setting myproxy server endpoint: using myproxy.cern.ch")
55			common.logger.debug(e)
56			self.proxyServer= 'myproxy.cern.ch'
57			self.group = cfg_params.get("GRID.group", None)
58			self.role = cfg_params.get("GRID.role", None)
59			self.VO = cfg_params.get('GRID.virtual_organization','cms')
60
61	ewv	1.10	try:
62			tmp = cfg_params['CMSSW.datasetpath']
63			if tmp.lower() == 'none':
64			self.datasetPath = None
65			self.selectNoInput = 1
66			else:
67			self.datasetPath = tmp
68			self.selectNoInput = 0
69			except KeyError:
70			msg = "Error: datasetpath not defined "
71			raise CrabException(msg)
72
73	ewv	1.32	self.checkProxy()
74
75	ewv	1.10	return
76	ewv	1.1
77	ewv	1.23	def envUniqueID(self):
78	ewv	1.29	taskHash = sha1(common._db.queryTask('name')).hexdigest()
79	spiga	1.22	id = "https://" + socket.gethostname() + '/' + taskHash + "/${NJob}"
80			return id
81	ewv	1.7
82	ewv	1.10	def sched_parameter(self, i, task):
83			"""
84			Return scheduler-specific parameters
85			"""
86	ewv	1.27	req = ''
87			if self.EDG_addJdlParam:
88			if self.EDG_addJdlParam[-1] == '':
89			self.EDG_addJdlParam = self.EDG_addJdlParam[:-1]
90			for p in self.EDG_addJdlParam:
91			req += p.strip()+';\n'
92	ewv	1.1
93	ewv	1.27	return req
94	ewv	1.1
95
96	ewv	1.10	def realSchedParams(self, cfg_params):
97			"""
98			Return dictionary with specific parameters, to use with real scheduler
99			"""
100	ewv	1.1
101	ewv	1.10	tmpDir = os.path.join(common.work_space.shareDir(),'.condor_temp')
102	ewv	1.13	tmpDir = os.path.join(common.work_space.shareDir(),'.condor_temp')
103			jobDir = common.work_space.jobDir()
104			params = {'tmpDir':tmpDir,
105			'jobDir':jobDir}
106	ewv	1.10	return params
107	ewv	1.3
108	ewv	1.2
109	ewv	1.10	def listMatch(self, seList, full):
110			"""
111			Check the compatibility of available resources
112			"""
113	ewv	1.2
114	ewv	1.16	return [True]
115	ewv	1.3
116	ewv	1.7
117	ewv	1.10	def decodeLogInfo(self, fileName):
118			"""
119			Parse logging info file and return main info
120			"""
121	ewv	1.7
122	ewv	1.10	import CondorGLoggingInfo
123			loggingInfo = CondorGLoggingInfo.CondorGLoggingInfo()
124			reason = loggingInfo.decodeReason(fileName)
125			return reason
126	ewv	1.7
127	ewv	1.3
128	ewv	1.14	def wsCopyOutput(self):
129			"""
130			Write a CopyResults part of a job script, e.g.
131			to copy produced output into a storage element.
132			"""
133			txt = self.wsCopyOutput_comm()
134			return txt
135
136
137	ewv	1.10	def wsExitFunc(self):
138			"""
139			Returns the part of the job script which runs prior to exit
140			"""
141	ewv	1.3
142	ewv	1.10	txt = '\n'
143			txt += '#\n'
144			txt += '# EXECUTE THIS FUNCTION BEFORE EXIT \n'
145			txt += '#\n\n'
146	ewv	1.3
147	ewv	1.10	txt += 'func_exit() { \n'
148			txt += self.wsExitFunc_common()
149	spiga	1.8
150	ewv	1.10	txt += ' tar zcvf ${out_files}.tgz ${final_list}\n'
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)