fix: job wrapper status now that the watchdog kills the payload

src/DIRAC/WorkloadManagementSystem/Agent/PushJobAgent.py

@@ -741,11 +741,17 @@ def postProcessJob(self, job: JobWrapper, payloadResults):
                     return
 
                 job.jobReport.setJobParameter("Error Message", result["Message"], sendFlag=False)
-                job.jobReport.setJobStatus(
-                    status=JobStatus.FAILED, minorStatus=JobMinorStatus.EXCEPTION_DURING_EXEC, sendFlag=False
-                )
-                job.sendFailoverRequest()
-                job.sendJobAccounting(status=JobStatus.FAILED, minorStatus=JobMinorStatus.EXCEPTION_DURING_EXEC)
+                # If postProcess already stamped FAILED with a specific minor status (e.g. a
+                # watchdog reason), preserve it, as JobWrapperUtilities.executePayload does.
+                if job.wmsMajorStatus == JobStatus.FAILED:
+                    job.sendFailoverRequest()
+                    job.sendJobAccounting()
+                else:
+                    job.jobReport.setJobStatus(
+                        status=JobStatus.FAILED, minorStatus=JobMinorStatus.EXCEPTION_DURING_EXEC, sendFlag=False
+                    )
+                    job.sendFailoverRequest()
+                    job.sendJobAccounting(status=JobStatus.FAILED, minorStatus=JobMinorStatus.EXCEPTION_DURING_EXEC)
                 shutil.rmtree(job.jobIDPath)
                 return
         except Exception as exc:  # pylint: disable=broad-except

src/DIRAC/WorkloadManagementSystem/Agent/test/Test_Agent_PushJobAgent.py

@@ -542,3 +542,35 @@ def fakeDump(filename):
     assert reloaded.getOption("/DIRAC/VirtualOrganization") == "lhcb"
 
     shutil.rmtree("job", ignore_errors=True)
+
+
+@pytest.mark.parametrize(
+    "wmsMajorStatus, expectedAccounting",
+    [
+        # postProcess already reported the failure (e.g. the Watchdog's reason): keep it
+        ("Failed", {}),
+        # postProcess failed without reporting anything: fall back to the generic status
+        ("Running", {"status": "Failed", "minorStatus": JobMinorStatus.EXCEPTION_DURING_EXEC}),
+    ],
+)
+def test_postProcessJob_keeps_the_status_postProcess_reported(mocker, tmp_path, wmsMajorStatus, expectedAccounting):
+    """A failed postProcess is not overwritten with EXCEPTION_DURING_EXEC when it already stamped FAILED."""
+    agent = _bareAgent(mocker)
+    mocker.patch("DIRAC.WorkloadManagementSystem.Agent.PushJobAgent.shutil.rmtree")
+
+    job = Mock()
+    job.jobID = 1
+    job.jobIDPath = tmp_path
+    job.wmsMajorStatus = wmsMajorStatus
+    job.postProcess.return_value = S_ERROR(f"Payload killed by watchdog: {JobMinorStatus.JOB_EXCEEDED_CPU}")
+
+    agent.postProcessJob(job, {})
+
+    overwritten = [
+        call
+        for call in job.jobReport.setJobStatus.call_args_list
+        if call.kwargs.get("minorStatus") == JobMinorStatus.EXCEPTION_DURING_EXEC
+    ]
+    assert bool(overwritten) is bool(expectedAccounting)
+    job.sendFailoverRequest.assert_called_once_with()
+    job.sendJobAccounting.assert_called_once_with(**expectedAccounting)

src/DIRAC/WorkloadManagementSystem/JobWrapper/JobWrapper.py

@@ -544,41 +544,53 @@ def postProcess(
         """This method is called after the payload has finished running."""
         self.log.info(f"Job Wrapper is starting the post processing phase for job {self.jobID}")
 
-        # if the execution thread didn't complete
-        if payloadStatus is None and not payloadExecutorError:
+        # process() and postProcess() may run on different machines: propagate the
+        # CPU figures to executionResults early so the accounting record sees them
+        # even if we hit one of the early-return failure paths below.
+        if cpuTimeConsumed and not self.executionResults.get("CPU"):
+            self.executionResults["CPU"] = cpuTimeConsumed
+
+        # No clean exit code: identify the cause (watchdog > executor > generic)
+        # Watchdog kills (killChild -> SIGTERM/SIGKILL) may leave payloadStatus=None,
+        # so checking watchdogError first preserves its specific reason.
+        if payloadStatus is None:
+            self.__logWatchdogStats(watchdogStats)
+            if payloadOutput and self.executionResults.get("CPU"):
+                self.__sendFinalStdOut(payloadOutput)
+
+            if watchdogError:
+                self.log.error("Payload killed by watchdog", watchdogError)
+                self.__report(status=JobStatus.FAILED, minorStatus=watchdogError, sendFlag=True)
+                self.__setJobParam("ApplicationError", watchdogError, sendFlag=True)
+                return S_ERROR(f"Payload killed by watchdog: {watchdogError}")
+            if payloadExecutorError:
+                self.log.error("Failed to execute the payload", payloadExecutorError)
+                self.__report(status=JobStatus.FAILED, minorStatus=JobMinorStatus.APP_THREAD_FAILED, sendFlag=True)
+                self.__setJobParam("ApplicationError", "None reported", sendFlag=True)
+                return S_ERROR(payloadExecutorError)
             self.log.error("Application thread did not complete")
             self.__report(status=JobStatus.FAILED, minorStatus=JobMinorStatus.APP_THREAD_NOT_COMPLETE, sendFlag=True)
             self.__setJobParam("ApplicationError", JobMinorStatus.APP_THREAD_NOT_COMPLETE, sendFlag=True)
             return S_ERROR("No outputs generated from job execution")
 
-        # If the execution thread got an error (not a payload error)
+        # payloadStatus is set: payload exited with a real exit code.
+
         if payloadExecutorError:
             self.log.error("Failed to execute the payload", payloadExecutorError)
             self.__report(status=JobStatus.FAILED, minorStatus=JobMinorStatus.APP_THREAD_FAILED, sendFlag=True)
-            applicationErrorStatus = "None reported"
-            if payloadStatus:
-                applicationErrorStatus = str(payloadStatus)
-            self.__setJobParam("ApplicationError", applicationErrorStatus, sendFlag=True)
-
-        # This might happen if process() and postProcess() are called on different machines
-        if cpuTimeConsumed and not self.executionResults.get("CPU"):
-            self.executionResults["CPU"] = cpuTimeConsumed
+            self.__setJobParam(
+                "ApplicationError", str(payloadStatus) if payloadStatus else "None reported", sendFlag=True
+            )
 
         if watchdogError:
             self.__report(status=JobStatus.FAILED, minorStatus=watchdogError, sendFlag=True)
 
-        if watchdogStats:
-            self.log.info(
-                "Statistics collected by the Watchdog:\n ",
-                "\n  ".join(["%s: %s" % items for items in watchdogStats.items()]),
-            )  # can be an iterator
-
-        if payloadStatus is None:
-            return S_ERROR("No outputs generated from job execution")
+        self.__logWatchdogStats(watchdogStats)
 
-        # Send final heartbeat of a configurable number of lines here
-        self.log.verbose("Sending final application standard output heartbeat")
-        self.__sendFinalStdOut(payloadOutput)
+        # Send final heartbeat of a configurable number of lines here.
+        if self.executionResults.get("CPU"):
+            self.log.verbose("Sending final application standard output heartbeat")
+            self.__sendFinalStdOut(payloadOutput)
         self.log.verbose(f"Execution thread status = {payloadStatus}")
 
         self.log.info("Checking directory contents after execution:")
@@ -591,6 +603,7 @@ def postProcess(
             # no timeout and exit code is 0
             self.log.info(res["Value"][1])
 
+        # Non-zero exit without a watchdog reason: report a generic application error.
         if not watchdogError and payloadStatus != 0:
             self.__report(status=JobStatus.COMPLETING, minorStatus=JobMinorStatus.APP_ERRORS, sendFlag=True)
 
@@ -702,6 +715,15 @@ def __getCPUHMS(self, cpuTime):
         self.log.verbose(f"Human readable CPU time is: {humanTime}")
         return S_OK((cpuTime, humanTime))
 
+    #############################################################################
+    def __logWatchdogStats(self, watchdogStats: dict):
+        """Log the stats collected by the Watchdog, if any."""
+        if watchdogStats:
+            self.log.info(
+                "Statistics collected by the Watchdog:\n ",
+                "\n  ".join(["%s: %s" % items for items in watchdogStats.items()]),
+            )
+
     #############################################################################
     def resolveInputData(self):
         """Input data is resolved here using a VO specific plugin module."""

src/DIRAC/WorkloadManagementSystem/JobWrapper/JobWrapperUtilities.py

@@ -217,13 +217,23 @@ def executePayload(job: JobWrapper) -> bool:
             )
             job.sendJobAccounting(status=rescheduleResult, minorStatus=JobMinorStatus.JOB_WRAPPER_EXECUTION)
             return False
-        gLogger.exception("Job failed in execution phase")
         job.jobReport.setJobParameter("Error Message", repr(exc), sendFlag=False)
-        job.jobReport.setJobStatus(
-            status=JobStatus.FAILED, minorStatus=JobMinorStatus.EXCEPTION_DURING_EXEC, sendFlag=False
-        )
-        job.sendFailoverRequest()
-        job.sendJobAccounting(status=JobStatus.FAILED, minorStatus=JobMinorStatus.EXCEPTION_DURING_EXEC)
+        # If postProcess already stamped FAILED with a specific minor status (e.g. a
+        # watchdog reason like JOB_EXCEEDED_CPU), preserve it. Otherwise (postProcess
+        # bypassed) fall back to EXCEPTION_DURING_EXEC so the job isn't left RUNNING.
+        if job.wmsMajorStatus == JobStatus.FAILED:
+            gLogger.error("Job failed in execution phase", repr(exc))
+            job.sendFailoverRequest()
+            job.sendJobAccounting()
+        else:
+            gLogger.exception("Job failed in execution phase")
+            job.jobReport.setJobStatus(
+                status=JobStatus.FAILED, minorStatus=JobMinorStatus.EXCEPTION_DURING_EXEC, sendFlag=False
+            )
+            job.sendFailoverRequest()
+            # setJobStatus on jobReport does not propagate to the JobWrapper's
+            # wmsMajorStatus / wmsMinorStatus, so pass them explicitly.
+            job.sendJobAccounting(status=JobStatus.FAILED, minorStatus=JobMinorStatus.EXCEPTION_DURING_EXEC)
         return False
     except Exception as exc:  # pylint: disable=broad-except
         gLogger.exception("Job raised exception during execution phase")

src/DIRAC/WorkloadManagementSystem/JobWrapper/test/Test_JobWrapper.py

@@ -530,6 +530,128 @@ def test_postProcess_watchdog_error(setup_job_wrapper, mocker, mock_report_and_s
     assert report_args[-1]["minorStatus"] == payloadResult["watchdogError"]
 
 
+def test_postProcess_no_status_executor_and_watchdog_priority(setup_job_wrapper, mocker, mock_report_and_set_param):
+    """payloadStatus=None with both executor error and watchdog error set.
+
+    Watchdog must win -- it's the proximate cause when both are present (the kill
+    is what made the executor record an error in the first place).
+    """
+    jw = setup_job_wrapper()
+    report_args, set_param_args, report_side_effect, set_param_side_effect = mock_report_and_set_param
+
+    mocker.patch.object(jw, "_JobWrapper__report", side_effect=report_side_effect)
+    mocker.patch.object(jw, "_JobWrapper__setJobParam", side_effect=set_param_side_effect)
+
+    payloadResult = {
+        "payloadStatus": None,
+        "payloadOutput": None,
+        "payloadExecutorError": "systemCall failed",
+        "cpuTimeConsumed": None,
+        "watchdogError": JobMinorStatus.JOB_EXCEEDED_CPU,
+        "watchdogStats": None,
+    }
+
+    result = jw.postProcess(**payloadResult)
+    assert not result["OK"]
+    assert "Payload killed by watchdog" in result["Message"]
+    assert report_args[-1]["minorStatus"] == JobMinorStatus.JOB_EXCEEDED_CPU
+    # The executor-error branch's APP_THREAD_FAILED must NOT appear -- watchdog wins.
+    assert not any(call.get("minorStatus") == JobMinorStatus.APP_THREAD_FAILED for call in report_args)
+
+
+@pytest.mark.parametrize("payloadStatus", [0, DErrno.EWMSRESC, DErrno.EWMSRESC & 255])
+def test_postProcess_watchdog_verdict_wins_over_exit_code(
+    setup_job_wrapper, mocker, mock_report_and_set_param, payloadStatus
+):
+    """The Watchdog killed the payload, which still managed to exit 0 or ask to be rescheduled.
+
+    A payload stopped for exceeding a limit (CPU, memory, time left) did not complete its
+    work, whatever it returned on the way out: the Watchdog's reason stands.
+    """
+    jw = setup_job_wrapper()
+    report_args, set_param_args, report_side_effect, set_param_side_effect = mock_report_and_set_param
+
+    mocker.patch.object(jw, "_JobWrapper__report", side_effect=report_side_effect)
+    mocker.patch.object(jw, "_JobWrapper__setJobParam", side_effect=set_param_side_effect)
+
+    payloadResult = {
+        "payloadStatus": payloadStatus,
+        "payloadOutput": "",
+        "payloadExecutorError": None,
+        "cpuTimeConsumed": [100, 200, 300, 400, 500],
+        "watchdogError": JobMinorStatus.JOB_EXCEEDED_CPU,
+        "watchdogStats": {"LastUpdateCPU(s)": "100", "MemoryUsed(MB)": "100"},
+    }
+    jw.executionResults["CPU"] = payloadResult["cpuTimeConsumed"]
+
+    assert jw.postProcess(**payloadResult)["OK"]
+    assert report_args[-1]["status"] == JobStatus.FAILED
+    assert report_args[-1]["minorStatus"] == JobMinorStatus.JOB_EXCEEDED_CPU
+    assert jw.failedFlag
+
+
+def test_postProcess_watchdog_killed_payload_with_partial_output(setup_job_wrapper, mocker, mock_report_and_set_param):
+    """Watchdog killed the payload but partial output and CPU figures are available.
+
+    The wrapper should still send the partial output as the final heartbeat -- the
+    user often wants to see how far the payload got before the kill.
+    """
+    jw = setup_job_wrapper()
+    report_args, set_param_args, report_side_effect, set_param_side_effect = mock_report_and_set_param
+
+    mocker.patch.object(jw, "_JobWrapper__report", side_effect=report_side_effect)
+    mocker.patch.object(jw, "_JobWrapper__setJobParam", side_effect=set_param_side_effect)
+    sendFinal = mocker.patch.object(jw, "_JobWrapper__sendFinalStdOut")
+
+    payloadResult = {
+        "payloadStatus": None,
+        "payloadOutput": "Last log line before SIGTERM\n",
+        "payloadExecutorError": None,
+        "cpuTimeConsumed": [100, 200, 300, 400, 500],
+        "watchdogError": JobMinorStatus.JOB_EXCEEDED_CPU,
+        "watchdogStats": {"LastUpdateCPU(s)": "100"},
+    }
+    jw.executionResults["CPU"] = payloadResult["cpuTimeConsumed"]
+
+    result = jw.postProcess(**payloadResult)
+
+    assert not result["OK"]
+    sendFinal.assert_called_once_with(payloadResult["payloadOutput"])
+    # Watchdog status still reported correctly.
+    assert report_args[-1]["minorStatus"] == JobMinorStatus.JOB_EXCEEDED_CPU
+
+
+def test_postProcess_watchdog_killed_payload(setup_job_wrapper, mocker, mock_report_and_set_param):
+    """Watchdog killed the payload mid-flight: payloadStatus=None + watchdogError set.
+
+    Regression test for the path where killChild leaves no clean exit code: the
+    watchdog's verdict must be reported as the minor status, not silently lost
+    behind APP_THREAD_NOT_COMPLETE / "No outputs generated".
+    """
+    jw = setup_job_wrapper()
+    report_args, set_param_args, report_side_effect, set_param_side_effect = mock_report_and_set_param
+
+    mocker.patch.object(jw, "_JobWrapper__report", side_effect=report_side_effect)
+    mocker.patch.object(jw, "_JobWrapper__setJobParam", side_effect=set_param_side_effect)
+
+    payloadResult = {
+        "payloadStatus": None,
+        "payloadOutput": None,
+        "payloadExecutorError": None,
+        "cpuTimeConsumed": None,
+        "watchdogError": JobMinorStatus.JOB_EXCEEDED_CPU,
+        "watchdogStats": None,
+    }
+
+    result = jw.postProcess(**payloadResult)
+    assert not result["OK"]
+    assert "Payload killed by watchdog" in result["Message"]
+    assert JobMinorStatus.JOB_EXCEEDED_CPU in result["Message"]
+    assert report_args[-1]["status"] == JobStatus.FAILED
+    assert report_args[-1]["minorStatus"] == JobMinorStatus.JOB_EXCEEDED_CPU
+    assert set_param_args[-1][0][1] == JobMinorStatus.JOB_EXCEEDED_CPU
+
+
 def test_postProcess_executor_failed_no_status(setup_job_wrapper, mocker, mock_report_and_set_param):
     """Test the postProcess method of the JobWrapper class: executor failed and no status defined."""
     jw = setup_job_wrapper()