diff --git a/changelog/70213.fixed.md b/changelog/70213.fixed.md new file mode 100644 index 000000000000..b40787ffec9e --- /dev/null +++ b/changelog/70213.fixed.md @@ -0,0 +1 @@ +Made deltaproxy's ``parallel_startup`` path behave like the serial one. A sub-proxy whose initialisation raised (for example a ``proxytype`` that does not load) propagated out of ``asyncio.gather`` and aborted the control proxy, killing the salt-proxy daemon and every healthy sub-proxy with it. Sub-proxy schedules, beacons and subprocess cleanup were also registered on a throwaway asyncio loop created in a worker thread and never run, so under ``parallel_startup`` they never fired and each sub-proxy leaked an event loop. diff --git a/salt/metaproxy/deltaproxy.py b/salt/metaproxy/deltaproxy.py index cb2b8e2c9a09..6916d0c47b26 100644 --- a/salt/metaproxy/deltaproxy.py +++ b/salt/metaproxy/deltaproxy.py @@ -59,6 +59,33 @@ log = logging.getLogger(__name__) +async def gather_subproxies(waitfor, ids): + """ + Await every sub-proxy initialisation coroutine and return the results of + the ones that succeeded. + + A sub-proxy that fails to initialise must not stop its siblings from + loading. Without ``return_exceptions=True`` a single bad ``proxytype`` + raises ``KeyError`` out of the proxy loader, propagates through + ``asyncio.gather``, and aborts the control proxy's ``post_master_init``, + which kills the salt-proxy daemon and takes every healthy sub-proxy down + with it. The non-parallel startup path has always caught per sub-proxy + and carried on; this keeps the parallel path equivalent. + """ + results = await asyncio.gather(*waitfor, return_exceptions=True) + collected = [] + for _id, result in zip(ids, results): + if isinstance(result, BaseException): + log.error( + "An exception occured during initialization for %s, skipping: %s", + _id, + result, + ) + continue + collected.append(result) + return collected + + async def post_master_init(self, master): """ Function to finish init after a deltaproxy proxy @@ -354,11 +381,7 @@ async def post_master_init(self, master): ) ) - try: - results = await asyncio.gather(*waitfor) - except Exception as exc: # pylint: disable=broad-except - log.error("Errors loading sub proxies: %s", exc) - raise + results = await gather_subproxies(waitfor, self.opts["proxy"].get("ids", [])) _failed = self.opts["proxy"].get("ids", [])[:] for sub_proxy_data in results: @@ -1167,15 +1190,34 @@ def threaded_subproxy_tune_in(proxy_minion): """ loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) - return subproxy_tune_in(proxy_minion) + try: + return subproxy_tune_in(proxy_minion) + finally: + # The worker thread is about to go away. Drop the throwaway loop + # instead of leaking its file descriptors for the life of the + # process, once per sub-proxy. + asyncio.set_event_loop(None) + loop.close() def subproxy_tune_in(proxy_minion, start=True): """ Tunein sub proxy minions """ - proxy_minion.setup_scheduler() - proxy_minion.setup_beacons() - proxy_minion.add_periodic_callback("cleanup", proxy_minion.cleanup_subprocesses) + + def _start_periodics(): + proxy_minion.setup_scheduler() + proxy_minion.setup_beacons() + proxy_minion.add_periodic_callback("cleanup", proxy_minion.cleanup_subprocesses) + + # ``PeriodicCallback.start()`` binds to ``IOLoop.current()``. Under + # ``parallel_startup`` this function runs on a ThreadPoolExecutor worker + # whose current loop is a throwaway that is never run, so starting the + # timers here would bind every one of them to a dead loop and the + # sub-proxy would silently get no schedule, no beacons and no subprocess + # cleanup. Hand the registration to the sub-proxy's own io_loop -- the + # loop that is actually run -- via the thread-safe + # ``call_soon_threadsafe``, so the timers bind to that loop instead. + proxy_minion.io_loop.call_soon_threadsafe(_start_periodics) proxy_minion._state_run() return proxy_minion diff --git a/tests/pytests/unit/metaproxy/test_deltaproxy.py b/tests/pytests/unit/metaproxy/test_deltaproxy.py index ded184c23daf..876fa63602da 100644 --- a/tests/pytests/unit/metaproxy/test_deltaproxy.py +++ b/tests/pytests/unit/metaproxy/test_deltaproxy.py @@ -229,3 +229,117 @@ def test_subproxy_post_master_init_packs_per_minion_grains( # control proxy stores the right grains in ``self.deltaproxy_opts``. assert result1["proxy_opts"]["grains"]["serial_number"] == "SN-AAA-001" assert result2["proxy_opts"]["grains"]["serial_number"] == "SN-BBB-002" + + +# --------------------------------------------------------------------------- +# parallel_startup must behave like serial startup +# --------------------------------------------------------------------------- + + +def test_gather_subproxies_skips_the_one_that_failed(): + """ + One sub-proxy failing to initialise must not take the others down. + + A bad ``proxytype`` makes the proxy loader raise ``KeyError`` on + ``.init``. Before the fix that exception propagated out of + ``asyncio.gather`` and aborted the control proxy's ``post_master_init``, + killing the salt-proxy daemon and every healthy sub-proxy with it, while + the non-parallel branch skipped the failure and carried on. + """ + + async def _ok(value): + return value + + async def _boom(): + raise KeyError("nosuchproxytype_xyz.init") + + loop = tornado.ioloop.IOLoop() + try: + collected = loop.run_sync( + lambda: deltaproxy.gather_subproxies( + [_ok("A"), _boom(), _ok("C")], + ["minionA", "minionB", "minionC"], + ) + ) + finally: + loop.close() + + # The healthy sub-proxies survive and the failure is dropped, rather than + # the whole coroutine raising. + assert collected == ["A", "C"] + + +def test_gather_subproxies_passes_everything_through_when_all_succeed(): + """ + Inverse of the above: with no failures nothing may be dropped or + reordered, so the skip path cannot silently eat healthy sub-proxies. + """ + + async def _ok(value): + return value + + loop = tornado.ioloop.IOLoop() + try: + collected = loop.run_sync( + lambda: deltaproxy.gather_subproxies( + [_ok("A"), _ok("B"), _ok("C")], + ["minionA", "minionB", "minionC"], + ) + ) + finally: + loop.close() + + assert collected == ["A", "B", "C"] + + +def test_subproxy_tune_in_starts_periodics_on_the_running_loop(): + """ + ``PeriodicCallback.start()`` binds to ``IOLoop.current()``. Under + ``parallel_startup`` ``subproxy_tune_in`` runs on a ThreadPoolExecutor + worker whose current asyncio loop is a throwaway that is never run, so + starting the timers inline bound every one of them to a dead loop and the + sub-proxy silently got no schedule, no beacons and no subprocess cleanup. + + The registration must therefore be handed to the sub-proxy's own + ``io_loop`` instead of being run inline. + """ + calls = [] + + class _FakeSubProxy: + def __init__(self): + self.io_loop = MagicMock() + + def setup_scheduler(self): + calls.append("scheduler") + + def setup_beacons(self): + calls.append("beacons") + + def add_periodic_callback(self, name, method): + calls.append(f"periodic:{name}") + + def cleanup_subprocesses(self): + pass + + def _state_run(self): + calls.append("state_run") + + proxy_minion = _FakeSubProxy() + deltaproxy.subproxy_tune_in(proxy_minion) + + # Inverse must-not: none of the periodic registrations may happen inline, + # because inline means "bound to whatever loop this thread happens to + # have", which is the dead one under parallel_startup. + assert "scheduler" not in calls + assert "beacons" not in calls + assert "periodic:cleanup" not in calls + + # They must instead be queued onto the sub-proxy's own loop, thread-safely. + assert proxy_minion.io_loop.call_soon_threadsafe.called + queued = proxy_minion.io_loop.call_soon_threadsafe.call_args[0][0] + + # And running what was queued must perform all three registrations. + # ``_state_run`` still happens inline, so it lands ahead of them. + assert calls == ["state_run"] + queued() + assert calls == ["state_run", "scheduler", "beacons", "periodic:cleanup"]