[v8r0] fix (WMS): do not report non-finite job parameters

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,14 +69,27 @@ 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
             return self.sendStoredJobParameters()
 
         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"""
 

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)]

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

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"]:

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

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")