Some more additional changes to the ZMQHandler

changelog/74.improvement.rst

@@ -0,0 +1 @@
+Some more additional changes to the ZMQHandler to make sure it's resources are cleaned when terminating

src/saltfactories/utils/saltext/log_handlers/pytest_log_handler.py

@@ -250,7 +250,7 @@ def start(self):
 
         self.pid = os.getpid()
 
-    def stop(self):
+    def stop(self, flush=True):
         if self._exiting:
             return
 
@@ -266,8 +266,12 @@ def stop(self):
 
         try:
             if self.pusher is not None and not self.pusher.closed:
-                # Give it 1.5 seconds to flush any messages in it's queue
-                self.pusher.close(1500)
+                if flush:
+                    # Give it 1.5 seconds to flush any messages in it's queue
+                    linger = 1500
+                else:
+                    linger = 0
+                self.pusher.close(linger)
                 self.pusher = None
             if self.context is not None and not self.context.closed:
                 self.context.term()
@@ -282,7 +286,13 @@ def stop(self):
             sys.stderr.flush()
             raise
         finally:
-            self.context = self.pusher = self.pid = None
+            if self.pusher is not None and not self.pusher.closed:
+                self.pusher.close(0)
+                self.pusher = None
+            if self.context is not None and not self.context.closed:
+                self.context.term()
+                self.context = None
+            self.pid = None
 
     def format(self, record):
         msg = super(ZMQHandler, self).format(record)
@@ -348,6 +358,7 @@ def emit(self, record):
         except (SystemExit, KeyboardInterrupt):  # pragma: no cover pylint: disable=try-except-raise
             # Catch and raise SystemExit and KeyboardInterrupt so that we can handle
             # all other exception below
+            self.stop(flush=False)
             raise
         except Exception:  # pragma: no cover pylint: disable=broad-except
             self.handleError(record)
@@ -366,7 +377,7 @@ def close(self):
         Tidy up any resources used by the handler.
         """
         # The logging machinery has asked to stop this handler
-        self.stop()
+        self.stop(flush=False)
         # self._exiting should already be True, nonetheless, we set it here
         # too to ensure the handler doesn't get a chance to restart itself
         self._exiting = True

tests/functional/utils/saltext/test_log_handlers.py

@@ -187,15 +187,13 @@ def test_all_messages_received_multiprocessing(tempfiles, salt_factories, caplog
         def log_from_child_process(idx, parent_pid, evt):
             evt.set()  # process started, ready to start another one
             import os
-            import time
             import logging
             log = logging.getLogger("foo")
             for idx in range(1, {calls} + 1):
                 log.debug("Foo(Child of pid %s):%s:%s", parent_pid, idx, os.getpid())
             exit(0)
 
         def log_from_process(pidx, evt):
-            evt.set()  # process started, ready to start another one
 
             import os
             import logging
@@ -215,6 +213,8 @@ def log_from_process(pidx, evt):
                 cevt.clear()
                 time.sleep(0.25)
 
+            evt.set()  # process started, ready to start another one
+
             log = logging.getLogger("foo")
             for idx in range(1, {calls} + 1):
                 log.debug("Foo:%s:%s", idx, os.getpid())