Fix App driver stop take 3

doc/newsfragments/3041_changed.app_stop_refactor.rst

@@ -0,0 +1 @@
+Make sure when stop() is called on App type driver, we clean up all orphaned processes.

testplan/common/entity/base.py

@@ -60,9 +60,6 @@ def __init__(self, parent: Optional["Entity"] = None):
         self.__dict__["_resources"] = OrderedDict()
         self.__dict__["start_exceptions"] = OrderedDict()
         self.__dict__["stop_exceptions"] = OrderedDict()
-        # warnings for info display only, no interference needed (really?)
-        self.__dict__["start_warnings"] = OrderedDict()
-        self.__dict__["stop_warnings"] = OrderedDict()
 
     def add(self, item: "Resource", uid: Optional[str] = None) -> str:
         """
@@ -177,17 +174,6 @@ def all_status(self, target) -> bool:
             for resource in self._resources
         )
 
-    def _record_resource_warning(self, message, resource, msg_store):
-        fetch_msg = "\n".join(resource.fetch_error_log())
-
-        msg = message.format(
-            resource=resource,
-            traceback_exc=traceback.format_exc(),
-            fetch_msg=fetch_msg,
-        )
-        resource.logger.warning(msg)
-        msg_store[resource] = "WARNING: " + msg
-
     def _record_resource_exception(self, message, resource, msg_store):
         fetch_msg = "\n".join(resource.fetch_error_log())
 
@@ -219,7 +205,6 @@ def start(self):
                     msg_store=self.start_exceptions,
                 )
 
-                # XXX: in the case of failover, put exc in warnings instead?
                 failover = resource.failover()
                 if failover:
                     self._resources[resource.uid()] = failover
@@ -296,34 +281,20 @@ def stop(self, is_reversed=False):
         if is_reversed is True:
             resources = resources[::-1]
 
-        # XXX: with the legacy design, we cannot really distinguish exceptions
-        # XXX: raised by ``stopping`` from those raised by ``stopped_check``,
-        # XXX: we have this ambiguity regarding ``stop``.
-
         # Stop all resources
         resources_to_wait_for: List[Resource] = []
         for resource in resources:
             try:
                 resource.stop()
-            except Exception as e:
-                if isinstance(e, tuple(resource.SUPPRESSED_STOP_EXC)):
-                    self._record_resource_warning(
-                        message="While stopping resource {resource}"
-                        " (mitigated by forcefully stopping resource)"
-                        ":\n{traceback_exc}\n{fetch_msg}",
-                        resource=resource,
-                        msg_store=self.stop_warnings,
-                    )
-                else:
-                    self._record_resource_exception(
-                        message="While stopping resource {resource}"
-                        ":\n{traceback_exc}\n{fetch_msg}",
-                        resource=resource,
-                        msg_store=self.stop_exceptions,
-                    )
-
+            except Exception:
+                self._record_resource_exception(
+                    message="While stopping resource {resource}"
+                    ":\n{traceback_exc}\n{fetch_msg}",
+                    resource=resource,
+                    msg_store=self.stop_exceptions,
+                )
                 # Resource status should be STOPPED even it failed to stop
-                resource.force_stopped()
+                resource.force_stop()
             else:
                 if (
                     resource.async_start
@@ -336,24 +307,15 @@ def stop(self, is_reversed=False):
         for resource in resources_to_wait_for:
             try:
                 resource.wait(resource.STATUS.STOPPED)
-            except Exception as e:
-                if isinstance(e, tuple(resource.SUPPRESSED_STOP_EXC)):
-                    self._record_resource_warning(
-                        message="While waiting for resource {resource} to stop"
-                        " (mitigated by forcefully stopping resource)"
-                        ":\n{traceback_exc}\n{fetch_msg}",
-                        resource=resource,
-                        msg_store=self.stop_warnings,
-                    )
-                else:
-                    self._record_resource_exception(
-                        message="While waiting for resource {resource} to stop"
-                        ":\n{traceback_exc}\n{fetch_msg}",
-                        resource=resource,
-                        msg_store=self.stop_exceptions,
-                    )
+            except Exception:
+                self._record_resource_exception(
+                    message="While waiting for resource {resource} to stop"
+                    ":\n{traceback_exc}\n{fetch_msg}",
+                    resource=resource,
+                    msg_store=self.stop_exceptions,
+                )
                 # Resource status should be STOPPED even it failed to stop
-                resource.force_stopped()
+                resource.force_stop()
             resource.logger.info("%s stopped", resource)
 
     def stop_in_pool(self, pool, is_reversed=False):
@@ -393,7 +355,7 @@ def stop_in_pool(self, pool, is_reversed=False):
                 resource.logger.info("%s stopped", resource)
             else:
                 # Resource status should be STOPPED even it failed to stop
-                resource.force_stopped()
+                resource.force_stop()
 
     def _log_exception(self, resource, func, exception_record):
         """
@@ -1385,7 +1347,6 @@ class Resource(Entity):
 
     CONFIG = ResourceConfig
     STATUS = ResourceStatus
-    SUPPRESSED_STOP_EXC = []
 
     def __init__(self, **options):
         super(Resource, self).__init__(**options)
@@ -1562,9 +1523,9 @@ def _wait_stopped(self, timeout: Optional[float] = None):
 
         :param timeout: timeout in seconds
         """
-        self._after_stopped()
+        self._mark_stopped()
 
-    def _after_stopped(self):
+    def _mark_stopped(self):
         """
         Common logic after a successful Resource stop.
         """
@@ -1617,7 +1578,7 @@ def restart(self):
         if self.async_start:
             self.wait(self.STATUS.STARTED)
 
-    def force_stopped(self):
+    def force_stop(self):
         """
         Change the status to STOPPED (e.g. exception raised).
         """

testplan/common/utils/process.py

@@ -184,6 +184,54 @@ def kill_process_psutil(
     return alive
 
 
+def kill_proc_and_child_procs(
+    proc: subprocess.Popen,
+    child_procs: List[psutil.Process],
+    log: Callable[[BaseException], None] = None,
+    timeout: int = 5,
+) -> None:
+    """
+    Kill a process and its child processes.
+
+    This function attempts to kill a given process and its child processes.
+    It first kills the main process, waits for it to terminate, and then
+    attempts to kill any remaining child processes.
+
+    :param proc: The main process to kill.
+    :type proc: subprocess.Popen
+    :param child_procs: A list of child processes to kill.
+    :type child_procs: List[psutil.Process]
+    :param log: A callable to log exceptions, defaults to None.
+    :type log: Callable[[BaseException], None], optional
+    :param timeout: The timeout in seconds to wait for processes to terminate, defaults to 5.
+    :type timeout: int, optional
+    :return: The return code of the main process if it was terminated, otherwise None.
+    :rtype: Union[int, None]
+    """
+
+    if proc is not None and proc.poll() is None:
+        try:
+            proc.kill()
+            proc.wait()
+            wait_process_clean(proc.pid, timeout=timeout)
+        except Exception as exc:
+            if log:
+                log(exc)
+
+    if child_procs:
+        # we did not send sig term to child processes
+        # thus no point to wait for them to terminate
+        _, alive = psutil.wait_procs(child_procs, timeout=0)
+        for p in alive:
+            try:
+                p.kill()
+            except psutil.NoSuchProcess:
+                pass  # already reaped
+            except Exception as exc:
+                if log:
+                    log(exc)
+
+
 def cleanup_child_procs(
     procs: List[psutil.Process],
     timeout: float,
@@ -206,17 +254,6 @@ def cleanup_child_procs(
             log(exc)
 
 
-def any_alive_child_procs(procs: List[psutil.Process]) -> bool:
-    for p in procs:
-        try:
-            # orphaned zombie process should be reaped by pid 1
-            if p.is_running() and p.status() != psutil.STATUS_ZOMBIE:
-                return True
-        except psutil.NoSuchProcess:
-            pass
-    return False
-
-
 DEFAULT_CLOSE_FDS = platform.system() != "Windows"
 
 

testplan/testing/base.py

@@ -464,9 +464,6 @@ def _start_resource(self) -> None:
             self._record_driver_connection(case_report)
         case_report.pass_if_empty()
 
-        if self.resources.start_warnings:
-            for msg in self.resources.start_warnings.values():
-                case_report.logger.warning(msg)
         if self.resources.start_exceptions:
             for msg in self.resources.start_exceptions.values():
                 case_report.logger.error(msg)
@@ -490,9 +487,6 @@ def _stop_resource(self, is_reversed=True) -> None:
             )
         case_report.pass_if_empty()
 
-        if self.resources.stop_warnings:
-            for msg in self.resources.stop_warnings.values():
-                case_report.logger.warning(msg)
         if self.resources.stop_exceptions:
             for msg in self.resources.stop_exceptions.values():
                 case_report.logger.error(msg)

testplan/testing/environment/base.py

@@ -206,77 +206,38 @@ def stop(self, is_reversed=False):
                     self._pocketwatches[driver.uid()].record_start()
                     driver.stop()
                 except Exception as e:
-                    do_suppress = isinstance(
-                        e, tuple(driver.SUPPRESSED_STOP_EXC)
+                    self._record_resource_exception(
+                        message="While stopping driver {resource}"
+                        ":\n{traceback_exc}\n{fetch_msg}",
+                        resource=driver,
+                        msg_store=self.stop_exceptions,
                     )
-                    if do_suppress:
-                        self._record_resource_warning(
-                            message="While stopping driver {resource}"
-                            " (mitigated by forcefully stopping driver)"
-                            ":\n{traceback_exc}\n{fetch_msg}",
-                            resource=driver,
-                            msg_store=self.stop_warnings,
-                        )
-                    else:
-                        self._record_resource_exception(
-                            message="While stopping driver {resource}"
-                            ":\n{traceback_exc}\n{fetch_msg}",
-                            resource=driver,
-                            msg_store=self.stop_exceptions,
-                        )
                     # driver status should be STOPPED even it failed to stop
-                    driver.force_stopped()
+                    driver.force_stop()
                     driver.logger.info("%s force stopped", driver)
                     self._rt_dependency.mark_processing(driver)
-                    if do_suppress:
-                        self._rt_dependency.mark_processed(driver)
-                    else:
-                        self._rt_dependency.mark_failed_to_process(driver)
+                    self._rt_dependency.mark_failed_to_process(driver)
                 else:
                     self._rt_dependency.mark_processing(driver)
 
             # check current drivers
             for driver in self._rt_dependency.drivers_processing():
                 watch: DriverPocketwatch = self._pocketwatches[driver.uid()]
                 try:
-                    if time.time() >= watch.start_time + watch.total_wait:
-                        # we got a timed-out here
-                        raise TimeoutException(
-                            f"Timeout when stopping {driver}. "
-                            f"{TimeoutExceptionInfo(watch.start_time).msg()}"
-                        )
-                    res = None
-                    if watch.should_check():
-                        res = driver.stopped_check()
-                    if res:
-                        driver._after_stopped()
+                    if driver.stopped_check_with_watch(watch):
+                        driver._mark_stopped()
                         driver.logger.info("%s stopped", driver)
                         self._rt_dependency.mark_processed(driver)
-                except Exception as e:
-                    do_suppress = isinstance(
-                        e, tuple(driver.SUPPRESSED_STOP_EXC)
+                except Exception:
+                    self._record_resource_exception(
+                        message="While waiting for driver {resource} to stop:\n"
+                        "{traceback_exc}\n{fetch_msg}",
+                        resource=driver,
+                        msg_store=self.stop_exceptions,
                     )
-                    if do_suppress:
-                        self._record_resource_warning(
-                            message="While waiting for driver {resource} to stop"
-                            " (mitigated by forcefully stopping driver)"
-                            ":\n{traceback_exc}\n{fetch_msg}",
-                            resource=driver,
-                            msg_store=self.stop_warnings,
-                        )
-                    else:
-                        self._record_resource_exception(
-                            message="While waiting for driver {resource} to stop"
-                            ":\n{traceback_exc}\n{fetch_msg}",
-                            resource=driver,
-                            msg_store=self.stop_exceptions,
-                        )
                     # driver status should be STOPPED even it failed to stop
-                    driver.force_stopped()
+                    driver.force_stop()
                     driver.logger.info("%s force stopped", driver)
-                    if do_suppress:
-                        self._rt_dependency.mark_processed(driver)
-                    else:
-                        self._rt_dependency.mark_failed_to_process(driver)
+                    self._rt_dependency.mark_failed_to_process(driver)
 
             time.sleep(MINIMUM_CHECK_INTERVAL)