3.14.2.2

PILOTVERSION

@@ -1 +1 @@
-3.14.1.35
+3.14.2.2

pilot/common/errorcodes.py

@@ -193,6 +193,7 @@ class ErrorCodes:
     ALLOCATIONERROR = 1387
     XRDACCESSRESTRICTED = 1388  # XRootD [3010] FullyRestricted / proxy scope too narrow
     NOTIMELEFTFORNEWJOB = 1389  # set when the pilot ends without running a job since too little time remained
+    SETUPTIMEDOUT = 1390  # the containerised payload setup verification command did not finish within its time limit
 
     _error_messages = {
         GENERALERROR: "General pilot error, consult batch log",
@@ -346,6 +347,7 @@ class ErrorCodes:
         ALLOCATIONERROR: "Failed to allocate memory for transform execution (cling JIT failure)",
         XRDACCESSRESTRICTED: "XRootD access restricted: authorisation denied (proxy scope too narrow)",
         NOTIMELEFTFORNEWJOB: "Insufficient time remaining to start a new job",
+        SETUPTIMEDOUT: "Payload setup verification timed out",
     }
 
     put_error_codes = [1135, 1136, 1137, 1141, 1152, 1181]
@@ -487,8 +489,16 @@ def resolve_transform_error(self, exit_code: int, stderr: str) -> tuple[int, str
             "Singularity is not installed": self.SINGULARITYNOTINSTALLED,
             "Apptainer is not installed": self.APPTAINERNOTINSTALLED,
             "cannot create directory": self.MKDIR,
-            "General payload setup verification error": self.SETUPFAILURE,
         }
+        # Note: "General payload setup verification error" used to be a key in
+        # the map above. It is not an apptainer/singularity message at all, but
+        # the pilot's own placeholder for "the setup verification failed and
+        # there was no output to explain why" - so the pilot ended up
+        # pattern-matching its own text and reporting SETUPFAILURE for it.
+        # Confirmed in production (job 7291003889, 2026-09-02): a setup
+        # verification that was timed out by execute() (COMMANDTIMEDOUT) had its
+        # exit code reclassified to SETUPFAILURE because the placeholder was
+        # passed in as "stderr", hiding the fact that the command had hung.
 
         # Apptainer CLI version-incompatibility patterns: ALRB's
         # apptainerFunctions.sh probes the binary at job start with
@@ -522,13 +532,6 @@ def resolve_transform_error(self, exit_code: int, stderr: str) -> tuple[int, str
             "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.
         # Return immediately on the first match: the matched pattern is
         # authoritative regardless of the numeric exit code.  (The previous
@@ -546,22 +549,30 @@ def get_key_by_value(d: dict, value: str) -> str:
                 if error_message in stderr:
                     return error_code, error_message
 
-        # Handle specific exit codes
-        key = get_key_by_value({**error_map, **ambiguous_apptainer_patterns}, exit_code)
+        # Nothing was found in stderr. The remaining decisions are based on the
+        # numeric exit code alone, so there is no error message to report: the
+        # callers log any returned message as "found apptainer error in stderr",
+        # which must not be said about a string that was never in stderr.
+        # (This used to be a reverse look-up of the pattern maps by error code,
+        # which fabricated a message for every exit code that happened to equal
+        # one of the mapped pilot error codes - e.g. an empty payload.stderr
+        # with exit_code=SETUPFAILURE produced "found apptainer error in
+        # stderr: General payload setup verification error".)
         if exit_code == 2:
-            return self.LSETUPTIMEDOUT, key
+            return self.LSETUPTIMEDOUT, ""
         if exit_code == 3:
-            return self.REMOTEFILEOPENTIMEDOUT, key
-        if exit_code == 251:
-            return self.UNKNOWNTRFFAILURE, key
-        if exit_code == -1:
-            return self.UNKNOWNTRFFAILURE, key
-        if exit_code == self.COMMANDTIMEDOUT:
-            return exit_code, key
+            return self.REMOTEFILEOPENTIMEDOUT, ""
+        if exit_code in {251, -1}:
+            return self.UNKNOWNTRFFAILURE, ""
+        if exit_code >= 1000:
+            # already a pilot error code (e.g. COMMANDTIMEDOUT set by execute()
+            # on a time-out): keep the specific code instead of replacing it
+            # with the generic PAYLOADEXECUTIONFAILURE fallback
+            return exit_code, ""
         if exit_code != 0:
-            return self.PAYLOADEXECUTIONFAILURE, key
+            return self.PAYLOADEXECUTIONFAILURE, ""
 
-        return exit_code, key  # Return original exit code if no specific error is found
+        return exit_code, ""  # Return original exit code if no specific error is found
 
     def extract_stderr_error(self, stderr: str) -> str:
         """Extract the ERROR message from the payload stderr.

pilot/control/job.py

@@ -4224,8 +4224,11 @@ def job_monitor(queues: namedtuple, traces: Any, args: object) -> None:  # noqa:
     # fail-safe to be able to report a job that is still running after the pilot has been ordered to abort
     final_job = None
 
-    # overall loop counter (ignoring the fact that more than one job may be running)
-    n = 0
+    # per-job monitor loop counters: a single job_monitor thread serves every job
+    # of a multi-job pilot, so the counter must be keyed by job id to restart at
+    # #1 for each new payload (it used to be a single counter for the lifetime of
+    # the thread, incremented even while no job was being monitored)
+    loop_counters = {}
     cont = True
     while cont:
 
@@ -4346,7 +4349,9 @@ def job_monitor(queues: namedtuple, traces: Any, args: object) -> None:  # noqa:
                             logger.debug('killing payload processes')
                             kill_processes(jobs[i].pid)
 
-                    logger.info(f"monitor loop #{n}: job {i}:{current_id} is in state \'{jobs[i].state}\'")
+                    loop_counters[current_id] = loop_counters.get(current_id, 0) + 1
+                    logger.info(f"monitor loop #{loop_counters[current_id]}: job {i}:{current_id} "
+                                f"is in state \'{jobs[i].state}\'")
                     if jobs[i].state in {'finished', 'failed'}:
                         logger.info('will abort job monitoring soon since job state=%s (job is still in queue)', jobs[i].state)
                         if args.workflow == 'stager':  # abort interactive stager pilot, this will trigger an abort of all threads
@@ -4425,8 +4430,6 @@ def job_monitor(queues: namedtuple, traces: Any, args: object) -> None:  # noqa:
         elif os.environ.get('PILOT_JOB_STATE') == 'stagein':
             logger.info('job monitoring is waiting for stage-in to finish')
 
-        n += 1
-
         if abort_job:
             logger.warning('cannot recover job monitoring - aborting pilot')
             args.graceful_stop.set()

pilot/control/payload.py

@@ -17,11 +17,11 @@
 # under the License.
 #
 # Authors:
-# - Mario Lassnig, mario.lassnig@cern.ch, 2016-2017
+# - Mario Lassnig, mario.lassnig@cern.ch, 2016-17
 # - Daniel Drizhuk, d.drizhuk@gmail.com, 2017
 # - Tobias Wegner, tobias.wegner@cern.ch, 2017
-# - Paul Nilsson, paul.nilsson@cern.ch, 2017-2026
-# - Wen Guan, wen.guan@cern.ch, 2017-2018
+# - Paul Nilsson, paul.nilsson@cern.ch, 2017-26
+# - Wen Guan, wen.guan@cern.ch, 2017-18
 
 """Functions for handling the payload."""
 
@@ -292,6 +292,12 @@ def execute_payloads(queues: namedtuple, traces: Any, args: object) -> None:  #
                 logger.warning(f'failed to open payload stdout/err: {error}')
                 out = None
                 err = None
+            # set the internal job state as well: send_state() only updates
+            # job.serverstate, so without this the job monitor would report an
+            # empty state until run_payload() sets 'running' - which can be many
+            # minutes later if the setup verification is slow (and never for jobs
+            # without input files, which do not get the 'stagein' state either)
+            set_pilot_state(job=job, state='starting')
             send_state(job, args, 'starting')
 
             # note: when sending a state change to the server, the server might respond with 'tobekilled'

pilot/control/payloads/generic.py

@@ -892,9 +892,55 @@ def resolve_payload_exit_code(exit_code: int, stdout: str, stderr: str) -> int:
         return exit_code
 
     @staticmethod
-    def resolve_setup_verification_result(exit_code: int, stdout: str, diagnostics: str) -> tuple[int, str]:
+    def collect_setup_diagnostics(stdout: str, stderr: str, stdout_filename: str,
+                                  stderr_filename: str) -> tuple[str, str, str]:
+        """Collect all available output from a failed setup verification.
+
+        ``execute()`` returns whatever it captured in memory, which for a
+        timed-out command is an empty stdout plus the reason for the kill in
+        stderr, while the child process writes its own output to
+        ``setup.stdout``/``setup.stderr``. Either source can be empty, so each
+        stream is only read back from file when it is missing from memory:
+        reading both unconditionally overwrote the in-memory stderr - the only
+        description of the failure - with an empty file (the container writes
+        everything to stdout, leaving setup.stderr empty).
+
+        Args:
+            stdout: stdout as returned by ``execute()`` (may be empty).
+            stderr: stderr as returned by ``execute()`` (may be empty).
+            stdout_filename: path to the setup stdout file.
+            stderr_filename: path to the setup stderr file.
+
+        Returns:
+            tuple[str, str, str]: (stdout, stderr, diagnostics), where
+            *diagnostics* is stderr followed by stdout, or a placeholder when
+            no output at all could be found.
+        """
+        if not stdout and os.path.exists(stdout_filename):
+            stdout = read_file(stdout_filename)
+        if not stderr and os.path.exists(stderr_filename):
+            stderr = read_file(stderr_filename)
+
+        diagnostics = stderr + stdout
+        if not diagnostics:
+            # note: this text must not match any pattern in
+            # ErrorCodes.resolve_transform_error(), or the pilot will end up
+            # pattern-matching its own placeholder
+            diagnostics = "setup verification failed without any output (check setup logs)"
+
+        return stdout, stderr, diagnostics
+
+    @staticmethod
+    def resolve_setup_verification_result(exit_code: int, stdout: str, diagnostics: str,
+                                          stderr: str = "") -> tuple[int, str]:
         """Resolve the final exit code/diagnostics for a setup verification run.
 
+        A ``COMMANDTIMEDOUT`` exit code means ``execute()`` had to kill the
+        command because it never finished (e.g. a container startup stalling
+        on CVMFS). Nothing was produced to pattern-match in that case, so the
+        time-out is reported as such (``SETUPTIMEDOUT``) rather than being run
+        through the apptainer/singularity stderr patterns.
+
         The setup verification command (see ``run()``) always appends
         ``echo "Done."`` as its final statement. ALRB separately probes the
         apptainer binary at job start with ``apptainer buildcfg ...``, a
@@ -911,14 +957,30 @@ def resolve_setup_verification_result(exit_code: int, stdout: str, diagnostics:
         Args:
             exit_code: raw exit code from the setup verification command.
             stdout: captured stdout from the setup verification command.
-            diagnostics: combined diagnostic text (typically stdout + stderr).
+            diagnostics: combined diagnostic text (typically stderr + stdout).
+            stderr: captured stderr, used on its own for the time-out
+                diagnostics since it holds the reason the command was killed.
 
         Returns:
             tuple[int, str]: (exit_code, diagnostics), overridden to (0, "")
             when the failure is determined to be the benign buildcfg-probe
             artifact; otherwise the (possibly reclassified) exit code and
             formatted diagnostics.
         """
+        if exit_code == errors.COMMANDTIMEDOUT:
+            logger.warning("the setup verification command was timed out - reporting it as a setup time-out")
+            # keep the diagnostics short: format_diagnostics() truncates to 256
+            # characters and the setup command string alone is longer than that,
+            # which would push the actual time-out out of the reported message
+            found = re.search(r"timed out after ([\d.]+) seconds", stderr or diagnostics)
+            reason = (
+                f"the containerised setup command did not finish within {round(float(found.group(1)))} s "
+                f"and was killed"
+                if found
+                else "the containerised setup command did not finish and was killed"
+            )
+            return errors.SETUPTIMEDOUT, errors.format_diagnostics(errors.SETUPTIMEDOUT, reason)
+
         _exit_code, error_message = errors.resolve_transform_error(exit_code, diagnostics)
         if error_message:
             logger.warning(f"found apptainer error in stderr: {error_message}")
@@ -991,26 +1053,25 @@ def run(self) -> tuple[int, str]:  # noqa: C901
                     job=self.__job,
                     timeout=_setup_verify_timeout,
                 )
-                if exit_code:
-                    logger.warning(f"setup returned exit code={exit_code}")
-                    diagnostics = stderr + stdout if stdout and stderr else ""
-                    if not diagnostics:
-                        stdout = read_file(stdout_filename)
-                        stderr = read_file(stderr_filename)
-                    diagnostics = (
-                        stderr + stdout
-                        if stdout and stderr
-                        else "General payload setup verification error (check setup logs)"
-                    )
-                    exit_code, diagnostics = self.resolve_setup_verification_result(exit_code, stdout, diagnostics)
-                    if exit_code:
-                        return exit_code, diagnostics
+                # the command has finished (or been killed), so close the output
+                # files before they are read back - and before any early return
+                # below, which used to leak both file objects
                 if out:
                     out.close()
                     logger.debug(f"closed {stdout_filename}")
                 if err:
                     err.close()
                     logger.debug(f"closed {stderr_filename}")
+                if exit_code:
+                    logger.warning(f"setup returned exit code={exit_code}")
+                    stdout, stderr, diagnostics = self.collect_setup_diagnostics(
+                        stdout, stderr, stdout_filename, stderr_filename
+                    )
+                    exit_code, diagnostics = self.resolve_setup_verification_result(
+                        exit_code, stdout, diagnostics, stderr=stderr
+                    )
+                    if exit_code:
+                        return exit_code, diagnostics
             except Exception as error:
                 diagnostics = f"could not execute: {error}"
                 logger.error(diagnostics)