CRAB/python/SchedulerCondor.py

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

__revision__ = "$Id: SchedulerCondor.py,v 1.28 2009/12/12 16:35:39 spiga Exp $"
__version__ = "$Revision: 1.28 $"

from SchedulerLocal  import SchedulerLocal
from crab_exceptions import CrabException

import common
import os
import socket
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.return_data   = 0
        self.copy_data     = 0

        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)

        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.return_data = cfg_params.get('USER.return_data', 0)
        self.copy_data   = cfg_params.get("USER.copy_data", 0)

        self.proxyValid = 0
        self.dontCheckProxy = int(cfg_params.get("GRID.dont_check_proxy", 0))
        self.proxyServer = cfg_params.get("GRID.proxy_server", 'myproxy.cern.ch')
        common.logger.debug('Setting myproxy server to ' + self.proxyServer)

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

        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.29
Committed:	Wed Dec 16 17:37:28 2009 UTC (15 years, 4 months ago) by ewv
Content type:	text/x-python
Branch:	MAIN
Changes since 1.28:	+7 -4 lines
Log Message:	More transparent sha -> hashlib transition
#	User	Rev	Content
1	ewv	1.10	"""
2			Implements the vanilla (local) Condor scheduler
3			"""
4
5	ewv	1.29	__revision__ = "$Id: SchedulerCondor.py,v 1.28 2009/12/12 16:35:39 spiga Exp $"
6			__version__ = "$Revision: 1.28 $"
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.29	try:
15			from hashlib import sha1
16			except:
17			from sha import sha as sha1
18	ewv	1.1
19			class SchedulerCondor(SchedulerLocal) :
20	ewv	1.10	"""
21			Class to implement the vanilla (local) Condor scheduler
22	ewv	1.14	Naming convention: Methods starting with 'ws' provide
23			the corresponding part of the job script
24			('ws' stands for 'write script').
25	ewv	1.10	"""
26
27			def __init__(self):
28			SchedulerLocal.__init__(self,"CONDOR")
29			self.datasetPath = None
30			self.selectNoInput = None
31	ewv	1.19	self.return_data = 0
32			self.copy_data = 0
33
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			try:
46			tmp = cfg_params['CMSSW.datasetpath']
47			if tmp.lower() == 'none':
48			self.datasetPath = None
49			self.selectNoInput = 1
50			else:
51			self.datasetPath = tmp
52			self.selectNoInput = 0
53			except KeyError:
54			msg = "Error: datasetpath not defined "
55			raise CrabException(msg)
56
57	ewv	1.15	self.return_data = cfg_params.get('USER.return_data', 0)
58			self.copy_data = cfg_params.get("USER.copy_data", 0)
59
60	ewv	1.24	self.proxyValid = 0
61	spiga	1.26	self.dontCheckProxy = int(cfg_params.get("GRID.dont_check_proxy", 0))
62			self.proxyServer = cfg_params.get("GRID.proxy_server", 'myproxy.cern.ch')
63	spiga	1.25	common.logger.debug('Setting myproxy server to ' + self.proxyServer)
64	ewv	1.24
65	spiga	1.26	self.group = cfg_params.get("GRID.group", None)
66			self.role = cfg_params.get("GRID.role", None)
67			self.VO = cfg_params.get('GRID.virtual_organization', 'cms')
68	ewv	1.15
69	ewv	1.24	self.checkProxy()
70	ewv	1.15
71	ewv	1.10	return
72	ewv	1.1
73	ewv	1.23	def envUniqueID(self):
74	ewv	1.29	taskHash = sha1(common._db.queryTask('name')).hexdigest()
75	spiga	1.22	id = "https://" + socket.gethostname() + '/' + taskHash + "/${NJob}"
76			return id
77	ewv	1.7
78	ewv	1.10	def sched_parameter(self, i, task):
79			"""
80			Return scheduler-specific parameters
81			"""
82	ewv	1.27	req = ''
83			if self.EDG_addJdlParam:
84			if self.EDG_addJdlParam[-1] == '':
85			self.EDG_addJdlParam = self.EDG_addJdlParam[:-1]
86			for p in self.EDG_addJdlParam:
87			req += p.strip()+';\n'
88	ewv	1.1
89	ewv	1.27	return req
90	ewv	1.1
91
92	ewv	1.10	def realSchedParams(self, cfg_params):
93			"""
94			Return dictionary with specific parameters, to use with real scheduler
95			"""
96	ewv	1.1
97	ewv	1.10	tmpDir = os.path.join(common.work_space.shareDir(),'.condor_temp')
98	ewv	1.13	tmpDir = os.path.join(common.work_space.shareDir(),'.condor_temp')
99			jobDir = common.work_space.jobDir()
100			params = {'tmpDir':tmpDir,
101			'jobDir':jobDir}
102	ewv	1.10	return params
103	ewv	1.3
104	ewv	1.2
105	ewv	1.10	def listMatch(self, seList, full):
106			"""
107			Check the compatibility of available resources
108			"""
109	ewv	1.2
110	ewv	1.16	return [True]
111	ewv	1.3
112	ewv	1.7
113	ewv	1.10	def decodeLogInfo(self, fileName):
114			"""
115			Parse logging info file and return main info
116			"""
117	ewv	1.7
118	ewv	1.10	import CondorGLoggingInfo
119			loggingInfo = CondorGLoggingInfo.CondorGLoggingInfo()
120			reason = loggingInfo.decodeReason(fileName)
121			return reason
122	ewv	1.7
123	ewv	1.3
124	ewv	1.14	def wsCopyOutput(self):
125			"""
126			Write a CopyResults part of a job script, e.g.
127			to copy produced output into a storage element.
128			"""
129			txt = self.wsCopyOutput_comm()
130			return txt
131
132
133	ewv	1.10	def wsExitFunc(self):
134			"""
135			Returns the part of the job script which runs prior to exit
136			"""
137	ewv	1.3
138	ewv	1.10	txt = '\n'
139			txt += '#\n'
140			txt += '# EXECUTE THIS FUNCTION BEFORE EXIT \n'
141			txt += '#\n\n'
142	ewv	1.3
143	ewv	1.10	txt += 'func_exit() { \n'
144			txt += self.wsExitFunc_common()
145	spiga	1.8
146	ewv	1.10	txt += ' tar zcvf ${out_files}.tgz ${final_list}\n'
147	ewv	1.12	txt += ' cp ${out_files}.tgz $_CONDOR_SCRATCH_DIR/\n'
148			txt += ' cp crab_fjr_$NJob.xml $_CONDOR_SCRATCH_DIR/\n'
149	ewv	1.3
150	ewv	1.10	txt += ' exit $job_exit_code\n'
151			txt += '}\n'
152	ewv	1.3
153	ewv	1.10	return txt
154	ewv	1.5
155	ewv	1.10	def wsInitialEnvironment(self):
156			"""
157			Returns part of a job script which does scheduler-specific work.
158			"""
159	ewv	1.5
160	ewv	1.10	txt = '\n# Written by SchedulerCondor::wsInitialEnvironment\n'
161			txt += 'echo "Beginning environment"\n'
162			txt += 'printenv \| sort\n'
163	ewv	1.5
164	ewv	1.10	txt += 'middleware='+self.name()+' \n'
165	ewv	1.14	txt += 'if [ -e /opt/d-cache/srm/bin ]; then\n'
166			txt += ' export PATH=${PATH}:/opt/d-cache/srm/bin\n'
167			txt += 'fi\n'
168
169	ewv	1.10	txt += """
170	ewv	1.6	if [ $_CONDOR_SCRATCH_DIR ] && [ -d $_CONDOR_SCRATCH_DIR ]; then
171	ewv	1.12	echo "cd to Condor scratch directory: $_CONDOR_SCRATCH_DIR"
172	ewv	1.6	if [ -e ../default.tgz ] ;then
173			echo "Found ISB in parent directory (Local Condor)"
174			cp ../default.tgz $_CONDOR_SCRATCH_DIR
175			fi
176			cd $_CONDOR_SCRATCH_DIR
177			fi
178			"""
179	ewv	1.5
180	ewv	1.10	return txt
181	ewv	1.27
182
183			def sched_fix_parameter(self):
184			"""
185			Returns string with requirements and scheduler-specific parameters
186			"""
187
188			if self.EDG_requirements:
189			req = self.EDG_requirements
190			taskReq = {'commonRequirements':req}
191			common._db.updateTask_(taskReq)