feat(cluster_management): kill all leftover processes

It can happen that leftover processes from previous cluster instance
prevent new cluster instance from starting. This change makes sure to
kill all the leftover processes.

- Introduced `netstat_tools` module to handle netstat operations.
- Enhanced process killing logic to handle leftover processes more robustly.
- Moved `_kill_supervisor` and `_get_netstat_out` functions to the new module.
- Updated `ClusterGetter` to use the new `netstat_tools` functions.

Author: mkoura

cardano_node_tests/cluster_management/cluster_getter.py

@@ -14,6 +14,7 @@
 from _pytest.config import Config
 
 from cardano_node_tests.cluster_management import common
+from cardano_node_tests.cluster_management import netstat_tools
 from cardano_node_tests.cluster_management import resources
 from cardano_node_tests.cluster_management import resources_management
 from cardano_node_tests.utils import artifacts
@@ -36,35 +37,6 @@ def _xdist_sleep(
         """No need to sleep if tests are running on a single worker."""
 
 
-def _kill_supervisor(instance_num: int) -> None:
-    """Attempt to kill the `supervisord` process."""
-    try:
-        netstat = helpers.run_command("netstat -plnt").decode().splitlines()
-    except Exception:
-        return
-
-    port_num = (
-        cluster_nodes.get_cluster_type().cluster_scripts.get_instance_ports(instance_num).supervisor
-    )
-    port_str = f":{port_num}"
-
-    for line in netstat:
-        if port_str not in line:
-            continue
-        line_p = line.replace("  ", " ").strip()
-        pid = line_p.split()[-1].split("/")[0]
-        os.kill(int(pid), 15)
-        return
-
-
-def _get_netstat_out() -> str:
-    """Get output of the `netstat` command."""
-    try:
-        return helpers.run_command("netstat -plnt").decode()
-    except Exception:
-        return ""
-
-
 @dataclasses.dataclass
 class _ClusterGetStatus:
     """Intermediate status while trying to `get` suitable cluster instance."""
@@ -198,10 +170,7 @@ def _respin(self, start_cmd: str = "", stop_cmd: str = "") -> bool:  # noqa: C90
             except Exception as err:
                 self.log(f"c{self.cluster_instance_num}: failed to stop cluster:\n{err}")
 
-            _kill_supervisor(self.cluster_instance_num)
-
-            # Give the cluster time to stop
-            time.sleep(5)
+            netstat_tools.kill_old_cluster(instance_num=self.cluster_instance_num)
 
             # Save artifacts only when produced during this test run
             if cluster_running_file.exists() or i > 0:
@@ -229,7 +198,7 @@ def _respin(self, start_cmd: str = "", stop_cmd: str = "") -> bool:  # noqa: C90
             if _cluster_started:
                 break
         else:
-            netstat_out = _get_netstat_out()
+            netstat_out = netstat_tools.get_netstat_out()
             self.log(
                 f"c{self.cluster_instance_num}: failed to start cluster:\n{excp}"
                 f"\nnetstat:\n{netstat_out}"

cardano_node_tests/cluster_management/netstat_tools.py

@@ -0,0 +1,84 @@
+"""Functions based on `netstat`."""
+
+import logging
+import os
+import re
+import time
+
+from cardano_node_tests.utils import cluster_nodes
+from cardano_node_tests.utils import helpers
+
+LOGGER = logging.getLogger(__name__)
+
+
+def get_netstat_out() -> str:
+    """Get output of the `netstat` command."""
+    try:
+        return helpers.run_command(
+            "netstat -pant | grep -E 'LISTEN|TIME_WAIT|CLOSE_WAIT|FIN_WAIT'"
+        ).decode()
+    except Exception as excp:
+        LOGGER.error(f"Failed to fetch netstat output: {excp}")  # noqa: TRY400
+        return ""
+
+
+def kill_old_cluster(instance_num: int) -> None:  # noqa: C901
+    """Attempt to kill all processes left over from a previous cluster instance."""
+
+    def _get_netstat_split() -> list[str]:
+        return get_netstat_out().splitlines()
+
+    def _get_pid(line: str) -> int | None:
+        try:
+            pid_str = line.replace("  ", " ").strip().split()[-1].split("/")[0]
+            return int(pid_str)
+        except (IndexError, ValueError):
+            return None
+
+    def _try_kill(pid: int) -> None:
+        try:
+            os.kill(pid, 15)
+        except Exception as excp:
+            LOGGER.error(f"Failed to kill leftover process PID {pid}: {excp}")  # noqa: TRY400
+            return
+
+    port_nums = cluster_nodes.get_cluster_type().cluster_scripts.get_instance_ports(instance_num)
+    port_strs = [
+        f":{p}"
+        for p in (
+            port_nums.supervisor,
+            port_nums.webserver,
+            port_nums.submit_api,
+            *port_nums.node_ports,
+        )
+    ]
+
+    # Attempt to kill the `supervisord` process first. If successful, this will also kill all the
+    # processes started by supervisor.
+    port_supervisor_str = port_strs[0]
+    for line in _get_netstat_split():
+        if port_supervisor_str not in line:
+            continue
+        pid = _get_pid(line)
+        if pid:
+            LOGGER.info(f"Killing supervisor process: PID {pid}")
+            _try_kill(pid)
+        time.sleep(5)
+        break
+
+    # Kill all the leftover processes, if possible, and wait for them to finish
+    ports_re = re.compile(r"|".join(re.escape(p) for p in port_strs))
+    for _ in range(5):
+        found = False
+        for line in _get_netstat_split():
+            if not ports_re.search(line):
+                continue
+            found = True
+            pid = _get_pid(line)
+            if pid:
+                LOGGER.info(f"Killing leftover process: PID {pid}")
+                _try_kill(pid)
+            time.sleep(5)
+            break
+        if not found:
+            break