Merge pull request #8179 from aldbr/main_FIX_watchdog-killing-payload

[9.0] fix(wms): Watchdog does not kill payload properly and runs endlessly

Author: fstagni

src/DIRAC/Core/Utilities/Subprocess.py

@@ -268,18 +268,33 @@ def killChild(self, recursive=True):
 
         :param boolean recursive: flag to kill all descendants
         """
-
-        parent = psutil.Process(self.childPID)
-        children = parent.children(recursive=recursive)
-        children.append(parent)
-        for p in children:
+        pgid = os.getpgid(self.childPID)
+        if pgid != os.getpgrp():
             try:
-                p.send_signal(signal.SIGTERM)
-            except psutil.NoSuchProcess:
+                # Child is in its own group: kill the group
+                os.killpg(pgid, signal.SIGTERM)
+            except OSError:
+                # Process is already dead
                 pass
-        _gone, alive = psutil.wait_procs(children, timeout=10)
-        for p in alive:
-            p.kill()
+        else:
+            # No separate group: walk the tree
+            parent = psutil.Process(self.childPID)
+            procs = parent.children(recursive=recursive)
+            procs.append(parent)
+            for p in procs:
+                try:
+                    p.terminate()
+                except psutil.NoSuchProcess:
+                    pass
+            _gone, alive = psutil.wait_procs(procs, timeout=10)
+            # Escalate any survivors
+            for p in alive:
+                try:
+                    p.kill()
+                except psutil.NoSuchProcess:
+                    pass
+
+        self.childKilled = True
 
     def pythonCall(self, function, *stArgs, **stKeyArgs):
         """call python function :function: with :stArgs: and :stKeyArgs:"""
@@ -405,7 +420,9 @@ def __readFromSystemCommandOutput(self, fd, bufferIndex):
             self.killChild()
             return self.__generateSystemCommandError(1, f"{retDict['Message']} for '{self.cmdSeq}' call")
 
-    def systemCall(self, cmdSeq, callbackFunction=None, shell=False, env=None, preexec_fn=None):
+    def systemCall(
+        self, cmdSeq, callbackFunction=None, shell=False, env=None, preexec_fn=None, start_new_session=False
+    ):
         """system call (no shell) - execute :cmdSeq:"""
 
         if shell:
@@ -426,6 +443,7 @@ def systemCall(self, cmdSeq, callbackFunction=None, shell=False, env=None, preex
                 env=env,
                 universal_newlines=True,
                 preexec_fn=preexec_fn,
+                start_new_session=start_new_session,
             )
             self.childPID = self.child.pid
         except OSError as v:

src/DIRAC/WorkloadManagementSystem/JobWrapper/JobWrapper.py

@@ -1584,7 +1584,9 @@ def run(self):
         start = time.time()
         initialStat = os.times()
         log.verbose("Cmd called", self.cmd)
-        output = self.spObject.systemCall(self.cmd, env=self.exeEnv, callbackFunction=self.sendOutput, shell=True)
+        output = self.spObject.systemCall(
+            self.cmd, env=self.exeEnv, callbackFunction=self.sendOutput, shell=True, start_new_session=True
+        )
         log.verbose(f"Output of system call within execution thread: {output}")
         self.executionResults["Thread"] = output
         timing = time.time() - start

src/DIRAC/WorkloadManagementSystem/JobWrapper/test/Test_JobWrapper.py

@@ -296,6 +296,30 @@ def test_processFailedSubprocess(mocker):
     assert not result["Value"]["watchdogStats"]
 
 
+@pytest.mark.slow
+def test_processKilledSubprocess(mocker):
+    """Test the process method of the JobWrapper class: the job is stalled and is killed by the Watchdog."""
+    jw = JobWrapper()
+    jw.jobArgs = {"CPUTime": 100, "Memory": 1}
+
+    mocker.patch.object(jw, "_JobWrapper__report")
+    mocker.patch.object(jw, "_JobWrapper__setJobParam")
+
+    mock_progress_call = mocker.patch("DIRAC.WorkloadManagementSystem.JobWrapper.Watchdog.Watchdog._checkProgress")
+    mock_progress_call.return_value = S_ERROR("Job is stalled!")
+
+    with tempfile.NamedTemporaryFile(delete=True) as std_out, tempfile.NamedTemporaryFile(delete=True) as std_err:
+        jw.outputFile = std_out.name
+        jw.errorFile = std_err.name
+        result = jw.process("sleep 20", {})
+
+    assert result["OK"]
+    assert result["Value"]["payloadStatus"] == 15  # SIGTERM
+    assert not result["Value"]["payloadOutput"]
+    assert not result["Value"]["payloadExecutorError"]
+    assert result["Value"]["watchdogError"] == "Job is stalled!"  # Error message from the watchdog
+
+
 @pytest.mark.slow
 def test_processQuickExecutionNoWatchdog(mocker):
     """Test the process method of the JobWrapper class: the payload is too fast to start the watchdog."""