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.74
Committed:	Mon Mar 28 10:02:26 2011 UTC (14 years, 1 month ago) by spiga
Content type:	text/x-python
Branch:	MAIN
CVS Tags:	CRAB_2_7_8_pre2, CRAB_2_7_8_dash3
Changes since 1.73:	+1 -6 lines
Log Message:	fix for #76950 does not apply. Rolling it back
#	User	Rev	Content
1	farinafa	1.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	farinafa	1.13	from crab_util import *
13	farinafa	1.1	import common
14	spiga	1.34	import Scram
15			from ProdCommon.Credential.CredentialAPI import CredentialAPI
16	farinafa	1.1	from xml.dom import minidom
17	farinafa	1.54	import os, sys
18	farinafa	1.2	import commands
19	ewv	1.45	import traceback
20	spiga	1.55	from Downloader import Downloader
21	farinafa	1.1
22	farinafa	1.54	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	spiga	1.65	elif os.environ["SCRAM_ARCH"].split("_")[1].find('32')>1:
25	slacapra	1.60	try:
26			from CRAB_Server_API_1_1 import CRAB_Server_Session as C_AS_Session
27	spiga	1.65	except :
28	spiga	1.71	try:
29			from CRAB_Server_API_36X import CRAB_Server_Session as C_AS_Session
30	spiga	1.72	except :
31	spiga	1.71	from CRAB_Server_API_311X import CRAB_Server_Session as C_AS_Session
32	spiga	1.65	else:
33	spiga	1.66	try:
34			from lib64.CRAB_Server_API_1_1 import CRAB_Server_Session as C_AS_Session
35			except:
36	spiga	1.71	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	farinafa	1.54
41	farinafa	1.1	class ServerCommunicator:
42			"""
43	ewv	1.45	Common interface for the interaction between the Crab client and the server Web Service
44	farinafa	1.1	"""
45	farinafa	1.2	def __init__(self, serverName, serverPort, cfg_params, proxyPath=None):
46	farinafa	1.1	"""
47			Open the communication with an Analysis Server by passing the server URL and the port
48			"""
49	spiga	1.27
50	spiga	1.61	self.ServerTwiki = 'https://twiki.cern.ch/twiki/bin/view/CMS/SWGuideCrabServerForUsers#Server_available_for_users'
51	ewv	1.45
52	farinafa	1.1	self.asSession = C_AS_Session(serverName, serverPort)
53			self.cfg_params = cfg_params
54			self.userSubj = ''
55			self.serverName = serverName
56	spiga	1.31	credentialType = 'Proxy'
57	ewv	1.45	if common.scheduler.name().upper() in ['CAF','LSF']:
58	spiga	1.31	credentialType = 'Token'
59	spiga	1.27	CliServerParams(self)
60	farinafa	1.11	self.crab_task_name = common.work_space.topDir().split('/')[-2] # nice task name "crab_0_..."
61
62	spiga	1.40	configAPI = {'credential' : credentialType, \
63			'logger' : common.logger() }
64	ewv	1.45
65			CredAPI = CredentialAPI( configAPI )
66	spiga	1.31	try:
67	ewv	1.45	self.userSubj = CredAPI.getSubject()
68	spiga	1.31	except Exception, err:
69	spiga	1.35	common.logger.debug("Getting Credential Subject: " +str(traceback.format_exc()))
70	spiga	1.31	raise CrabException("Error Getting Credential Subject")
71	spiga	1.34
72			self.scram=Scram.Scram(cfg_params)
73	farinafa	1.1	###################################################
74			# Interactions with the server
75			###################################################
76
77	spiga	1.47	# wmbs
78			def submitNewTask(self, blTaskName, blXml, rng, TotJob,taskType='fullySpecified'):
79	farinafa	1.1	"""
80			_submitNewTask_
81	ewv	1.45	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	farinafa	1.1	"""
87
88	spiga	1.47
89	farinafa	1.1	if not blXml:
90			raise CrabException('Error while serializing task object to XML')
91			return -2
92
93	farinafa	1.2	if not blTaskName:
94	farinafa	1.1	raise CrabException('Error while extracting the Task Unique Name string')
95			return -3
96
97			cmdXML = None
98	spiga	1.47	cmdXML = self._createXMLcommand(blTaskName, 'submit', rng, newTaskAddIns=True,jobs=TotJob,type=taskType)
99
100	farinafa	1.1	if not cmdXML:
101			raise CrabException('Error while creating the Command XML')
102			return -4
103
104	ewv	1.45	ret = -1
105	farinafa	1.2	ret = self.asSession.transferTaskAndSubmit(blXml, cmdXML, blTaskName)
106	ewv	1.45
107	farinafa	1.1	if ret == 0:
108			# success
109	farinafa	1.11	logMsg = 'Task %s successfully submitted to server %s'%(self.crab_task_name, self.serverName)
110	spiga	1.35	common.logger.info(logMsg+'\n')
111	farinafa	1.1	else:
112	ewv	1.45	self.checkServerResponse(ret)
113	spiga	1.27
114			return ret
115
116	ewv	1.45	def checkServerResponse(self, ret):
117	spiga	1.27	"""
118			analyze the server return codes
119			"""
120	ewv	1.45
121	spiga	1.27	logMsg = ''
122			if ret == 10:
123	slacapra	1.38	# 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	spiga	1.27	elif ret == 14:
127	ewv	1.45	# Draining
128	slacapra	1.38	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	spiga	1.27	elif ret == 101:
132	slacapra	1.38	# 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	spiga	1.27	elif ret == 11:
135	slacapra	1.38	# failed to push message in DB
136	farinafa	1.50	logMsg = 'Server unable to release messages into DB. Task %s won\'t be submitted.'%self.crab_task_name
137	spiga	1.27	elif ret == 12:
138	slacapra	1.38	# 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	spiga	1.27	elif ret == 20:
142	slacapra	1.38	# failed to push message in PA
143	farinafa	1.50	logMsg = 'Server unable to release messages to other components. Task %s won\'t be submitted.'%self.crab_task_name
144	spiga	1.27	elif ret == 22:
145	slacapra	1.38	# failed SOAP communication
146			logMsg = 'Error during SOAP communication with server %s'%self.serverName
147	spiga	1.27	elif ret == 33:
148	slacapra	1.38	# 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	spiga	1.27	else:
152	ewv	1.45	logMsg = 'Unexpected return code from server %s: %d'%(self.serverName, ret)
153	farinafa	1.1
154	spiga	1.27	# print loggings
155			if logMsg != '':
156	slacapra	1.38	# reset server choice
157			opsToBeSaved={'serverName' : '' }
158			common._db.updateTask_(opsToBeSaved)
159	ewv	1.45	common.logger.info(logMsg)
160	farinafa	1.1	return ret
161	ewv	1.45
162	farinafa	1.2	def subsequentJobSubmit(self, blTaskName, rng):
163	farinafa	1.1	"""
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	farinafa	1.2	return self._genericCommand('submit', blTaskName, rng)
173	farinafa	1.1
174	farinafa	1.2	def killJobs(self, blTaskName, rng):
175	farinafa	1.1	"""
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	farinafa	1.2	return self._genericCommand('kill', blTaskName, rng)
184	farinafa	1.1
185	riahi	1.52	def StopWorkflow(self, blTaskName):
186			"""
187			_StopWorkflow_
188			"""
189			return self._genericCommand('StopWorkflow', blTaskName, 'all')
190
191	farinafa	1.42	def cleanTask(self, blTaskName):
192	farinafa	1.1	"""
193	farinafa	1.42	_cleanTask_
194	farinafa	1.1	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	farinafa	1.42	return self._genericCommand('clean', blTaskName, 'all')
201	farinafa	1.1
202	farinafa	1.23	def getStatus(self, blTaskName, statusFile=None, statusFamilyType='status'):
203	farinafa	1.1	"""
204			_getStatus_
205	farinafa	1.23	Retrieve the task status from the server. It can recover any kind of status (version, loggingInfos,...)
206			"""
207	farinafa	1.1
208			# fill the filename
209			filename = str(statusFile)
210
211	farinafa	1.2	if not blTaskName:
212	farinafa	1.1	raise CrabException('Exception while getting the task unique name')
213			return ''
214
215			# get the data and fill the file content
216	farinafa	1.23	statusMsg = self.asSession.getTaskStatus(statusFamilyType, blTaskName)
217	farinafa	1.2	if 'Error:' in statusMsg[:6] or len(statusMsg)==0:
218	farinafa	1.58	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	farinafa	1.2
224			if statusFile is not None:
225			f = open(statusFile, 'w')
226			f.write(statusMsg)
227			f.close()
228	ewv	1.45	return statusFile
229	farinafa	1.2	return statusMsg
230	farinafa	1.1
231	farinafa	1.8	def outputRetrieved(self, blTaskName, rng):
232	farinafa	1.1	"""
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	farinafa	1.8	return self._genericCommand('outputRetrieved', blTaskName, rng)
241	farinafa	1.1
242	farinafa	1.49	def checkDrainMode(self, blTaskName='null'):
243			return self.getStatus( blTaskName, statusFamilyType='isServerDrained')
244
245	farinafa	1.2	def postMortemInfos(self, blTaskName, rng):
246	farinafa	1.1	"""
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	ewv	1.45	# get the status in
255	farinafa	1.1	raise NotImplementedError
256			return None
257
258			###################################################
259			# Auxiliary methods
260			###################################################
261	ewv	1.45
262	spiga	1.74	def _createXMLcommand(self, taskUName, cmdSpec='status', rng='all', newTaskAddIns=False, flavour='analysis', type='fullySpecified',jobs='-1'):
263	fanzago	1.73
264	farinafa	1.1	xmlString = ''
265			cfile = minidom.Document()
266	ewv	1.45
267	spiga	1.27	ver = common.prog_version_str
268	farinafa	1.1	node = cfile.createElement("TaskCommand")
269			node.setAttribute("Task", str(taskUName) )
270			node.setAttribute("Subject", str(self.userSubj) )
271			node.setAttribute("Command", str(cmdSpec) )
272			node.setAttribute("Range", str(rng) )
273	spiga	1.29	node.setAttribute("TotJob", str(jobs) )
274	ewv	1.45	node.setAttribute("Scheduler", str(self.cfg_params['CRAB.scheduler']) )
275	farinafa	1.24	node.setAttribute("Flavour", str(flavour) )
276	ewv	1.45	node.setAttribute("Type", str(type) )
277			node.setAttribute("ClientVersion", str(ver) )
278	farinafa	1.10
279	spiga	1.30	## Only Temporary. it should be at Server level
280	spiga	1.55	removeBList = self.cfg_params.get("GRID.remove_default_blacklist", 0 )
281	spiga	1.57	blackAnaOps = None
282	spiga	1.55	if int(removeBList) == 0:
283	spiga	1.56	blacklist = Downloader("http://cmsdoc.cern.ch/cms/LCG/crab/config/")
284	spiga	1.55	result = blacklist.config("site_black_list.conf")
285			if result != None:
286			blackAnaOps = result
287			common.logger.debug("Enforced black list: %s "%str(blacklist))
288			else:
289			common.logger.info("WARNING: Skipping default black list!")
290	spiga	1.30
291	farinafa	1.62	try:
292			myproxyserver = Downloader("http://cmsdoc.cern.ch/cms/LCG/crab/config/").config("myproxy_server.conf")
293	farinafa	1.63	myproxyserver = myproxyserver.strip()
294	farinafa	1.62	if myproxyserver is None:
295			raise CrabException("myproxy_server.conf retrieved but empty")
296			except Exception, e:
297			common.logger.info("Problem setting myproxy server endpoint: using myproxy.cern.ch")
298			common.logger.debug(e)
299			myproxyserver = 'myproxy.cern.ch'
300
301	farinafa	1.1	# create a mini-cfg to be transfered to the server
302			miniCfg = {}
303	farinafa	1.3
304			## migrate CE/SE infos
305	spiga	1.37	miniCfg['EDG.ce_white_list'] = ""
306	spiga	1.36	if 'GRID.ce_white_list' in self.cfg_params:
307	spiga	1.37	miniCfg['EDG.ce_white_list'] = str( self.cfg_params['GRID.ce_white_list'] )
308	farinafa	1.1
309	spiga	1.55	miniCfg['EDG.ce_black_list'] = blackAnaOps
310	spiga	1.36	if 'GRID.ce_black_list' in self.cfg_params:
311	spiga	1.57	if blackAnaOps:
312	spiga	1.37	miniCfg['EDG.ce_black_list'] += ", "
313			miniCfg['EDG.ce_black_list'] += str( self.cfg_params['GRID.ce_black_list'] )
314	farinafa	1.1
315	spiga	1.37	miniCfg['EDG.se_white_list'] = ""
316	spiga	1.36	if 'GRID.se_white_list' in self.cfg_params:
317	spiga	1.37	miniCfg['EDG.se_white_list'] = str( self.cfg_params['GRID.se_white_list'] )
318	spiga	1.36
319	spiga	1.37	miniCfg['EDG.se_black_list'] = ""
320	spiga	1.36	if 'GRID.se_black_list' in self.cfg_params:
321	spiga	1.37	miniCfg['EDG.se_black_list'] = str( self.cfg_params['GRID.se_black_list'] )
322	spiga	1.36
323	spiga	1.37	miniCfg['EDG.group'] = ""
324	spiga	1.36	if 'GRID.group' in self.cfg_params:
325	spiga	1.37	miniCfg['EDG.group'] = str( self.cfg_params['GRID.group'] )
326	spiga	1.36
327	spiga	1.37	miniCfg['EDG.role'] = ""
328	spiga	1.36	if 'GRID.role' in self.cfg_params:
329	spiga	1.37	miniCfg['EDG.role'] = str( self.cfg_params['GRID.role'] )
330	spiga	1.33
331	ewv	1.45	miniCfg['cfgFileNameCkSum'] = makeCksum(common.work_space.cfgFileName())
332	farinafa	1.1	if 'cfgFileNameCkSum' in self.cfg_params:
333			miniCfg['cfgFileNameCkSum'] = str(self.cfg_params['cfgFileNameCkSum'])
334
335			miniCfg['CRAB.se_remote_dir'] = ''
336			if 'CRAB.se_remote_dir' in self.cfg_params:
337	ewv	1.45	miniCfg['CRAB.se_remote_dir'] = str(self.cfg_params['CRAB.se_remote_dir'])
338	farinafa	1.1
339	spiga	1.41	miniCfg['CAF.queue'] = self.cfg_params.get('CAF.queue','cmscaf1nw')
340			miniCfg['CAF.resources'] = self.cfg_params.get('CAF.resource', 'cmscaf')
341	spiga	1.61	miniCfg['CAF.group'] = self.cfg_params.get('CAF.group', None)
342	farinafa	1.3	## JDL requirements specific data. Scheduler dependant
343	spiga	1.37	miniCfg['EDG.max_wall_time'] = self.cfg_params.get('GRID.max_wall_clock_time', None)
344			miniCfg['EDG.max_cpu_time'] = self.cfg_params.get('GRID.max_cpu_time', '130')
345	farinafa	1.62	miniCfg['proxyServer'] = myproxyserver
346	spiga	1.36	miniCfg['VO'] = self.cfg_params.get('GRID.virtual_organization', 'cms')
347	spiga	1.64	miniCfg['EDG_retry_count'] = 0
348	farinafa	1.70	miniCfg['EDG_shallow_retry_count'] = -1
349	farinafa	1.13
350	farinafa	1.14	## Additional field for DashBoard
351			miniCfg['CMSSW.datasetpath'] = self.cfg_params.get('CMSSW.datasetpath', 'None')
352
353			## Additional fields for Notification by the server
354	farinafa	1.15	miniCfg['eMail'] = self.cfg_params.get('USER.email', None)
355	farinafa	1.20	miniCfg['threshold'] = self.cfg_params.get('USER.thresholdlevel', 100)
356	farinafa	1.3
357	spiga	1.34	miniCfg['CMSSW_version'] = self.scram.getSWVersion()
358	spiga	1.48	#WMBS
359			miniCfg['feeder'] = self.cfg_params.get('WMBS.feeder', 'Feeder')
360	riahi	1.68	miniCfg['parent_task'] = self.cfg_params.get('WMBS.parent_task', '')
361	riahi	1.53
362			miniCfg['processing'] = self.cfg_params.get('WMBS.processing', 'bulk')
363	riahi	1.59	miniCfg['startrun'] = self.cfg_params.get('WMBS.startrun', 'None')
364	riahi	1.53
365	spiga	1.48	miniCfg['splitting_algorithm'] = self.cfg_params.get('WMBS.splitting_algorithm','FileBased')
366	riahi	1.51	miniCfg['split_per_job'] = self.cfg_params.get('WMBS.split_per_job','files_per_job')
367	riahi	1.68	miniCfg['split_value'] = self.cfg_params.get('WMBS.split_value',1000000)
368	spiga	1.48
369			miniCfg['CMSSW_version'] = self.scram.getSWVersion()
370	ewv	1.45
371	farinafa	1.1	## put here other fields if needed
372			node.setAttribute("CfgParamDict", str(miniCfg) )
373			cfile.appendChild(node)
374			xmlString += str(cfile.toprettyxml())
375			return xmlString
376
377	farinafa	1.2	def _genericCommand(self, cmd, blTaskName, rng):
378			if not blTaskName:
379	farinafa	1.1	raise CrabException('Error while extracting the Task Unique Name string')
380			return -2
381
382			cmdXML = None
383	farinafa	1.2	cmdXML = self._createXMLcommand(blTaskName, cmd, rng)
384	farinafa	1.1	if not cmdXML:
385			raise CrabException('Error while creating the Command XML')
386			return -3
387	ewv	1.45
388	farinafa	1.1	ret = -1
389	farinafa	1.2	ret = self.asSession.sendCommand(cmdXML, blTaskName)
390	farinafa	1.1	logMsg = ''
391	ewv	1.45	debugMsg = ''
392	farinafa	1.1	if ret == 0:
393			# success
394	farinafa	1.22	debugMsg = 'Command successfully sent to server %s for task %s'%(self.serverName, self.crab_task_name)
395	farinafa	1.1	else:
396	spiga	1.27	self.checkServerResponse(ret)
397	farinafa	1.1	return ret
398	ewv	1.45