From 5fc3f1c4987332b096ba45117080b7628912b311 Mon Sep 17 00:00:00 2001 From: aldbr Date: Tue, 25 Aug 2026 13:27:10 +0200 Subject: [PATCH] fix (WMS): reserve the post-processing margin before advertising the slot The JobAgent advertised the whole of the remaining slot to the matcher and published it in /LocalSite/CPUTimeLeft, but the payload never got all of it: the JobWrapper still has to upload the outputs and the logs once the payload stops, and the Watchdog was quietly holding StopMargin back at the far end for that. Nothing upstream knew. A job whose declared CPUTime just fits the slot is matched into it and then killed before it finishes, and an elastic payload sizing itself against /LocalSite/CPUTimeLeft over-commits by the same margin. Either way the work done so far is lost, never having been uploaded. Take the margin off once, in JobAgent.initialize(), where the slot enters the agent. Every later use of it -- the filling-mode check, the CPU work advertised to the matcher, /LocalSite/CPUTimeLeft and so the payload itself -- then works from what a payload may actually consume, without any of them knowing a reserve exists. The Watchdog no longer deducts it, the budget it reads being already net, and stops the payload when that budget is spent. PushJobAgent is left alone: it post-processes on the server, so its remote allocation runs the payload and nothing else. Restore the graceful stop, on that same budget. A payload killed at the end of its budget loses whatever it had produced but not written; an event generator stopped mid-event loses the whole file. The JDL options for this are still parsed, but the code behind them went with MJF in 1968c85c6d and nothing has sent the signal since. The slot budget is a better clock than MJF was: it is the one __checkTimeLeft already counts down, so the warning and the hard stop cannot disagree. StopSigRegex also matches the command line again, as it did before 0e67f781de, so the application is signalled rather than the shells and wrappers sharing its process tree -- and only within this JobWrapper's own descendants, so other jobs on the node are untouched however well they match. Both thresholds are now in CPU work rather than seconds: StopSigStartWork and StopSigFinishWork replace StopSigStartSeconds and StopSigFinishSeconds. How much work a payload has done, and how much it needs to stop, are questions about computation, and the same computation costs more wall clock on a slower node -- so one figure means the same thing across a heterogeneous fleet where a figure in seconds does not. StopSigFinishWork has no default, leaving the graceful stop off until a value is given; StopSigStartWork defaults to it, so a payload is never stopped having done less work than stopping it costs. All the options may now be set in the CS as well as per job, since a JDL is fixed at submission and cannot reach work already in the system. Unchanged for any job that does not ask: the signal is sent only where StopSigRegex and StopSigFinishWork are both set, and only once. Deleted along the way are wallClockCheckSeconds and wallClockCheckCount, which throttled reading MJF's files from disk and have nothing left to throttle. Documented in a new Allocated time management page: where the budget comes from and in which units, who may spend which part of it, and how to measure the values rather than guess them. BEGINRELEASENOTES *WorkloadManagement FIX: the JobAgent reserves the JobWrapper's StopMargin before advertising the slot to the matcher and publishing it in /LocalSite/CPUTimeLeft, so that jobs are neither matched nor sized against time the post-processing will need. The Watchdog no longer deducts it a second time. NEW: the Watchdog can again signal a payload to wind down before its budget runs out (StopSigRegex, StopSigNumber, StopSigStartWork, StopSigFinishWork), now configurable in Systems/WorkloadManagement//JobWrapper as well as per job in the JDL. CHANGE: StopSigStartSeconds and StopSigFinishSeconds are replaced by StopSigStartWork and StopSigFinishWork, expressed in CPU work. CHANGE: new "Allocated time management" page in the Administrator Guide. ENDRELEASENOTES Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01MXoFjUCdmqT3oP6FEwc8K7 --- .../WorkloadManagement/JobWrapper/index.rst | 48 ++++ .../WorkloadManagement/allocatedTime.rst | 186 ++++++++++++ .../Systems/WorkloadManagement/index.rst | 1 + .../Agent/JobAgent.py | 16 +- .../Agent/test/Test_Agent_JobAgent.py | 24 +- .../JobWrapper/Watchdog.py | 142 +++++++-- .../JobWrapper/test/Test_Watchdog.py | 271 +++++++++++++++++- 7 files changed, 645 insertions(+), 43 deletions(-) create mode 100644 docs/source/AdministratorGuide/Systems/WorkloadManagement/allocatedTime.rst diff --git a/docs/source/AdministratorGuide/Configuration/ConfReference/Systems/WorkloadManagement/JobWrapper/index.rst b/docs/source/AdministratorGuide/Configuration/ConfReference/Systems/WorkloadManagement/JobWrapper/index.rst index 85b11dd2607..fa2f9638762 100644 --- a/docs/source/AdministratorGuide/Configuration/ConfReference/Systems/WorkloadManagement/JobWrapper/index.rst +++ b/docs/source/AdministratorGuide/Configuration/ConfReference/Systems/WorkloadManagement/JobWrapper/index.rst @@ -1,3 +1,5 @@ +.. _cs-JobWrapper: + Systems / WorkloadManagement / / JobWrapper - Sub-subsection ============================================================================ @@ -31,3 +33,49 @@ The options used to configure JobWrapper are showed in the table below: +----------------------+-------------------------------------------------+------------------------------+ | *OutputSandboxLimit* | Limit of sandbox output expressed in MB | OutputSandboxLimit = 10 | +----------------------+-------------------------------------------------+------------------------------+ +| *StopMargin* | Wall-clock seconds at the end of the batch | StopMargin = 300 | +| | slot reserved for uploading the outputs and | | +| | the logs. Deducted once, by the agent that | | +| | publishes the slot budget, so consumers of | | +| | /LocalSite/CPUTimeLeft must not deduct it | | +| | again | | ++----------------------+-------------------------------------------------+------------------------------+ + +Graceful stop +------------- + +A payload killed when its slot runs out loses whatever it had produced but not yet +written. An application that knows how to wind down can be signalled instead, early +enough to finish its current unit of work and write its output; the Watchdog stops it +anyway once the budget is spent, so these only buy it the chance. + +The same four options may be set per job in the JDL, under the same names, which takes +precedence. Setting them here is what reaches work already submitted, whose JDL is fixed. + ++------------------------+-------------------------------------------------+------------------------------+ +| **Name** | **Description** | **Example** | ++------------------------+-------------------------------------------------+------------------------------+ +| *StopSigRegex* | Regular expression matched against the | StopSigRegex = Gauss | +| | command line of the payload processes, | | +| | naming the application to signal. Unset | | +| | means the mechanism is off | | ++------------------------+-------------------------------------------------+------------------------------+ +| *StopSigNumber* | Signal to send to the matching processes. | StopSigNumber = 10 | +| | Default 2 (SIGINT) | | ++------------------------+-------------------------------------------------+------------------------------+ +| *StopSigStartWork* | How much work a payload must have done | StopSigStartWork = 8370 | +| | before it is worth interrupting: below it | | +| | there is nothing to save. Defaults to | | +| | StopSigFinishWork, so a payload is never | | +| | stopped having done less work than stopping | | +| | it costs | | ++------------------------+-------------------------------------------------+------------------------------+ +| *StopSigFinishWork* | How much computation the application needs | StopSigFinishWork = 16740 | +| | to wind down: finish the unit of work in | | +| | progress, close and write its output. In | | +| | CPU work rather than seconds, so that one | | +| | figure means the same on a fast and a slow | | +| | node; divided by this node's | | +| | CPUNormalizationFactor before use. The | | +| | graceful stop does nothing until it is set | | ++------------------------+-------------------------------------------------+------------------------------+ diff --git a/docs/source/AdministratorGuide/Systems/WorkloadManagement/allocatedTime.rst b/docs/source/AdministratorGuide/Systems/WorkloadManagement/allocatedTime.rst new file mode 100644 index 00000000000..72072919e3a --- /dev/null +++ b/docs/source/AdministratorGuide/Systems/WorkloadManagement/allocatedTime.rst @@ -0,0 +1,186 @@ +.. _allocated_time: + +========================================= +Allocated time management +========================================= + +A pilot holds a batch slot for a fixed length of time. Everything that has to happen in +that slot -- matching a job, running the payload, uploading what it produced -- has to fit, +and the pieces are sized by components that must agree on how much is left. This page +describes where that budget comes from, who may spend which part of it, and which options +control it. + +.. contents:: :local: + +Where the budget comes from +=========================== + +The pilot establishes two numbers before any job is matched: how fast this worker node is, +and how long the batch system will let it run. + +.. graphviz:: + + digraph seeding { + rankdir=LR; node [shape=box, fontsize=10]; edge [fontsize=9]; + bench [label="dirac-wms-cpu-normalization\n(benchmarks the node)"]; + norm [label="/LocalSite/\nCPUNormalizationFactor", shape=cylinder]; + batch [label="batch system\n(sacct, qstat, ...)"]; + cs [label="queue maxCPUTime\nin the CS", shape=note]; + queue [label="dirac-wms-get-queue-cpu-time"]; + left [label="/LocalSite/CPUTimeLeft", shape=cylinder]; + + bench -> norm; + batch -> queue [label="seconds left, if reachable"]; + cs -> queue [label="otherwise, minutes x 60"]; + norm -> left [label="multiplied in", style=dashed]; + queue -> left [label="seconds"]; + } + +``dirac-wms-get-queue-cpu-time`` tries three sources in order, and takes the first that +answers: + +#. **The batch system**, through the TimeLeft utility -- ``sacct``, ``qstat``, ``bjobs`` + and so on, selected by ``/LocalSite/BatchSystemInfo``. This is the only source that + knows how much of the slot has already been used. It is unavailable to a payload + running inside a container, which is why the figure is written to the configuration + rather than recomputed later. +#. **The queue's** ``maxCPUTime`` **in the CS**, under + ``/Resources/Sites///CEs//Queues//``. If no queue is identified, + the smallest ``maxCPUTime`` among that CE's queues is used instead. +#. ``/Resources/Computing/CEDefaults/MaxCPUTime``, a global fallback. + +.. warning:: + + These are not in the same unit. ``maxCPUTime`` on a queue is in **minutes** and is + multiplied by 60; ``CEDefaults/MaxCPUTime`` is in **seconds** and is not. A queue + configured as though it were seconds will hand out slots sixty times too long. + +The result, in seconds, is multiplied by ``CPUNormalizationFactor`` and stored as +``/LocalSite/CPUTimeLeft``. That field is therefore **CPU work**, not seconds: wall-clock +seconds times the power of the node. + +.. note:: + + ``CPUNormalizationFactor`` comes from a benchmark run on the node itself, divided by + ``Operations/JobScheduling/CPUNormalizationCorrection``. Everything downstream is + proportional to it, so an inaccurate benchmark scales the whole budget with it. + +Who may spend which part +======================== + +There is one writer of ``/LocalSite/CPUTimeLeft`` after the pilot: the JobAgent. Everything +else reads what it publishes. + +.. graphviz:: + + digraph budget { + rankdir=LR; node [shape=box, fontsize=10]; edge [fontsize=9]; + agent [label="JobAgent", style=filled, fillcolor="#e8e8e8"]; + cfg [label="/LocalSite/CPUTimeLeft", shape=cylinder]; + matcher [label="Matcher"]; + wd [label="Watchdog"]; + payload [label="payload\n(elastic sizing)"]; + + cfg -> agent [label="read once at\ninitialize()"]; + agent -> cfg [label="republished each cycle,\nStopMargin already off"]; + agent -> matcher [label="CPU work advertised\nby the CE"]; + cfg -> wd [label="read at job start"]; + cfg -> payload [label="read by the application"]; + } + +The JobAgent subtracts ``StopMargin`` once, when it first reads the slot. Everything +downstream therefore works from what a payload may actually *consume*, and none of them +needs to know a reserve exists. **Consumers must not subtract it again.** + +Then the end of a slot looks like this: + +.. graphviz:: + + digraph timeline { + rankdir=LR; node [shape=box, fontsize=10, width=1.4]; edge [fontsize=9]; + run [label="payload runs", style=filled, fillcolor="#dff0d8"]; + wind [label="winding down", style=filled, fillcolor="#fcf8e3"]; + up [label="uploads", style=filled, fillcolor="#f2dede"]; + end [label="slot ends", shape=plaintext]; + + run -> wind [label="StopSigNumber sent"]; + wind -> up [label="payload exits,\nor is killed"]; + up -> end; + } + +``StopSigFinishWork`` and ``StopMargin`` both reserve time at the end of the slot, but they +are not interchangeable: + +=================== =============================== ========================== ============== +Reserve Whose work When Unit +=================== =============================== ========================== ============== +StopSigFinishWork the payload's, finishing the before the payload exits CPU work + unit of work in progress +StopMargin the JobWrapper's, uploading after the payload exits wall clock + the outputs and the logs +=================== =============================== ========================== ============== + +They also differ in reach. ``StopMargin`` applies to every job and shrinks the budget that +the matcher and the payload are told about. ``StopSigFinishWork`` changes no budget at all: +it only decides how early to tap an opted-in payload on the shoulder, and comes out of that +payload's own share. + +The units differ because the work differs. Uploading a file takes the same wall clock on any +node, so ``StopMargin`` is in seconds. How much computation a payload has done, and how much +it needs to stop, are both questions about work, so the ``StopSig`` thresholds are in CPU work +and are divided by the node's ``CPUNormalizationFactor`` before use -- which means one figure means the same thing across +a heterogeneous fleet, where a figure in seconds would not: + +========= ========== ============== ========== +CPU work slow node typical node fast node +========= ========== ============== ========== +30 3 s 1 s 1 s +200 20 s 7 s 4 s +1000 100 s 36 s 20 s +8000 800 s 287 s 160 s +========= ========== ============== ========== + +The Watchdog stops the payload when the published budget is spent, whether or not it was +signalled first, so a payload that ignores the signal still cannot eat into the uploads. + +Choosing the values +=================== + +All of these live in ``/Systems/WorkloadManagement//JobWrapper`` and are listed in +:ref:`the configuration reference `. The same names may be set per job in the +JDL, which takes precedence -- but a JDL is fixed at submission, so only the CS reaches work +that is already in the system. + +``StopMargin`` + Measure it rather than guess. The JobWrapper reports ``Completing`` the moment the + payload exits and ``Done`` when it has finished uploading, so the span between those two + records in the JobLoggingDB is exactly what this has to cover. Take a high percentile + rather than the mean, and remember that jobs killed *during* post-processing never reach + ``Done`` and so are absent from the sample: the true requirement is if anything larger + than what is measured. + +``StopSigRegex`` + A regular expression matched against the command line of this job's payload processes. + It names the application to signal and nothing else: signalling the shells, wrappers and + container entrypoints that share the process tree is at best useless and at worst kills + the job outright. Only descendants of this JobWrapper are considered, so other jobs on + the same node are never affected however well their command lines match. Leaving it + unset disables the graceful stop. + +``StopSigNumber`` + Whichever signal the application handles. Verify it rather than assume: an application + that does not handle the signal is killed by it, losing exactly the work the mechanism + exists to save. + +``StopSigStartWork`` + How much work a payload must have done before it is worth interrupting. Below that there + is nothing to save, and a payload matched into a slot shorter than its own wind-down + would otherwise be signalled at once and report success having produced nothing. It + defaults to ``StopSigFinishWork``, so a payload is never stopped having done less work + than stopping it costs. + +``StopSigFinishWork`` + How much computation the application needs to wind down -- typically the cost of one unit + of work, plus what it takes to close and write its output. The graceful stop does nothing + until this is set. Err high: reserving too much costs a few units of work at the end of a + slot, while reserving too little costs everything the payload had not yet written. diff --git a/docs/source/AdministratorGuide/Systems/WorkloadManagement/index.rst b/docs/source/AdministratorGuide/Systems/WorkloadManagement/index.rst index 2b0ab2319a7..ec246a72cc0 100644 --- a/docs/source/AdministratorGuide/Systems/WorkloadManagement/index.rst +++ b/docs/source/AdministratorGuide/Systems/WorkloadManagement/index.rst @@ -87,5 +87,6 @@ The following sections add some detail for the WMS systems. JobsPriorities JobsMatching tagsAndJobs + allocatedTime multiProcessorJobs InputDataResolution diff --git a/src/DIRAC/WorkloadManagementSystem/Agent/JobAgent.py b/src/DIRAC/WorkloadManagementSystem/Agent/JobAgent.py index 3860b6e3ea1..2d4970af692 100755 --- a/src/DIRAC/WorkloadManagementSystem/Agent/JobAgent.py +++ b/src/DIRAC/WorkloadManagementSystem/Agent/JobAgent.py @@ -108,8 +108,15 @@ def initialize(self): if not result["OK"]: return result - # Read initial CPU work left from config (seeded by pilot via dirac-wms-get-queue-cpu-time) - self.initCPUWork = gConfig.getValue("/LocalSite/CPUTimeLeft", self.initCPUWork) + # This is the factor to convert raw CPU to Normalized units (based on the CPU Model) + self.cpuPower = gConfig.getValue("/LocalSite/CPUNormalizationFactor", self.cpuPower) + + # Read initial CPU work left from config (seeded by pilot via dirac-wms-get-queue-cpu-time), + # less the tail of the slot the JobWrapper needs for the uploads. Taking it off here, once, + # leaves every later use of the slot working from what a payload may actually consume. + # An unbenchmarked node has no cpuPower to express the margin in, and reserves nothing. + stopMargin = gConfig.getValue("/Systems/WorkloadManagement/JobWrapper/StopMargin", 300) + self.initCPUWork = max(0.0, gConfig.getValue("/LocalSite/CPUTimeLeft", 0.0) - stopMargin * self.cpuPower) self.cpuWorkLeft = self.initCPUWork self.initTime = time.time() @@ -118,8 +125,6 @@ def initialize(self): self.pilotReference = gConfig.getValue("/LocalSite/PilotReference", self.pilotReference) self.defaultProxyLength = gConfig.getValue("/Registry/DefaultProxyLifeTime", self.defaultProxyLength) # Agent options - # This is the factor to convert raw CPU to Normalized units (based on the CPU Model) - self.cpuPower = gConfig.getValue("/LocalSite/CPUNormalizationFactor", self.cpuPower) self.jobSubmissionDelay = self.am_getOption("SubmissionDelay", self.jobSubmissionDelay) self.fillingMode = self.am_getOption("FillingModeFlag", self.fillingMode) self.minimumCPUWork = self.am_getOption("MinimumTimeLeft", self.minimumCPUWork) @@ -404,6 +409,9 @@ def _computeCPUWorkLeft(self): via dirac-wms-get-queue-cpu-time). The elapsed wall-clock time is multiplied by the CPU power to get the consumed CPU work. + What is counted down is what a payload may consume: initialize() has already taken + the StopMargin off. + :return: cpu work left (wall-clock time left * cpu power) """ elapsed = time.time() - self.initTime diff --git a/src/DIRAC/WorkloadManagementSystem/Agent/test/Test_Agent_JobAgent.py b/src/DIRAC/WorkloadManagementSystem/Agent/test/Test_Agent_JobAgent.py index 95ba8fb64d3..9d731156a39 100644 --- a/src/DIRAC/WorkloadManagementSystem/Agent/test/Test_Agent_JobAgent.py +++ b/src/DIRAC/WorkloadManagementSystem/Agent/test/Test_Agent_JobAgent.py @@ -8,9 +8,10 @@ from pathlib import Path import pytest +from unittest.mock import patch from DIRAC.Core.Security.X509Chain import X509Chain # pylint: disable=import-error -from DIRAC import S_ERROR, S_OK, gLogger +from DIRAC import S_ERROR, S_OK, gConfig, gLogger from DIRAC.Resources.Computing.ComputingElementFactory import ComputingElementFactory from DIRAC.Resources.Computing.test.Test_PoolComputingElement import badJobScript, jobScript from DIRAC.WorkloadManagementSystem.Agent.JobAgent import JobAgent @@ -175,6 +176,27 @@ def test__computeCPUWorkLeft(mocker, initCPUWork, cpuPower, elapsedSeconds, expe assert abs(result - expectedTimeLeft) < 10 +def test_the_upload_margin_is_reserved_once_for_the_whole_slot(mocker): + """Filling mode: two jobs matched into one slot share one margin, not one each. + + initialize() carves it out of initCPUWork and the cycles only count down from there, + so the second match is short by the time the first used and by nothing else. + """ + power, slotSeconds, margin = 27.9, 3600, 300 + mocker.patch("DIRAC.WorkloadManagementSystem.Agent.JobAgent.AgentModule.__init__") + jobAgent = JobAgent("Test", "Test1") + jobAgent.log = gLogger + jobAgent.cpuPower = power + # as initialize() computes it, from the pilot's /LocalSite/CPUTimeLeft + jobAgent.initCPUWork = max(0.0, slotSeconds * power - margin * power) + + jobAgent.initTime = time.time() + assert jobAgent._computeCPUWorkLeft() / power == pytest.approx(slotSeconds - margin, abs=2) + + jobAgent.initTime = time.time() - 600 # second match, ten minutes in + assert jobAgent._computeCPUWorkLeft() / power == pytest.approx(slotSeconds - 600 - margin, abs=2) + + @pytest.mark.parametrize( "cpuWorkLeft, fillingMode, expectedResult", [ diff --git a/src/DIRAC/WorkloadManagementSystem/JobWrapper/Watchdog.py b/src/DIRAC/WorkloadManagementSystem/JobWrapper/Watchdog.py index 652c6ef938f..0f8800fd59f 100755 --- a/src/DIRAC/WorkloadManagementSystem/JobWrapper/Watchdog.py +++ b/src/DIRAC/WorkloadManagementSystem/JobWrapper/Watchdog.py @@ -13,6 +13,7 @@ import getpass import math import os +import re import socket import time from pathlib import Path @@ -32,8 +33,9 @@ class Watchdog: ############################################################################# def __init__(self, pid, exeThread, spObject, jobCPUTime, memoryLimit=0, processors=1, jobArgs={}): """Constructor, takes system flag as argument.""" - self.stopSigStartSeconds = int(jobArgs.get("StopSigStartSeconds", 1800)) # 30 minutes - self.stopSigFinishSeconds = int(jobArgs.get("StopSigFinishSeconds", 1800)) # 30 minutes + self.jobArgs = jobArgs + self.stopSigStartWork = int(jobArgs.get("StopSigStartWork", 0)) # normalized units + self.stopSigFinishWork = int(jobArgs.get("StopSigFinishWork", 0)) # normalized units self.stopSigNumber = int(jobArgs.get("StopSigNumber", 2)) # SIGINT self.stopSigRegex = jobArgs.get("StopSigRegex", None) self.stopSigSent = False @@ -68,7 +70,6 @@ def __init__(self, pid, exeThread, spObject, jobCPUTime, memoryLimit=0, processo self.pollingTime = 10 # 10 seconds self.checkingTime = 30 * 60 # 30 minute period self.minCheckingTime = 20 * 60 # 20 mins - self.wallClockCheckSeconds = 5 * 60 # 5 minutes self.maxWallClockTime = 3 * 24 * 60 * 60 # e.g. 4 days self.jobPeekFlag = 1 # on / off self.minDiskSpace = 10 # MB @@ -78,12 +79,10 @@ def __init__(self, pid, exeThread, spObject, jobCPUTime, memoryLimit=0, processo self.minCPUWallClockRatio = 5 # ratio age self.nullCPULimit = 5 # After 5 sample times return null CPU consumption kill job self.checkCount = 0 - self.wallClockCheckCount = 0 self.nullCPUCount = 0 self.initialWallClockLeft = 0 self.wallClockLeft = 0 - self.stopMargin = 300 # seconds of wall-clock time to reserve for post-processing self.cpuPower = 1.0 self.processors = processors @@ -127,15 +126,13 @@ def initialize(self): ) self.checkingTime = self.minCheckingTime - self.cpuPower = gConfig.getValue("/LocalSite/CPUNormalizationFactor", 1.0) - self.stopMargin = gConfig.getValue(self.section + "/StopMargin", self.stopMargin) + self.__resolveStopSigOptions() - # Read CPU work left from config (written by JobAgent) and convert to wall-clock seconds + self.cpuPower = gConfig.getValue("/LocalSite/CPUNormalizationFactor", 1.0) + # CPU work left, as published by the JobAgent and so already net of the StopMargin + # it set aside for the uploads. Converted to the wall clock this counts down. cpuWorkLeft = gConfig.getValue("/LocalSite/CPUTimeLeft", 0) - if cpuWorkLeft and self.cpuPower: - self.initialWallClockLeft = cpuWorkLeft / self.cpuPower - else: - self.initialWallClockLeft = 0 + self.initialWallClockLeft = cpuWorkLeft / self.cpuPower if cpuWorkLeft and self.cpuPower else 0 return S_OK() @@ -176,14 +173,6 @@ def execute(self): self.log.info("Process to monitor has completed, Watchdog will exit.") return S_OK("Ended") - # WallClock checks every self.wallClockCheckSeconds, but only if StopSigRegex is defined in JDL - if ( - not self.stopSigSent - and self.stopSigRegex is not None - and (time.time() - self.initialValues["StartTime"]) > self.wallClockCheckSeconds * self.wallClockCheckCount - ): - self.wallClockCheckCount += 1 - # Check time left on every poll cycle (cheap: just reads wall-clock) if self.testTimeLeft: result = self.__checkTimeLeft() @@ -726,21 +715,122 @@ def __checkTimeLeft(self): The initial wall-clock time left is read from the local configuration at initialization (written by the JobAgent). A simple countdown is then used to determine the remaining time. - Returns S_ERROR when the remaining wall-clock time drops below the configurable StopMargin - (default: 300s), leaving enough time for post-processing (output upload, cleanup, etc.). + That budget is already net of the ``StopMargin`` the JobAgent set aside for the + uploads, so the payload is stopped as soon as it is exhausted. + + Returning S_ERROR gets the payload killed, which is too late to ask it to wind + down: that takes time, and what is left by then belongs to the uploads. Hence the + second, earlier threshold on the S_OK path. """ if not self.initialWallClockLeft: return S_OK("TimeLeft not available") - elapsed = time.time() - self.initialValues["StartTime"] - wallClockLeft = self.initialWallClockLeft - elapsed - self.wallClockLeft = wallClockLeft + self.wallClockLeft = self.initialWallClockLeft - (time.time() - self.initialValues["StartTime"]) - if wallClockLeft < self.stopMargin: + if self.wallClockLeft <= 0: return S_ERROR(JobMinorStatus.JOB_EXCEEDED_CPU) + self._askPayloadToStop() return S_OK() + ############################################################################# + def __resolveStopSigOptions(self): + """Take from the CS the graceful-stop settings the JDL did not give. + + A JDL is fixed when the job is submitted, so only the CS can reach work already in + the system. + """ + for option, attribute in ( + ("StopSigStartWork", "stopSigStartWork"), + ("StopSigFinishWork", "stopSigFinishWork"), + ("StopSigNumber", "stopSigNumber"), + ("StopSigRegex", "stopSigRegex"), + ): + if option not in self.jobArgs: + setattr(self, attribute, gConfig.getValue(f"{self.section}/{option}", getattr(self, attribute))) + + ############################################################################# + def _askPayloadToStop(self): + """Ask the payload to stop while there is still time for it to do so. + + A payload killed when its budget runs out loses whatever it had produced but not + written: an event generator stopped mid-event loses the whole file. A job that + knows how to wind down names its application in ``StopSigRegex`` and is sent + ``StopSigNumber`` early enough to finish and write. ``__checkTimeLeft`` remains the + backstop for one that ignores it. + + Always returns S_OK: failing to signal is not a reason to fail the job. + """ + if self.stopSigSent or not self.stopSigRegex or not self.stopSigFinishWork: + return S_OK() + if not self.initialWallClockLeft or not self.cpuPower: + return S_OK() + + # Both thresholds are in CPU work, not seconds: how much a payload has done, and how + # much it needs to stop, are questions about computation, and the same computation + # costs more wall clock on a slower node. + windDown = self.stopSigFinishWork / self.cpuPower + + # Nothing worth saving in a payload that has not done as much work as stopping it + # costs. Without this, one matched into a slot shorter than its wind-down would be + # signalled at once and report success having produced nothing. + startWork = self.stopSigStartWork or self.stopSigFinishWork + if (time.time() - self.initialValues["StartTime"]) < startWork / self.cpuPower: + return S_OK() + if self.wallClockLeft > windDown: + return S_OK() + + self.stopSigSent = True # sent once, whether or not the payload acts on it + signalled = self._signalPayload() + self.log.info( + "Asked the payload to stop while it still can", + f"signal={self.stopSigNumber} processes={signalled} " + f"wallClockLeft={self.wallClockLeft:.0f}s windDown={windDown:.0f}s", + ) + return S_OK() + + ############################################################################# + def _signalPayload(self): + """Send ``StopSigNumber`` to the descendants matching ``StopSigRegex``. + + Only descendants of this JobWrapper, so other jobs sharing the node are untouched + however well their command lines match. Unlike killChild this also spares the parent + -- the JobWrapper still has the outputs to upload -- and does not escalate to + SIGKILL: the point is to ask, and to leave time to answer. + + The regex picks the application out of the shells, wrappers and container entrypoints + sharing its tree, which must not be signalled in its place. + + :return: the number of processes signalled + :rtype: int + """ + try: + children = psutil.Process(self.wrapperPID).children(recursive=True) + except psutil.NoSuchProcess: + return 0 + + try: + matcher = re.compile(self.stopSigRegex) + except re.error as exc: + self.log.error("Not a usable StopSigRegex, signalling nothing", f"{self.stopSigRegex!r}: {exc}") + return 0 + + signalled = 0 + for child in children: + try: + if not matcher.search(" ".join(child.cmdline()) or child.name()): + continue + child.send_signal(self.stopSigNumber) + signalled += 1 + except psutil.NoSuchProcess: + continue + except psutil.Error as exc: + self.log.warn("Could not signal payload process", f"pid={child.pid}: {exc}") + + if not signalled: + self.log.warn("StopSigRegex matched no payload process", f"{self.stopSigRegex!r} of {len(children)}") + return signalled + ############################################################################# def __getUsageSummary(self): """Returns average load, memory etc. over execution of job thread""" diff --git a/src/DIRAC/WorkloadManagementSystem/JobWrapper/test/Test_Watchdog.py b/src/DIRAC/WorkloadManagementSystem/JobWrapper/test/Test_Watchdog.py index 8501b567466..a329bfb3069 100644 --- a/src/DIRAC/WorkloadManagementSystem/JobWrapper/test/Test_Watchdog.py +++ b/src/DIRAC/WorkloadManagementSystem/JobWrapper/test/Test_Watchdog.py @@ -1,11 +1,19 @@ """unit test for Watchdog.py""" import os +import signal +import subprocess +import sys import time from unittest.mock import MagicMock, patch +import psutil +import pytest + # sut from DIRAC.WorkloadManagementSystem.JobWrapper.Watchdog import Watchdog +SECTION = "/Systems/WorkloadManagement/JobWrapper" + mock_exeThread = MagicMock() mock_spObject = MagicMock() @@ -42,11 +50,11 @@ def test__performChecksFull(): class TestCheckTimeLeft: """Tests for the simplified wall-clock countdown time-left logic.""" - def _make_watchdog(self, initialWallClockLeft=0, stopMargin=300, cpuPower=10.0): + def _make_watchdog(self, initialWallClockLeft=0, cpuPower=10.0): + """A Watchdog holding a budget the JobAgent has already taken StopMargin off.""" pid = os.getpid() wd = Watchdog(pid, mock_exeThread, mock_spObject, 5000) wd.initialWallClockLeft = initialWallClockLeft - wd.stopMargin = stopMargin wd.cpuPower = cpuPower wd.initialValues = {"StartTime": time.time()} wd.testTimeLeft = 1 @@ -60,17 +68,27 @@ def test_time_left_not_available(self): def test_plenty_of_time_left(self): """When there's plenty of time, the check should pass.""" - wd = self._make_watchdog(initialWallClockLeft=3600, stopMargin=300) + wd = self._make_watchdog(initialWallClockLeft=3600) result = wd._Watchdog__checkTimeLeft() assert result["OK"] is True assert wd.wallClockLeft > 3000 - def test_below_stop_margin(self): - """When wall-clock left drops below stop margin, the check should fail.""" - wd = self._make_watchdog(initialWallClockLeft=3600, stopMargin=300) - # Pretend the job started 3500 seconds ago (only 100s left, below 300s margin) + def test_budget_not_exhausted(self): + """The payload keeps the whole published budget: it is already net of the margin. + + Under the old behaviour the Watchdog took another StopMargin off here and stopped + the job with 100 s still on its clock. + """ + wd = self._make_watchdog(initialWallClockLeft=3600) wd.initialValues["StartTime"] = time.time() - 3500 result = wd._Watchdog__checkTimeLeft() + assert result["OK"] is True + + def test_budget_exhausted(self): + """Once the published budget runs out, what is left of the slot is the reserve.""" + wd = self._make_watchdog(initialWallClockLeft=3600) + wd.initialValues["StartTime"] = time.time() - 3700 + result = wd._Watchdog__checkTimeLeft() assert not result["OK"] def test_time_left_updates_heartbeat_value(self): @@ -80,11 +98,10 @@ def test_time_left_updates_heartbeat_value(self): # wallClockLeft should be approximately 3600s assert wd.wallClockLeft > 3500 - def test_exact_stop_margin_boundary(self): - """When wall-clock left equals stop margin, the check should fail (< not <=).""" - wd = self._make_watchdog(initialWallClockLeft=1000, stopMargin=300) - # 699s elapsed → 301s left, which is not < 300 → should pass - wd.initialValues["StartTime"] = time.time() - 699 + def test_boundary(self): + """A second still on the clock is a second the payload may use.""" + wd = self._make_watchdog(initialWallClockLeft=1000) + wd.initialValues["StartTime"] = time.time() - 998 result = wd._Watchdog__checkTimeLeft() assert result["OK"] is True @@ -104,3 +121,233 @@ def test_initialize_reads_config(self, mock_gConfig): # 36000 / 10.0 = 3600 wall-clock seconds assert wd.initialWallClockLeft == 3600.0 + + +class TestConcurrentJobsInOneSlot: + """A PoolComputingElement runs several jobs side by side in the same batch slot.""" + + SLOT_SECONDS = 3600 + MARGIN = 300 + + @pytest.mark.parametrize( + "secondsIntoSlot, stillRunning", + [(SLOT_SECONDS - MARGIN - 60, True), (SLOT_SECONDS - MARGIN + 60, False)], + ) + def test_jobs_matched_at_different_times_stop_together(self, secondsIntoSlot, stillRunning): + """Both end at the slot's end, not at their own start plus a whole slot. + + The JobAgent publishes what is left *now*, so the job matched ten minutes later is + handed a budget shorter by exactly those ten minutes. + """ + now = time.time() + for startedAt in (0, 600): # two matches, ten minutes apart + wd = Watchdog(os.getpid(), MagicMock(), MagicMock(), 5000) + wd.initialWallClockLeft = self.SLOT_SECONDS - startedAt - self.MARGIN + wd.initialValues = {"StartTime": now - (secondsIntoSlot - startedAt)} + wd.testTimeLeft = 1 + assert wd._Watchdog__checkTimeLeft()["OK"] is stillRunning, f"job matched at {startedAt}s" + + +#: A job that knows how to wind down: signal it 10 min out, once it has run 5 min. +#: 16740 normalized units is 600 s of wall clock on a node benchmarked at 27.9. +GRACEFUL = {"StopSigRegex": "gauss", "StopSigFinishWork": 16740} + + +@patch.object(Watchdog, "_signalPayload") +class TestGracefulStop: + """Signalling the payload before its budget runs out. Opt-in through StopSigRegex.""" + + def _make_watchdog(self, jobArgs=None, initialWallClockLeft=3600, elapsed=0): + wd = Watchdog(os.getpid(), MagicMock(), MagicMock(), 5000, jobArgs=jobArgs or {}) + wd.initialWallClockLeft = initialWallClockLeft + wd.initialValues = {"StartTime": time.time() - elapsed} + wd.testTimeLeft = 1 + wd.exeThread.is_alive.return_value = True + wd.cpuPower = 27.9 + return wd + + @pytest.mark.parametrize( + "cpuPower, elapsed, expectSignalled", + [ + # 16740 units buys 600 s here, so 700 s of budget left is still comfortable... + (27.9, 2900, False), + (27.9, 3100, True), + # ...but on a node half as fast the same work needs 1200 s, and 700 s is not + # enough. A figure in seconds could not tell those two nodes apart. + (13.95, 2900, True), + ], + ) + def test_the_wind_down_is_budgeted_in_cpu_work(self, mock_signal, cpuPower, elapsed, expectSignalled): + """Finishing the unit of work in progress costs more wall clock on a slower node.""" + wd = self._make_watchdog(jobArgs=GRACEFUL, initialWallClockLeft=3600, elapsed=elapsed) + wd.cpuPower = cpuPower + wd._Watchdog__checkTimeLeft() + assert wd.stopSigSent is expectSignalled + + def test_signals_the_payload_with_budget_still_on_the_clock(self, mock_signal): + """3600 s of budget, 3000 gone: the payload is asked to stop, and told with 600 left. + + Those 600 s are the point: an application asked to stop needs time to stop, so the + ask cannot wait for the budget to run out. + """ + wd = self._make_watchdog(jobArgs=GRACEFUL, elapsed=3000) + wd.execute() + assert wd.stopSigSent is True + assert wd.wallClockLeft == pytest.approx(GRACEFUL["StopSigFinishWork"] / 27.9, abs=2) + mock_signal.assert_called_once_with() + + @pytest.mark.parametrize( + "jobArgs, initialWallClockLeft, elapsed, why", + [ + ({"StopSigFinishWork": 16740}, 3600, 3000, "no StopSigRegex, so opted out"), + ({**GRACEFUL, "StopSigRegex": ""}, 3600, 3000, "an empty regex opts out, it does not match everything"), + (GRACEFUL, 3600, 600, "plenty of budget left"), + (GRACEFUL, 300, 60, "barely started: less work done than stopping it would cost"), + (GRACEFUL, 0, 3000, "no budget published, so no deadline to act on"), + ], + ) + def test_stays_silent(self, mock_signal, jobArgs, initialWallClockLeft, elapsed, why): + wd = self._make_watchdog(jobArgs=jobArgs, initialWallClockLeft=initialWallClockLeft, elapsed=elapsed) + wd._Watchdog__checkTimeLeft() + assert wd.stopSigSent is False, why + mock_signal.assert_not_called() + + @pytest.mark.parametrize( + "elapsed, expectSignalled", + [ + # 16740 units is 600 s here, so the payload must have run 600 s to have done as + # much work as stopping it will cost. 500 s in, it has not, however near the end. + (500, False), + (700, True), + ], + ) + def test_the_start_guard_defaults_to_the_cost_of_stopping(self, mock_signal, elapsed, expectSignalled): + """Unset, StopSigStartWork is what winding down costs: below that there is nothing to save. + + Without it a payload matched into a slot shorter than its own wind-down would be + signalled immediately and report success having produced nothing. + """ + wd = self._make_watchdog(jobArgs=GRACEFUL, initialWallClockLeft=1000, elapsed=elapsed) + assert wd._Watchdog__checkTimeLeft()["OK"] is True + assert wd.stopSigSent is expectSignalled + + def test_signals_once_then_leaves_the_hard_stop_to_finish(self, mock_signal): + """Later polls must not keep signalling, and a payload that ignores it is still killed.""" + wd = self._make_watchdog(jobArgs=GRACEFUL, elapsed=3000) + for _ in range(5): + assert wd._Watchdog__checkTimeLeft()["OK"] is True + assert mock_signal.call_count == 1 + + wd.initialValues["StartTime"] = time.time() - 3700 + assert not wd._Watchdog__checkTimeLeft()["OK"] + + @pytest.mark.parametrize( + "cs, jobArgs, expected", + [ + ({}, {}, (None, 2, 0, 0)), # off, on the built-in defaults + ( # the CS alone turns it on: no JDL change, so productions already submitted are covered + { + f"{SECTION}/StopSigRegex": "Gauss", + f"{SECTION}/StopSigNumber": 10, + f"{SECTION}/StopSigStartWork": 8370, + f"{SECTION}/StopSigFinishWork": 5580, + }, + {}, + ("Gauss", 10, 8370, 5580), + ), + # a job that knows better than the site-wide setting still gets its way + ( + {f"{SECTION}/StopSigRegex": "Gauss"}, + {"StopSigRegex": "myApp", "StopSigNumber": "1"}, + ("myApp", 1, 0, 0), + ), + # and can opt out of it + ({f"{SECTION}/StopSigRegex": "Gauss"}, {"StopSigRegex": ""}, ("", 2, 0, 0)), + ], + ) + def test_settings_come_from_the_jdl_then_the_cs(self, mock_signal, cs, jobArgs, expected): + with patch("DIRAC.WorkloadManagementSystem.JobWrapper.Watchdog.gConfig") as watchdogCfg: + watchdogCfg.getValue.side_effect = lambda key, default=None: cs.get(key, default) + wd = Watchdog(os.getpid(), MagicMock(), MagicMock(), 5000, jobArgs=jobArgs) + wd.calibrate() + wd.initialize() + + assert ( + wd.stopSigRegex, + wd.stopSigNumber, + wd.stopSigStartWork, + wd.stopSigFinishWork, + ) == expected + + +class TestSignalPayload: + """Which processes the signal reaches, and which it must not.""" + + @staticmethod + def _standIn(*payloads): + """A stand-in JobWrapper that spawns the given payloads and outlives them.""" + spawns = "; ".join(f"subprocess.Popen({list(p)!r})" for p in payloads) + wrapper = subprocess.Popen([sys.executable, "-c", f"import subprocess, time; {spawns}; time.sleep(60)"]) + deadline = time.time() + 30 + children = [] + while len(children) < len(payloads) and time.time() < deadline: + children = psutil.Process(wrapper.pid).children(recursive=True) + time.sleep(0.05) + assert len(children) == len(payloads), "the stand-in never spawned its payloads" + return wrapper, {" ".join(c.cmdline()): c for c in children} + + @staticmethod + def _watchdog(pid, nameRegex): + wd = Watchdog(pid, MagicMock(), MagicMock(), 5000, jobArgs={"StopSigRegex": nameRegex}) + wd.stopSigNumber = int(signal.SIGTERM) + return wd + + @staticmethod + def _stopped(process): + deadline = time.time() + 30 + while process.is_running() and process.status() != psutil.STATUS_ZOMBIE and time.time() < deadline: + time.sleep(0.05) + return not process.is_running() or process.status() == psutil.STATUS_ZOMBIE + + def test_signals_the_named_application_only(self): + """Not the parent, which still has the outputs to upload, nor its siblings.""" + wrapper, payloads = self._standIn(["sleep", "61"], ["sleep", "62"]) + try: + assert self._watchdog(wrapper.pid, "sleep 61")._signalPayload() == 1 + assert self._stopped(payloads["sleep 61"]), "the named payload ignored the signal" + assert payloads["sleep 62"].is_running(), "a process the regex did not name was signalled" + assert wrapper.poll() is None, "the wrapper was signalled too" + finally: + wrapper.kill() + wrapper.wait(timeout=30) + + def test_leaves_other_jobs_on_the_node_alone(self): + """A node runs many jobs whose command lines all match. Only our own tree is ours. + + The search starts from this JobWrapper's pid, so another job's payload is out of + reach however well it matches. + """ + ours, ourPayloads = self._standIn(["sleep", "61"]) + theirs, theirPayloads = self._standIn(["sleep", "61"]) + try: + assert self._watchdog(ours.pid, "sleep 61")._signalPayload() == 1 + assert self._stopped(ourPayloads["sleep 61"]) + assert theirPayloads["sleep 61"].is_running(), "signalled another job's payload" + finally: + for wrapper in (ours, theirs): + wrapper.kill() + wrapper.wait(timeout=30) + + @pytest.mark.parametrize("nameRegex", ["no-such-payload", "[unclosed"]) + def test_signals_nothing_it_cannot_name(self, nameRegex): + """A regex matching nothing, or not compiling, must not fall back to signalling all.""" + wrapper, payloads = self._standIn(["sleep", "61"]) + try: + assert self._watchdog(wrapper.pid, nameRegex)._signalPayload() == 0 + assert payloads["sleep 61"].is_running() + finally: + wrapper.kill() + wrapper.wait(timeout=30) + + def test_an_exited_payload_is_not_an_error(self): + assert self._watchdog(999999999, "anything")._signalPayload() == 0