Merge pull request #174 from PanDAWMS/next

3.10.4.12

Author: PalNilsson

PILOTVERSION

@@ -1 +1 @@
-3.10.3.82
+3.10.4.12

pilot.py

@@ -45,7 +45,7 @@
     shell_exit_code,
 )
 from pilot.util.batchsystem import is_htcondor_version_sufficient
-# from pilot.util.cgroups import create_cgroup
+from pilot.util.cgroups import create_cgroup
 from pilot.util.config import config
 from pilot.util.constants import (
     get_pilot_version,
@@ -811,8 +811,7 @@ def set_environment_variables():
 
     # create a cgroup for the pilot
     if pilot_cache.use_cgroups:
-        pass
-        # _ = create_cgroup()
+        _ = create_cgroup()
 
 
 def wrap_up() -> int:
@@ -961,7 +960,7 @@ def send_workernode_map(
     try:
         data = get_workernode_map(site)
     except Exception as e:
-        logger.warning(f"exception caught: {e}")
+        logger.warning(f"exception caught when calling get_workernode_map(): {e}")
     else:
         send_update("api/v1/pilot/update_worker_node", data, url, port, ipv=internet_protocol_version, max_attempts=1)
 

pilot/api/data.py

@@ -1086,8 +1086,8 @@ def prepare_destinations(self, files: list, activities: list or str, alt_exclude
         :param alt_exclude: global list of destinations that should be excluded / not used for alternative stage-out
         :return: updated fspec entries (list).
         """
-        if alt_exclude is None:  # to bypass pylint complaint if declared as [] above
-            alt_exclude = []
+
+        alt_exclude = list(alt_exclude or [])
 
         if not self.infosys.queuedata:  # infosys is not initialized: not able to fix destination if need, nothing to do
             return files
@@ -1120,7 +1120,7 @@ def resolve_alt_destination(primary, exclude=None):
 
             cur = storages.index(primary) if primary in storages else 0
             inext = (cur + 1) % len(storages)  # cycle storages, take the first elem when reach end
-            exclude = set([primary] + list(exclude if exclude is not None else []))
+            exclude = set([primary] + list(exclude or []))
             alt = None
             for attempt in range(len(exclude) or 1):  # apply several tries to jump exclude entries (in case of dublicated data will stack)
                 inext = (cur + 1) % len(storages)  # cycle storages, start from the beginning when reach end

pilot/common/errorcodes.py

@@ -186,6 +186,7 @@ class ErrorCodes:
     PROXYTOOSHORT = 1382  # used at the beginning of the pilot to indicate that the proxy is too short
     STAGEOUTAUTHENTICATIONFAILURE = 1383
     QUEUENOTSETUPFORCONTAINERS = 1384
+    NOJOBSINPANDA = 1385  # internally used code
 
     _error_messages = {
         GENERALERROR: "General pilot error, consult batch log",
@@ -333,7 +334,7 @@ class ErrorCodes:
         PROXYTOOSHORT: "Proxy is too short",
         STAGEOUTAUTHENTICATIONFAILURE: "Authentication failure during stage-out",
         QUEUENOTSETUPFORCONTAINERS: "Queue is not set up for containers",
-
+        NOJOBSINPANDA: "No jobs in PanDA",
     }
 
     put_error_codes = [1135, 1136, 1137, 1141, 1152, 1181]
@@ -344,11 +345,11 @@ def reset_pilot_errors(self):
         ErrorCodes.pilot_error_codes = []
         ErrorCodes.pilot_error_diags = []
 
-    def get_kill_signal_error_code(self, signal: str) -> int:
+    def get_kill_signal_error_code(self, signal_name: str) -> int:
         """
         Match a kill signal with a corresponding Pilot error code.
 
-        :param signal: signal name (str).
+        :param signal_name: signal name (str).
         :return: Pilot error code (int).
         """
         signals_dictionary = {
@@ -358,9 +359,10 @@ def get_kill_signal_error_code(self, signal: str) -> int:
             "SIGXCPU": self.SIGXCPU,
             "SIGUSR1": self.SIGUSR1,
             "SIGBUS": self.SIGBUS,
+            "SIGINT": self.SIGINT,
         }
 
-        return signals_dictionary.get(signal, self.KILLSIGNAL)
+        return signals_dictionary.get(signal_name, self.KILLSIGNAL)
 
     def get_error_message(self, errorcode: int) -> str:
         """
@@ -445,13 +447,17 @@ def report_errors(self) -> str:
 
         return report
 
-    def resolve_transform_error(self, exit_code: int, stderr: str) -> int:
+    def resolve_transform_error(self, exit_code: int, stderr: str) -> tuple[int, str]:
         """
         Assign a pilot error code to a specific transform error.
 
-        :param exit_code: transform exit code (int)
-        :param stderr: transform stderr (str)
-        :return: pilot error code (int).
+        Args:
+            exit_code (int): Transform exit code.
+            stderr (str): Transform stderr.
+
+        Returns:
+            int: Pilot error code.
+            str: Error message if extracted from stderr, otherwise an empty string.
         """
         error_map = {
             "Not mounting requested bind point": self.SINGULARITYBINDPOINTFAILURE,
@@ -464,28 +470,51 @@ def resolve_transform_error(self, exit_code: int, stderr: str) -> int:
             "Apptainer is not installed": self.APPTAINERNOTINSTALLED,
             "cannot create directory": self.MKDIR,
             "General payload setup verification error": self.SETUPFAILURE,
+            "No such file or directory": self.NOSUCHFILE,
         }
 
+        def get_key_by_value(d: dict, value: str) -> str:
+            """Return the key corresponding to a given value."""
+            for k, v in d.items():
+                if v == value:
+                    return k
+            return ""
+
         # Check if stderr contains any known error messages
+        apptainer_codes = {
+            self.SINGULARITYBINDPOINTFAILURE,
+            self.SINGULARITYNOLOOPDEVICES,
+            self.SINGULARITYIMAGEMOUNTFAILURE,
+            self.SINGULARITYIMAGEMOUNTFAILURE,
+            self.SINGULARITYGENERALFAILURE,
+            self.SINGULARITYFAILEDUSERNAMESPACE,
+            self.SINGULARITYNOTINSTALLED,
+            self.APPTAINERNOTINSTALLED
+        }
         for error_message, error_code in error_map.items():
             if error_message in stderr:
-                return error_code
+                # only allow overwriting exit code 0 for specific errors (read: apptainer)
+                if exit_code == 0 and error_code in apptainer_codes:
+                    return error_code, error_message
+                else:
+                    continue
 
         # Handle specific exit codes
+        key = get_key_by_value(error_map, exit_code)
         if exit_code == 2:
-            return self.LSETUPTIMEDOUT
+            return self.LSETUPTIMEDOUT, key
         if exit_code == 3:
-            return self.REMOTEFILEOPENTIMEDOUT
+            return self.REMOTEFILEOPENTIMEDOUT, key
         if exit_code == 251:
-            return self.UNKNOWNTRFFAILURE
+            return self.UNKNOWNTRFFAILURE, key
         if exit_code == -1:
-            return self.UNKNOWNTRFFAILURE
+            return self.UNKNOWNTRFFAILURE, key
         if exit_code == self.COMMANDTIMEDOUT:
-            return exit_code
+            return exit_code, key
         if exit_code != 0:
-            return self.PAYLOADEXECUTIONFAILURE
+            return self.PAYLOADEXECUTIONFAILURE, key
 
-        return exit_code  # Return original exit code if no specific error is found
+        return exit_code, key  # Return original exit code if no specific error is found
 
     def extract_stderr_error(self, stderr: str) -> str:
         """

pilot/common/pilotcache.py

@@ -33,7 +33,7 @@ def __init__(self):
             self.use_cgroups = None  # for process management
             self.cgroups = {}  # for process management
             self.proxy_lifetime = 0
-            self.queuedata = {}
+            self.queuedata = None
             self.pilot_version = None
             self.pilot_work_dir = None
             self.pilot_source_dir = None

pilot/control/data.py

@@ -1114,7 +1114,7 @@ def generate_fileinfo(job: JobData) -> dict:
                 f'{checksum_type}': entry.checksum.get(config.File.checksum_type),
                 'surl': entry.turl
             }
-            if entry.is_altstaged:
+            if True or entry.is_altstaged:  # always report output RSE (ATLASPANDA-604)
                 dat['endpoint'] = entry.ddmendpoint
 
             fileinfo[entry.lfn] = dat

pilot/control/job.py

@@ -807,7 +807,8 @@ def get_data_structure(job: Any, state: str, args: Any, xml: str = "", metadata:
         try:
             readfrac = data.get('totRBYTES') / _totalsize
         except (TypeError, ZeroDivisionError) as exc:
-            logger.warning(f"failed to calculate {data.get('totRBYTES')}/{_totalsize}: {exc}")
+            logger.warning(f"failed to calculate totRBYTES / total size of input files = {data.get('totRBYTES')}/{_totalsize}: {exc}")
+            logger.warning('will not report readbyterate')
             readfrac = None
         else:
             readfrac = float_to_rounded_string(readfrac, precision=2)
@@ -1860,11 +1861,76 @@ def get_job_definition(queues: namedtuple, args: object) -> dict or str:
         if abort:
             res = None  # None will trigger 'fatal' error and will finish the pilot
         else:
-            res = get_job_definition_from_server(args, taskid=taskid)
+            if delay_to_get_job():
+                res = None
+            else:
+                res = get_job_definition_from_server(args, taskid=taskid)
 
     return res
 
 
+def get_load_factor() -> float:
+    """
+    Get the pilot load factor.
+    """
+    try:
+        load_factor = os.environ.get('PILOT_LOAD_FACTOR', None)
+        if load_factor:
+            try:
+                load_factor = float(load_factor)
+            except Exception as exc:      # NOQA F841
+                load_factor = None
+        if not load_factor:
+            load_factor = float(config.Pilot.load_factor)
+        return load_factor
+    except Exception as exc:
+        logger.warning(f'delay_to_get_job caught exception: {exc}')
+    return 1.2
+
+
+def is_delay_to_get_job_enabled() -> bool:
+    """
+    To check if it's enabled to delay to get jobs based on load.
+    """
+    try:
+        enabled = os.environ.get('PILOT_ENABLE_DELAY_TO_GET_JOB', 'false').lower() == 'true'
+        if enabled:
+            return enabled
+        enabled = bool(config.Pilot.enable_delay_to_get_job)
+        return enabled
+    except Exception as exc:
+        logger.warning(f'is_delay_to_get_job_enabled caught exception: {exc}')
+    return False
+
+
+def delay_to_get_job() -> bool:
+    """
+    Delay to get job if the machine load is too high.
+
+    This code checks the system's load average and compares it to a calculated maximum load based on the number
+    of CPU cores and a configurable load factor. If the 1-minute load average exceeds the maximum load, it logs
+    the condition and delays job retrieval.
+
+    :return: bool.
+    """
+    try:
+        if not is_delay_to_get_job_enabled():
+            return False
+
+        load1, load5, load15 = os.getloadavg()
+        num_cores = os.cpu_count()
+        load_factor = get_load_factor()
+        max_load = num_cores * load_factor
+        logger.info(f"num_cores: {num_cores}, load_factor: {load_factor}, load1: {load1}, load5: {load5}, load15: {load15}")
+        if load1 > max_load:
+            logger.info(f"load1 ({load1}) > max_load ({max_load}) (num_cores * load_factor). delay to getjob.")
+            return True
+        return False
+    except Exception as exc:
+        logger.warning(f'delay_to_get_job caught exception: {exc}')
+    return False
+
+
 def get_message_from_mb(args: Any) -> dict:
     """
     Get a message from a message broker.
@@ -2222,6 +2288,10 @@ def retrieve(queues: namedtuple, traces: Any, args: object):  # noqa: C901
             getjob_failures += 1
             if getjob_failures >= get_nr_getjob_failures(args.getjob_failures, args.harvester_submitmode):
                 logger.warning(f'did not get a job -- max number of job request failures reached: {getjob_failures} (setting graceful_stop)')
+                if getjob_failures >= 5 and pilot_cache.queuedata.resource_type.lower() == 'hpc':
+                    logger.warning('setting error code to NOJOBSINPANDA on HPC resource')
+                    traces.pilot['error_code'] = errors.NOJOBSINPANDA
+
                 args.graceful_stop.set()
                 break
 
@@ -2239,6 +2309,10 @@ def retrieve(queues: namedtuple, traces: Any, args: object):  # noqa: C901
             getjob_failures += 1
             if getjob_failures >= get_nr_getjob_failures(args.getjob_failures, args.harvester_submitmode):
                 logger.warning(f'did not get a job -- max number of job request failures reached: {getjob_failures}')
+                if getjob_failures >= 5 and pilot_cache.queuedata.resource_type.lower() == 'hpc':
+                    logger.warning('setting error code to NOJOBSINPANDA on HPC resource')
+                    traces.pilot['error_code'] = errors.NOJOBSINPANDA
+
                 args.graceful_stop.set()
                 break
 

pilot/control/payload.py

@@ -273,10 +273,14 @@ def execute_payloads(queues: namedtuple, traces: Any, args: object):  # noqa: C9
             if exit_code and exit_code > 1000:  # pilot error code, add to list
                 logger.warning(f'pilot error code received (code={exit_code}, diagnostics=\n{diagnostics})')
                 job.piloterrorcodes, job.piloterrordiags = errors.add_error_code(exit_code, msg=diagnostics)
+            else:
+                if exit_code >= 0:
+                    job.transexitcode = exit_code % 255
+                else:
+                    logger.warning(f'pilot error code received a negative transform exit code={exit_code} - will not set transexitcode')
 
             logger.debug(f'run() returned exit_code={exit_code}')
             set_cpu_consumption_time(job)
-            job.transexitcode = exit_code % 255
             if out:
                 out.close()
             if err:
@@ -321,6 +325,11 @@ def execute_payloads(queues: namedtuple, traces: Any, args: object):  # noqa: C9
                 else:
                     job.piloterrorcodes, job.piloterrordiags = errors.add_error_code(errors.INTERNALPILOTPROBLEM, msg=error)
 
+            if exit_code < 0 and not job.piloterrorcodes:
+                # exit code < 0 means that the payload was killed, e.g. by a signal
+                logger.warning("it seems the payload was killed but no error code was assigned yet - setting SIGTERM error")
+                job.piloterrorcodes, job.piloterrordiags = errors.add_error_code(errors.SIGTERM)
+
             if job.piloterrorcodes:
                 exit_code_interpret = 1
 
@@ -667,7 +676,7 @@ def set_cpu_consumption_time(job: JobData):
     logger.info(f'CPU consumption time: {cpuconsumptiontime} {job.cpuconsumptionunit} (rounded to {job.cpuconsumptiontime} {job.cpuconsumptionunit})')
 
 
-def perform_initial_payload_error_analysis(job: JobData, exit_code: int):
+def perform_initial_payload_error_analysis(job: JobData, exit_code: int):  # noqa: C901
     """
     Perform an initial analysis of the payload.
 
@@ -678,6 +687,8 @@ def perform_initial_payload_error_analysis(job: JobData, exit_code: int):
     """
     if exit_code != 0:
         logger.warning(f'main payload execution returned non-zero exit code: {exit_code}')
+    if exit_code < 0:
+        logger.warning("payload was killed (negative exit code)")
 
     # check if the transform has produced an error report
     path = os.path.join(job.workdir, config.Payload.error_report)
@@ -700,7 +711,13 @@ def perform_initial_payload_error_analysis(job: JobData, exit_code: int):
     path = os.path.join(job.workdir, config.Payload.payloadstderr)
     if os.path.exists(path):
         stderr = read_file(path)
-        exit_code = errors.resolve_transform_error(exit_code, stderr)
+        _exit_code, error_message = errors.resolve_transform_error(exit_code, stderr)
+        if error_message:
+            logger.warning(f"found apptainer error in stderr: {error_message}")
+            if exit_code == 0 and _exit_code != 0:
+                logger.warning("will overwrite trf exit code 0 due to previous error")
+        exit_code = _exit_code
+
     else:
         stderr = ''
         logger.info(f'file does not exist: {path}')

pilot/control/payloads/generic.py

@@ -860,7 +860,13 @@ def run(self) -> tuple[int, str]:  # noqa: C901
                         else "General payload setup verification error (check setup logs)"
                     )
                     # check for special errors in the output
-                    exit_code = errors.resolve_transform_error(exit_code, diagnostics)
+                    _exit_code, error_message = errors.resolve_transform_error(exit_code, diagnostics)
+                    if error_message:
+                        logger.warning(f"found apptainer error in stderr: {error_message}")
+                        if exit_code == 0 and _exit_code != 0:
+                            logger.warning("will overwrite trf exit code 0 due to previous error")
+                    exit_code = _exit_code
+
                     diagnostics = errors.format_diagnostics(exit_code, diagnostics)
                     return exit_code, diagnostics
                 if out:

pilot/info/queuedata.py

@@ -63,6 +63,7 @@ class QueueData(BaseData):
     platform = ""     # cmtconfig value
     container_options = ""  # singularity only options? to be reviewed and forced to be a dict (support options for other containers?)
     container_type = {}  # dict of container names by user as a key
+    resource_type = ""  # type of resource, e.g. 'grid', 'hpc'
     copytools = None
     acopytools = None
 
@@ -104,7 +105,7 @@ class QueueData(BaseData):
     _keys = {int: ['timefloor', 'maxwdir', 'pledgedcpu', 'es_stageout_gap',
                    'corecount', 'maxrss', 'maxtime', 'maxinputsize', 'memkillgrace'],
              str: ['name', 'type', 'appdir', 'catchall', 'platform', 'container_options', 'container_type',
-                   'resource', 'state', 'status', 'site', 'environ'],
+                   'resource', 'state', 'status', 'site', 'environ', 'resource_type'],
              dict: ['copytools', 'acopytools', 'astorages', 'aprotocols', 'acopytools_schemas', 'params'],
              bool: ['allow_lan', 'allow_wan', 'direct_access_lan', 'direct_access_wan', 'is_cvmfs', 'use_pcache']
              }