Refactored kill/killpg mechanism to be unified and actually fail on n… (#625)

* Refactored kill/killpg mechanism to be unified and actually fail on not process not found

* Implementation change refactor and clarifications

Author: ArneTR

lib/process_helpers.py

@@ -2,36 +2,28 @@
 import os
 import subprocess
 
-def kill_ps(ps_to_kill):
-    print('Killing processes')
-    for ps_info in ps_to_kill:
-        pid = ps_info['ps'].pid
-        print(f"Trying to kill {ps_info['cmd']} with PID: {pid}")
-        try:
-            if ps_info['ps_group'] is True:
-                try:
-                    ps_group_id = os.getpgid(pid)
-                    print(f" with process group {ps_group_id}")
-                    os.killpg(os.getpgid(pid), signal.SIGTERM)
-                    try:
-                        ps_info['ps'].wait(timeout=5)
-                    except subprocess.TimeoutExpired:
-                        # If the process hasn't gracefully exited after 5 seconds we kill it
-                        os.killpg(os.getpgid(pid), signal.SIGKILL)
-                except ProcessLookupError:
-                    # process may be not have been in a process group
-                    print(f"Could not find process-group for {pid}")
-            # always, just in case the calling process (typically the shell) did not die
-            os.kill(pid, signal.SIGTERM)
-            try:
-                ps_info['ps'].wait(timeout=5)
-            except subprocess.TimeoutExpired:
-                # If the process hasn't gracefully exited after 5 seconds we kill it
-                os.kill(pid, signal.SIGKILL)
+def kill_pg(ps, cmd):
+    pgid = os.getpgid(ps.pid)
+    print(f"Trying to kill {cmd} with PGID: {pgid}")
+
+    os.killpg(pgid, signal.SIGTERM)
+    try:
+        ps.wait(timeout=10)
+    except subprocess.TimeoutExpired as exc:
+        # If the process hasn't gracefully exited after 5 seconds we kill it
+        os.killpg(pgid, signal.SIGKILL)
+        raise RuntimeError(f"Killed the process {cmd} with SIGKILL. This could lead to corrupted data!") from exc
 
-        except ProcessLookupError:
-            # process may already have ended or been killed in the process group
-            print(f"Could not find process {pid}")
+def kill_ps(ps, cmd):
+    pid = ps.pid
+    print(f"Trying to kill {cmd} with PID: {pid}")
+
+    os.kill(pid, signal.SIGTERM)
+    try:
+        ps.wait(timeout=10)
+    except subprocess.TimeoutExpired as exc:
+        os.kill(pid, signal.SIGKILL)
+        raise RuntimeError(f"Killed the process {cmd} with SIGKILL. This could lead to corrupted data!") from exc
 
 # currently unused
 def timeout(process, cmd: str, duration: int):

metric_providers/base.py

@@ -1,12 +1,11 @@
 import os
 from pathlib import Path
 import subprocess
-import signal
-import sys
 from io import StringIO
 import pandas
 
 from lib.system_checks import ConfigurationCheckError
+from lib import process_helpers
 
 class MetricProviderConfigurationError(ConfigurationCheckError):
     pass
@@ -146,20 +145,6 @@ def start_profiling(self, containers=None):
     def stop_profiling(self):
         if self._ps is None:
             return
-        try:
-            print(f"Killing process with id: {self._ps.pid}")
-            ps_group_id = os.getpgid(self._ps.pid)
-            print(f" and process group {ps_group_id}")
-            os.killpg(ps_group_id, signal.SIGTERM)
-            try:
-                self._ps.wait(timeout=5)
-            except subprocess.TimeoutExpired:
-                # If the process hasn't gracefully exited after 5 seconds we kill it
-                os.killpg(ps_group_id, signal.SIGKILL)
-                print("Killed the process with SIGKILL. This could lead to corrupted metric log files!")
-
-        except ProcessLookupError:
-            print(f"Could not find process-group for {self._ps.pid}",
-                    file=sys.stderr)  # process/-group may have already closed
 
+        process_helpers.kill_pg(self._ps, self._metric_provider_executable)
         self._ps = None

runner.py

@@ -1058,7 +1058,7 @@ def run_flows(self):
                         if  stdout_behaviour == subprocess.PIPE:
                             os.set_blocking(ps.stdout.fileno(), False)
 
-                        self.__ps_to_kill.append({'ps': ps, 'cmd': inner_el['command'], 'ps_group': False})
+                        self.__ps_to_kill.append({'ps': ps, 'cmd': inner_el['command']})
                     else:
                         print(f"Process should be synchronous. Alloting {config['measurement']['flow-process-runtime']}s runtime ...")
                         ps = subprocess.run(
@@ -1103,7 +1103,11 @@ def stop_metric_providers(self):
             if stderr_read is not None and str(stderr_read):
                 errors.append(f"Stderr on {metric_provider.__class__.__name__} was NOT empty: {stderr_read}")
 
-            metric_provider.stop_profiling()
+            # pylint: disable=broad-exception-caught
+            try:
+                metric_provider.stop_profiling()
+            except Exception as exc:
+                errors.append(f"Could not stop profiling on {metric_provider.__class__.__name__}: {str(exc)}")
 
             df = metric_provider.read_metrics(self._run_id, self.__containers)
             if isinstance(df, int):
@@ -1124,7 +1128,17 @@ def stop_metric_providers(self):
 
     def read_and_cleanup_processes(self):
         print(TerminalColors.HEADER, '\nReading process stdout/stderr (if selected) and cleaning them up', TerminalColors.ENDC)
-        process_helpers.kill_ps(self.__ps_to_kill)
+
+        ps_errors = []
+        for ps in self.__ps_to_kill:
+            try:
+                process_helpers.kill_ps(ps['ps'], ps['cmd'])
+            except ProcessLookupError as exc: # Process might have done expected exit already. However all other errors shall bubble
+                ps_errors.append(f"Could not kill {ps['cmd']}. Exception: {exc}")
+        if ps_errors:
+            raise RuntimeError(ps_errors)
+        self.__ps_to_kill = [] # we need to clear, so we do not kill twice later
+
         for ps in self.__ps_to_read:
             if ps['detach']:
                 stdout, stderr = ps['ps'].communicate(timeout=5)
@@ -1240,7 +1254,12 @@ def cleanup(self):
 
         print('Stopping metric providers')
         for metric_provider in self.__metric_providers:
-            metric_provider.stop_profiling()
+            # pylint: disable=broad-exception-caught
+            try:
+                metric_provider.stop_profiling()
+            except Exception as exc:
+                error_helpers.log_error(f"Could not stop profiling on {metric_provider.__class__.__name__}: {str(exc)}")
+
 
         print('Stopping containers')
         for container_id in self.__containers:
@@ -1257,7 +1276,16 @@ def cleanup(self):
 
         self.remove_docker_images()
 
-        process_helpers.kill_ps(self.__ps_to_kill)
+        ps_errors = []
+        for ps in self.__ps_to_kill:
+            try:
+                process_helpers.kill_ps(ps['ps'], ps['cmd'])
+            except ProcessLookupError as exc: # Process might have done expected exit already. However all other errors shall bubble
+                ps_errors.append(f"Could not kill {ps['cmd']}. Exception: {exc}")
+        if ps_errors:
+            raise RuntimeError(ps_errors)
+
+
         print(TerminalColors.OKBLUE, '-Cleanup gracefully completed', TerminalColors.ENDC)
 
         self.__notes_helper = Notes()