diff --git a/changelog/69930.fixed.md b/changelog/69930.fixed.md new file mode 100644 index 00000000000..2646422699b --- /dev/null +++ b/changelog/69930.fixed.md @@ -0,0 +1 @@ +Wired the ``ipc_write_buffer`` master option through to the TCP transport in 3008.x. The option remained in the config schema after the legacy ``salt.transport.ipc`` module was removed but was no longer read by any code path, so setting it in ``master.conf`` had no effect. It now caps the per-stream Tornado outbound ``max_write_buffer_size`` on both ``PubServer`` (event-bus subscribers, plaintext and SSL-delayed paths) and ``SaltMessageServer`` (request/reply clients), matching the semantics of the legacy IPC module's per-connection cap. The default (unset / ``0``) preserves the existing unlimited-buffer behavior; operators opt in by setting an explicit byte value. diff --git a/salt/transport/tcp.py b/salt/transport/tcp.py index 961a85c5f22..6ccb2a99419 100644 --- a/salt/transport/tcp.py +++ b/salt/transport/tcp.py @@ -737,11 +737,17 @@ def post_fork(self, message_handler, io_loop, **kwargs): ctx = None if self.ssl is not None: ctx = salt.transport.base.ssl_context(self.ssl, server_side=True) + # See issue #69930: pass the configured cap through to the + # per-stream Tornado outbound write buffer. ``ipc_write_buffer`` + # is the legacy option name kept for master.conf compatibility; + # it was a no-op on 3008.x until this wiring was added. + max_write_buffer_size = self.opts.get("ipc_write_buffer") or None if USE_LOAD_BALANCER: self.req_server = LoadBalancerWorker( self.socket_queue, self.handle_message, ssl_options=ctx, + max_write_buffer_size=max_write_buffer_size, ) else: if salt.utils.platform.is_windows(): @@ -754,6 +760,7 @@ def post_fork(self, message_handler, io_loop, **kwargs): self.handle_message, ssl_options=ctx, io_loop=io_loop, + max_write_buffer_size=max_write_buffer_size, ) self.req_server.add_socket(self._socket) self._socket.listen(self.backlog) @@ -806,6 +813,11 @@ class SaltMessageServer(tornado.tcpserver.TCPServer): def __init__(self, message_handler, *args, **kwargs): io_loop = kwargs.pop("io_loop", None) or tornado.ioloop.IOLoop.current() + # ``ipc_write_buffer`` (the legacy option name preserved for + # backwards-compat with ``master.conf``) caps the per-stream + # Tornado outbound write buffer. ``0`` / ``None`` == unlimited + # (Tornado default), matching prior behavior. + self.max_write_buffer_size = kwargs.pop("max_write_buffer_size", None) or None self._closing = False super().__init__(*args, **kwargs) self.io_loop = io_loop @@ -823,6 +835,12 @@ async def handle_stream( # pylint: disable=arguments-differ,invalid-overridden- Handle incoming streams and add messages to the incoming queue """ log.trace("Req client %s connected", address) + if self.max_write_buffer_size: + # See issue #69930: cap the outbound IOStream buffer per accepted + # request/reply client so a slow consumer can't grow it without + # bound. Tornado's ``TCPServer`` builds the ``IOStream`` before + # dispatching to ``handle_stream``, so we set the attribute here. + stream.max_write_buffer_size = self.max_write_buffer_size self.clients.append((stream, address)) unpacker = salt.utils.msgpack.Unpacker() try: @@ -1384,11 +1402,28 @@ def handle_stream(self, stream, address): self._validate_ssl_and_add_client(stream, address) ) return + self._apply_write_buffer_cap(stream) client = Subscriber(stream, address) self.clients.add(client) stream.set_close_callback(self._discard_on_close(client)) self.io_loop.create_task(self._stream_read(client)) + def _apply_write_buffer_cap(self, stream): + """ + Cap the accepted stream's outbound write buffer per ``ipc_write_buffer``. + + See issue #69930: the legacy ``salt.transport.ipc`` module was + removed in 3008.x but the ``ipc_write_buffer`` opt remained in + the config schema. Without this cap, Tornado defaults the + per-stream write buffer to unlimited, so a slow / blocked + event-bus subscriber lets the master's outbound bytearray grow + without bound (RSS growth observed on prod masters under event + burst). ``0`` / falsy preserves prior behavior (unlimited). + """ + cap = self.opts.get("ipc_write_buffer") or None + if cap: + stream.max_write_buffer_size = cap + async def _validate_ssl_and_add_client(self, stream, address): """ Validate SSL handshake completed successfully before accepting client. @@ -1409,6 +1444,7 @@ async def _validate_ssl_and_add_client(self, stream, address): return # Successfully got cert - add client + self._apply_write_buffer_cap(stream) client = Subscriber(stream, address) self.clients.add(client) stream.set_close_callback(self._discard_on_close(client)) diff --git a/tests/pytests/unit/transport/test_tcp.py b/tests/pytests/unit/transport/test_tcp.py index 0d588bfd87d..e0f0a00089a 100644 --- a/tests/pytests/unit/transport/test_tcp.py +++ b/tests/pytests/unit/transport/test_tcp.py @@ -1507,3 +1507,170 @@ def closed(self): # The "boom" was dropped by the except-log-and-continue guard; the # other two got through. assert handled == ["ok1", "ok2"] + + +# --------------------------------------------------------------------------- +# issue #69930: ipc_write_buffer wired through to per-stream cap. +# --------------------------------------------------------------------------- + + +async def test_salt_message_server_applies_ipc_write_buffer(master_opts): + """ + ``SaltMessageServer.handle_stream`` must set the accepted stream's + ``max_write_buffer_size`` to the ``ipc_write_buffer`` value passed + in. Without this wiring (regression on 3008.x after the legacy + ``salt.transport.ipc`` module was dropped), setting + ``ipc_write_buffer`` in ``master.conf`` was a no-op and the + outbound IOStream buffer grew without bound under slow-consumer + conditions. See issue #69930. + """ + + def handler(stream, body, header): # pylint: disable=unused-argument + return None + + cap = 12345 + server = salt.transport.tcp.SaltMessageServer(handler, max_write_buffer_size=cap) + + class Stream: + def __init__(self): + self.max_write_buffer_size = None + + def read_bytes(self, *args, **kwargs): + raise tornado.iostream.StreamClosedError() + + stream = Stream() + await server.handle_stream(stream, "client-cap") + + assert stream.max_write_buffer_size == cap + + +async def test_salt_message_server_no_cap_by_default(master_opts): + """ + Not passing ``max_write_buffer_size`` (or passing 0) must leave the + stream untouched -- preserves Tornado's default (unlimited) and + matches prior behavior when ``ipc_write_buffer`` is not set in + ``master.conf``. + """ + + def handler(stream, body, header): # pylint: disable=unused-argument + return None + + server = salt.transport.tcp.SaltMessageServer(handler) + assert server.max_write_buffer_size is None + + server_zero = salt.transport.tcp.SaltMessageServer(handler, max_write_buffer_size=0) + assert server_zero.max_write_buffer_size is None + + class Stream: + def __init__(self): + self.max_write_buffer_size = "sentinel" + + def read_bytes(self, *args, **kwargs): + raise tornado.iostream.StreamClosedError() + + stream = Stream() + await server.handle_stream(stream, "client-nocap") + # Untouched -- the sentinel is still there. + assert stream.max_write_buffer_size == "sentinel" + + +def test_pub_server_applies_ipc_write_buffer(master_opts, io_loop): + """ + ``PubServer.handle_stream`` must set the accepted stream's + ``max_write_buffer_size`` to ``opts['ipc_write_buffer']`` when set. + See issue #69930. + """ + master_opts["ipc_write_buffer"] = 54321 + server = salt.transport.tcp.PubServer(master_opts, io_loop=io_loop) + + class Stream: + def __init__(self): + self.max_write_buffer_size = None + self.socket = MagicMock() + self.socket.getpeercert.return_value = None + self._closed = False + + def set_close_callback(self, cb): + pass + + def close(self): + self._closed = True + + def closed(self): + return self._closed + + stream = Stream() + try: + with patch.object( + server, "_stream_read", MagicMock(return_value=None) + ), patch.object(server.io_loop, "create_task"): + server.handle_stream(stream, ("127.0.0.1", 12345)) + finally: + server.close() + + assert stream.max_write_buffer_size == 54321 + + +def test_pub_server_no_cap_when_ipc_write_buffer_zero(master_opts, io_loop): + """ + ``ipc_write_buffer == 0`` (the default when the operator hasn't + opted in) must leave the stream's ``max_write_buffer_size`` + untouched -- preserving Tornado's unlimited-write-buffer default. + """ + master_opts["ipc_write_buffer"] = 0 + server = salt.transport.tcp.PubServer(master_opts, io_loop=io_loop) + + class Stream: + def __init__(self): + self.max_write_buffer_size = "sentinel" + self.socket = MagicMock() + self.socket.getpeercert.return_value = None + self._closed = False + + def set_close_callback(self, cb): + pass + + def close(self): + self._closed = True + + def closed(self): + return self._closed + + stream = Stream() + try: + with patch.object( + server, "_stream_read", MagicMock(return_value=None) + ), patch.object(server.io_loop, "create_task"): + server.handle_stream(stream, ("127.0.0.1", 12345)) + finally: + server.close() + + assert stream.max_write_buffer_size == "sentinel" + + +def test_pub_server_apply_write_buffer_cap_helper(master_opts, io_loop): + """ + ``_apply_write_buffer_cap`` is the shared helper used by both the + plaintext ``handle_stream`` path and the SSL-delayed + ``_validate_ssl_and_add_client`` path. Verify the helper's contract + directly so both call sites are covered. + """ + master_opts["ipc_write_buffer"] = 99999 + server = salt.transport.tcp.PubServer(master_opts, io_loop=io_loop) + + class Stream: + max_write_buffer_size = None + + stream = Stream() + server._apply_write_buffer_cap(stream) + assert stream.max_write_buffer_size == 99999 + + master_opts["ipc_write_buffer"] = 0 + server2 = salt.transport.tcp.PubServer(master_opts, io_loop=io_loop) + + class Stream2: + max_write_buffer_size = "sentinel" + + stream2 = Stream2() + server2._apply_write_buffer_cap(stream2) + assert stream2.max_write_buffer_size == "sentinel"