diff --git a/src/DIRAC/WorkloadManagementSystem/Client/JobReport.py b/src/DIRAC/WorkloadManagementSystem/Client/JobReport.py index 6af63f6d1c1..c1a852e6dab 100644 --- a/src/DIRAC/WorkloadManagementSystem/Client/JobReport.py +++ b/src/DIRAC/WorkloadManagementSystem/Client/JobReport.py @@ -2,6 +2,8 @@ It's an interface to JobStateUpdateClient, used when bulk submission is needed. """ import datetime +import decimal +import math from collections import defaultdict from DIRAC import S_OK, S_ERROR, gLogger @@ -10,6 +12,33 @@ from DIRAC.WorkloadManagementSystem.Client.JobStateUpdateClient import JobStateUpdateClient +def isFiniteParameterValue(value): + """Check that a job parameter value holds no non-finite number. + + Non-finite floats (NaN, +/-Infinity) have no representation in JSON: they make + the encoded payload invalid, which the receiving side rejects outright. Values + are inspected recursively, since a parameter may well be a container. + + :param value: any job parameter value + :return: False if a NaN or an infinity is found anywhere in `value` + """ + # bool and int are exact and always finite, and math.isfinite() would raise + # OverflowError on a large enough int + if isinstance(value, int): + return True + if isinstance(value, (float, decimal.Decimal)): + try: + return math.isfinite(value) + except (TypeError, ValueError): + # e.g. decimal.Decimal("sNaN"), which cannot even be converted to float + return False + if isinstance(value, dict): + return all(isFiniteParameterValue(item) for item in value.values()) + if isinstance(value, (list, tuple, set)): + return all(isFiniteParameterValue(item) for item in value) + return True + + class JobReport: """ .. class:: JobReport @@ -24,6 +53,7 @@ def __init__(self, jobid, source=""): self.source = source if not source: self.source = "Job_%d" % self.jobID + self.log = gLogger.getSubLogger(self.__class__.__name__) def setJob(self, jobID): """Set the job ID for which to send reports""" @@ -57,17 +87,13 @@ def setApplicationStatus(self, appStatus, sendFlag=True): def setJobParameter(self, par_name, par_value, sendFlag=True): """Set job parameter for jobID""" - self.jobParameters.append((par_name, par_value)) - if sendFlag and self.jobID: - # and send - return self.sendStoredJobParameters() - - return S_OK() + return self.setJobParameters([(par_name, par_value)], sendFlag) def setJobParameters(self, parameters, sendFlag=True): """Set job parameters for jobID""" for pname, pvalue in parameters: - self.jobParameters.append((pname, pvalue)) + if self._isValidParameterValue(pname, pvalue): + self.jobParameters.append((pname, pvalue)) if sendFlag and self.jobID: # and send @@ -75,6 +101,18 @@ def setJobParameters(self, parameters, sendFlag=True): return S_OK() + def _isValidParameterValue(self, par_name, par_value): + """Check that a parameter value can be reported. + + Non-finite floats (NaN, +/-Infinity) cannot be represented in JSON nor + stored in the job parameters backends, so they are dropped here with a + warning rather than failing the whole parameters update. + """ + if not isFiniteParameterValue(par_value): + self.log.warn("Dropping non-finite value for job parameter", f"{par_name} = {par_value}") + return False + return True + def sendStoredStatusInfo(self): """Send the job status information stored in the internal cache""" diff --git a/src/DIRAC/WorkloadManagementSystem/Client/test/Test_JobReport.py b/src/DIRAC/WorkloadManagementSystem/Client/test/Test_JobReport.py index dde79275899..2aa3ba149cf 100644 --- a/src/DIRAC/WorkloadManagementSystem/Client/test/Test_JobReport.py +++ b/src/DIRAC/WorkloadManagementSystem/Client/test/Test_JobReport.py @@ -1,10 +1,14 @@ """Test for JobReport""" # pylint: disable=missing-docstring +import decimal +import math from unittest.mock import MagicMock +import pytest + # sut -from DIRAC.WorkloadManagementSystem.Client.JobReport import JobReport +from DIRAC.WorkloadManagementSystem.Client.JobReport import JobReport, isFiniteParameterValue def test_jobReport(mocker): @@ -22,3 +26,63 @@ def test_jobReport(mocker): res = jr.setJobParameters([("par_3", "value_3"), ("par_4", "value_4")], sendFlag=False) print(jr.jobParameters) jr.dump() + + +@pytest.mark.parametrize( + "value, expected", + [ + # finite values of every shape are kept + (0.0, True), + (9.5, True), + (-1, True), + (True, True), + (10**400, True), # too large for a float, but exact and finite + ("nan", True), # a string that merely looks like one, e.g. an application message + ([1.0, 2.0], True), + ({"a": {"b": [1, 2.5]}}, True), + (None, True), + (decimal.Decimal("1.5"), True), + # non-finite ones are rejected, wherever they sit + (math.nan, False), + (math.inf, False), + (-math.inf, False), + ([1.0, math.nan], False), + ((1.0, math.inf), False), + ({"nested": [math.nan]}, False), + ({"a": {"b": {"c": -math.inf}}}, False), + (decimal.Decimal("NaN"), False), + (decimal.Decimal("Infinity"), False), + ], +) +def test_isFiniteParameterValue(value, expected): + """Containers are inspected recursively, mirroring the check on the diracx side.""" + assert isFiniteParameterValue(value) is expected + + +def test_jobReportDropsNonFiniteParameters(mocker): + """Non-finite floats cannot be represented in JSON nor stored in the backends.""" + mocker.patch("DIRAC.WorkloadManagementSystem.Client.JobStateUpdateClient", side_effect=MagicMock()) + + jr = JobReport(123) + res = jr.setJobParameter("LoadAverage", math.nan, sendFlag=False) + assert res["OK"] + res = jr.setJobParameters( + [("MemoryUsed(MB)", math.inf), ("DiskSpace(MB)", -math.inf), ("CPUNormalizationFactor", 9.5)], + sendFlag=False, + ) + assert res["OK"] + assert jr.jobParameters == [("CPUNormalizationFactor", 9.5)] + + +def test_jobReportDropsNonFiniteParametersInContainers(mocker): + """A single non-finite value nested in a container invalidates the whole payload.""" + mocker.patch("DIRAC.WorkloadManagementSystem.Client.JobStateUpdateClient", side_effect=MagicMock()) + + jr = JobReport(123) + res = jr.setJobParameter("NodeInformation", {"LoadAverage": math.nan}, sendFlag=False) + assert res["OK"] + res = jr.setJobParameter("Samples", [1.0, 2.0, math.inf], sendFlag=False) + assert res["OK"] + res = jr.setJobParameter("InitialValues", {"DiskSpace": 1024.0}, sendFlag=False) + assert res["OK"] + assert jr.jobParameters == [("InitialValues", {"DiskSpace": 1024.0})] diff --git a/src/DIRAC/WorkloadManagementSystem/JobWrapper/JobWrapper.py b/src/DIRAC/WorkloadManagementSystem/JobWrapper/JobWrapper.py index 53c803b3400..c0cca072f70 100755 --- a/src/DIRAC/WorkloadManagementSystem/JobWrapper/JobWrapper.py +++ b/src/DIRAC/WorkloadManagementSystem/JobWrapper/JobWrapper.py @@ -54,7 +54,7 @@ from DIRAC.WorkloadManagementSystem.Client.JobMonitoringClient import JobMonitoringClient from DIRAC.WorkloadManagementSystem.Client.JobManagerClient import JobManagerClient from DIRAC.WorkloadManagementSystem.Client.SandboxStoreClient import SandboxStoreClient -from DIRAC.WorkloadManagementSystem.Client.JobReport import JobReport +from DIRAC.WorkloadManagementSystem.Client.JobReport import JobReport, isFiniteParameterValue from DIRAC.WorkloadManagementSystem.Client import JobStatus from DIRAC.WorkloadManagementSystem.Client import JobMinorStatus @@ -1450,6 +1450,14 @@ def __report(self, status="", minorStatus="", sendFlag=False): ############################################################################# def __setJobParam(self, name, value, sendFlag=False): """Wraps around setJobParameter of JobReport client""" + # The value is stringified below, which would hide a non-finite number from + # JobReport (str(float("nan")) is just "nan"), so check it here while it is + # still a number. Nothing is added, but sendFlag is honoured so that anything + # already accumulated is still flushed. + if not isFiniteParameterValue(value): + self.log.warn("Dropping non-finite value for job parameter", f"{name} = {value}") + return self.jobReport.setJobParameters([], sendFlag) + jobParam = self.jobReport.setJobParameter(str(name), str(value), sendFlag) if not jobParam["OK"]: self.log.warn("Failed setting job parameter", jobParam["Message"]) diff --git a/src/DIRAC/WorkloadManagementSystem/JobWrapper/Watchdog.py b/src/DIRAC/WorkloadManagementSystem/JobWrapper/Watchdog.py index 7ff8a46d3b8..c99b8e49e90 100755 --- a/src/DIRAC/WorkloadManagementSystem/JobWrapper/Watchdog.py +++ b/src/DIRAC/WorkloadManagementSystem/JobWrapper/Watchdog.py @@ -14,7 +14,6 @@ import datetime import errno import getpass -import math import os import re import socket @@ -32,6 +31,7 @@ from DIRAC.Core.Utilities.Subprocess import getChildrenPIDs from DIRAC.Resources.Computing.BatchSystems.TimeLeft.TimeLeft import TimeLeft from DIRAC.WorkloadManagementSystem.Client import JobMinorStatus +from DIRAC.WorkloadManagementSystem.Client.JobReport import isFiniteParameterValue from DIRAC.WorkloadManagementSystem.Client.JobStateUpdateClient import JobStateUpdateClient @@ -801,39 +801,35 @@ def __timeLeft(self): ############################################################################# def __getUsageSummary(self): - """Returns average load, memory etc. over execution of job thread""" + """Returns average load, memory etc. over execution of job thread + + Parameters for which no sample was collected (e.g. because the job + ended before the first Watchdog cycle) are omitted from the summary: + NaN cannot be represented in JSON nor stored in the backends. + + The differential parameters also need their baseline: calibrate() only + records one when the corresponding probe succeeded, while a later check + cycle may well have collected samples. + """ summary = {} # CPUConsumed - if "CPUConsumed" in self.parameters: - cpuList = self.parameters["CPUConsumed"] - if cpuList: - hmsCPU = cpuList[-1] - rawCPU = self.__convertCPUTime(hmsCPU) - if rawCPU["OK"]: - summary["LastUpdateCPU(s)"] = rawCPU["Value"] - else: - summary["LastUpdateCPU(s)"] = math.nan + if self.parameters.get("CPUConsumed"): + hmsCPU = self.parameters["CPUConsumed"][-1] + rawCPU = self.__convertCPUTime(hmsCPU) + if rawCPU["OK"]: + summary["LastUpdateCPU(s)"] = rawCPU["Value"] # DiskSpace - if "DiskSpace" in self.parameters: + if self.parameters.get("DiskSpace") and "DiskSpace" in self.initialValues: space = self.parameters["DiskSpace"] - if space: - summary["DiskSpace(MB)"] = max(abs(float(space[-1]) - float(self.initialValues["DiskSpace"])), 0.0) - else: - summary["DiskSpace(MB)"] = math.nan + summary["DiskSpace(MB)"] = max(abs(float(space[-1]) - float(self.initialValues["DiskSpace"])), 0.0) # MemoryUsed - if "MemoryUsed" in self.parameters: + if self.parameters.get("MemoryUsed") and "MemoryUsed" in self.initialValues: memory = self.parameters["MemoryUsed"] - if memory: - summary["MemoryUsed(MB)"] = abs(float(memory[-1]) - float(self.initialValues["MemoryUsed"])) - else: - summary["MemoryUsed(MB)"] = math.nan + summary["MemoryUsed(MB)"] = abs(float(memory[-1]) - float(self.initialValues["MemoryUsed"])) # LoadAverage - if "LoadAverage" in self.parameters: + if self.parameters.get("LoadAverage"): laList = self.parameters["LoadAverage"] - if laList: - summary["LoadAverage"] = sum(laList) / len(laList) - else: - summary["LoadAverage"] = math.nan + summary["LoadAverage"] = sum(laList) / len(laList) result = self.__getWallClockTime() if not result["OK"]: @@ -921,8 +917,20 @@ def __setJobParamList(self, value): self.log.info("Running without JOBID so parameters will not be reported") return S_OK() jobID = os.environ["JOBID"] - jobParam = JobStateUpdateClient().setJobParameters(int(jobID), value) - self.log.verbose(f"setJobParameters({jobID},{value})") + # The Watchdog talks to the JobStateUpdate service directly rather than through + # JobReport, so it has to drop non-finite values itself: a single one of them + # makes the whole encoded payload invalid, losing every other parameter with it + keptValue = [] + for name, paramValue in value: + if isFiniteParameterValue(paramValue): + keptValue.append((name, paramValue)) + else: + self.log.warn("Dropping non-finite value for job parameter", f"{name} = {paramValue}") + if not keptValue: + return S_OK() + + jobParam = JobStateUpdateClient().setJobParameters(int(jobID), keptValue) + self.log.verbose(f"setJobParameters({jobID},{keptValue})") if not jobParam["OK"]: self.log.warn(jobParam["Message"]) diff --git a/src/DIRAC/WorkloadManagementSystem/JobWrapper/test/Test_JobWrapper.py b/src/DIRAC/WorkloadManagementSystem/JobWrapper/test/Test_JobWrapper.py index 4fb1521f718..ac0e934771b 100644 --- a/src/DIRAC/WorkloadManagementSystem/JobWrapper/test/Test_JobWrapper.py +++ b/src/DIRAC/WorkloadManagementSystem/JobWrapper/test/Test_JobWrapper.py @@ -1,5 +1,6 @@ """ Test class for JobWrapper """ +import math import os import shutil import pytest @@ -48,6 +49,21 @@ def test_InputData(mocker): assert res["OK"] +def test_setJobParamDropsNonFiniteValues(mocker): + """The value is stringified on its way to JobReport, which would hide a NaN from it.""" + mocker.patch( + "DIRAC.WorkloadManagementSystem.JobWrapper.JobWrapper.getSystemSection", side_effect=getSystemSectionMock + ) + mocker.patch("DIRAC.WorkloadManagementSystem.JobWrapper.JobWrapper.ModuleFactory", side_effect=MagicMock()) + + jw = JobWrapper() + jw._JobWrapper__setJobParam("NormCPUTime(s)", math.nan) + jw._JobWrapper__setJobParam("ScaledCPUTime(s)", math.inf) + jw._JobWrapper__setJobParam("TotalCPUTime(s)", 12.5) + + assert jw.jobReport.jobParameters == [("TotalCPUTime(s)", "12.5")] + + def test_performChecks(): wd = Watchdog( pid=os.getpid(), diff --git a/src/DIRAC/WorkloadManagementSystem/JobWrapper/test/Test_Watchdog.py b/src/DIRAC/WorkloadManagementSystem/JobWrapper/test/Test_Watchdog.py index 8dc07b2c3d2..eeb46c4c693 100644 --- a/src/DIRAC/WorkloadManagementSystem/JobWrapper/test/Test_Watchdog.py +++ b/src/DIRAC/WorkloadManagementSystem/JobWrapper/test/Test_Watchdog.py @@ -1,5 +1,6 @@ """ unit test for Watchdog.py """ +import math import os from unittest.mock import MagicMock @@ -37,3 +38,72 @@ def test__performChecksFull(): assert res["OK"] is True res = wd._performChecks() assert res["OK"] is True + + +def test__getUsageSummaryNoSamples(monkeypatch): + """A job ending before the first Watchdog cycle must not report non-finite parameters.""" + monkeypatch.delenv("JOBID", raising=False) + pid = os.getpid() + wd = Watchdog(pid, mock_exeThread, mock_spObject, 5000) + res = wd.calibrate() + assert res["OK"] is True + + # No check cycle has run yet, so all the sampling lists are still empty + wd._Watchdog__getUsageSummary() + + for name in ("LastUpdateCPU(s)", "DiskSpace(MB)", "MemoryUsed(MB)", "LoadAverage"): + assert name not in wd.currentStats + assert all(math.isfinite(value) for value in wd.currentStats.values()), wd.currentStats + + +def test__getUsageSummaryWithSamples(monkeypatch): + """Once samples have been collected, the summary must actually report them.""" + monkeypatch.delenv("JOBID", raising=False) + pid = os.getpid() + wd = Watchdog(pid, mock_exeThread, mock_spObject, 5000) + assert wd.calibrate()["OK"] is True + assert wd._performChecks()["OK"] is True + + wd._Watchdog__getUsageSummary() + + # LoadAverage is sampled unconditionally, the others depend on the probes + assert "LoadAverage" in wd.currentStats + assert {"WallClockTime(s)", "ScaledCPUTime(s)"} <= set(wd.currentStats) + assert all(math.isfinite(value) for value in wd.currentStats.values()), wd.currentStats + + +def test__getUsageSummaryWithoutCalibrationBaseline(monkeypatch): + """Samples without a baseline must be skipped, not raise. + + calibrate() only records initialValues[DiskSpace]/[MemoryUsed] when the probe + succeeded, while a later check cycle collects samples regardless. + """ + monkeypatch.delenv("JOBID", raising=False) + pid = os.getpid() + wd = Watchdog(pid, mock_exeThread, mock_spObject, 5000) + assert wd.calibrate()["OK"] is True + + # as if both probes had failed at calibration time + wd.initialValues.pop("DiskSpace", None) + wd.initialValues.pop("MemoryUsed", None) + + assert wd._performChecks()["OK"] is True + wd._Watchdog__getUsageSummary() + + for name in ("DiskSpace(MB)", "MemoryUsed(MB)"): + assert name not in wd.currentStats + assert all(math.isfinite(value) for value in wd.currentStats.values()), wd.currentStats + + +def test__setJobParamListDropsNonFiniteValues(monkeypatch, mocker): + """The Watchdog reports directly to the service, so it filters values itself.""" + monkeypatch.setenv("JOBID", "123") + mockClient = MagicMock() + mockClient().setJobParameters.return_value = {"OK": True, "Value": ""} + mocker.patch("DIRAC.WorkloadManagementSystem.JobWrapper.Watchdog.JobStateUpdateClient", mockClient) + + pid = os.getpid() + wd = Watchdog(pid, mock_exeThread, mock_spObject, 5000) + wd._Watchdog__setJobParamList([("LoadAverage", math.nan), ("MemoryUsed(MB)", 1024.0)]) + + mockClient().setJobParameters.assert_called_once_with(123, [("MemoryUsed(MB)", 1024.0)])