Class for a single CrabRequest e This class represents one crab3 task/request. More...

Public Member Functions
def	__init__ (self, taskname="", crab_config="", crabController=None, initUpdate=True, debuglevel="ERROR", datasetpath="", localDir="", outlfn="")
	The object constructor. More...

def	crab_folder (self)

def	crabConfig (self)
	Function to access crab config object or read it if unititalized. More...

def	crabFolder (self)

def	datasetpath (self)

def	handleNoState (self)
	Function to handle Task which received NOSTATE status. More...

def	isData (self)
	Property function to find out if task runs on data. More...

def	readLogArch (self, logArchName)
	Function to read log info from log.tar.gz. More...

def	resubmit_failed (self)
	Function to resubmit failed jobs in tasks. More...

def	test_print (self)

def	update (self)
	Function to update Task in associated Jobs. More...

def	updateJobStats (self, dCacheFileList=None)
	Function to update JobStatistics. More...

Public Attributes
	debug

	failureReason

	finalFiles

	isUpdating

	jobs

	lastUpdate

	localDir

	log

	maxjobnumber

	name

	nComplete

	nCooloff

	nFailed

	nFinished

	nIdle

	nJobs

	nRunning

	nTransferring

	nUnsubmitted

	outlfn

	resubmitCount

	state

	taskId

	totalEvents

	uuid

Private Attributes
	_crabConfig

	_crabFolder

	_datasetpath_default

	_isData

Detailed Description

Class for a single CrabRequest e This class represents one crab3 task/request.

Definition at line 372 of file crabFunctions.py.

Constructor & Destructor Documentation

def crabFunctions.CrabTask.__init__	(	self,
		taskname = `""`,
		crab_config = `""`,
		crabController = `None`,
		initUpdate = `True`,
		debuglevel = `"ERROR"`,
		datasetpath = `""`,
		localDir = `""`,
		outlfn = `""`
	)

The object constructor.

Parameters

self	The object pointer.
taskname	The object pointer.
initUpdate	Flag if crab status should be called when an instance is created

Definition at line 387 of file crabFunctions.py.

                  outlfn = "" ,):
 
         # crab config as a python object should only be used via .config
         self._crabConfig = None
 
         self._crabFolder = None
 
         if taskname:
             self.name = taskname
         else:
             if not crab_config:
                 raise ValueError("Either taskname or crab_config needs to be set")
             if not os.path.exists( crab_config):
                 raise IOError("File %s not found" % crab_config )
             self.name = crab_config
             self.name = self.crabConfig.General.requestName
         self.uuid = uuid.uuid4()
         #~ self.lock = multiprocessing.Lock()
         #setup logging
         self.log = logging.getLogger( 'crabTask' )
         self.log.setLevel(logging._levelNames[ debuglevel ])
         self.jobs = {}
         self.localDir = localDir
         self.outlfn = outlfn
         self.isUpdating = False
         self.taskId = -1
         #variables for statistics
         self.nJobs = 0
         self.state = "NOSTATE"
         self.maxjobnumber = 0
         self.nUnsubmitted   = 0
         self.nIdle = 0
         self.nRunning = 0
         self.nTransferring    = 0
         self.nCooloff    = 0
         self.nFailed    = 0
         self.nFinished    = 0
         self.nComplete    = 0
         self.failureReason = None
         self.lastUpdate = datetime.datetime.now().strftime( "%Y-%m-%d_%H.%M.%S" )
 
         self._isData = None
         self.resubmitCount = 0
 
         self.debug = False
 
         self.finalFiles = []
         self.totalEvents = 0
 
 
         self._datasetpath_default = datasetpath
 
         #start with first updates
         if initUpdate:
             self.update()
             self.updateJobStats()
 

Member Function Documentation

def crabFunctions.CrabTask.crab_folder ( self )

Definition at line 507 of file crabFunctions.py.

Referenced by crabFunctions.CrabTask.update().

     def crab_folder(self):
         return os.path.join( self.crabConfig.General.workArea,
             "crab_" + self.crabConfig.General.requestName)

def crabFunctions.CrabTask.crabConfig ( self )

Function to access crab config object or read it if unititalized.

Parameters

self	CrabTask The object pointer.

Definition at line 465 of file crabFunctions.py.

References crabFunctions.CrabTask._crabConfig, AlignableObjectId::entry.name, preexistingValidation.PreexistingValidation.name, alignment.Alignment.name, XMLProcessor::_loaderBaseConfig.name, genericValidation.GenericValidation.name, h4DSegm.name, TrackerSectorStruct.name, classes.MonitorData.name, MuonGeometrySanityCheckPoint.name, classes.OutputData.name, h2DSegm.name, geometry.Structure.name, plotscripts.SawTeethFunction.name, crabFunctions.CrabTask.name, and hTMaxCell.name.

     def crabConfig( self ):
         if self._crabConfig is None:
             crab = CrabController()
             self._crabConfig = crab.readCrabConfig( self.name )
         return self._crabConfig
 

def crabFunctions.CrabTask.crabFolder ( self )

Definition at line 480 of file crabFunctions.py.

References crabFunctions.CrabTask._crabFolder, AlignableObjectId::entry.name, preexistingValidation.PreexistingValidation.name, alignment.Alignment.name, XMLProcessor::_loaderBaseConfig.name, genericValidation.GenericValidation.name, h4DSegm.name, TrackerSectorStruct.name, classes.MonitorData.name, MuonGeometrySanityCheckPoint.name, classes.OutputData.name, h2DSegm.name, geometry.Structure.name, plotscripts.SawTeethFunction.name, crabFunctions.CrabTask.name, and hTMaxCell.name.

     def crabFolder( self ):
         if not self._crabFolder is None: return self._crabFolder
         crab = CrabController()
         if os.path.exists( os.path.join( self.crabConfig.General.workArea, crab._prepareFoldername( self.name ) ) ):
             self._crabFolder = os.path.join( self.crabConfig.General.workArea, crab._prepareFoldername( self.name ) )
             return self._crabFolder
         alternative_path = os.path.join(os.path.cwd(), crab._prepareFoldername( self.name ) )
         if os.path.exists( alternative_path ):
             self._crabFolder = alternative_path
             return self._crabFolder
         self.log.error( "Unable to find folder for Task")
         return ""
 

def crabFunctions.CrabTask.datasetpath ( self )

Definition at line 472 of file crabFunctions.py.

References crabFunctions.CrabTask._datasetpath_default.

     def datasetpath( self ):
         try:
             return self.crabConfig.Data.inputDataset
         except:
             pass
         return self._datasetpath_default
 

def crabFunctions.CrabTask.handleNoState ( self )

Function to handle Task which received NOSTATE status.

Parameters

self	CrabTask The object pointer.

Definition at line 542 of file crabFunctions.py.

References AlignableObjectId::entry.name, preexistingValidation.PreexistingValidation.name, alignment.Alignment.name, XMLProcessor::_loaderBaseConfig.name, genericValidation.GenericValidation.name, h4DSegm.name, TrackerSectorStruct.name, classes.MonitorData.name, MuonGeometrySanityCheckPoint.name, classes.OutputData.name, h2DSegm.name, geometry.Structure.name, plotscripts.SawTeethFunction.name, crabFunctions.CrabTask.name, hTMaxCell.name, crabFunctions.CrabTask.resubmitCount, CastorLedAnalysis.state, HcalLedAnalysis.state, HcalPedestalAnalysis.state, CastorPedestalAnalysis.state, and crabFunctions.CrabTask.state.

     def handleNoState( self ):
         crab = CrabController()
         if "The CRAB3 server backend could not resubmit your task because the Grid scheduler answered with an error." in task.failureReason:
             # move folder and try it again
             cmd = 'mv %s bak_%s' %(crab._prepareFoldername( self.name ),crab._prepareFoldername( self.name ))
             p = subprocess.Popen(cmd,stdout=subprocess.PIPE, shell=True)#,shell=True,universal_newlines=True)
             (out,err) = p.communicate()
             self.state = "SHEDERR"
             configName = '%s_cfg.py' %(crab._prepareFoldername( self.name ))
             crab.submit( configName )
 
         elif task.failureReason is not None:
             self.state = "ERRHANDLE"
             crab.resubmit( self.name )
         self.resubmitCount += 1
 

def crabFunctions.CrabTask.isData ( self )

Property function to find out if task runs on data.

Parameters

self	CrabTask The object pointer.

Definition at line 448 of file crabFunctions.py.

References crabFunctions.CrabTask._isData.

     def isData( self ):
         if self._isData is None:
             try:
                 test = self.crabConfig.Data.lumiMask
                 self._isData = True
             except:
                 if self.name.startswith( "Data_" ):
                     self._isData = True
                 else:
                     self._isData = False
         return self._isData
 
 

def crabFunctions.CrabTask.readLogArch	(	self,
		logArchName
	)

Function to read log info from log.tar.gz.

Parameters

self	The object pointer.
logArchName	path to the compressed log file

Returns: a dictionary with parsed info

Definition at line 599 of file crabFunctions.py.

References createfilelist.int, edm.print(), and split.

     def readLogArch(self, logArchName):
         JobNumber = logArchName.split("/")[-1].split("_")[1].split(".")[0]
         log = {'readEvents' : 0}
         with tarfile.open( logArchName, "r") as tar:
             try:
                 JobXmlFile = tar.extractfile('FrameworkJobReport-%s.xml' % JobNumber)
                 root = ET.fromstring( JobXmlFile.read() )
                 for child in root:
                     if child.tag == 'InputFile':
                         for subchild in child:
                             if subchild.tag == 'EventsRead':
                                 nEvents = int(subchild.text)
                                 log.update({'readEvents' : nEvents})
                                 break
                         break
             except:
                 print("Can not parse / read %s" % logArchName)
         return log
 

def crabFunctions.CrabTask.resubmit_failed ( self )

Function to resubmit failed jobs in tasks.

Parameters

self	CrabTask The object pointer.

Definition at line 496 of file crabFunctions.py.

References crabFunctions.CrabTask.jobs, crabFunctions.CrabTask.lastUpdate, AlignableObjectId::entry.name, preexistingValidation.PreexistingValidation.name, alignment.Alignment.name, XMLProcessor::_loaderBaseConfig.name, genericValidation.GenericValidation.name, h4DSegm.name, TrackerSectorStruct.name, classes.MonitorData.name, MuonGeometrySanityCheckPoint.name, classes.OutputData.name, h2DSegm.name, geometry.Structure.name, plotscripts.SawTeethFunction.name, crabFunctions.CrabTask.name, and hTMaxCell.name.

     def resubmit_failed( self ):
         failedJobIds = []
         controller =  CrabController()
         for jobkey in self.jobs.keys():
             job = self.jobs[jobkey]
             if job['State'] == 'failed':
                 failedJobIds.append( job['JobIds'][-1] )
         controller.resubmit( self.name, joblist = failedJobIds )
         self.lastUpdate = datetime.datetime.now().strftime( "%Y-%m-%d_%H.%M.%S" )
 

def crabFunctions.CrabTask.test_print ( self )

Definition at line 558 of file crabFunctions.py.

References crabFunctions.CrabTask.uuid.

558 def test_print(self):

559 return self.uuid

crabFunctions.CrabTask.uuid

uuid

Definition: crabFunctions.py:403

crabFunctions.CrabTask.test_print

def test_print(self)

Definition: crabFunctions.py:558

def crabFunctions.CrabTask.update ( self )

Function to update Task in associated Jobs.

Parameters

self	CrabTask The object pointer.

Definition at line 513 of file crabFunctions.py.

     def update(self):
         #~ self.lock.acquire()
         self.log.debug( "Start update for task %s" % self.name )
         self.isUpdating = True
         controller =  CrabController()
         self.state = "UPDATING"
         # check if we should drop this sample due to missing info
 
         self.log.debug( "Try to get status for task" )
         self.state , self.jobs,self.failureReason = controller.status(self.crab_folder)
         self.log.debug( "Found state: %s" % self.state )
         if self.state=="FAILED":
             #try it once more
             time.sleep(2)
             self.state , self.jobs,self.failureReason = controller.status(self.crab_folder)
         self.nJobs = len(self.jobs)
         self.updateJobStats()
         if self.state == "NOSTATE":
             self.log.debug( "Trying to resubmit because of NOSTATE" )
             if self.resubmitCount < 3: self.self.handleNoState()
         # add to db if not
         # Final solution inf state not yet found
         self.isUpdating = False
         self.lastUpdate = datetime.datetime.now().strftime( "%Y-%m-%d_%H.%M.%S" )
         #~ self.lock.release()
 

def crabFunctions.CrabTask.updateJobStats	(	self,
		dCacheFileList = `None`
	)

Function to update JobStatistics.

Parameters

self	The object pointer.
dCacheFilelist	A list of files on the dCache

Definition at line 564 of file crabFunctions.py.

References any(), createfilelist.int, crabFunctions.CrabTask.jobs, AlignableObjectId::entry.name, preexistingValidation.PreexistingValidation.name, alignment.Alignment.name, XMLProcessor::_loaderBaseConfig.name, genericValidation.GenericValidation.name, h4DSegm.name, TrackerSectorStruct.name, classes.MonitorData.name, MuonGeometrySanityCheckPoint.name, classes.OutputData.name, h2DSegm.name, geometry.Structure.name, plotscripts.SawTeethFunction.name, crabFunctions.CrabTask.name, hTMaxCell.name, crabFunctions.CrabTask.nComplete, and edm.print().

Referenced by crabFunctions.CrabTask.update().

     def updateJobStats(self,dCacheFileList = None):
         jobKeys = sorted(self.jobs.keys())
         try:
             intJobkeys = [int(x) for x in jobKeys]
         except:
             print("error parsing job numers to int")
 
         #maxjobnumber = max(intJobkeys)
 
         stateDict = {'unsubmitted':0,'idle':0,'running':0,'transferring':0,'cooloff':0,'failed':0,'finished':0}
         nComplete = 0
 
         # loop through jobs
         for key in jobKeys:
             job = self.jobs[key]
              #check if all completed files are on decache
             for statekey in stateDict.keys():
                 if statekey in job['State']:
                     stateDict[statekey]+=1
                     # check if finished fails are found on dCache if dCacheFilelist is given
                     if dCacheFileList is not None:
                         outputFilename = "%s_%s"%( self.name, key)
                         if 'finished' in statekey and any(outputFilename in s for s in dCacheFileList):
                             nComplete +=1
 
         for state in stateDict:
             attrname = "n" + state.capitalize()
             setattr(self, attrname, stateDict[state])
         self.nComplete = nComplete
 