pandaharvester/harvesterstager/rse_direct_stager.py

0001 from pandaharvester.harvestercore import core_utils
0002
0003 from .base_stager import BaseStager
0004
0005 # logger
0006 baseLogger = core_utils.setup_logger("rse_direct_stager")
0007
0008
0009 # stager plugin with RSE + no data motion
0010 class RseDirectStager(BaseStager):
0011     """In the workflow for RseDirectStager, workers directly upload output files to RSE
0012     and thus there is no data motion in Harvester."""
0013
0014     # constructor
0015
0016     def __init__(self, **kwarg):
0017         BaseStager.__init__(self, **kwarg)
0018
0019     # check status
0020     def check_stage_out_status(self, jobspec):
0021         for fileSpec in jobspec.get_output_file_specs(skip_done=True):
0022             fileSpec.status = "finished"
0023         return True, ""
0024
0025     # trigger stage out
0026     def trigger_stage_out(self, jobspec):
0027         return True, ""
0028
0029     # zip output files
0030     def zip_output(self, jobspec):
0031         # make logger
0032         tmpLog = self.make_logger(baseLogger, f"PandaID={jobspec.PandaID}", method_name="zip_output")
0033         return self.simple_zip_output(jobspec, tmpLog)