Merge pull request #211 from PanDAWMS/next

3.14.0.22

Author: PalNilsson

.gitignore

@@ -13,3 +13,6 @@ build
 rucio_upload.json
 tools
 doc/_build/
+__pycache__/
+.coverage
+

PILOTVERSION

@@ -1 +1 @@
-3.13.3.3
+3.14.0.22

pilot/common/errorcodes.py

@@ -191,6 +191,7 @@ class ErrorCodes:
     NOJOBSINPANDA = 1385  # internally used code
     PANDAQUEUENOTONLINE = 1386
     ALLOCATIONERROR = 1387
+    XRDACCESSRESTRICTED = 1388  # XRootD [3010] FullyRestricted / proxy scope too narrow
 
     _error_messages = {
         GENERALERROR: "General pilot error, consult batch log",
@@ -286,7 +287,7 @@ class ErrorCodes:
         PAYLOADSIGSEGV: "SIGSEGV: Invalid memory reference or a segmentation fault",
         NONDETERMINISTICDDM: "Failed to construct SURL for non-deterministic ddm (update CRIC)",
         JSONRETRIEVALTIMEOUT: "JSON retrieval timed out",
-        MISSINGINPUTFILE: "Input file is missing in storage element",
+        MISSINGINPUTFILE: "Input file missing on storage",
         BLACKHOLE: "Black hole detected in file system (consult Pilot log)",
         NOREMOTESPACE: "No space left on device",
         SETUPFATAL: "Setup failed with a fatal exception (consult Payload log)",
@@ -342,6 +343,7 @@ class ErrorCodes:
         NOJOBSINPANDA: "No jobs in PanDA",
         PANDAQUEUENOTONLINE: "PanDA queue is not online",
         ALLOCATIONERROR: "Failed to allocate memory for transform execution (cling JIT failure)",
+        XRDACCESSRESTRICTED: "XRootD access restricted: authorisation denied (proxy scope too narrow)",
     }
 
     put_error_codes = [1135, 1136, 1137, 1141, 1152, 1181]
@@ -487,34 +489,53 @@ def resolve_transform_error(self, exit_code: int, stderr: str) -> tuple[int, str
             "No such file or directory": self.NOSUCHFILE,
         }
 
+        # Apptainer CLI version-incompatibility patterns: ALRB's
+        # apptainerFunctions.sh probes the binary at job start with
+        # 'apptainer buildcfg ...' purely to detect CLI capabilities, using
+        # flags (e.g. -B) that some apptainer builds' 'buildcfg' subcommand
+        # does not accept. This is a *different* subcommand from the one
+        # that actually launches the payload container ('apptainer exec'),
+        # which commonly does accept those same flags. Confirmed in
+        # production: this probe failed with "unknown shorthand flag" while
+        # the job's container started normally afterwards and the payload
+        # completed with trf exit code 0. So, unlike the patterns above,
+        # these two do not reliably indicate that the container failed to
+        # start, and must not override an already-successful (exit_code=0)
+        # result. They are only trusted as a genuine failure signal when the
+        # transform itself already reported a non-zero exit code, in which
+        # case they still provide a more specific diagnostic than the
+        # generic PAYLOADEXECUTIONFAILURE fallback.
+        ambiguous_apptainer_patterns = {
+            "unknown shorthand flag": self.SINGULARITYGENERALFAILURE,
+            "unknown flag:": self.SINGULARITYGENERALFAILURE,
+        }
+
         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
-        }
+        # Check if stderr contains any known error messages.
+        # Return immediately on the first match: the matched pattern is
+        # authoritative regardless of the numeric exit code.  (The previous
+        # guard "only return when exit_code == 0" meant that any non-zero
+        # exit code with a recognisable apptainer pattern fell through to the
+        # generic PAYLOADEXECUTIONFAILURE fallback below.)
         for error_message, error_code in error_map.items():
             if error_message in stderr:
-                # 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
+
+        # These patterns are only authoritative when the transform already
+        # failed (see comment above) - do not let them override exit_code=0.
+        if exit_code != 0:
+            for error_message, error_code in ambiguous_apptainer_patterns.items():
+                if error_message in stderr:
                     return error_code, error_message
-                else:
-                    continue
 
         # Handle specific exit codes
-        key = get_key_by_value(error_map, exit_code)
+        key = get_key_by_value({**error_map, **ambiguous_apptainer_patterns}, exit_code)
         if exit_code == 2:
             return self.LSETUPTIMEDOUT, key
         if exit_code == 3:

pilot/control/job.py

@@ -109,6 +109,7 @@
     SERVER_UPDATE_NOT_DONE
 )
 from pilot.util.container import execute
+from pilot.util.features import MachineFeatures
 from pilot.util.filehandling import (
     copy,
     create_symlink,
@@ -182,6 +183,12 @@
 logger = logging.getLogger(__name__)
 pilot_cache = get_pilot_cache()
 
+# Minimum time (s) that must remain before the MachineFeatures shutdowntime for the pilot to
+# fetch a new job. Chosen to comfortably exceed the monitor's own shutdowntime grace window
+# (10 * 60 = 600s, see pilot.control.monitor.run_shutdowntime_minute_check()) plus typical
+# job setup/stage-in/stage-out overhead, so a freshly-started job is not immediately killed.
+MIN_TIME_FOR_NEW_JOB = 1800
+
 
 def control(queues: namedtuple, traces: Any, args: object) -> None:
     """Set up job control threads.
@@ -760,7 +767,13 @@ def send_state(job: Any, args: Any, state: str, xml: str = "", metadata: str = "
 
     # Backchannel only makes sense on accepted responses
     if result.response is not None:
-        handle_backchannel_command(result.response, job, args, test_tobekilled=test_tobekilled)
+        # note: the api/v1 endpoint nests backchannel fields (command, pilotSecrets) inside
+        # response['data'], e.g. {'success': True, 'data': {'command': 'tobekilled', ..}}.
+        # Normalize the response before dispatching so that handle_backchannel_command() can
+        # find these fields regardless of whether the enveloped or legacy flat shape was returned
+        # (fixes: tobekilled/debug/softkill/nocleanup/pilotSecrets silently ignored for api/v1 responses).
+        backchannel_data = extract_backchannel_data(result.response)
+        handle_backchannel_command(backchannel_data, job, args, test_tobekilled=test_tobekilled)
 
     if final:
         os.environ["SERVER_UPDATE"] = SERVER_UPDATE_FINAL
@@ -811,11 +824,45 @@ def get_debug_command(cmd: str) -> tuple[bool, str]:
     return debug_mode, debug_command
 
 
+def extract_backchannel_data(res: dict) -> dict:
+    """Normalize a PanDA server update response into a flat backchannel dict.
+
+    The current REST API (``api/v1/pilot/update_job``) nests backchannel fields such as
+    ``command`` and ``pilotSecrets`` inside ``res['data']``, e.g.
+    ``{'success': True, 'message': '', 'data': {'StatusCode': 0, 'command': 'tobekilled'}}``.
+    Older/legacy server responses returned these fields directly at the top level. This
+    function merges both layers into a single flat dict so that
+    :func:`handle_backchannel_command` can look fields up in one place regardless of which
+    response shape was actually received.
+
+    Nested ``data`` fields take precedence over top-level fields on key collisions, since
+    ``data`` reflects the current API format.
+
+    Args:
+        res: raw server response (either the enveloped api/v1 shape or a legacy flat dict).
+
+    Returns:
+        dict: flattened dict containing both top-level and formerly-nested-under-'data' keys,
+            ready for backchannel command lookups.
+    """
+    if not isinstance(res, dict):
+        return {}
+
+    merged = dict(res)
+    data = res.get('data')
+    if isinstance(data, dict):
+        merged.update(data)
+
+    return merged
+
+
 def handle_backchannel_command(res: dict, job: Any, args: Any, test_tobekilled: bool = False) -> None:
     """Check if the server update contain any backchannel information. If so, update the job object.
 
     Args:
-        res: server response.
+        res: normalized server response (see extract_backchannel_data()) - a flat dict in which
+            'command' and 'pilotSecrets', if present, are looked up at the top level regardless
+            of whether the original server response nested them under 'data'.
         job: job object.
         args: pilot args object.
         test_tobekilled: emulate a tobekilled command.
@@ -1794,8 +1841,51 @@ def get_dispatcher_dictionary(args: Any, taskid: str = "") -> dict:
     return data
 
 
+def _time_until_shutdown(args: Any) -> Optional[int]:
+    """Return the number of seconds until the MachineFeatures shutdowntime, if known.
+
+    This mirrors the shutdowntime lookup performed by
+    pilot.control.monitor.run_shutdowntime_minute_check(), but is used here to decide
+    whether it is worth fetching a new job at all -- not to abort an already running one.
+
+    Args:
+        args: pilot arguments (used to determine time since pilot start, for the same
+            staleness check applied in run_shutdowntime_minute_check()).
+
+    Returns:
+        Optional[int]: seconds remaining until shutdowntime, or None if shutdowntime is not
+            set, not parseable, or refers to a time before the pilot started (stale value).
+    """
+    machinefeatures = MachineFeatures().get()
+    if not machinefeatures:
+        return None
+
+    _shutdowntime = machinefeatures.get('shutdowntime', None)
+    if not _shutdowntime:
+        return None
+
+    try:
+        shutdowntime = int(_shutdowntime)
+    except (TypeError, ValueError) as exc:
+        logger.warning(f'failed to convert shutdowntime: {exc}')
+        return None
+
+    now = int(time.time())
+    time_since_start = get_time_since_start(args)
+
+    # ignore shutdowntime if it predates pilot start (stale value) -- same convention as
+    # pilot.control.monitor.run_shutdowntime_minute_check()
+    if shutdowntime < (now - time_since_start):
+        logger.debug(f'shutdowntime ({shutdowntime}) was set before pilot started - ignoring it '
+                     f'(now - time since start = {now - time_since_start})')
+        return None
+
+    return shutdowntime - now
+
+
 def proceed_with_getjob(timefloor: int, starttime: int, jobnumber: int, getjob_requests: int, max_getjob_requests: int,  # noqa: C901
-                        should_update_server: bool, submitmode: str, harvester: bool, verify_proxy: bool, traces: Any) -> bool:  # noqa: C901
+                        should_update_server: bool, submitmode: str, harvester: bool, verify_proxy: bool, traces: Any,  # noqa: C901
+                        args: Any = None) -> bool:
     """Check if we can proceed with getJob.
 
     We may not proceed if we have run out of time (timefloor limit), if the proxy is too short, if disk space is too
@@ -1812,6 +1902,8 @@ def proceed_with_getjob(timefloor: int, starttime: int, jobnumber: int, getjob_r
         harvester: True if Harvester is used, False otherwise. Affects the max number of getjob reads from file.
         verify_proxy: True if the proxy should be verified. False otherwise.
         traces: traces object (to be able to propagate a proxy error all the way back to the wrapper).
+        args: pilot arguments, used to check the MachineFeatures shutdowntime before fetching a
+            new job (optional; the shutdowntime check is skipped if not provided).
 
     Returns:
         bool: True if pilot should proceed with getJob.
@@ -1874,6 +1966,16 @@ def proceed_with_getjob(timefloor: int, starttime: int, jobnumber: int, getjob_r
     if jobnumber > 0:
         logger.info(f'since timefloor={timefloor} s and only {currenttime - starttime} s has passed since launch, pilot can run another job')
 
+    # do not fetch a new job if the node is about to be reclaimed (MachineFeatures shutdowntime) --
+    # not relevant for the first job, which is handled separately by the batch system / wrapper
+    if jobnumber > 0 and args is not None:
+        remaining = _time_until_shutdown(args)
+        if remaining is not None and remaining < MIN_TIME_FOR_NEW_JOB:
+            return wrap_up_quickly(
+                f'insufficient time remaining before node shutdown ({remaining}s < '
+                f'{MIN_TIME_FOR_NEW_JOB}s minimum) - will not fetch another job'
+            )
+
     if harvester and jobnumber > 0:
         # unless it's the first job (which is preplaced in the init dir), instruct Harvester to place another job
         # in the init dir
@@ -2720,6 +2822,7 @@ def _wait_for_harvester_job_definition(timeout: int, poll_interval: float) -> bo
                 args.harvester,
                 args.verify_proxy,
                 traces,
+                args,
             )
         except Exception as exc:
             logger.warning(f"proceed_with_getjob() raised exception: {exc}")