From 550244c6d7f7c8c0c9fd9918b69a00fbee0dd183 Mon Sep 17 00:00:00 2001 From: Sean Whalen <44679+seanthegeek@users.noreply.github.com> Date: Fri, 11 Sep 2026 08:40:36 -0400 Subject: [PATCH] fix: roll back the search backends' default alias when output init fails (#906) _init_output_clients() is now all-or-nothing against the Elasticsearch and OpenSearch connection registries: either it returns the fully built client dict, or it closes everything it built and leaves both "default" aliases naming exactly what they named on entry. elastic.set_hosts()/opensearch.set_hosts() end in connections.create_connection(), which registers the new client under the process-wide "default" alias the moment it is constructed (elasticsearch/dsl/connections.py:81-88 and opensearchpy/connection/connections.py:87-95, as installed: 8.19.3 / 3.2.0). Everything that runs after that -- the index migration, and every output configured later -- can still fail. On the SIGHUP reload path _main() builds the replacement clients before closing the old ones and, when the build raises, logs "Config reload failed, continuing with previous config" and keeps the old opts. But the alias had already been handed to the new client, so every save -- elastic.py's Search and Document.save() calls, which resolve the "default" alias through elasticsearch/dsl/_sync/document.py:99-100 -- reached the new cluster while the index prefixes/suffixes and index_prefix_domain_map still came from the old configuration. The half-built client was never closed either, nor was any client built earlier in the same failed call (that half also leaked on every attempt of the startup retry loop). The handler tears down first and restores second -- with the teardown in a try/finally, since a second Ctrl-C landing in it propagates straight through _close_output_clients, which swallows only Exception -- and the two steps are not interchangeable. With an _ElasticsearchHandle already in `clients`, teardown closes its client and releases the alias -- which still names that client -- and the restore then re-registers the previous client into an unset alias. Restoring first would put the previous client back and only then close the handle, which rests the whole rollback on the handle declining to touch an alias that no longer names its own client: true only since #902, and a property of the handle rather than of this function. Tearing down first keeps the guarantee local. Both failure points are traced in the comment on the handler. Closing the discarded client is best-effort, and closing it twice is safe: Elasticsearch.close() -> Transport.close() closes each node's urllib3 pool (elastic_transport/_transport.py:499-504, _node/_http_urllib3.py:224-228, and urllib3 pool close() is a no-op once cleared), and OpenSearch's connection close() guards on `if self.pool` (opensearchpy/connection/http_urllib3.py:323-328). Taking the snapshot cannot itself open a connection: get_connection() lazily builds a client from kwargs left behind by configure() (dsl/connections.py:90-115), and parsedmarc never calls configure(). The guard catches BaseException rather than Exception because migrate_indexes() wraps every cluster call in `except Exception` and logs a warning (elastic.py:824-829, and again in each of the per-index loops that follow), so what escapes the Elasticsearch block after set_hosts() is, in practice, a KeyboardInterrupt landing in one of those calls -- which is what the regression tests inject at the SDK transport boundary. For the same reason, closing the discarded client swallows BaseException: an interrupt there must not cost the alias its hand-back, and the original failure is re-raised afterwards either way. The alias is not the only module-level state set_hosts() writes: elastic.set_hosts() also assigns elastic._SERVERLESS (elastic.py:620-622) -- before it constructs the client, so it can be stale even on a failure that never reached the registry -- and create_indexes() consults it (elastic.py:652-659) to decide whether to strip the shard settings Serverless rejects. It is snapshotted and restored alongside the alias. opensearch.py has no equivalent (no `global` statement in the file). Shape: the existing body keeps its indentation as _build_output_clients(), which fills a `clients` dict the caller passes in and is explicitly not transactional; the rollback lives in a short _init_output_clients() wrapper that owns that dict on both paths. Passing the dict in is what lets the wrapper close what was already built after the build raises. The public name is unchanged, so the tests and _main() call sites are untouched. Co-authored-by: Claude Fable 5.1 --- CHANGELOG.md | 7 +- parsedmarc/cli.py | 227 +++++++++++++++++++++- tests/test_cli.py | 476 +++++++++++++++++++++++++++++++++++++++++++--- 3 files changed, 667 insertions(+), 43 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 73500fac..1b364d6b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,14 +10,9 @@ ### Bug fixes - `find_unknown_base_reverse_dns.py`'s missing-file checks for `base_reverse_dns_map.csv` and the `known_unknown`/PSL-override lists printed a clean error message but fell through into an unhandled `FileNotFoundError` traceback instead of exiting. - -### Bug fixes - - **`parse_report_file()` now closes the file handle it opens itself for a path input if reading it raises.** When `input_` is a path, the function opened the file, read it, and closed it with no exception handling in between; an exception raised by `read()` (e.g. an `OSError` from the underlying storage) skipped the close, so the descriptor was left to be released only when Python's garbage collector eventually finalized the object — CPython's `io.IOBase.__del__` closes an unclosed file on finalization () — rather than being closed deterministically. This is the pattern CodeQL's `py/file-not-closed` query flags, found in a local code-quality scan. The path branch now opens the file with a `with` block, so the handle is closed on both the success and exception paths. A file-like object or bytes buffer supplied by the caller is unaffected: as before, it is closed only after a successful read, and left open if `read()` raises. - -### Bug fixes - - **A SIGHUP configuration reload no longer breaks every subsequent save to Elasticsearch or OpenSearch.** With `[elasticsearch]` or `[opensearch]` configured in watch mode, reloading the configuration re-registered the search client under the client library's `default` connection alias and then closed the previous run's clients. The close step re-resolved that alias instead of remembering the client it was created for, so it closed the *newly built* client and deleted the `default` alias outright — after which every report save failed with `KeyError: "There is no connection with alias 'default'."` until parsedmarc was restarted, and the original client was left open. Each backend's handle now closes the exact client it was created for and gives up the alias only while the alias still points at that client. Both the Elasticsearch and OpenSearch backends were affected. +- **A configuration reload that fails part-way no longer leaves reports being written to the new Elasticsearch or OpenSearch hosts under the old configuration.** The search client is registered under the client library's process-wide `default` connection alias as soon as it is constructed — before the index migration runs, and before the outputs configured after it are created. When a later step then failed on a SIGHUP reload — for example an `[opensearch]` section that cannot build its client (an unsupported `auth_type`, `awssigv4` without an `aws_region`, AWS credentials that will not load) — parsedmarc logged `Config reload failed, continuing with previous config` and kept the old configuration. But the alias had already been handed to the new client, so every subsequent save resolved it to the *new* hosts while the index prefixes, suffixes, and `index_prefix_domain_map` still came from the old configuration. Reports were silently written to a destination that was never successfully configured, with the log saying nothing had changed. The half-built client was never closed either, nor were the other clients (S3, Kafka, PostgreSQL, and so on) built earlier in the same failed reload — the same leak occurred on every attempt of the startup retry loop, which calls the same function. Building the output clients is now all-or-nothing: if any step fails, everything built so far is closed and the module-level state the search backends keep is put back — each configured backend's `default` alias restored to exactly the client it named beforehand, and the Elasticsearch `serverless` flag (which decides whether `number_of_shards`/`number_of_replicas` are sent when an index is created) to its old value — so a failed reload no longer changes where reports are written, or how indexes are created. ## 11.0.1 diff --git a/parsedmarc/cli.py b/parsedmarc/cli.py index cae792f1..1ff91921 100644 --- a/parsedmarc/cli.py +++ b/parsedmarc/cli.py @@ -1698,11 +1698,136 @@ def _migration_index_names( return names -def _init_output_clients(opts, index_prefix_domain_map=None): - """Create output clients based on current opts. +def _search_alias_snapshot() -> list[tuple[Any, Any, Any]]: + """Record the module-level state each search backend's set_hosts() writes. + + ``elastic.set_hosts()`` and ``opensearch.set_hosts()`` register the client + they build under their SDK's process-wide ``default`` connection alias, + and the save path resolves that alias on every write. + :func:`_init_output_clients` takes this snapshot before it touches either + registry, so that a failure part-way through can put back exactly what it + found. + + Returns: + list: One ``(module, client, serverless)`` triple per backend. + ``client`` is what that backend's ``default`` alias names, or ``None`` + when the alias is unset. ``serverless`` is ``elastic._SERVERLESS`` for + the Elasticsearch backend and ``None`` for OpenSearch, which has no + equivalent. Carrying the module itself, rather than a name to look it + up by, is what lets :func:`_restore_search_aliases` put each client + back into the registry it came from without a second lookup to + disagree with. A backend whose module is ``None`` -- its optional + extra is not installed, see the guarded imports at the top of this + module -- has no state to snapshot and is left out of the list + entirely. + """ + snapshot: list[tuple[Any, Any, Any]] = [] + for module in (elastic, opensearch): + if module is None: + continue + try: + # get_connection() does not only look the alias up: it also + # *constructs* a client from kwargs stashed by an earlier + # connections.configure() call. parsedmarc never calls + # configure() -- both set_hosts() implementations register their + # client with create_connection() -- so each registry's + # ``_kwargs`` stays empty and this can only return an + # already-registered client or raise KeyError. Taking the + # snapshot can never itself open a connection. + connection = module.connections.get_connection("default") + except KeyError: + connection = None + # The alias is not the only module-level state set_hosts() writes: + # elastic.set_hosts() also assigns elastic._SERVERLESS, which + # elastic.create_indexes() consults to decide whether to strip the + # shard settings Serverless rejects. Reaching into a sibling module's + # private is the same liberty this function already takes with + # ``connections``; without it, a failed reload that flipped + # ``[elasticsearch] serverless`` would pair the restored old client + # with the new config's flag. opensearch.py declares no module + # globals at all (no ``global`` statement in the file), so there is + # nothing to pair with it. + serverless = elastic._SERVERLESS if module is elastic else None + snapshot.append((module, connection, serverless)) + return snapshot + + +def _restore_search_aliases(snapshot: list[tuple[Any, Any, Any]]) -> None: + """Put the state in *snapshot* back the way it was when it was taken. + + ``elastic._SERVERLESS`` is put back first and unconditionally; the alias + then has three cases per backend. The alias still names the client it + named before, so there is nothing to do. A different client has taken it + over -- that client is + closed, and the previous client is registered again, or the alias is + removed outright when there was no previous client. Or the alias is unset + because a fully built handle released it during the teardown that runs + first -- nothing left to close, and the previous client is simply + registered again; this is the ordinary path when the failure came after a + search backend was fully built. + + Closing is best-effort and silent: the client being closed here is the + half-built one for the configuration that just failed, it is being + discarded either way, and a teardown error is not actionable -- while + handing the alias back is what keeps the still-running configuration + writing where it thinks it is, so it must happen either way. Closing the + same client twice is safe -- ``_close_output_clients`` may already have + closed it through its handle -- because both SDKs' ``close()`` are + idempotent: ``Elasticsearch.close()`` closes each node's urllib3 pool, + whose ``close()`` is a no-op once cleared, and ``OpenSearch.close()`` + guards on ``if self.pool``. + + Args: + snapshot (list): The return value of :func:`_search_alias_snapshot`. + """ + for module, previous, previous_serverless in snapshot: + if elastic is not None and module is elastic: + # Unconditionally, and before the alias: set_hosts() assigns + # _SERVERLESS before it constructs the client, so it can be stale + # even on a failure that never reached the registry. + elastic._SERVERLESS = previous_serverless + try: + current = module.connections.get_connection("default") + except KeyError: + current = None + if current is previous: + # Nothing took the alias over, including the common case of + # both being None. Leave it alone. + continue + if current is not None: + try: + current.close() + except BaseException: + # Best-effort; see the docstring. BaseException, not + # Exception, for the same reason the caller's teardown is + # wrapped in try/finally: a Ctrl-C landing in close() must + # not cost the alias its hand-back, and the exception the + # caller re-raises afterwards still reports the failure. + pass + if previous is None: + # Cannot raise KeyError: the alias was just resolved out of this + # registry's own ``_conns``, and closing a client does not touch + # the registry, so it is still there. + module.connections.remove_connection("default") + else: + module.connections.add_connection("default", previous) + + +def _build_output_clients(opts, clients, index_prefix_domain_map=None): + """Create output clients based on current opts, into *clients*. + + Deliberately not transactional: it fills *clients* as it goes and, when a + step fails, leaves behind both the clients it had already built and any + change ``elastic.set_hosts()``/``opensearch.set_hosts()`` made to their + SDKs' module-level state. Undoing that is :func:`_init_output_clients`'s + job, which is why *clients* is a parameter -- the caller owns the dict on + the failure path too, and can close what is in it. Call + :func:`_init_output_clients`, not this. Args: opts: Namespace of parsed configuration values. + clients (dict): The dict to fill, keyed by client name. Filled in + place, and also returned. index_prefix_domain_map (dict | None): The parsed ``general.index_prefix_domain_map``. ``None`` -- the default -- means multi-tenant prefixing is not configured, so Elasticsearch @@ -1710,13 +1835,13 @@ def _init_output_clients(opts, index_prefix_domain_map=None): from ``index_prefix``/``index_suffix``. Returns: - dict of client instances keyed by name. + dict: *clients*, filled. Raises: ConfigurationError: If a required output client cannot be created. + RuntimeError: If constructing an output client fails, chained to the + error the SDK raised. """ - clients = {} - # Each check below is deliberately outside the try/except that wraps # its constructor: those handlers re-raise everything as RuntimeError, # which would bury the install hint. @@ -1865,11 +1990,12 @@ def _init_output_clients(opts, index_prefix_domain_map=None): if opts.la_dce and loganalytics is None: raise ConfigurationError(_missing_extra_hint("log_analytics", "loganalytics")) - # Elasticsearch and OpenSearch mutate module-level global state via - # connections.create_connection(), which cannot be rolled back if a later - # step fails. Initialise them last so that all other clients are created - # successfully first; this minimizes the window for partial-init problems - # during config reload. + # Elasticsearch and OpenSearch mutate module-level global state, in two + # places rather than one: connections.create_connection() registers the + # new client under the ``default`` alias, and elastic.set_hosts() also + # assigns elastic._SERVERLESS. _init_output_clients() rolls both back if + # a later step fails. They are still initialized last, so that a failure + # in any other output happens before either registry has been touched. if opts.save_aggregate or opts.save_failure or opts.save_smtp_tls: # Scoped to the same condition as the constructors below, which is # also the condition under which process_reports() dereferences @@ -2024,6 +2150,87 @@ def _init_output_clients(opts, index_prefix_domain_map=None): return clients +def _init_output_clients(opts, index_prefix_domain_map=None): + """Create output clients based on current opts, all-or-nothing. + + Either every configured client is built and returned, or the clients built + so far are closed and the module-level state that + ``elastic.set_hosts()``/``opensearch.set_hosts()`` mutate -- each + backend's ``default`` connection alias, and ``elastic._SERVERLESS`` -- is + left exactly as it was on entry. + + That guarantee is what the SIGHUP reload in :func:`_main` needs. It builds + the replacement clients before closing the old ones and keeps running with + the old ``opts`` if the build fails. But ``set_hosts()`` registers its + client under the ``default`` alias as soon as it is constructed, well + before the rest of the initialization can fail -- on an output configured + later failing to build, or on a Ctrl-C landing in the index migration. + Without the rollback, such a reload left every subsequent save resolving + that alias to the *new* cluster while ``opts`` stayed old, and leaked the + half-built client. + + Args: + opts: Namespace of parsed configuration values. + index_prefix_domain_map (dict | None): The parsed + ``general.index_prefix_domain_map``; see + :func:`_build_output_clients`. + + Returns: + dict of client instances keyed by name. + + Raises: + ConfigurationError: If a required output client cannot be created. + RuntimeError: If constructing an output client fails, chained to the + error the SDK raised. + """ + previous_search_state = _search_alias_snapshot() + clients: dict[str, Any] = {} + + try: + return _build_output_clients( + opts, clients, index_prefix_domain_map=index_prefix_domain_map + ) + except BaseException: + # Teardown first, then restore. The order is deliberate, and the two + # steps are not interchangeable. Tracing the two points at which a + # failure can leave a ``default`` alias pointing at a new client: + # + # (1) Inside the Elasticsearch block after set_hosts(), before the + # handle exists. Teardown closes the outputs built earlier and + # leaves the alias alone -- nothing in ``clients`` owns it -- and + # the restore then closes the new client and re-registers the old + # one. Either order reaches that state. + # + # (2) In a later step, with _ElasticsearchHandle already in + # ``clients`` owning the new client that holds the alias. + # Teardown first: the handle closes its client and, seeing the + # alias still name that client, releases the alias; the restore + # then finds the alias unset and re-registers the old client, + # which was never closed. Restoring first would instead put the + # old client back and only then close the handle -- leaving the + # whole rollback resting on the handle declining to touch an + # alias that no longer names its own client. It does decline + # today, but that is a property of _ElasticsearchHandle.close(), + # not of this function: before #902 the handle re-resolved the + # alias at close time and would have deleted the registration + # restored a moment earlier, leaving every later save raising + # KeyError. Tearing down first keeps this guarantee local. + # + # BaseException, not Exception: elastic.migrate_indexes() catches + # Exception around every cluster call and logs a warning, so the + # failure that escapes the Elasticsearch block after set_hosts() is, + # in practice, a KeyboardInterrupt landing in one of them. And try/finally, + # because a second Ctrl-C arriving during the teardown propagates + # straight through _close_output_clients, which swallows only + # Exception; a plain statement sequence would then skip the restore + # and leave behind exactly the state this function exists to prevent. + try: + _close_output_clients(clients) + finally: + _restore_search_aliases(previous_search_state) + raise + + def _close_output_clients(clients): """Close output clients that hold persistent connections. diff --git a/tests/test_cli.py b/tests/test_cli.py index 3c934f29..0317b333 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -23,6 +23,7 @@ from typing import cast from unittest.mock import MagicMock, patch import httpx +import opensearchpy from azure.core.exceptions import ClientAuthenticationError from elasticsearch import Elasticsearch from kiota_abstractions.api_error import APIError @@ -4200,6 +4201,29 @@ class _RaisingSearchConnection(_FakeSearchConnection): raise RuntimeError("transport already closed") +def _isolate_default_alias(test_case, connections): + """Restore whatever the process-wide registry held under the ``default`` + alias, so a test that registers a client into a real Elasticsearch or + OpenSearch connection registry cannot leak into other tests regardless of + ordering.""" + try: + previous = connections.get_connection("default") + except KeyError: + previous = None + + def restore(): + try: + connections.remove_connection("default") + except KeyError: + # Already gone: the test under this cleanup removed the + # alias itself, which is the state we are restoring to. + pass + if previous is not None: + connections.add_connection("default", previous) + + test_case.addCleanup(restore) + + class TestSearchBackendHandles(unittest.TestCase): """_ElasticsearchHandle / _OpenSearchHandle must close the client they were built for, not whatever holds the ``default`` alias at close time. @@ -4215,30 +4239,9 @@ class TestSearchBackendHandles(unittest.TestCase): the original client was never closed. """ - def _isolate_default_alias(self, connections): - """Restore whatever the process-wide registry held under the - ``default`` alias, so these tests cannot leak into other tests - regardless of ordering.""" - try: - previous = connections.get_connection("default") - except KeyError: - previous = None - - def restore(): - try: - connections.remove_connection("default") - except KeyError: - # Already gone: the test under this cleanup removed the - # alias itself, which is the state we are restoring to. - pass - if previous is not None: - connections.add_connection("default", previous) - - self.addCleanup(restore) - def testElasticsearchHandleClosesItsOwnConnectionAfterReload(self): connections = parsedmarc.elastic.connections - self._isolate_default_alias(connections) + _isolate_default_alias(self, connections) old_conn = _FakeSearchConnection("old") new_conn = _FakeSearchConnection("new") @@ -4259,7 +4262,7 @@ class TestSearchBackendHandles(unittest.TestCase): def testElasticsearchHandleReleasesAliasOnShutdown(self): """The normal teardown path still frees the alias it owns.""" connections = parsedmarc.elastic.connections - self._isolate_default_alias(connections) + _isolate_default_alias(self, connections) conn = _FakeSearchConnection("only") connections.add_connection("default", cast(Elasticsearch, conn)) @@ -4280,7 +4283,7 @@ class TestSearchBackendHandles(unittest.TestCase): alias-release step is now a no-op, since get_connection() raises KeyError with the alias already gone.""" connections = parsedmarc.elastic.connections - self._isolate_default_alias(connections) + _isolate_default_alias(self, connections) conn = _RaisingSearchConnection("raiser") connections.add_connection("default", cast(Elasticsearch, conn)) @@ -4300,7 +4303,7 @@ class TestSearchBackendHandles(unittest.TestCase): def testOpenSearchHandleClosesItsOwnConnectionAfterReload(self): connections = opensearch_module.connections - self._isolate_default_alias(connections) + _isolate_default_alias(self, connections) old_conn = _FakeSearchConnection("old") new_conn = _FakeSearchConnection("new") @@ -4317,7 +4320,7 @@ class TestSearchBackendHandles(unittest.TestCase): def testOpenSearchHandleReleasesAliasOnShutdown(self): """The normal teardown path still frees the alias it owns.""" connections = opensearch_module.connections - self._isolate_default_alias(connections) + _isolate_default_alias(self, connections) conn = _FakeSearchConnection("only") connections.add_connection("default", conn) @@ -4336,7 +4339,7 @@ class TestSearchBackendHandles(unittest.TestCase): alias-release step is now a no-op, since get_connection() raises KeyError with the alias already gone.""" connections = opensearch_module.connections - self._isolate_default_alias(connections) + _isolate_default_alias(self, connections) conn = _RaisingSearchConnection("raiser") connections.add_connection("default", conn) @@ -4353,6 +4356,425 @@ class TestSearchBackendHandles(unittest.TestCase): self.assertEqual(conn.close_count, 2) +# A host nothing listens on. Every test below patches the SDK transport +# method that would talk to it, so the client is only ever constructed -- +# which opens no socket -- and no test can reach the network even if a +# patch is later removed by mistake. +_UNREACHABLE_HOSTS = ["127.0.0.1:9299"] + + +def _es_request_without_network(client, method, path, **kwargs): + """Stand-in for ``elasticsearch.Elasticsearch.perform_request``. + + Every cluster call ``migrate_indexes()`` makes -- through + ``Index.exists()``, ``client.count()`` and ``client.update_by_query()``, + and through the namespaced clients, which delegate to the + ``Elasticsearch`` instance's own ``perform_request`` -- funnels through + this one method, so it is the SDK boundary at which a migration can be + made to succeed without a cluster. + + A ``HEAD`` (``Index.exists()``) answers "no such index", which skips the + legacy ``published_policy.fo`` migration; anything else answers the + backfill's ``count`` query with zero matching documents, which skips the + ``update_by_query``. ``migrate_indexes()`` therefore returns having + logged nothing and changed nothing. + """ + if method == "HEAD": + return {} + return {"count": 0} + + +class TestInitOutputClientsRollback(unittest.TestCase): + """_init_output_clients() is all-or-nothing against the search backends' + process-wide ``default`` connection alias. + + ``elastic.set_hosts()`` / ``opensearch.set_hosts()`` register the client + they build under that alias the moment it is constructed, but the rest of + the initialization -- the index migration, and every output configured + after them -- can still fail. The SIGHUP reload in _main() builds the + replacement clients *before* closing the old ones and keeps the old opts + when the build raises ("Config reload failed, continuing with previous + config"), so a mutated alias meant the save path -- every ``Search`` and + ``Document.save()`` in parsedmarc/elastic.py, each of which resolves the + ``default`` alias through its SDK -- reached the *new* cluster while + ``opts`` and ``index_prefix_domain_map`` still described the old one: + reports written to a destination the operator + never successfully configured, with the log saying the previous config + was still in effect. The half-built client leaked with it, as did every + client built earlier in the same failed call. + """ + + def _close_spy(self, cls): + """Patch *cls*.close() with a spy that still closes, and return the + list it appends each closed client to. + + Patching the SDK client class -- not any parsedmarc function -- is + what makes "the client was closed" observable, and the real close() + still runs, so a client that is closed twice would still have to + survive it. + """ + closed = [] + real_close = cls.close + + def spy(client): + closed.append(client) + real_close(client) + + patcher = patch.object(cls, "close", spy) + patcher.start() + self.addCleanup(patcher.stop) + return closed + + def test_interrupt_after_set_hosts_restores_the_previous_elasticsearch_client(self): + """Failure inside the Elasticsearch block, before the handle exists. + + ``migrate_indexes()`` runs once ``set_hosts()`` has already handed the + ``default`` alias to the new client, and ``clients["elasticsearch"]`` + is assigned only after it returns -- so a failure here leaves a + registered client that no handle owns. It has to be a BaseException: + ``elastic.migrate_indexes()`` wraps every cluster call in + ``except Exception`` and logs a warning (parsedmarc/elastic.py), so + what escapes this block is a Ctrl-C landing in one of them, not a + connection error. + """ + connections = parsedmarc.elastic.connections + _isolate_default_alias(self, connections) + old_conn = _FakeSearchConnection("old") + connections.add_connection("default", cast(Elasticsearch, old_conn)) + closed = self._close_spy(Elasticsearch) + registered = [] + + def interrupt(client, *args, **kwargs): + # Whatever set_hosts() registered a moment ago: the client the + # rollback is responsible for. + registered.append(connections.get_connection("default")) + raise KeyboardInterrupt + + opts = _output_client_opts( + save_aggregate=True, + elasticsearch_hosts=_UNREACHABLE_HOSTS, + elasticsearch_timeout=1.0, + ) + with patch.object(Elasticsearch, "perform_request", interrupt): + with self.assertRaises(KeyboardInterrupt): + parsedmarc.cli._init_output_clients(opts) + + self.assertEqual(len(registered), 1) + new_conn = registered[0] + self.assertIsNot(new_conn, old_conn) + self.assertIs(connections.get_connection("default"), old_conn) + self.assertEqual(closed, [new_conn]) + self.assertEqual(old_conn.close_count, 0) + + def test_interrupt_after_set_hosts_restores_the_previous_opensearch_client(self): + """The same for OpenSearch, whose alias lives in a second registry. + + The rollback ranges over both backends, and each has its own module, + registry and client class; a rollback that snapshotted or restored + one of them twice would pass the Elasticsearch test above. + """ + connections = opensearch_module.connections + _isolate_default_alias(self, connections) + old_conn = _FakeSearchConnection("old") + connections.add_connection("default", old_conn) + closed = self._close_spy(opensearchpy.OpenSearch) + registered = [] + + def interrupt(transport, *args, **kwargs): + registered.append(connections.get_connection("default")) + raise KeyboardInterrupt + + opts = _output_client_opts( + save_aggregate=True, + opensearch_hosts=_UNREACHABLE_HOSTS, + opensearch_timeout=1.0, + ) + with patch.object(opensearchpy.Transport, "perform_request", interrupt): + with self.assertRaises(KeyboardInterrupt): + parsedmarc.cli._init_output_clients(opts) + + self.assertEqual(len(registered), 1) + new_conn = registered[0] + self.assertIsNot(new_conn, old_conn) + self.assertIs(connections.get_connection("default"), old_conn) + self.assertEqual(closed, [new_conn]) + self.assertEqual(old_conn.close_count, 0) + + def test_failure_with_no_previous_alias_leaves_the_alias_unset(self): + """The startup case: there was no ``default`` alias to restore. + + Rolling back to "unset" is not the same as leaving the discarded + client in place. parsedmarc's own retry loop calls + _init_output_clients() again after a failed start, and elastic.py's + save path resolves the alias on every write, so a leftover + registration is a client for a configuration that never finished + being built. + """ + connections = parsedmarc.elastic.connections + _isolate_default_alias(self, connections) + try: + connections.remove_connection("default") + except KeyError: + # Nothing was registered: the startup state this test needs. + pass + closed = self._close_spy(Elasticsearch) + registered = [] + + def interrupt(client, *args, **kwargs): + registered.append(connections.get_connection("default")) + raise KeyboardInterrupt + + opts = _output_client_opts( + save_aggregate=True, + elasticsearch_hosts=_UNREACHABLE_HOSTS, + elasticsearch_timeout=1.0, + ) + with patch.object(Elasticsearch, "perform_request", interrupt): + with self.assertRaises(KeyboardInterrupt): + parsedmarc.cli._init_output_clients(opts) + + with self.assertRaises(KeyError): + connections.get_connection("default") + self.assertEqual(closed, registered) + + def test_failure_in_a_later_output_closes_everything_already_built(self): + """Elasticsearch fully built, then a later step fails. + + ``[opensearch] auth_type = awssigv4`` with no ``aws_region`` is a real + config error, raised by ``opensearch.set_hosts()`` before it + constructs anything -- and OpenSearch is the only output initialized + after Elasticsearch, so this is the shape of the reload that leaves an + _ElasticsearchHandle in ``clients``, owning the client that now holds + the alias. The handle must be closed (that is what releases the alias + again), the Kafka client built earlier in the same call must be closed + too, and both aliases must end up naming what they named on entry. + + The Elasticsearch client is closed *exactly* once: the teardown does + it through the handle, and the restore then finds the alias already + released rather than closing the same client a second time. + """ + es_connections = parsedmarc.elastic.connections + os_connections = opensearch_module.connections + _isolate_default_alias(self, es_connections) + _isolate_default_alias(self, os_connections) + old_es = _FakeSearchConnection("old es") + old_os = _FakeSearchConnection("old os") + es_connections.add_connection("default", cast(Elasticsearch, old_es)) + os_connections.add_connection("default", old_os) + closed = self._close_spy(Elasticsearch) + built = [] + + def request(client, method, path, **kwargs): + # The client set_hosts() registered, recorded on its first + # cluster call, so the assertions below can name it. + if not built: + built.append(es_connections.get_connection("default")) + return _es_request_without_network(client, method, path, **kwargs) + + opts = _output_client_opts( + save_aggregate=True, + kafka_hosts=["kafka.example.com:9092"], + elasticsearch_hosts=_UNREACHABLE_HOSTS, + elasticsearch_timeout=1.0, + opensearch_hosts=_UNREACHABLE_HOSTS, + opensearch_auth_type="awssigv4", + ) + with ( + patch("parsedmarc.kafkaclient.KafkaProducer") as mock_producer, + patch.object(Elasticsearch, "perform_request", request), + ): + with self.assertRaises(RuntimeError) as context: + parsedmarc.cli._init_output_clients(opts) + + self.assertIn("aws_region", str(context.exception)) + mock_producer.return_value.close.assert_called_once() + self.assertEqual(built, closed) + self.assertEqual(len(closed), 1) + self.assertIs(es_connections.get_connection("default"), old_es) + self.assertIs(os_connections.get_connection("default"), old_os) + self.assertEqual(old_es.close_count, 0) + self.assertEqual(old_os.close_count, 0) + + def test_a_failed_build_does_not_leave_the_serverless_flag_flipped(self): + """The alias is not the only module-level state ``set_hosts()`` writes. + + ``elastic.set_hosts()`` also assigns ``elastic._SERVERLESS``, which + ``elastic.create_indexes()`` consults to decide whether to strip the + ``number_of_shards``/``number_of_replicas`` settings Elastic Cloud + Serverless rejects with HTTP 400. A reload that turns + ``[elasticsearch] serverless`` on and then fails would otherwise pair + the restored old client -- a normal cluster -- with the new flag, and + the next index created on it would silently lose its shard settings. + + Both halves are observed: the flag really is flipped while the build + runs (otherwise the assertion after it would hold for the wrong + reason), and it is back to its old value once the rollback is done. + """ + es_connections = parsedmarc.elastic.connections + _isolate_default_alias(self, es_connections) + old_es = _FakeSearchConnection("old es") + es_connections.add_connection("default", cast(Elasticsearch, old_es)) + self.addCleanup( + setattr, parsedmarc.elastic, "_SERVERLESS", parsedmarc.elastic._SERVERLESS + ) + parsedmarc.elastic._SERVERLESS = False + during = [] + + def request(client, method, path, **kwargs): + # The first cluster call happens after set_hosts() has written + # both the alias and the flag. + if not during: + during.append(parsedmarc.elastic._SERVERLESS) + return _es_request_without_network(client, method, path, **kwargs) + + opts = _output_client_opts( + save_aggregate=True, + elasticsearch_hosts=_UNREACHABLE_HOSTS, + elasticsearch_timeout=1.0, + elasticsearch_serverless=True, + opensearch_hosts=_UNREACHABLE_HOSTS, + opensearch_auth_type="awssigv4", + ) + with patch.object(Elasticsearch, "perform_request", request): + with self.assertRaises(RuntimeError): + parsedmarc.cli._init_output_clients(opts) + + self.assertEqual(during, [True]) + self.assertIs(parsedmarc.elastic._SERVERLESS, False) + self.assertIs(es_connections.get_connection("default"), old_es) + + def test_an_interrupt_during_teardown_still_restores_the_alias(self): + """A second Ctrl-C, landing in the teardown of the first one's rollback. + + The rollback exists because a KeyboardInterrupt can escape the + Elasticsearch block, and an operator holding the key sends more than + one. ``_close_output_clients()`` swallows only ``Exception``, and so + does ``_ElasticsearchHandle.close()``, so an interrupt raised by the + client's own ``close()`` propagates out of the teardown. Run as a + plain statement sequence, that would skip the hand-back and leave the + alias on the new cluster -- the very state the rollback is for -- so + the teardown is wrapped in ``try``/``finally`` and the discarded + client's ``close()`` swallows BaseException. + """ + es_connections = parsedmarc.elastic.connections + os_connections = opensearch_module.connections + _isolate_default_alias(self, es_connections) + _isolate_default_alias(self, os_connections) + old_es = _FakeSearchConnection("old es") + old_os = _FakeSearchConnection("old os") + es_connections.add_connection("default", cast(Elasticsearch, old_es)) + os_connections.add_connection("default", old_os) + attempted = [] + + def interrupting_close(client): + attempted.append(client) + raise KeyboardInterrupt + + # Elasticsearch is built, then [opensearch] fails to build its + # client -- the same later-step failure as the test above -- and the + # interrupt arrives while the Elasticsearch handle is being closed. + opts = _output_client_opts( + save_aggregate=True, + elasticsearch_hosts=_UNREACHABLE_HOSTS, + elasticsearch_timeout=1.0, + opensearch_hosts=_UNREACHABLE_HOSTS, + opensearch_auth_type="awssigv4", + ) + with ( + patch.object(Elasticsearch, "perform_request", _es_request_without_network), + patch.object(Elasticsearch, "close", interrupting_close), + ): + with self.assertRaises(KeyboardInterrupt): + parsedmarc.cli._init_output_clients(opts) + + # Twice: once through the handle, and once more by the rollback, + # which still finds the alias unreleased because the handle's close() + # never got that far. + self.assertEqual(len(attempted), 2) + self.assertIs(es_connections.get_connection("default"), old_es) + self.assertIs(os_connections.get_connection("default"), old_os) + self.assertEqual(old_es.close_count, 0) + + def test_a_discarded_client_that_cannot_be_closed_still_returns_the_alias(self): + """Closing the discarded client is best-effort; the hand-back is not. + + A client whose transport is already broken raises from ``close()``. + Restoring the alias is the half that keeps the configuration still in + force writing to the cluster it was configured for, so it must not be + skipped because the client being thrown away could not be closed. + """ + connections = parsedmarc.elastic.connections + _isolate_default_alias(self, connections) + old_conn = _FakeSearchConnection("old") + connections.add_connection("default", cast(Elasticsearch, old_conn)) + attempted = [] + + def failing_close(client): + attempted.append(client) + raise RuntimeError("transport already closed") + + def interrupt(client, *args, **kwargs): + raise KeyboardInterrupt + + opts = _output_client_opts( + save_aggregate=True, + elasticsearch_hosts=_UNREACHABLE_HOSTS, + elasticsearch_timeout=1.0, + ) + with ( + patch.object(Elasticsearch, "perform_request", interrupt), + patch.object(Elasticsearch, "close", failing_close), + ): + with self.assertRaises(KeyboardInterrupt): + parsedmarc.cli._init_output_clients(opts) + + self.assertEqual(len(attempted), 1) + self.assertIs(connections.get_connection("default"), old_conn) + + def test_successful_init_hands_the_alias_to_the_new_client(self): + """The negative half: a successful call rolls nothing back. + + The alias must name the client ``set_hosts()`` just built -- it is + what the save path resolves -- the previous client must be left + untouched for _main() to close on its own terms, and the returned + handle must own the new client, which is what makes + _close_output_clients() close it later. Ownership is observed by + closing the handle and watching which client's close() runs, not by + reading the handle's attributes. + """ + connections = parsedmarc.elastic.connections + _isolate_default_alias(self, connections) + # set_hosts() assigns elastic._SERVERLESS, and on the success path + # nothing rolls it back, so restore it here or it leaks into the + # rest of the run. + self.addCleanup( + setattr, parsedmarc.elastic, "_SERVERLESS", parsedmarc.elastic._SERVERLESS + ) + old_conn = _FakeSearchConnection("old") + connections.add_connection("default", cast(Elasticsearch, old_conn)) + closed = self._close_spy(Elasticsearch) + + opts = _output_client_opts( + save_aggregate=True, + elasticsearch_hosts=_UNREACHABLE_HOSTS, + elasticsearch_timeout=1.0, + ) + with patch.object( + Elasticsearch, "perform_request", _es_request_without_network + ): + clients = parsedmarc.cli._init_output_clients(opts) + new_conn = connections.get_connection("default") + + self.assertIsNot(new_conn, old_conn) + self.assertEqual(closed, []) + self.assertEqual(old_conn.close_count, 0) + self.assertEqual(sorted(clients), ["elasticsearch"]) + + clients["elasticsearch"].close() + + self.assertEqual(closed, [new_conn]) + + def _domain_map_tls_reports(): """Four SMTP TLS reports for the index_prefix_domain_map tests: two whose policy domains fold to the mapped base domain example.com (one of