CRAB/python/ServerCommunicator.py

#!/usr/bin/env python
"""
_ServerCommunicator_

"""

__version__ = "$Id"
__revision__ = "$Revision"
__author__ = "farinafa@cern.ch"

from crab_exceptions import *
from crab_util import *
import common
import Scram
from ProdCommon.Credential.CredentialAPI import CredentialAPI
from xml.dom import minidom
import os, sys
import commands
import traceback
from Downloader import Downloader

if (sys.version_info[0] + .1 * sys.version_info[1]) < 2.6 : 
    from CRAB_Server_API import CRAB_Server_Session as C_AS_Session
elif os.environ["SCRAM_ARCH"].split("_")[1].find('32')>1:
    try:
        from CRAB_Server_API_1_1 import CRAB_Server_Session as C_AS_Session
    except :
        try:
            from CRAB_Server_API_36X import CRAB_Server_Session as C_AS_Session 
         except :
            from CRAB_Server_API_311X import CRAB_Server_Session as C_AS_Session 
else:
    try:
        from lib64.CRAB_Server_API_1_1 import CRAB_Server_Session as C_AS_Session
    except:
        try:
            from lib64.CRAB_Server_API_36X_amd64 import CRAB_Server_Session as C_AS_Session
        except:
            from lib64.CRAB_Server_API_311X_amd64 import CRAB_Server_Session as C_AS_Session

class ServerCommunicator:
    """
    Common interface for the interaction between the Crab client and the server Web Service
    """
    def __init__(self, serverName, serverPort, cfg_params, proxyPath=None):
        """
        Open the communication with an Analysis Server by passing the server URL and the port
        """

        self.ServerTwiki = 'https://twiki.cern.ch/twiki/bin/view/CMS/SWGuideCrabServerForUsers#Server_available_for_users'

        self.asSession = C_AS_Session(serverName, serverPort)
        self.cfg_params = cfg_params
        self.userSubj = ''
        self.serverName = serverName
        credentialType = 'Proxy'
        if common.scheduler.name().upper() in ['CAF','LSF']:
            credentialType = 'Token'
        CliServerParams(self)
        self.crab_task_name = common.work_space.topDir().split('/')[-2] # nice task name "crab_0_..."

        configAPI = {'credential' : credentialType, \
                     'logger' : common.logger() }

        CredAPI =  CredentialAPI( configAPI )
        try:
            self.userSubj = CredAPI.getSubject()
        except Exception, err:
            common.logger.debug("Getting Credential Subject: " +str(traceback.format_exc()))
            raise CrabException("Error Getting Credential Subject")

        self.scram=Scram.Scram(cfg_params)
###################################################
    # Interactions with the server
###################################################

# wmbs
    def submitNewTask(self, blTaskName, blXml, rng, TotJob,taskType='fullySpecified'):
        """
        _submitNewTask_
        Send a new task to the server to be submitted.

        Accepts in input:
             - the bossLite object representing the task (jobs are assumed to be RunningJobs)
             - the range of the submission as specified by the user at the command line
        """


        if not blXml:
            raise CrabException('Error while serializing task object to XML')
            return -2

        if not blTaskName:
            raise CrabException('Error while extracting the Task Unique Name string')
            return -3

        cmdXML = None
        cmdXML = self._createXMLcommand(blTaskName, 'submit', rng, newTaskAddIns=True,jobs=TotJob,type=taskType)

        if not cmdXML:
            raise CrabException('Error while creating the Command XML')
            return -4

        ret = -1
        ret = self.asSession.transferTaskAndSubmit(blXml, cmdXML, blTaskName)

        if ret == 0:
             # success
             logMsg = 'Task %s successfully submitted to server %s'%(self.crab_task_name, self.serverName)
             common.logger.info(logMsg+'\n')
        else:
             self.checkServerResponse(ret)

        return ret

    def checkServerResponse(self, ret):
        """
        analyze the server return codes
        """

        logMsg = ''
        if ret == 10:
            # overlaod
            logMsg = 'Error The server %s refused to accept the task %s because it is overloaded\n'%(self.serverName, self.crab_task_name)
            logMsg += '\t For Further infos please contact the server Admin: %s'%self.server_admin
        elif ret == 14:
            # Draining
            logMsg  = 'Error The server %s refused to accept the task %s because it is Draining out\n'%(self.serverName, self.crab_task_name)
            logMsg += '\t remaining jobs due to scheduled maintainence\n'
            logMsg += '\t For Further infos please contact the server Admin: %s'%self.server_admin
        elif ret == 101:
            # overlaod
            logMsg = 'Error The server %s refused the submission %s because you asked a too large task. Please submit by range'%(self.serverName, self.crab_task_name)
        elif ret == 11:
            # failed to push message in DB
            logMsg = 'Server unable to release messages into DB. Task %s won\'t be submitted.'%self.crab_task_name
        elif ret == 12:
            # failed SOAP communication
            logMsg = 'Error The server %s refused to accept the task %s. It could be under maintainance. \n'%(self.serverName, self.crab_task_name)
            logMsg += '\t For Further infos please contact the server Admin: %s'%self.server_admin
        elif ret == 20:
            # failed to push message in PA
            logMsg = 'Server unable to release messages to other components. Task %s won\'t be submitted.'%self.crab_task_name
        elif ret == 22:
            # failed SOAP communication
            logMsg = 'Error during SOAP communication with server %s'%self.serverName
        elif ret == 33:
            # uncompatible client version
            logMsg  = 'Error You are using a wrong client version for server: %s\n'%self.serverName
            logMsg += '\t For further informations about "Servers available for users" please check here:\n \t%s '%self.ServerTwiki
        else:
            logMsg = 'Unexpected return code from server %s: %d'%(self.serverName, ret)

        # print loggings
        if logMsg != '':
            # reset server choice
            opsToBeSaved={'serverName' : '' }
            common._db.updateTask_(opsToBeSaved)
            common.logger.info(logMsg)
        return ret

    def subsequentJobSubmit(self, blTaskName, rng):
        """
        _subsequentJobSubmit_
        Let the submission of other jobs of a task that has been already sent to a server.
        This method is used for subsequent submission of ranged sets of jobs.

        Accepts in input:
             - the bossLite object representing the task (jobs are assumed to be RunningJobs)
             - the range of the submission as specified by the user at the command line
        """
        return self._genericCommand('submit', blTaskName, rng)

    def killJobs(self, blTaskName, rng):
        """
        _killJobs_
        Send a kill command to one or more jobs running on the server.

        Accepts in input:
             - the bossLite object representing the task (jobs are assumed to be RunningJobs)
             - the range of the submission as specified by the user at the command line
        """
        return self._genericCommand('kill', blTaskName, rng)

    def StopWorkflow(self, blTaskName):
        """
        _StopWorkflow_
        """
        return self._genericCommand('StopWorkflow', blTaskName, 'all')

    def cleanTask(self, blTaskName):
        """
        _cleanTask_
        Force the server to clean the jobs on the server.

        Accepts in input:
             - the bossLite object representing the task (jobs are assumed to be RunningJobs)
             - the range of the submission as specified by the user at the command line
        """
        return self._genericCommand('clean', blTaskName, 'all')

    def getStatus(self, blTaskName, statusFile=None, statusFamilyType='status'):
        """
        _getStatus_
        Retrieve the task status from the server. It can recover any kind of status (version, loggingInfos,...)
        """

        # fill the filename
        filename = str(statusFile)

        if not blTaskName:
            raise CrabException('Exception while getting the task unique name')
            return ''

        # get the data and fill the file content
        statusMsg = self.asSession.getTaskStatus(statusFamilyType, blTaskName)
        if 'Error:' in  statusMsg[:6] or len(statusMsg)==0:
             if statusFamilyType != 'isServerDrained':
                 raise CrabException('Error in retrieving task %s status from server %s'%(self.crab_task_name, self.serverName) )
             else:
                 statusFile = None
                 statusMMsg = ""

        if statusFile is not None:
            f = open(statusFile, 'w')
            f.write(statusMsg)
            f.close()
            return statusFile
        return statusMsg

    def outputRetrieved(self, blTaskName, rng):
        """
        _getJobsOutput_
        Get from the server the output file locations to be transfered back.

        Accepts in input:
             - the bossLite object representing the task (jobs are assumed to be RunningJobs)
             - the range of the submission as specified by the user at the command line
        """
        return self._genericCommand('outputRetrieved', blTaskName, rng)

    def checkDrainMode(self, blTaskName='null'):
        return self.getStatus( blTaskName, statusFamilyType='isServerDrained')

    def postMortemInfos(self, blTaskName, rng):
        """
        _postMortemInfos_
        Retrieve the job postmortem information from the server.

        Accepts in input:
             - the bossLite object representing the task (jobs are assumed to be RunningJobs)
             - the range of the submission as specified by the user at the command line
        """
        # get the status in
        raise NotImplementedError
        return None

###################################################
    # Auxiliary methods
###################################################

    def _createXMLcommand(self, taskUName, cmdSpec='status', rng='all', newTaskAddIns=False, flavour='analysis', type='fullySpecified',jobs='-1'):
        xmlString = ''
        cfile = minidom.Document()

        ver = common.prog_version_str
        node = cfile.createElement("TaskCommand")
        node.setAttribute("Task", str(taskUName) )
        node.setAttribute("Subject", str(self.userSubj) )
        node.setAttribute("Command", str(cmdSpec) )
        node.setAttribute("Range", str(rng) )
        node.setAttribute("TotJob", str(jobs) )
        node.setAttribute("Scheduler", str(self.cfg_params['CRAB.scheduler']) )
        node.setAttribute("Flavour", str(flavour) )
        node.setAttribute("Type", str(type) )
        node.setAttribute("ClientVersion", str(ver) )

        ## Only Temporary. it should be at Server level
        removeBList = self.cfg_params.get("GRID.remove_default_blacklist", 0 )
        blackAnaOps = None 
        if int(removeBList) == 0:
            blacklist = Downloader("http://cmsdoc.cern.ch/cms/LCG/crab/config/")
            result = blacklist.config("site_black_list.conf")
            if result != None:
                blackAnaOps = result
            common.logger.debug("Enforced black list: %s "%str(blacklist))
        else:
            common.logger.info("WARNING: Skipping default black list!")

        try:
            myproxyserver = Downloader("http://cmsdoc.cern.ch/cms/LCG/crab/config/").config("myproxy_server.conf")
            myproxyserver = myproxyserver.strip()
            if myproxyserver 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)
            myproxyserver = 'myproxy.cern.ch'

        # create a mini-cfg to be transfered to the server
        miniCfg = {}

        ## migrate CE/SE infos
        miniCfg['EDG.ce_white_list'] = ""
        if 'GRID.ce_white_list' in self.cfg_params:
            miniCfg['EDG.ce_white_list'] = str( self.cfg_params['GRID.ce_white_list'] )

        miniCfg['EDG.ce_black_list'] = blackAnaOps
        if 'GRID.ce_black_list' in self.cfg_params:
            if blackAnaOps:
                miniCfg['EDG.ce_black_list'] += ", "
            miniCfg['EDG.ce_black_list'] += str( self.cfg_params['GRID.ce_black_list'] )

        miniCfg['EDG.se_white_list'] = ""
        if 'GRID.se_white_list' in self.cfg_params:
            miniCfg['EDG.se_white_list'] = str( self.cfg_params['GRID.se_white_list'] )

        miniCfg['EDG.se_black_list'] = ""
        if 'GRID.se_black_list' in self.cfg_params:
            miniCfg['EDG.se_black_list'] = str( self.cfg_params['GRID.se_black_list'] )

        miniCfg['EDG.group'] = ""
        if 'GRID.group' in self.cfg_params:
            miniCfg['EDG.group'] = str( self.cfg_params['GRID.group'] )

        miniCfg['EDG.role'] = ""
        if 'GRID.role' in self.cfg_params:
            miniCfg['EDG.role'] = str( self.cfg_params['GRID.role'] )

        miniCfg['cfgFileNameCkSum'] = makeCksum(common.work_space.cfgFileName())
        if 'cfgFileNameCkSum' in self.cfg_params:
            miniCfg['cfgFileNameCkSum'] = str(self.cfg_params['cfgFileNameCkSum'])

        miniCfg['CRAB.se_remote_dir'] = ''
        if 'CRAB.se_remote_dir' in self.cfg_params:
            miniCfg['CRAB.se_remote_dir'] = str(self.cfg_params['CRAB.se_remote_dir'])

        miniCfg['CAF.queue'] = self.cfg_params.get('CAF.queue','cmscaf1nw')
        miniCfg['CAF.resources'] = self.cfg_params.get('CAF.resource', 'cmscaf')
        miniCfg['CAF.group'] = self.cfg_params.get('CAF.group', None)
        ## JDL requirements specific data. Scheduler dependant
        miniCfg['EDG.max_wall_time'] = self.cfg_params.get('GRID.max_wall_clock_time', None)
        miniCfg['EDG.max_cpu_time'] = self.cfg_params.get('GRID.max_cpu_time', '130')
        miniCfg['proxyServer'] = myproxyserver 
        miniCfg['VO'] = self.cfg_params.get('GRID.virtual_organization', 'cms')
        miniCfg['EDG_retry_count'] = 0
        miniCfg['EDG_shallow_retry_count'] = -1

        ## Additional field for DashBoard
        miniCfg['CMSSW.datasetpath'] = self.cfg_params.get('CMSSW.datasetpath', 'None')

        ## Additional fields for Notification by the server
        miniCfg['eMail'] = self.cfg_params.get('USER.email', None)
        miniCfg['threshold'] = self.cfg_params.get('USER.thresholdlevel', 100)

        miniCfg['CMSSW_version'] = self.scram.getSWVersion()
#WMBS
        miniCfg['feeder'] = self.cfg_params.get('WMBS.feeder', 'Feeder')
        miniCfg['parent_task'] = self.cfg_params.get('WMBS.parent_task', '')

        miniCfg['processing'] = self.cfg_params.get('WMBS.processing', 'bulk')
        miniCfg['startrun'] = self.cfg_params.get('WMBS.startrun', 'None')

        miniCfg['splitting_algorithm'] = self.cfg_params.get('WMBS.splitting_algorithm','FileBased')
        miniCfg['split_per_job'] = self.cfg_params.get('WMBS.split_per_job','files_per_job')
        miniCfg['split_value'] = self.cfg_params.get('WMBS.split_value',1000000)

        miniCfg['CMSSW_version'] = self.scram.getSWVersion()

        ## put here other fields if needed
        node.setAttribute("CfgParamDict", str(miniCfg) )
        cfile.appendChild(node)
        xmlString += str(cfile.toprettyxml())
        return xmlString

    def _genericCommand(self, cmd, blTaskName, rng):
        if not blTaskName:
            raise CrabException('Error while extracting the Task Unique Name string')
            return -2

        cmdXML = None
        cmdXML = self._createXMLcommand(blTaskName, cmd, rng)
        if not cmdXML:
            raise CrabException('Error while creating the Command XML')
            return -3

        ret = -1
        ret = self.asSession.sendCommand(cmdXML, blTaskName)
        logMsg = ''
        debugMsg = ''
        if ret == 0:
             # success
             debugMsg = 'Command successfully sent to server %s for task %s'%(self.serverName, self.crab_task_name)
        else:
             self.checkServerResponse(ret)
        return ret

Revision:	1.71
Committed:	Tue Feb 22 16:36:22 2011 UTC (14 years, 2 months ago) by spiga
Content type:	text/x-python
Branch:	MAIN
Changes since 1.70:	+8 -2 lines
Log Message:	add compatibility boost 1.44
#	Content
1	#!/usr/bin/env python
2	"""
3	_ServerCommunicator_
4
5	"""
6
7	__version__ = "$Id"
8	__revision__ = "$Revision"
9	__author__ = "farinafa@cern.ch"
10
11	from crab_exceptions import *
12	from crab_util import *
13	import common
14	import Scram
15	from ProdCommon.Credential.CredentialAPI import CredentialAPI
16	from xml.dom import minidom
17	import os, sys
18	import commands
19	import traceback
20	from Downloader import Downloader
21
22	if (sys.version_info[0] + .1 * sys.version_info[1]) < 2.6 :
23	from CRAB_Server_API import CRAB_Server_Session as C_AS_Session
24	elif os.environ["SCRAM_ARCH"].split("_")[1].find('32')>1:
25	try:
26	from CRAB_Server_API_1_1 import CRAB_Server_Session as C_AS_Session
27	except :
28	try:
29	from CRAB_Server_API_36X import CRAB_Server_Session as C_AS_Session
30	except :
31	from CRAB_Server_API_311X import CRAB_Server_Session as C_AS_Session
32	else:
33	try:
34	from lib64.CRAB_Server_API_1_1 import CRAB_Server_Session as C_AS_Session
35	except:
36	try:
37	from lib64.CRAB_Server_API_36X_amd64 import CRAB_Server_Session as C_AS_Session
38	except:
39	from lib64.CRAB_Server_API_311X_amd64 import CRAB_Server_Session as C_AS_Session
40
41	class ServerCommunicator:
42	"""
43	Common interface for the interaction between the Crab client and the server Web Service
44	"""
45	def __init__(self, serverName, serverPort, cfg_params, proxyPath=None):
46	"""
47	Open the communication with an Analysis Server by passing the server URL and the port
48	"""
49
50	self.ServerTwiki = 'https://twiki.cern.ch/twiki/bin/view/CMS/SWGuideCrabServerForUsers#Server_available_for_users'
51
52	self.asSession = C_AS_Session(serverName, serverPort)
53	self.cfg_params = cfg_params
54	self.userSubj = ''
55	self.serverName = serverName
56	credentialType = 'Proxy'
57	if common.scheduler.name().upper() in ['CAF','LSF']:
58	credentialType = 'Token'
59	CliServerParams(self)
60	self.crab_task_name = common.work_space.topDir().split('/')[-2] # nice task name "crab_0_..."
61
62	configAPI = {'credential' : credentialType, \
63	'logger' : common.logger() }
64
65	CredAPI = CredentialAPI( configAPI )
66	try:
67	self.userSubj = CredAPI.getSubject()
68	except Exception, err:
69	common.logger.debug("Getting Credential Subject: " +str(traceback.format_exc()))
70	raise CrabException("Error Getting Credential Subject")
71
72	self.scram=Scram.Scram(cfg_params)
73	###################################################
74	# Interactions with the server
75	###################################################
76
77	# wmbs
78	def submitNewTask(self, blTaskName, blXml, rng, TotJob,taskType='fullySpecified'):
79	"""
80	_submitNewTask_
81	Send a new task to the server to be submitted.
82
83	Accepts in input:
84	- the bossLite object representing the task (jobs are assumed to be RunningJobs)
85	- the range of the submission as specified by the user at the command line
86	"""
87
88
89	if not blXml:
90	raise CrabException('Error while serializing task object to XML')
91	return -2
92
93	if not blTaskName:
94	raise CrabException('Error while extracting the Task Unique Name string')
95	return -3
96
97	cmdXML = None
98	cmdXML = self._createXMLcommand(blTaskName, 'submit', rng, newTaskAddIns=True,jobs=TotJob,type=taskType)
99
100	if not cmdXML:
101	raise CrabException('Error while creating the Command XML')
102	return -4
103
104	ret = -1
105	ret = self.asSession.transferTaskAndSubmit(blXml, cmdXML, blTaskName)
106
107	if ret == 0:
108	# success
109	logMsg = 'Task %s successfully submitted to server %s'%(self.crab_task_name, self.serverName)
110	common.logger.info(logMsg+'\n')
111	else:
112	self.checkServerResponse(ret)
113
114	return ret
115
116	def checkServerResponse(self, ret):
117	"""
118	analyze the server return codes
119	"""
120
121	logMsg = ''
122	if ret == 10:
123	# overlaod
124	logMsg = 'Error The server %s refused to accept the task %s because it is overloaded\n'%(self.serverName, self.crab_task_name)
125	logMsg += '\t For Further infos please contact the server Admin: %s'%self.server_admin
126	elif ret == 14:
127	# Draining
128	logMsg = 'Error The server %s refused to accept the task %s because it is Draining out\n'%(self.serverName, self.crab_task_name)
129	logMsg += '\t remaining jobs due to scheduled maintainence\n'
130	logMsg += '\t For Further infos please contact the server Admin: %s'%self.server_admin
131	elif ret == 101:
132	# overlaod
133	logMsg = 'Error The server %s refused the submission %s because you asked a too large task. Please submit by range'%(self.serverName, self.crab_task_name)
134	elif ret == 11:
135	# failed to push message in DB
136	logMsg = 'Server unable to release messages into DB. Task %s won\'t be submitted.'%self.crab_task_name
137	elif ret == 12:
138	# failed SOAP communication
139	logMsg = 'Error The server %s refused to accept the task %s. It could be under maintainance. \n'%(self.serverName, self.crab_task_name)
140	logMsg += '\t For Further infos please contact the server Admin: %s'%self.server_admin
141	elif ret == 20:
142	# failed to push message in PA
143	logMsg = 'Server unable to release messages to other components. Task %s won\'t be submitted.'%self.crab_task_name
144	elif ret == 22:
145	# failed SOAP communication
146	logMsg = 'Error during SOAP communication with server %s'%self.serverName
147	elif ret == 33:
148	# uncompatible client version
149	logMsg = 'Error You are using a wrong client version for server: %s\n'%self.serverName
150	logMsg += '\t For further informations about "Servers available for users" please check here:\n \t%s '%self.ServerTwiki
151	else:
152	logMsg = 'Unexpected return code from server %s: %d'%(self.serverName, ret)
153
154	# print loggings
155	if logMsg != '':
156	# reset server choice
157	opsToBeSaved={'serverName' : '' }
158	common._db.updateTask_(opsToBeSaved)
159	common.logger.info(logMsg)
160	return ret
161
162	def subsequentJobSubmit(self, blTaskName, rng):
163	"""
164	_subsequentJobSubmit_
165	Let the submission of other jobs of a task that has been already sent to a server.
166	This method is used for subsequent submission of ranged sets of jobs.
167
168	Accepts in input:
169	- the bossLite object representing the task (jobs are assumed to be RunningJobs)
170	- the range of the submission as specified by the user at the command line
171	"""
172	return self._genericCommand('submit', blTaskName, rng)
173
174	def killJobs(self, blTaskName, rng):
175	"""
176	_killJobs_
177	Send a kill command to one or more jobs running on the server.
178
179	Accepts in input:
180	- the bossLite object representing the task (jobs are assumed to be RunningJobs)
181	- the range of the submission as specified by the user at the command line
182	"""
183	return self._genericCommand('kill', blTaskName, rng)
184
185	def StopWorkflow(self, blTaskName):
186	"""
187	_StopWorkflow_
188	"""
189	return self._genericCommand('StopWorkflow', blTaskName, 'all')
190
191	def cleanTask(self, blTaskName):
192	"""
193	_cleanTask_
194	Force the server to clean the jobs on the server.
195
196	Accepts in input:
197	- the bossLite object representing the task (jobs are assumed to be RunningJobs)
198	- the range of the submission as specified by the user at the command line
199	"""
200	return self._genericCommand('clean', blTaskName, 'all')
201
202	def getStatus(self, blTaskName, statusFile=None, statusFamilyType='status'):
203	"""
204	_getStatus_
205	Retrieve the task status from the server. It can recover any kind of status (version, loggingInfos,...)
206	"""
207
208	# fill the filename
209	filename = str(statusFile)
210
211	if not blTaskName:
212	raise CrabException('Exception while getting the task unique name')
213	return ''
214
215	# get the data and fill the file content
216	statusMsg = self.asSession.getTaskStatus(statusFamilyType, blTaskName)
217	if 'Error:' in statusMsg[:6] or len(statusMsg)==0:
218	if statusFamilyType != 'isServerDrained':
219	raise CrabException('Error in retrieving task %s status from server %s'%(self.crab_task_name, self.serverName) )
220	else:
221	statusFile = None
222	statusMMsg = ""
223
224	if statusFile is not None:
225	f = open(statusFile, 'w')
226	f.write(statusMsg)
227	f.close()
228	return statusFile
229	return statusMsg
230
231	def outputRetrieved(self, blTaskName, rng):
232	"""
233	_getJobsOutput_
234	Get from the server the output file locations to be transfered back.
235
236	Accepts in input:
237	- the bossLite object representing the task (jobs are assumed to be RunningJobs)
238	- the range of the submission as specified by the user at the command line
239	"""
240	return self._genericCommand('outputRetrieved', blTaskName, rng)
241
242	def checkDrainMode(self, blTaskName='null'):
243	return self.getStatus( blTaskName, statusFamilyType='isServerDrained')
244
245	def postMortemInfos(self, blTaskName, rng):
246	"""
247	_postMortemInfos_
248	Retrieve the job postmortem information from the server.
249
250	Accepts in input:
251	- the bossLite object representing the task (jobs are assumed to be RunningJobs)
252	- the range of the submission as specified by the user at the command line
253	"""
254	# get the status in
255	raise NotImplementedError
256	return None
257
258	###################################################
259	# Auxiliary methods
260	###################################################
261
262	def _createXMLcommand(self, taskUName, cmdSpec='status', rng='all', newTaskAddIns=False, flavour='analysis', type='fullySpecified',jobs='-1'):
263	xmlString = ''
264	cfile = minidom.Document()
265
266	ver = common.prog_version_str
267	node = cfile.createElement("TaskCommand")
268	node.setAttribute("Task", str(taskUName) )
269	node.setAttribute("Subject", str(self.userSubj) )
270	node.setAttribute("Command", str(cmdSpec) )
271	node.setAttribute("Range", str(rng) )
272	node.setAttribute("TotJob", str(jobs) )
273	node.setAttribute("Scheduler", str(self.cfg_params['CRAB.scheduler']) )
274	node.setAttribute("Flavour", str(flavour) )
275	node.setAttribute("Type", str(type) )
276	node.setAttribute("ClientVersion", str(ver) )
277
278	## Only Temporary. it should be at Server level
279	removeBList = self.cfg_params.get("GRID.remove_default_blacklist", 0 )
280	blackAnaOps = None
281	if int(removeBList) == 0:
282	blacklist = Downloader("http://cmsdoc.cern.ch/cms/LCG/crab/config/")
283	result = blacklist.config("site_black_list.conf")
284	if result != None:
285	blackAnaOps = result
286	common.logger.debug("Enforced black list: %s "%str(blacklist))
287	else:
288	common.logger.info("WARNING: Skipping default black list!")
289
290	try:
291	myproxyserver = Downloader("http://cmsdoc.cern.ch/cms/LCG/crab/config/").config("myproxy_server.conf")
292	myproxyserver = myproxyserver.strip()
293	if myproxyserver is None:
294	raise CrabException("myproxy_server.conf retrieved but empty")
295	except Exception, e:
296	common.logger.info("Problem setting myproxy server endpoint: using myproxy.cern.ch")
297	common.logger.debug(e)
298	myproxyserver = 'myproxy.cern.ch'
299
300	# create a mini-cfg to be transfered to the server
301	miniCfg = {}
302
303	## migrate CE/SE infos
304	miniCfg['EDG.ce_white_list'] = ""
305	if 'GRID.ce_white_list' in self.cfg_params:
306	miniCfg['EDG.ce_white_list'] = str( self.cfg_params['GRID.ce_white_list'] )
307
308	miniCfg['EDG.ce_black_list'] = blackAnaOps
309	if 'GRID.ce_black_list' in self.cfg_params:
310	if blackAnaOps:
311	miniCfg['EDG.ce_black_list'] += ", "
312	miniCfg['EDG.ce_black_list'] += str( self.cfg_params['GRID.ce_black_list'] )
313
314	miniCfg['EDG.se_white_list'] = ""
315	if 'GRID.se_white_list' in self.cfg_params:
316	miniCfg['EDG.se_white_list'] = str( self.cfg_params['GRID.se_white_list'] )
317
318	miniCfg['EDG.se_black_list'] = ""
319	if 'GRID.se_black_list' in self.cfg_params:
320	miniCfg['EDG.se_black_list'] = str( self.cfg_params['GRID.se_black_list'] )
321
322	miniCfg['EDG.group'] = ""
323	if 'GRID.group' in self.cfg_params:
324	miniCfg['EDG.group'] = str( self.cfg_params['GRID.group'] )
325
326	miniCfg['EDG.role'] = ""
327	if 'GRID.role' in self.cfg_params:
328	miniCfg['EDG.role'] = str( self.cfg_params['GRID.role'] )
329
330	miniCfg['cfgFileNameCkSum'] = makeCksum(common.work_space.cfgFileName())
331	if 'cfgFileNameCkSum' in self.cfg_params:
332	miniCfg['cfgFileNameCkSum'] = str(self.cfg_params['cfgFileNameCkSum'])
333
334	miniCfg['CRAB.se_remote_dir'] = ''
335	if 'CRAB.se_remote_dir' in self.cfg_params:
336	miniCfg['CRAB.se_remote_dir'] = str(self.cfg_params['CRAB.se_remote_dir'])
337
338	miniCfg['CAF.queue'] = self.cfg_params.get('CAF.queue','cmscaf1nw')
339	miniCfg['CAF.resources'] = self.cfg_params.get('CAF.resource', 'cmscaf')
340	miniCfg['CAF.group'] = self.cfg_params.get('CAF.group', None)
341	## JDL requirements specific data. Scheduler dependant
342	miniCfg['EDG.max_wall_time'] = self.cfg_params.get('GRID.max_wall_clock_time', None)
343	miniCfg['EDG.max_cpu_time'] = self.cfg_params.get('GRID.max_cpu_time', '130')
344	miniCfg['proxyServer'] = myproxyserver
345	miniCfg['VO'] = self.cfg_params.get('GRID.virtual_organization', 'cms')
346	miniCfg['EDG_retry_count'] = 0
347	miniCfg['EDG_shallow_retry_count'] = -1
348
349	## Additional field for DashBoard
350	miniCfg['CMSSW.datasetpath'] = self.cfg_params.get('CMSSW.datasetpath', 'None')
351
352	## Additional fields for Notification by the server
353	miniCfg['eMail'] = self.cfg_params.get('USER.email', None)
354	miniCfg['threshold'] = self.cfg_params.get('USER.thresholdlevel', 100)
355
356	miniCfg['CMSSW_version'] = self.scram.getSWVersion()
357	#WMBS
358	miniCfg['feeder'] = self.cfg_params.get('WMBS.feeder', 'Feeder')
359	miniCfg['parent_task'] = self.cfg_params.get('WMBS.parent_task', '')
360
361	miniCfg['processing'] = self.cfg_params.get('WMBS.processing', 'bulk')
362	miniCfg['startrun'] = self.cfg_params.get('WMBS.startrun', 'None')
363
364	miniCfg['splitting_algorithm'] = self.cfg_params.get('WMBS.splitting_algorithm','FileBased')
365	miniCfg['split_per_job'] = self.cfg_params.get('WMBS.split_per_job','files_per_job')
366	miniCfg['split_value'] = self.cfg_params.get('WMBS.split_value',1000000)
367
368	miniCfg['CMSSW_version'] = self.scram.getSWVersion()
369
370	## put here other fields if needed
371	node.setAttribute("CfgParamDict", str(miniCfg) )
372	cfile.appendChild(node)
373	xmlString += str(cfile.toprettyxml())
374	return xmlString
375
376	def _genericCommand(self, cmd, blTaskName, rng):
377	if not blTaskName:
378	raise CrabException('Error while extracting the Task Unique Name string')
379	return -2
380
381	cmdXML = None
382	cmdXML = self._createXMLcommand(blTaskName, cmd, rng)
383	if not cmdXML:
384	raise CrabException('Error while creating the Command XML')
385	return -3
386
387	ret = -1
388	ret = self.asSession.sendCommand(cmdXML, blTaskName)
389	logMsg = ''
390	debugMsg = ''
391	if ret == 0:
392	# success
393	debugMsg = 'Command successfully sent to server %s for task %s'%(self.serverName, self.crab_task_name)
394	else:
395	self.checkServerResponse(ret)
396	return ret
397