python.PostProcessing Namespace Reference

Classes
class	PostponeProcessing
class	PostProcessingError
class	PostProcessingStep

Functions
	runPostProcStep (taskman, taskDict, oldStatus, previousSteps, step, postprocLib, jobName)
	doPostProcessing (taskman, taskDict, postprocSteps, postprocLib, forceRun=False, jobName=None)

Variables
str	__author__ = 'Juerg Beringer'
str	__version__ = '$Id $'

Detailed Description

This module defines the generic infrastructure for task postprocessing.

Function Documentation

doPostProcessing()

python.PostProcessing.doPostProcessing	(		taskman,
			taskDict,
			postprocSteps,
			postprocLib,
			forceRun = False,
			jobName = None )

Execute a series of postprocessing steps. The task status is updated before
   and after the postprocessing. Log files are handled by each postprocessing step
   individually. Returns a list of executed postprocessing steps.

Definition at line 55 of file PostProcessing.py.

def doPostProcessing(taskman,taskDict,postprocSteps,postprocLib,forceRun=False,jobName=None):
    """Execute a series of postprocessing steps. The task status is updated before
       and after the postprocessing. Log files are handled by each postprocessing step
       individually. Returns a list of executed postprocessing steps."""
    # NOTE: At present, we don't use per-job postprocessing steps. Should this feature
    #       be removed?
    dsName = taskDict['DSNAME']
    taskName = taskDict['TASKNAME']
    prePostProcStatus = taskDict['STATUS']
    executedSteps = []
 
    # Check if there's anything to do
    if not postprocSteps:
        return []   # nothing to do
 
    #update again from DataBase
    prePostProcStatus = taskman.getStatus(dsName,taskName)
    # Don't run postprocessing if status is already postprocessing
    if prePostProcStatus>=TaskManager.StatusCodes['POSTPROCRUNNING'] and not forceRun:
        print ('Exiting postprocessing without doing anything: task %s/%s status is %s\n' % (dsName,taskName,getKey(TaskManager.StatusCodes,prePostProcStatus)))
        return []
    
    # Start postprocessing
    taskman.setStatus(dsName,taskName,TaskManager.StatusCodes['POSTPROCRUNNING'])
    if jobName:
        print ('Postprocessing for task %s/%s - job %s\nOld status: %s\n' % (dsName,taskName,jobName,getKey(TaskManager.StatusCodes,prePostProcStatus)))
    else:
        print ('Postprocessing for task %s/%s\nOld status: %s\n' % (dsName,taskName,getKey(TaskManager.StatusCodes,prePostProcStatus)))
 
    # Get list of postprocessing status files that we may have to remove later
    if jobName:
        postprocStamps = glob.glob('%s/%s/%s/*.status.POSTPROCESSING' % (dsName,taskName,jobName))
    else:
        postprocStamps = glob.glob('%s/%s/*/*.status.POSTPROCESSING' % (dsName,taskName))
 
    # Do each specified processing step. Postprocessing may be interrupted either deliberately
    # by a postprocessing step or due to errors.
    try:
        for step in postprocSteps:
            executedSteps = runPostProcStep(taskman,taskDict,prePostProcStatus,executedSteps,step,postprocLib,jobName)
 
    except PostponeProcessing as e:
        # Stop postprocessing chain w/o error. New status will be determined below if not
        # specified in the exception.
        if e.newStatus:
            taskman.setStatus(dsName,taskName,e.newStatus)
            print ('%i step(s) completed successfully: ' % len(executedSteps),executedSteps,'\n')
            return executedSteps
 
    except PostProcessingError as e:
        print (e)
        if e.newStatus:
            taskman.setStatus(dsName,taskName,e.newStatus)
            print ('Executed steps: ',e.executedSteps)
            return e.executedSteps
        else:
            if (taskDict['NJOBS_SUBMITTED']+taskDict['NJOBS_RUNNING']) > 0:
                # Not all the jobs have run, so ignore error and postpone to later
                pass
 
            else:
                # All the jobs have run, so nothing new in the future
                taskman.setStatus(dsName,taskName,TaskManager.StatusCodes['POSTPROCFAILED'])
                print ('Executed steps: ',e.executedSteps)
                return e.executedSteps
 
    except Exception as e:
        # Any other postprocessing error. Task status becomes POSTPROCFAILED.
        print (e)
        taskman.setStatus(dsName,taskName,TaskManager.StatusCodes['POSTPROCFAILED'])
        print ('Executed steps: ',executedSteps)
        return executedSteps
 
    else:
        # After successful postprocessing, delete POSTPROCESSING status files
        # and mark corresponding jobs as completed
        for p in postprocStamps:
            os.system('rm -f %s' % p)
            basename = p[:-15]
            os.system('touch %s.COMPLETED' % basename)
 
    # Redetermine current status from files on disk
    a = TaskAnalyzer('.',dsName,taskName)
    if a.analyzeFiles():
        a.updateStatus(taskman)
    else:
        # Postprocessing could have deleted the task and migrated it to tape
        taskman.setDiskStatus(dsName,taskName,TaskManager.OnDiskCodes['DELETED'])
 
    print ('%i step(s) completed successfully: ' % len(executedSteps),executedSteps,'\n')
    return executedSteps
 
 

runPostProcStep()

python.PostProcessing.runPostProcStep	(		taskman,
			taskDict,
			oldStatus,
			previousSteps,
			step,
			postprocLib,
			jobName )

Instantiate and run a single postprocessing step.

Definition at line 42 of file PostProcessing.py.

def runPostProcStep(taskman,taskDict,oldStatus,previousSteps,step,postprocLib,jobName):
    """Instantiate and run a single postprocessing step."""
    if hasattr(postprocLib,step):
        print ('...',step,'\n')
        # First reload taskDict - a previous step might have updated the info
        taskDict = taskman.getTaskDict(taskDict['DSNAME'],taskDict['TASKNAME'])
        postprocClass = getattr(postprocLib,step)(taskman,taskDict,oldStatus,previousSteps,postprocLib,jobName)
        postprocClass.run()
        return postprocClass.executedSteps
    else:
        raise PostProcessingError('ERROR: undefined postprocessing step: '+step,previousSteps,TaskManager.StatusCodes['POSTPROCFAILED'])
 
 

Variable Documentation

__author__

str python.PostProcessing.__author__ = 'Juerg Beringer'

private

Definition at line 8 of file PostProcessing.py.

__version__

str python.PostProcessing.__version__ = '$Id $'

private

Definition at line 9 of file PostProcessing.py.