[integ-test-framework] Terminate processes where tests have finished

Some processes could hang in `sleeping` status forever. The suspected reason is that when tests fail, sometimes network connections are not closed, causing the parent process to hang in `sleeping`.

Prior to this commit, we have a 4-hour timeout for processes. This commit adds another check to read the Pytest logs to terminate processes where tests have finished for more than 30 minutes.

Signed-off-by: Hanwen <hanwenli@amazon.com>

Author: hanwen-cluster

tests/integration-tests/framework/fixture_utils.py

@@ -9,6 +9,7 @@
 import logging
 import os
 import pickle
+import re
 import time
 from dataclasses import dataclass, field
 from inspect import getfullargspec, isgeneratorfunction
@@ -53,13 +54,15 @@ def __init__(
         fixture_func_args: tuple,
         fixture_func_kwargs: dict,
         xdist_worker_id_and_pid: str,
+        log_file: str,
     ):
         self.name = name
         self.shared_save_location = shared_save_location
         self.fixture_func = fixture_func
         self.fixture_func_args = fixture_func_args
         self.fixture_func_kwargs = fixture_func_kwargs
         self.xdist_worker_id_and_pid = xdist_worker_id_and_pid
+        self.log_file = log_file
         self._lock_file = shared_save_location / f"{name}.lock"
         self._fixture_file = shared_save_location / f"{name}.fixture"
         self._generator = None
@@ -95,6 +98,19 @@ def _terminate_process(pid: int):
         except Exception:
             logging.error("Error terminating process %s.", pid)
 
+    @staticmethod
+    def _completed_in_the_past(minutes: int, line: str):
+        if line is None or "- Completed test" not in line:
+            return False
+        regex = r"^(.*?) \- .*"
+        re_result = re.search(regex, line)
+        if re_result:
+            date = re_result.group(1)
+            try:
+                return (time.time() - time.mktime(time.strptime(date, "%Y-%m-%d %H:%M:%S,%f"))) / 60 > minutes
+            except Exception:
+                return False
+
     def release(self):
         """
         Release a shared fixture.
@@ -119,6 +135,16 @@ def release(self):
             )
             time.sleep(30)
 
+            last_message_of_each_proc = {}
+            with open(self.log_file, "r") as f:
+                lines = f.readlines()
+                for line in lines:
+                    regex = r"^.* \- .* \- (\d+) - .*"
+                    re_result = re.search(regex, line)
+                    if re_result:
+                        pid = re.search(regex, line).group(1)
+                        last_message_of_each_proc[pid] = line
+
             with FileLock(self._lock_file):
                 for worker in data.currently_using_processes.copy():
                     pid = int(worker.split(" ")[1])
@@ -132,6 +158,11 @@ def release(self):
                             data.counter,
                         )
                         self._save_fixture_data(data)
+                    elif self._completed_in_the_past(30, last_message_of_each_proc.get(str(pid))):
+                        logging.warning(
+                            "%s is sleeping but the test has been finished. Terminating the process... ", worker
+                        )
+                        self._terminate_process(pid)
                 data = self._load_fixture_data()
 
             if time.time() > timeout:
@@ -222,6 +253,7 @@ def _xdist_session_fixture(request, *args, **kwargs):
                 fixture_func_args=args,
                 fixture_func_kwargs=kwargs,
                 xdist_worker_id_and_pid=f"{xdist_worker_id}: {pid}",
+                log_file=request.config.getoption("tests_log_file"),
             )
             try:
                 yield shared_fixture.acquire().fixture_return_value