[3007.x] Add started event to PublishServer

changelog/66931.fixed.md

@@ -0,0 +1,4 @@
+transports.tcp: ensure pull path is being used before attempting chmod.
+The fix prevents an unnecessary traceback when the TCP transport is
+not using unix sockets. No functionaly has changed as the traceback
+occurs when an async task was about to exit anyway.

changelog/66993.fixed.md

@@ -0,0 +1 @@
+Salt master waits for publish servers while starting up.

salt/master.py

@@ -813,13 +813,21 @@ def start(self):
             for _, opts in iter_transport_opts(self.opts):
                 chan = salt.channel.server.PubServerChannel.factory(opts)
                 chan.pre_fork(self.process_manager, kwargs={"secrets": SMaster.secrets})
+                if not chan.transport.started.wait(60):
+                    raise salt.exceptions.SaltMasterError(
+                        "Publish server did not start within 60 seconds. Something went wrong.",
+                    )
                 pub_channels.append(chan)
 
             log.info("Creating master event publisher process")
             ipc_publisher = salt.channel.server.MasterPubServerChannel.factory(
                 self.opts
             )
             ipc_publisher.pre_fork(self.process_manager)
+            if not ipc_publisher.transport.started.wait(30):
+                raise salt.exceptions.SaltMasterError(
+                    "IPC publish server did not start within 30 seconds. Something went wrong."
+                )
             self.process_manager.add_process(
                 EventMonitor,
                 args=[self.opts, ipc_publisher],

salt/transport/tcp.py

@@ -10,7 +10,6 @@
 import errno
 import logging
 import multiprocessing
-import os
 import queue
 import select
 import socket
@@ -1150,7 +1149,13 @@ class TCPPuller:
     """
 
     def __init__(
-        self, host=None, port=None, path=None, io_loop=None, payload_handler=None
+        self,
+        host=None,
+        port=None,
+        path=None,
+        mode=0o600,
+        io_loop=None,
+        payload_handler=None,
     ):
         """
         Create a new Tornado IPC server
@@ -1170,6 +1175,7 @@ def __init__(
         self.host = host
         self.port = port
         self.path = path
+        self.mode = mode
         self._started = False
         self.payload_handler = payload_handler
 
@@ -1187,7 +1193,7 @@ def start(self):
         # Start up the ioloop
         if self.path:
             log.trace("IPCServer: binding to socket: %s", self.path)
-            self.sock = tornado.netutil.bind_unix_socket(self.path)
+            self.sock = tornado.netutil.bind_unix_socket(self.path, self.mode)
         else:
             log.trace("IPCServer: binding to socket: %s:%s", self.host, self.port)
             self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
@@ -1331,6 +1337,7 @@ def __init__(
         pull_path_perms=0o600,
         pub_path_perms=0o600,
         ssl=None,
+        started=None,
     ):
         self.opts = opts
         self.pub_sock = None
@@ -1343,6 +1350,10 @@ def __init__(
         self.pull_path_perms = pull_path_perms
         self.pub_path_perms = pub_path_perms
         self.ssl = ssl
+        if started is None:
+            self.started = multiprocessing.Event()
+        else:
+            self.started = started
 
     @property
     def topic_support(self):
@@ -1362,6 +1373,8 @@ def __getstate__(self):
             "pull_path": self.pull_path,
             "pub_path_perms": self.pub_path_perms,
             "pull_path_perms": self.pull_path_perms,
+            "ssl": self.ssl,
+            "started": self.started,
         }
 
     def publish_daemon(
@@ -1414,8 +1427,9 @@ async def publisher(
                 "Publish server binding pub to %s ssl=%r", self.pub_path, self.ssl
             )
             with salt.utils.files.set_umask(0o177):
-                sock = tornado.netutil.bind_unix_socket(self.pub_path)
-                os.chmod(self.pub_path, self.pub_path_perms)
+                sock = tornado.netutil.bind_unix_socket(
+                    self.pub_path, self.pub_path_perms
+                )
         else:
             log.debug(
                 "Publish server binding pub to %s:%s ssl=%r",
@@ -1444,18 +1458,18 @@ async def publisher(
             pull_host = self.pull_host
             pull_port = self.pull_port
 
-        self.pull_sock = TCPPuller(
-            host=self.pull_host,
-            port=self.pull_port,
-            path=self.pull_path,
-            io_loop=io_loop,
-            payload_handler=publish_payload,
-        )
-
-        # Securely create socket
         with salt.utils.files.set_umask(0o177):
+            self.pull_sock = TCPPuller(
+                host=self.pull_host,
+                port=self.pull_port,
+                path=self.pull_path,
+                mode=self.pull_path_perms,
+                io_loop=io_loop,
+                payload_handler=publish_payload,
+            )
+            # Securely create socket
             self.pull_sock.start()
-            os.chmod(self.pull_path, self.pull_path_perms)
+        self.started.set()
 
     def pre_fork(self, process_manager):
         """

salt/transport/ws.py

@@ -263,6 +263,7 @@ def __init__(
         pull_path_perms=0o600,
         pub_path_perms=0o600,
         ssl=None,
+        started=None,
     ):
         self.opts = opts
         self.pub_host = pub_host
@@ -279,6 +280,10 @@ def __init__(
         self.pub_writer = None
         self.pub_reader = None
         self._connecting = None
+        if started is None:
+            self.started = multiprocessing.Event()
+        else:
+            self.started = started
 
     @property
     def topic_support(self):
@@ -298,6 +303,8 @@ def __getstate__(self):
             "pull_path": self.pull_path,
             "pull_path_perms": self.pull_path_perms,
             "pub_path_perms": self.pub_path_perms,
+            "ssl": self.ssl,
+            "started": self.started,
         }
 
     def publish_daemon(
@@ -375,6 +382,7 @@ async def publisher(
             self.puller = await asyncio.start_server(
                 self.pull_handler, self.pull_host, self.pull_port
             )
+        self.started.set()
         while self._run.is_set():
             await asyncio.sleep(0.3)
         await self.server.stop()

salt/transport/zeromq.py

@@ -7,6 +7,7 @@
 import errno
 import hashlib
 import logging
+import multiprocessing
 import os
 import signal
 import sys
@@ -854,6 +855,7 @@ def __init__(
         pull_path=None,
         pull_path_perms=0o600,
         pub_path_perms=0o600,
+        started=None,
     ):
         self.opts = opts
         self.pub_host = pub_host
@@ -878,10 +880,31 @@ def __init__(
         self.daemon_pub_sock = None
         self.daemon_pull_sock = None
         self.daemon_monitor = None
+        if started is None:
+            self.started = multiprocessing.Event()
+        else:
+            self.started = started
 
     def __repr__(self):
         return f"<PublishServer pub_uri={self.pub_uri} pull_uri={self.pull_uri} at {hex(id(self))}>"
 
+    def __setstate__(self, state):
+        self.__init__(**state)
+
+    def __getstate__(self):
+        return {
+            "opts": self.opts,
+            "pub_host": self.pub_host,
+            "pub_port": self.pub_port,
+            "pub_path": self.pub_path,
+            "pull_host": self.pull_host,
+            "pull_port": self.pull_port,
+            "pull_path": self.pull_path,
+            "pub_path_perms": self.pub_path_perms,
+            "pull_path_perms": self.pull_path_perms,
+            "started": self.started,
+        }
+
     def publish_daemon(
         self,
         publish_payload,
@@ -954,6 +977,7 @@ async def publisher(self, publish_payload, ioloop=None):
             self.daemon_pub_sock,
             self.daemon_monitor,
         ) = self._get_sockets(self.daemon_context, ioloop)
+        self.started.set()
         while True:
             try:
                 package = await self.daemon_pull_sock.recv()

tests/pytests/functional/channel/test_server.py

@@ -59,6 +59,8 @@ def transport_ids(value):
 @pytest.fixture(
     params=[
         "ws",
+        "tcp",
+        "zeromq",
     ],
     ids=transport_ids,
 )
@@ -173,6 +175,8 @@ def test_pub_server_channel(
         master_config,
     )
     server_channel.pre_fork(process_manager)
+    if not server_channel.transport.started.wait(30):
+        pytest.fail("Server channel did not start within 30 seconds.")
     req_server_channel = salt.channel.server.ReqServerChannel.factory(master_config)
     req_server_channel.pre_fork(process_manager)
 

tests/pytests/unit/transport/test_tcp.py

@@ -685,3 +685,74 @@ async def test_pub_server_publish_payload_closed_stream(master_opts, io_loop):
     server.clients = {client}
     await server.publish_payload(package, topic_list)
     assert server.clients == set()
+
+
+async def test_pub_server_paths_no_perms(master_opts, io_loop):
+    def publish_payload(payload):
+        return payload
+
+    pubserv = salt.transport.tcp.PublishServer(
+        master_opts,
+        pub_host="127.0.0.1",
+        pub_port=5151,
+        pull_host="127.0.0.1",
+        pull_port=5152,
+    )
+    assert pubserv.pull_path is None
+    assert pubserv.pub_path is None
+    with patch("os.chmod") as p:
+        await pubserv.publisher(publish_payload)
+        assert p.call_count == 0
+
+
+@pytest.mark.skip_on_windows()
+async def test_pub_server_publisher_pull_path_perms(master_opts, io_loop, tmp_path):
+    def publish_payload(payload):
+        return payload
+
+    pull_path = str(tmp_path / "pull.ipc")
+    pull_path_perms = 0o664
+    pubserv = salt.transport.tcp.PublishServer(
+        master_opts,
+        pub_host="127.0.0.1",
+        pub_port=5151,
+        pull_host=None,
+        pull_port=None,
+        pull_path=pull_path,
+        pull_path_perms=pull_path_perms,
+    )
+    assert pubserv.pull_path == pull_path
+    assert pubserv.pull_path_perms == pull_path_perms
+    assert pubserv.pull_host is None
+    assert pubserv.pull_port is None
+    with patch("os.chmod") as p:
+        await pubserv.publisher(publish_payload)
+        assert p.call_count == 1
+        assert p.call_args.args == (pubserv.pull_path, pubserv.pull_path_perms)
+
+
+@pytest.mark.skip_on_windows()
+async def test_pub_server_publisher_pub_path_perms(master_opts, io_loop, tmp_path):
+    def publish_payload(payload):
+        return payload
+
+    pub_path = str(tmp_path / "pub.ipc")
+    pub_path_perms = 0o664
+    pubserv = salt.transport.tcp.PublishServer(
+        master_opts,
+        pub_host=None,
+        pub_port=None,
+        pub_path=pub_path,
+        pub_path_perms=pub_path_perms,
+        pull_host="127.0.0.1",
+        pull_port=5151,
+        pull_path=None,
+    )
+    assert pubserv.pub_path == pub_path
+    assert pubserv.pub_path_perms == pub_path_perms
+    assert pubserv.pub_host is None
+    assert pubserv.pub_port is None
+    with patch("os.chmod") as p:
+        await pubserv.publisher(publish_payload)
+        assert p.call_count == 1
+        assert p.call_args.args == (pubserv.pub_path, pubserv.pub_path_perms)