control/payloads/eventservicemerge.py

0001 #!/usr/bin/env python
0002 # Licensed under the Apache License, Version 2.0 (the "License");
0003 # you may not use this file except in compliance with the License.
0004 # You may obtain a copy of the License at
0005 # http://www.apache.org/licenses/LICENSE-2.0
0006 #
0007 # Authors:
0008 # - Wen Guan, wen.guan@cern.ch, 2018
0009 # - Paul Nilsson, paul.nilsson@cern.ch, 2020-2021
0010
0011 import os
0012
0013 from pilot.control.payloads import generic
0014 from pilot.util.container import execute
0015
0016 import logging
0017 logger = logging.getLogger(__name__)
0018
0019
0020 class Executor(generic.Executor):
0021     def __init__(self, args, job, out, err, traces):
0022         super(Executor, self).__init__(args, job, out, err, traces)
0023
0024     def untar_file(self, lfn, job):
0025
0026         pfn = os.path.join(job.workdir, lfn)
0027         command = "tar -xf %s -C %s" % (pfn, job.workdir)
0028         logger.info("untar file: %s", command)
0029         exit_code, stdout, stderr = execute(command)
0030         logger.info("exit_code: %s, stdout: %s, stderr: %s\n", exit_code, stdout, stderr)
0031
0032     def utility_before_payload(self, job):
0033         """
0034         Functions to run before payload
0035         Note: this function updates job.jobparams (process_writetofile() call)
0036
0037         :param job: job object
0038         """
0039
0040         logger.info("untar input tar files for eventservicemerge job")
0041         for fspec in job.indata:
0042             if fspec.is_tar:
0043                 self.untar_file(fspec.lfn, job)
0044
0045         logger.info("Processing writeToFile for eventservicemerge job")
0046         job.process_writetofile()
0047
0048         super(Executor, self).utility_before_payload(job)