CRAB/python/SchedulerCondor.py

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

__revision__ = "$Id: SchedulerCondor.py,v 1.21 2009/03/06 16:54:42 spiga Exp $"
__version__ = "$Revision: 1.21 $"

from SchedulerLocal  import SchedulerLocal
from crab_exceptions import CrabException

import common
import os
import socket
import sha

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)

        if int(self.copy_data) == 1:

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

            self.group = cfg_params.get("EDG.group", None)
            self.role  = cfg_params.get("EDG.role", None)
            self.VO    = cfg_params.get('EDG.virtual_organization', 'cms')

            self.checkProxy()

        self.role  = None

        return

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

    def sched_parameter(self, i, task):
        """
        Return scheduler-specific parameters
        """

        index = int(common._db.nJobs()) - 1
        schedParam = ''

        for i in range(index):
            pass

        return schedParam


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