python.trfValidation Namespace Reference

Classes
class	ignorePatterns
	Class of patterns that can be ignored from athena logfiles. More...
class	logFileReport
	A class holding report information from scanning a logfile This is pretty much a virtual class, fill in the specific methods when you know what type of logfile you are dealing with. More...
class	athenaLogFileReport
	Logfile suitable for scanning logfiles with an athena flavour, i.e., lines of the form "SERVICE LOGLEVEL MESSAGE". More...
class	scriptLogFileReport
class	eventMatch
	Small class used for vailiadating event counts between input and output files. More...

Functions
	returnIntegrityOfFile (file, functionName, **kwargs)
	return integrity of file using appropriate validation function @ detail This method returns the integrity of a specified file using a @ specified validation function.
	performStandardFileValidation (dictionary, io, parallelMode=False, multithreadedMode=False)
	perform standard file validation @ detail This method performs standard file validation in either serial or @ parallel and updates file integrity metadata.

Variables
	msg = logging.getLogger(__name__)

Function Documentation

performStandardFileValidation()

performStandardFileValidation	(		dictionary,
			io,
			parallelMode = False,
			multithreadedMode = False )

perform standard file validation @ detail This method performs standard file validation in either serial or @ parallel and updates file integrity metadata.

Definition at line 731 of file trfValidation.py.

def performStandardFileValidation(dictionary, io, parallelMode = False, multithreadedMode=False):
    if io == "output":
        if multithreadedMode:
            os.environ['TRF_MULTITHREADED_VALIDATION'] = 'TRUE'
    if parallelMode is False:
        msg.info('Starting legacy (serial) file validation')
        for (key, arg) in dictionary.items():
            if not isinstance(arg, argFile):
                continue
            if not arg.io == io:
                continue
            if arg.auxiliaryFile:
                continue
            
            msg.info('Validating data type %s...', key)
    
            for fname in arg.value:
                msg.info('Validating file %s...', fname)
    
                if io == "output":
                    msg.info('{0}: Testing corruption...'.format(fname))
                    if arg.getSingleMetadata(fname, 'integrity') is True:
                        msg.info('Corruption test passed.')
                    elif arg.getSingleMetadata(fname, 'integrity') is False:
                        msg.error('Corruption test failed.')
                        raise trfExceptions.TransformValidationException(trfExit.nameToCode('TRF_EXEC_VALIDATION_FAIL'), 'File %s did not pass corruption test' % fname)
                    elif arg.getSingleMetadata(fname, 'integrity') == 'UNDEFINED':
                        msg.info('No corruption test defined.')
                    elif arg.getSingleMetadata(fname, 'integrity') is None:
                        msg.error('Could not check for file integrity')
                        raise trfExceptions.TransformValidationException(trfExit.nameToCode('TRF_EXEC_VALIDATION_FAIL'), 'File %s might be missing' % fname)
                    else:    
                        msg.error('Unknown rc from corruption test.')
                        raise trfExceptions.TransformValidationException(trfExit.nameToCode('TRF_EXEC_VALIDATION_FAIL'), 'File %s did not pass corruption test' % fname)
    
    
                msg.info('{0}: Testing event count...'.format(fname))
                if arg.getSingleMetadata(fname, 'nentries') is not None:
                    msg.info('Event counting test passed ({0!s} events).'.format(arg.getSingleMetadata(fname, 'nentries')))
                else:    
                    msg.error('Event counting test failed.')
                    raise trfExceptions.TransformValidationException(trfExit.nameToCode('TRF_EXEC_VALIDATION_FAIL'), 'File %s did not pass corruption test' % fname)
                  
    
                msg.info('{0}: Checking if guid exists...'.format(fname))
                if arg.getSingleMetadata(fname, 'file_guid') is None:
                    msg.error('Guid could not be determined.')
                    raise trfExceptions.TransformValidationException(trfExit.nameToCode('TRF_EXEC_VALIDATION_FAIL'), 'File %s did not pass corruption test' % fname)
                elif arg.getSingleMetadata(fname, 'file_guid') == 'UNDEFINED':
                    msg.info('Guid not defined.')
                else:
                    msg.info('Guid is %s', arg.getSingleMetadata(fname, 'file_guid'))
        msg.info('Stopping legacy (serial) file validation')
    elif parallelMode is True:
        msg.info('Starting parallel file validation')
        # Create lists of files and args. These lists are to be used with zip in
        # order to check and update file integrity metadata as appropriate.
        fileList = []
        argList = []
        # Create a list of the integrity functions for files.
        integrityFunctionList = []
        # Create a list for collation of file validation jobs for submission to
        # the parallel job processor.
        jobs = []
        msg.debug('Collating list of files for validation')
        for (key, arg) in dictionary.items():
            if not isinstance(arg, argFile):
                continue
            if not arg.io == io:
                continue
            for fname in arg.value:
                msg.debug('Appending file {fileName} to list of files for validation'.format(fileName = str(fname)))
                # Append the current file to the file list.
                fileList.append(fname)
                # Append the current arg to the arg list.
                argList.append(arg)
                # Append the current integrity function name to the integrity
                # function list if it exists. If it does not exist, raise an
                # exception.
                if io == "output":
                    try:
                        integrityFunctionList.append(arg.integrityFunction)
                    except AttributeError as e:
                        errmsg = f'Validation function for file {fname} of type'\
                            f' {type(arg).__name__!r} not available for parallel file validation: {e}'
                        msg.error(errmsg)
                        raise trfExceptions.TransformValidationException(
                            trfExit.nameToCode('TRF_EXEC_VALIDATION_FAIL'), errmsg)
                    # Compose a job for validation of the current file using the
                    # appropriate validation function, which is derived from the
                    # associated data attribute arg.integrityFunction.
                    jobs.append(
                        trfUtils.Job(
                            name = "validation of file {fileName}".format(
                                fileName = str(fname)),
                            workFunction = returnIntegrityOfFile,
                            workFunctionKeywordArguments = {
                                'file': fname,
                                'functionName': arg.integrityFunction,
                                'level': msg.getEffectiveLevel(),
                            },
                            workFunctionTimeout = 600
                        )
                    )
        # Contain the file validation jobs in a job group for submission to the
        # parallel job processor.
        if io == "output":
            jobGroup1 = trfUtils.JobGroup(
                name = "standard file validation",
                jobs = jobs
            )
            # Prepare the parallel job processor.
            parallelJobProcessor1 = trfUtils.ParallelJobProcessor(numberOfProcesses=len(jobs))
            # Submit the file validation jobs to the parallel job processor.
            msg.info('Submitting file validation jobs to parallel job processor')
            parallelJobProcessor1.submit(jobSubmission = jobGroup1)
            resultsList = parallelJobProcessor1.getResults()
            msg.info('Parallel file validation complete')
            # Update file metadata with integrity results using the lists fileList,
            # argList and resultsList.
            msg.info('Processing file integrity results')
            for currentFile, currentArg, currentIntegrityFunction, currentResult in zip(fileList, argList, integrityFunctionList, resultsList):
                msg.info('{IO} file {fileName} has integrity status {integrityStatus} as determined by integrity function {integrityFunction}'.format(
                    IO = str(io),
                    fileName = str(currentFile),
                    integrityStatus = str(currentResult),
                    integrityFunction = str(currentIntegrityFunction)
                ))
                # If the first (Boolean) element of the result tuple for the current
                # file is True, update the integrity metadata. If it is False, raise
                # an exception.
                if currentResult[0] is True:
                    msg.info('Updating integrity metadata for file {fileName}'.format(fileName = str(currentFile)))
                    currentArg._setMetadata(files=[currentFile,], metadataKeys={'integrity': currentResult[0]})
                else:
                    exceptionMessage = "{IO} file validation failure on file {fileName} with integrity status {integrityStatus} as determined by integrity function {integrityFunction}".format(
                        IO = str(io),
                        fileName = str(currentFile),
                        integrityStatus = str(currentResult),
                        integrityFunction = str(currentIntegrityFunction)
                    )
                    msg.error("exception message: {exceptionMessage}".format(
                        exceptionMessage = exceptionMessage
                    ))
                    exitCodeName = 'TRF_OUTPUT_FILE_VALIDATION_FAIL'
                    raise trfExceptions.TransformValidationException(
                        trfExit.nameToCode(exitCodeName),
                        exceptionMessage
                    )
                # Perform a check to determine if the file integrity metadata is
                # correct.
                if currentArg.getSingleMetadata(currentFile, metadataKey = 'integrity', populate = False) == currentResult[0]:
                    msg.debug("file integrity metadata update successful")
                else:
                    msg.error("file integrity metadata update unsuccessful")
 
        metadataKeys = ('nentries', 'file_guid')
        msg.info(f"{', '.join(fileList)}: Checking {', '.join(map(repr, metadataKeys))} ...")
        metadata = {fname: arg.getMetadata(fname, metadataKeys=metadataKeys)[fname]
                    for fname, arg in zip(fileList, argList, strict=True)}
        success = {fname: md for fname, md in metadata.items() if None not in md.values()}
        if len(success):
            lines = [
                f"{fname}: {' '.join(f'{k}={v}' for k, v in md.items())}"
                for fname, md in success.items()
            ]
            msg.info("Checked\n\t" + "\n\t".join(lines))
        if len(success) != len(metadata):
            missing = ", ".join(fname for fname in metadata if fname not in success)
            keys = '" and/or "'.join(metadataKeys)
            errmsg = f'{missing}: Could not determine "{keys}"'
            msg.error(errmsg)
            raise trfExceptions.TransformValidationException(trfExit.nameToCode('TRF_EXEC_VALIDATION_FAIL'), errmsg)
        msg.info('Stopping parallel file validation')
 
 

returnIntegrityOfFile()

returnIntegrityOfFile	(		file,
			functionName,
		**	kwargs )

return integrity of file using appropriate validation function @ detail This method returns the integrity of a specified file using a @ specified validation function.

Definition at line 705 of file trfValidation.py.

def returnIntegrityOfFile(file, functionName, **kwargs):
    try:
        import PyJobTransforms.trfFileValidationFunctions as trfFileValidationFunctions
    except Exception as exception:
        msg.error('Failed to import module PyJobTransforms.trfFileValidationFunctions with error {error}'.format(error = exception))
        raise
 
    import multiprocessing
 
    level = kwargs.get('level')
    if level is not None:
        if level < msg.getEffectiveLevel():
            msg.setLevel(level)
            msg.debug(f"Set logging level of {msg.name!r} to {logging.getLevelName(level)!r}")
 
    msg.debug(f"Current process: {multiprocessing.current_process().name}")
 
    validationFunction = getattr(trfFileValidationFunctions, functionName)
    args = ", ".join(f"{k}={v}" for k, v in kwargs.items())
    msg.debug(f"Calling {validationFunction.__name__}({file}, {args}) ")
    return validationFunction(file, **kwargs)
 
 

Variable Documentation

msg

python.trfValidation.msg = logging.getLogger(__name__)

Definition at line 18 of file trfValidation.py.