[3006.x] Warn on un-closed transport clients

changelog/65554.fixed.md

@@ -0,0 +1 @@
+Warn when an un-closed transport client is being garbage collected.

salt/transport/base.py

@@ -1,3 +1,6 @@
+import traceback
+import warnings
+
 import salt.ext.tornado.gen
 
 TRANSPORTS = (
@@ -94,14 +97,52 @@ def publish_client(opts, io_loop):
     raise Exception("Transport type not found: {}".format(ttype))
 
 
-class RequestClient:
+class TransportWarning(Warning):
+    """
+    Transport warning.
+    """
+
+
+class Transport:
+    def __init__(self, *args, **kwargs):
+        self._trace = "\n".join(traceback.format_stack()[:-1])
+        if not hasattr(self, "_closing"):
+            self._closing = False
+        if not hasattr(self, "_connect_called"):
+            self._connect_called = False
+
+    def connect(self, *args, **kwargs):
+        self._connect_called = True
+
+    # pylint: disable=W1701
+    def __del__(self):
+        """
+        Warn the user if the transport's close method was never called.
+
+        If the _closing attribute is missing we won't raise a warning. This
+        prevents issues when class's dunder init method is called with improper
+        arguments, and is later getting garbage collected. Users of this class
+        should take care to call super() and validate the functionality with a
+        test.
+        """
+        if getattr(self, "_connect_called") and not getattr(self, "_closing", True):
+            warnings.warn(
+                f"Unclosed transport! {self!r} \n{self._trace}",
+                TransportWarning,
+                source=self,
+            )
+
+    # pylint: enable=W1701
+
+
+class RequestClient(Transport):
     """
     The RequestClient transport is used to make requests and get corresponding
     replies from the RequestServer.
     """
 
     def __init__(self, opts, io_loop, **kwargs):
-        pass
+        super().__init__()
 
     @salt.ext.tornado.gen.coroutine
     def send(self, load, timeout=60):
@@ -116,7 +157,7 @@ def close(self):
         """
         raise NotImplementedError
 
-    def connect(self):
+    def connect(self):  # pylint: disable=W0221
         """
         Connect to the server / broker.
         """
@@ -197,13 +238,13 @@ def publish_daemon(
         raise NotImplementedError
 
 
-class PublishClient:
+class PublishClient(Transport):
     """
     The PublishClient receives messages from the PublishServer and runs a callback.
     """
 
     def __init__(self, opts, io_loop, **kwargs):
-        pass
+        super().__init__()
 
     def on_recv(self, callback):
         """
@@ -212,7 +253,9 @@ def on_recv(self, callback):
         raise NotImplementedError
 
     @salt.ext.tornado.gen.coroutine
-    def connect(self, publish_port, connect_callback=None, disconnect_callback=None):
+    def connect(  # pylint: disable=arguments-differ
+        self, publish_port, connect_callback=None, disconnect_callback=None
+    ):
         """
         Create a network connection to the the PublishServer or broker.
         """

salt/transport/tcp.py

@@ -213,6 +213,7 @@ class TCPPubClient(salt.transport.base.PublishClient):
     ttype = "tcp"
 
     def __init__(self, opts, io_loop, **kwargs):  # pylint: disable=W0231
+        super().__init__(opts, io_loop, **kwargs)
         self.opts = opts
         self.io_loop = io_loop
         self.message_client = None
@@ -228,14 +229,9 @@ def close(self):
             self.message_client.close()
             self.message_client = None
 
-    # pylint: disable=W1701
-    def __del__(self):
-        self.close()
-
-    # pylint: enable=W1701
-
     @salt.ext.tornado.gen.coroutine
     def connect(self, publish_port, connect_callback=None, disconnect_callback=None):
+        self._connect_called = True
         self.publish_port = publish_port
         self.message_client = MessageClient(
             self.opts,
@@ -1038,6 +1034,7 @@ class TCPReqClient(salt.transport.base.RequestClient):
     ttype = "tcp"
 
     def __init__(self, opts, io_loop, **kwargs):  # pylint: disable=W0231
+        super().__init__(opts, io_loop, **kwargs)
         self.opts = opts
         self.io_loop = io_loop
         parse = urllib.parse.urlparse(self.opts["master_uri"])
@@ -1054,9 +1051,11 @@ def __init__(self, opts, io_loop, **kwargs):  # pylint: disable=W0231
             source_ip=opts.get("source_ip"),
             source_port=opts.get("source_ret_port"),
         )
+        self._closing = False
 
     @salt.ext.tornado.gen.coroutine
     def connect(self):
+        self._connect_called = True
         yield self.message_client.connect()
 
     @salt.ext.tornado.gen.coroutine
@@ -1065,4 +1064,7 @@ def send(self, load, timeout=60):
         raise salt.ext.tornado.gen.Return(ret)
 
     def close(self):
+        if self._closing:
+            return
+        self._closing = True
         self.message_client.close()

salt/transport/zeromq.py

@@ -207,14 +207,16 @@ def __exit__(self, exc_type, exc_val, exc_tb):
     # TODO: this is the time to see if we are connected, maybe use the req channel to guess?
     @salt.ext.tornado.gen.coroutine
     def connect(self, publish_port, connect_callback=None, disconnect_callback=None):
+        self._connect_called = True
         self.publish_port = publish_port
         log.debug(
             "Connecting the Minion to the Master publish port, using the URI: %s",
             self.master_pub,
         )
         log.debug("%r connecting to %s", self, self.master_pub)
         self._socket.connect(self.master_pub)
-        connect_callback(True)
+        if connect_callback is not None:
+            connect_callback(True)
 
     @property
     def master_pub(self):
@@ -529,14 +531,8 @@ def connect(self):
         # wire up sockets
         self._init_socket()
 
-    # TODO: timeout all in-flight sessions, or error
     def close(self):
-        try:
-            if self._closing:
-                return
-        except AttributeError:
-            # We must have been called from __del__
-            # The python interpreter has nuked most attributes already
+        if self._closing:
             return
         else:
             self._closing = True
@@ -661,7 +657,10 @@ def monitor_callback(self, msg):
     def stop(self):
         if self._socket is None:
             return
-        self._socket.disable_monitor()
+        try:
+            self._socket.disable_monitor()
+        except zmq.Error:
+            pass
         self._socket = None
         self._monitor_socket = None
         if self._monitor_stream is not None:
@@ -880,24 +879,32 @@ class RequestClient(salt.transport.base.RequestClient):
     ttype = "zeromq"
 
     def __init__(self, opts, io_loop):  # pylint: disable=W0231
+        super().__init__(opts, io_loop)
         self.opts = opts
         master_uri = self.get_master_uri(opts)
         self.message_client = AsyncReqMessageClient(
             self.opts,
             master_uri,
             io_loop=io_loop,
         )
+        self._closing = False
+        self._connect_called = False
 
+    @salt.ext.tornado.gen.coroutine
     def connect(self):
+        self._connect_called = True
         self.message_client.connect()
 
     @salt.ext.tornado.gen.coroutine
     def send(self, load, timeout=60):
-        self.connect()
+        yield self.connect()
         ret = yield self.message_client.send(load, timeout=timeout)
         raise salt.ext.tornado.gen.Return(ret)
 
     def close(self):
+        if self._closing:
+            return
+        self._closing = True
         self.message_client.close()
 
     @staticmethod

tests/pytests/unit/transport/test_base.py

@@ -0,0 +1,21 @@
+"""
+Unit tests for salt.transport.base.
+"""
+import pytest
+
+import salt.transport.base
+
+pytestmark = [
+    pytest.mark.core_test,
+]
+
+
+def test_unclosed_warning():
+
+    transport = salt.transport.base.Transport()
+    assert transport._closing is False
+    assert transport._connect_called is False
+    transport.connect()
+    assert transport._connect_called is True
+    with pytest.warns(salt.transport.base.TransportWarning):
+        del transport

tests/pytests/unit/transport/test_zeromq.py

@@ -1498,3 +1498,31 @@ def test_pub_client_init(minion_opts, io_loop):
     client = salt.transport.zeromq.PublishClient(minion_opts, io_loop)
     client.send(b"asf")
     client.close()
+
+
+async def test_unclosed_request_client(minion_opts, io_loop):
+    minion_opts["master_uri"] = "tcp://127.0.0.1:4506"
+    client = salt.transport.zeromq.RequestClient(minion_opts, io_loop)
+    await client.connect()
+    try:
+        assert client._closing is False
+        with pytest.warns(salt.transport.base.TransportWarning):
+            client.__del__()
+    finally:
+        client.close()
+
+
+async def test_unclosed_publish_client(minion_opts, io_loop):
+    minion_opts["id"] = "minion"
+    minion_opts["__role"] = "minion"
+    minion_opts["master_ip"] = "127.0.0.1"
+    minion_opts["zmq_filtering"] = True
+    minion_opts["zmq_monitor"] = True
+    client = salt.transport.zeromq.PublishClient(minion_opts, io_loop)
+    await client.connect(2121)
+    try:
+        assert client._closing is False
+        with pytest.warns(salt.transport.base.TransportWarning):
+            client.__del__()
+    finally:
+        client.close()