pandaharvester/harvesterstager/dummy_stager.py

0001 import uuid
0002
0003 from pandaharvester.harvestercore import core_utils
0004
0005 from .base_stager import BaseStager
0006
0007 # logger
0008 _logger = core_utils.setup_logger("dummy_stager")
0009
0010
0011 # dummy plugin for stager
0012 class DummyStager(BaseStager):
0013     # constructor
0014     def __init__(self, **kwarg):
0015         BaseStager.__init__(self, **kwarg)
0016
0017     # check status
0018     def check_stage_out_status(self, jobspec):
0019         """Check the status of stage-out procedure. If staging-out is done synchronously in trigger_stage_out
0020         this method should always return True.
0021         Output files are available through jobspec.get_outfile_specs(skip_done=False) which gives
0022         a list of FileSpecs not yet done.
0023         FileSpec.attemptNr shows how many times the transfer was checked for the file.
0024         If the file was successfully transferred, status should be set to 'finished'.
0025         Or 'failed', if the file failed to be transferred. Once files are set to 'finished' or 'failed',
0026         jobspec.get_outfile_specs(skip_done=False) ignores them.
0027
0028         :param jobspec: job specifications
0029         :type jobspec: JobSpec
0030         :return: A tuple of return code (True: transfer success, False: fatal transfer failure,
0031                  None: on-going or temporary failure) and error dialog
0032         :rtype: (bool, string)
0033         """
0034         for fileSpec in jobspec.get_output_file_specs(skip_done=True):
0035             fileSpec.status = "finished"
0036         return True, ""
0037
0038     # trigger stage out
0039     def trigger_stage_out(self, jobspec):
0040         """Trigger the stage-out procedure for the job.
0041         Output files are available through jobspec.get_outfile_specs(skip_done=False) which gives
0042         a list of FileSpecs not yet done.
0043         FileSpec.attemptNr shows how many times transfer was tried for the file so far.
0044
0045         :param jobspec: job specifications
0046         :type jobspec: JobSpec
0047         :return: A tuple of return code (True: success, False: fatal failure, None: temporary failure)
0048                  and error dialog
0049         :rtype: (bool, string)
0050         """
0051         for fileSpec in jobspec.get_output_file_specs(skip_done=True):
0052             # fileSpec.objstoreID = 123
0053             # fileSpec.fileAttributes['guid']
0054             pass
0055         return True, ""
0056
0057     # zip output files
0058     def zip_output(self, jobspec):
0059         """OBSOLETE : zip functions should be implemented in zipper plugins.
0060         Zip output files. This method loops over jobspec.outFiles, which is a list of zip file's FileSpecs,
0061         to make a zip file for each zip file's FileSpec. FileSpec.associatedFiles is a list of FileSpecs of
0062         associated files to be zipped. The path of each associated file is available in associated
0063         file's FileSpec.path. Once zip files are made, their FileSpec.path, FileSpec.fsize and
0064         FileSpec.chksum need to be set.
0065
0066         :param jobspec: job specifications
0067         :type jobspec: JobSpec
0068         :return: A tuple of return code (True for success, False otherwise) and error dialog
0069         :rtype: (bool, string)
0070         """
0071         # make logger
0072         tmpLog = self.make_logger(_logger, f"PandaID={jobspec.PandaID}", method_name="zip_output")
0073         return self.simple_zip_output(jobspec, tmpLog)
0074
0075     # asynchronous zip output
0076     def async_zip_output(self, jobspec):
0077         """OBSOLETE : zip functions should be implemented in zipper plugins.
0078         Zip output files asynchronously. This method is followed by post_zip_output(),
0079         which is typically useful to trigger an asynchronous zipping mechanism such as batch job.
0080         This method loops over jobspec.outFiles, which is a list of zip file's FileSpecs, to make
0081         a zip file for each zip file's FileSpec. FileSpec.associatedFiles is a list of FileSpecs
0082         of associated files to be zipped. The path of each associated file is available in associated
0083         file's FileSpec.path.
0084
0085         :param jobspec: job specifications
0086         :type jobspec: JobSpec
0087         :return: A tuple of return code (True for success, False otherwise) and error dialog
0088         :rtype: (bool, string)
0089         """
0090         # make logger
0091         tmpLog = self.make_logger(_logger, f"PandaID={jobspec.PandaID}", method_name="zip_output")
0092         # set some ID which can be used for lookup in post_zip_output()
0093         groupID = str(uuid.uuid4())
0094         lfns = []
0095         for fileSpec in jobspec.outFiles:
0096             lfns.append(fileSpec.lfn)
0097         jobspec.set_groups_to_files({groupID: {"lfns": lfns, "groupStatus": "zipping"}})
0098         return True, ""
0099
0100     # post zipping
0101     def post_zip_output(self, jobspec):
0102         """OBSOLETE : zip functions should be implemented in zipper plugins.
0103         This method is executed after async_zip_output(), to do post-processing for zipping.
0104         Once zip files are made, this method needs to look over jobspec.outFiles to set their
0105         FileSpec.path, FileSpec.fsize, and FileSpec.chksum.
0106
0107         :param jobspec: job specifications
0108         :type jobspec: JobSpec
0109         :return: A tuple of return code (True for success, False otherwise) and error dialog
0110         :rtype: (bool, string)
0111         """
0112         # make logger
0113         tmpLog = self.make_logger(_logger, f"PandaID={jobspec.PandaID}", method_name="zip_output")
0114         # get groups for lookup
0115         groups = jobspec.get_groups_of_output_files()
0116         # do something with groupIDs
0117         pass
0118         # update file attributes
0119         for fileSpec in jobspec.outFiles:
0120             fileSpec.path = "/path/to/zip"
0121             fileSpec.fsize = 12345
0122             fileSpec.chksum = "66bb0985"
0123         return True, ""