diff --git a/src/DIRAC/WorkloadManagementSystem/Client/JobReport.py b/src/DIRAC/WorkloadManagementSystem/Client/JobReport.py index 6af63f6d1c1..c2b0f4e6440 100644 --- a/src/DIRAC/WorkloadManagementSystem/Client/JobReport.py +++ b/src/DIRAC/WorkloadManagementSystem/Client/JobReport.py @@ -2,6 +2,7 @@ It's an interface to JobStateUpdateClient, used when bulk submission is needed. """ import datetime +import math from collections import defaultdict from DIRAC import S_OK, S_ERROR, gLogger @@ -57,7 +58,8 @@ 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 self._isValidParameterValue(par_name, par_value): + self.jobParameters.append((par_name, par_value)) if sendFlag and self.jobID: # and send return self.sendStoredJobParameters() @@ -67,7 +69,8 @@ def setJobParameter(self, par_name, par_value, sendFlag=True): 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 +78,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 isinstance(par_value, float) and not math.isfinite(par_value): + gLogger.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..2825f927f75 100644 --- a/src/DIRAC/WorkloadManagementSystem/Client/test/Test_JobReport.py +++ b/src/DIRAC/WorkloadManagementSystem/Client/test/Test_JobReport.py @@ -22,3 +22,18 @@ def test_jobReport(mocker): res = jr.setJobParameters([("par_3", "value_3"), ("par_4", "value_4")], sendFlag=False) print(jr.jobParameters) jr.dump() + + +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", float("nan"), sendFlag=False) + assert res["OK"] + res = jr.setJobParameters( + [("MemoryUsed(MB)", float("inf")), ("DiskSpace(MB)", float("-inf")), ("CPUNormalizationFactor", 9.5)], + sendFlag=False, + ) + assert res["OK"] + assert jr.jobParameters == [("CPUNormalizationFactor", 9.5)] diff --git a/src/DIRAC/WorkloadManagementSystem/JobWrapper/JobWrapper.py b/src/DIRAC/WorkloadManagementSystem/JobWrapper/JobWrapper.py index 53c803b3400..8da4fbbfa95 100755 --- a/src/DIRAC/WorkloadManagementSystem/JobWrapper/JobWrapper.py +++ b/src/DIRAC/WorkloadManagementSystem/JobWrapper/JobWrapper.py @@ -16,6 +16,7 @@ import sys import time import datetime +import math import shutil import threading import tarfile @@ -124,6 +125,11 @@ def __init__(self, jobID=None, jobReport=None): self.boincUserID = gConfig.getValue("/LocalSite/BoincUserID", 0) self.pilotRef = gConfig.getValue("/LocalSite/PilotReference", "Unknown") self.cpuNormalizationFactor = gConfig.getValue("/LocalSite/CPUNormalizationFactor", 0.0) + if not math.isfinite(self.cpuNormalizationFactor): + self.log.error( + "Ignoring non-finite CPUNormalizationFactor from configuration", str(self.cpuNormalizationFactor) + ) + self.cpuNormalizationFactor = 0.0 self.bufferLimit = gConfig.getValue(self.section + "/BufferLimit", 10485760) self.defaultOutputSE = getDestinationSEList( gConfig.getValue("/Resources/StorageElementGroups/SE-USER", []), self.siteName @@ -225,6 +231,11 @@ def initialize(self, arguments): if not self.cpuNormalizationFactor: self.cpuNormalizationFactor = float(self.ceArgs.get("CPUNormalizationFactor", self.cpuNormalizationFactor)) + if not math.isfinite(self.cpuNormalizationFactor): + self.log.error( + "Ignoring non-finite CPUNormalizationFactor from CE parameters", str(self.cpuNormalizationFactor) + ) + self.cpuNormalizationFactor = 0.0 self.siteName = self.ceArgs.get("Site", self.siteName) # Prepare the working directory, cd to there, and copying eventual extra arguments in it diff --git a/src/DIRAC/WorkloadManagementSystem/JobWrapper/Watchdog.py b/src/DIRAC/WorkloadManagementSystem/JobWrapper/Watchdog.py index 7ff8a46d3b8..63cad8dcb9b 100755 --- a/src/DIRAC/WorkloadManagementSystem/JobWrapper/Watchdog.py +++ b/src/DIRAC/WorkloadManagementSystem/JobWrapper/Watchdog.py @@ -140,6 +140,9 @@ def initialize(self): # thus they need to be multiplied by a large enough factor self.fineTimeLeftLimit = gConfig.getValue(self.section + "/TimeLeftLimit", 150 * self.pollingTime) self.cpuPower = gConfig.getValue("/LocalSite/CPUNormalizationFactor", 1.0) + if not math.isfinite(self.cpuPower): + self.log.error("Ignoring non-finite CPUNormalizationFactor from configuration", str(self.cpuPower)) + self.cpuPower = 1.0 return S_OK() @@ -801,39 +804,31 @@ 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. + """ 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"): 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"): 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"]: diff --git a/src/DIRAC/WorkloadManagementSystem/JobWrapper/test/Test_Watchdog.py b/src/DIRAC/WorkloadManagementSystem/JobWrapper/test/Test_Watchdog.py index 8dc07b2c3d2..2d478fce6a0 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,19 @@ 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 diff --git a/src/DIRAC/WorkloadManagementSystem/scripts/dirac_wms_cpu_normalization.py b/src/DIRAC/WorkloadManagementSystem/scripts/dirac_wms_cpu_normalization.py index eb415f0ae5b..f52207857c8 100755 --- a/src/DIRAC/WorkloadManagementSystem/scripts/dirac_wms_cpu_normalization.py +++ b/src/DIRAC/WorkloadManagementSystem/scripts/dirac_wms_cpu_normalization.py @@ -12,6 +12,8 @@ DB12measured = 15.4 } """ +import math + from db12 import multiple_dirac_benchmark import DIRAC @@ -67,6 +69,11 @@ def main(): gLogger.info("Applying a correction on the CPU power:", corr) cpuPower = round(db12Result / corr, 1) + if not math.isfinite(cpuPower): + gLogger.error( + "Computed CPU power is not finite, falling back to 0.0", f"(db12Result={db12Result}, correction={corr})" + ) + cpuPower = 0.0 gLogger.notice(f"Estimated CPU power is {cpuPower:.1f} HS06")