diff --git a/src/DIRAC/WorkloadManagementSystem/Agent/PushJobAgent.py b/src/DIRAC/WorkloadManagementSystem/Agent/PushJobAgent.py index 24482430017..6d5d2c04e4b 100644 --- a/src/DIRAC/WorkloadManagementSystem/Agent/PushJobAgent.py +++ b/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 diff --git a/src/DIRAC/WorkloadManagementSystem/Agent/test/Test_Agent_PushJobAgent.py b/src/DIRAC/WorkloadManagementSystem/Agent/test/Test_Agent_PushJobAgent.py index 9c653941511..c6b5ada3098 100644 --- a/src/DIRAC/WorkloadManagementSystem/Agent/test/Test_Agent_PushJobAgent.py +++ b/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) diff --git a/src/DIRAC/WorkloadManagementSystem/JobWrapper/JobWrapper.py b/src/DIRAC/WorkloadManagementSystem/JobWrapper/JobWrapper.py index e9b12fa4e31..dd5411bb388 100755 --- a/src/DIRAC/WorkloadManagementSystem/JobWrapper/JobWrapper.py +++ b/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.""" diff --git a/src/DIRAC/WorkloadManagementSystem/JobWrapper/JobWrapperUtilities.py b/src/DIRAC/WorkloadManagementSystem/JobWrapper/JobWrapperUtilities.py index 947a7d0590e..57c74a6f529 100644 --- a/src/DIRAC/WorkloadManagementSystem/JobWrapper/JobWrapperUtilities.py +++ b/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") diff --git a/src/DIRAC/WorkloadManagementSystem/JobWrapper/test/Test_JobWrapper.py b/src/DIRAC/WorkloadManagementSystem/JobWrapper/test/Test_JobWrapper.py index 112e1140487..94da79e6cc6 100644 --- a/src/DIRAC/WorkloadManagementSystem/JobWrapper/test/Test_JobWrapper.py +++ b/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() diff --git a/src/DIRAC/WorkloadManagementSystem/JobWrapper/test/Test_JobWrapperUtilities.py b/src/DIRAC/WorkloadManagementSystem/JobWrapper/test/Test_JobWrapperUtilities.py new file mode 100644 index 00000000000..6028f8f1ad0 --- /dev/null +++ b/src/DIRAC/WorkloadManagementSystem/JobWrapper/test/Test_JobWrapperUtilities.py @@ -0,0 +1,89 @@ +"""Unit tests for JobWrapperUtilities.executePayload's failure-reporting paths. + +These complement the integration tests in Test_JobWrapperTemplate.py. +""" +from unittest.mock import MagicMock + +from DIRAC import S_ERROR +from DIRAC.Core.Utilities import DErrno +from DIRAC.WorkloadManagementSystem.Client import JobMinorStatus, JobStatus +from DIRAC.WorkloadManagementSystem.JobWrapper.JobWrapperUtilities import executePayload + + +def _make_mock_job(execute_result, wmsMajorStatus="Running"): + """Build a minimal JobWrapper-like mock that returns the given execute_result.""" + job = MagicMock() + job.jobID = 1 + job.execute.return_value = execute_result + job.wmsMajorStatus = wmsMajorStatus + job.jobReport = MagicMock() + return job + + +def test_executePayload_preserves_minor_status_when_postProcess_set_failed(): + """When postProcess set wmsMajorStatus=FAILED, executePayload must not overwrite + the minor status with EXCEPTION_DURING_EXEC. + + This is the path triggered by a watchdog kill: postProcess returns + S_ERROR("Payload killed by watchdog: ...") with the minor status already set + to (e.g.) JOB_EXCEEDED_CPU. The wrapper must respect that. + """ + job = _make_mock_job( + execute_result=S_ERROR(f"Payload killed by watchdog: {JobMinorStatus.JOB_EXCEEDED_CPU}"), + wmsMajorStatus=JobStatus.FAILED, + ) + + assert executePayload(job) is False + + # setJobStatus with EXCEPTION_DURING_EXEC must not have been called. + for call in job.jobReport.setJobStatus.call_args_list: + assert call.kwargs.get("minorStatus") != JobMinorStatus.EXCEPTION_DURING_EXEC + + # sendJobAccounting must be called with no minor-status override (preserves + # whatever wmsMinorStatus postProcess set on the job). + job.sendJobAccounting.assert_called_once_with() + + +def test_executePayload_falls_back_to_exception_when_postProcess_was_bypassed(): + """When postProcess never ran (e.g. process() returned S_ERROR before reaching + postProcess), wmsMajorStatus is still RUNNING and no minor-status was reported. + Fall back to EXCEPTION_DURING_EXEC so the user at least sees the job failed. + """ + job = _make_mock_job( + execute_result=S_ERROR("Payload process could not start after 5 seconds"), + wmsMajorStatus="Running", + ) + + assert executePayload(job) is False + + # The fallback override must have been called. + failed_calls = [ + call + for call in job.jobReport.setJobStatus.call_args_list + if call.kwargs.get("minorStatus") == JobMinorStatus.EXCEPTION_DURING_EXEC + ] + assert len(failed_calls) == 1 + assert failed_calls[0].kwargs["status"] == JobStatus.FAILED + + # sendJobAccounting must be called WITH explicit FAILED kwargs in the bypass case: + # setJobStatus on jobReport doesn't update wmsMajorStatus / wmsMinorStatus on the + # JobWrapper, so without these kwargs the accounting record would still show the + # stale "Running" / "Application" state. + job.sendJobAccounting.assert_called_once_with( + status=JobStatus.FAILED, minorStatus=JobMinorStatus.EXCEPTION_DURING_EXEC + ) + + +def test_executePayload_reschedule_path_unchanged(): + """EWMSRESC sub-branch returns early and is not affected by the clobber fix.""" + job = _make_mock_job( + execute_result=S_ERROR(DErrno.EWMSRESC, "Job will be rescheduled"), + wmsMajorStatus=JobStatus.FAILED, + ) + + assert executePayload(job) is False + + # The reschedule branch calls sendJobAccounting with explicit args. + job.sendJobAccounting.assert_called_once() + call = job.sendJobAccounting.call_args + assert call.kwargs.get("minorStatus") == JobMinorStatus.JOB_WRAPPER_EXECUTION