From b41230dc076cc05b7c9c01ce03098989a3fff2ac Mon Sep 17 00:00:00 2001 From: Sean Whalen <44679+seanthegeek@users.noreply.github.com> Date: Sat, 12 Sep 2026 13:45:53 -0400 Subject: [PATCH] Make the SIGHUP configuration reload atomic (#910) * fix: make the SIGHUP configuration reload atomic The reload built the replacement output clients, then immediately closed the old ones and swapped them in -- and only afterwards ran seven more fallible steps: load_reverse_dns_map() (which reloads the PSL overrides first), load_ip_db(), configure_ipinfo_api(), the setattr loop onto opts, _build_parser_config(), the watch-parameter int() conversions, and the log-level/FileHandler refresh. An exception in any of them was caught and logged as "Config reload failed, continuing with previous config", which was false: the new clients were already live and receiving reports, the old ones were closed and unrecoverable, opts could be half-applied, and both loaders empty their target before reading anything -- so a bad map path emptied REVERSE_DNS_MAP until some later uncached lookup happened to refill it from the previous configuration's paths (both get_service_from_reverse_dns_base_domain() and get_ip_address_info() reload a map they find empty), and a bad PSL overrides path emptied parsedmarc.utils.psl_overrides without touching the map, so nothing triggered that lazy reload and base domains were folded without the overrides until a restart. The reload is now three phases. Phase 1 stages every fallible step off to the side, in this order: _load_config/_parse_config, _init_output_clients, the derived values (_build_parser_config and the three watch-parameter int()s), load_reverse_dns_map into a fresh dict, load_ip_db, configure_ipinfo_api. The derived values run early because they write nothing at all, and the two loaders that write process-wide state run last, so as little as possible can fail behind them -- load_ip_db()'s download in particular rewrites the shared MMDB cache file before it assigns _IP_DB_PATH, and that is the one phase-1 side effect no snapshot can undo (what it leaves is the same refresh startup performs). The parsedmarc.utils globals the loaders assign are snapshotted by a new _utils_globals_snapshot(). Enumerated from the loaders' bodies, those are psl_overrides (cleared and repopulated in place, no `global` statement -- the list object itself is the state), _IP_DB_PATH (the only global load_ip_db assigns) and _IPINFO_API_TOKEN (the only one configure_ipinfo_api assigns, and assigned before the probe that can fail). _LAST_LOGGED_IP_DB_PATH is deliberately not covered: it is assigned by _get_ip_database_path(), which no loader the reload calls reaches, and it only decides whether the database path is logged again. Phase 2 commits with assignments only, REVERSE_DNS_MAP in place -- cli.py imports that name from parsedmarc, so rebinding it here would desync this module from parsedmarc.REVERSE_DNS_MAP and from the config.REVERSE_DNS_MAP that ParserConfig.__setstate__ rebinds to in a worker, and the ParserConfig staged in phase 1 already holds the pre-reload object. Phase 3 closes the replaced clients last and best effort, so a dying connection cannot undo the commit. On failure the replacement clients are closed, then the search backends' default alias is restored -- that order, because a fully built _ElasticsearchHandle releases the alias as it closes (see #906 and #902) -- and then the utils globals are put back. REVERSE_DNS_MAP, opts, clients, parser_config and the three watch parameters are never written before the commit, so there is nothing to undo for them; the per-step end state is traced in a comment at the rollback site, including which restores are load-bearing at which depth and which are defensive. configure_ipinfo_api's sys.exit(1) on an invalid token still bypasses all of this: SystemExit is not an Exception, and the process is leaving. Closing the replaced log file was the one statement in the commit block that could still raise (a full disk, a stale mount); it is now caught and reported after the handler is detached, since by that point the reloaded configuration is live and rolling back to report a failed reload would be a lie. Nothing runs concurrently with the swap: the reload happens in the main thread between watch_inbox() calls, after the mailbox backend has returned at a batch boundary, so no save callback is in flight, and parallel.py creates and tears down its ProcessPoolExecutor inside a single parallel_map() call -- its workers are separate processes with their own copies of REVERSE_DNS_MAP and the utils globals either way. The previous code mutated the same state at the same point, so this is not a change in exposure. #906 made _init_output_clients() itself all-or-nothing; this does the same for the whole reload. The docs already promised that a failed reload leaves "the previous configuration fully active"; that promise is now true, and usage.md spells out what it covers and the one thing it does not. Co-Authored-By: Claude Fable 5.1 * fix: open the replacement log file during reload staging, not after commit The logging refresh was the one fallible step left after the commit: phase 2 closed and removed the old FileHandler, then tried to open the new log_file inside a try/except that only warned. On an unwritable path the old handler was already gone, file logging was dead, opts.active_log_file was set to the new path anyway, and the reload was logged as successful. Because the old-versus-new comparison uses active_log_file, a later SIGHUP with the same path -- after the operator fixed permissions -- was a no-op, so only a restart recovered file logging. The replacement FileHandler is now constructed in phase 1, after the watch-parameter conversions and before the loaders (constructing it opens the file immediately, and that is the only thing in the refresh that can fail on a bad path). The rollback closes it if a later step fails; phase 2 only attaches it. An unwritable log_file is therefore a reload failure, with the previous log file still attached and receiving the traceback. Startup keeps its warning-only behaviour: there is no previous configuration to fall back to there. Found by Copilot's review of #910. Co-Authored-By: Claude Fable 5.1 * fix: retry a log file that could not be opened at startup on the next reload Startup recorded opts.active_log_file = opts.log_file even when opening the FileHandler failed (a warning there, since there is no previous configuration to keep). The SIGHUP reload only stages a replacement handler when the new config's log_file differs from active_log_file, so a reload naming the same path never retried the open: file logging stayed dead until restart, even after the operator fixed the directory or permissions. active_log_file now starts as None and is set only after the handler is attached, so an unchanged log_file that failed at startup is retried by the next reload, and one that opened at startup is not reopened. Co-Authored-By: Claude Fable 5.1 * fix: make the rollback's staged log handler close best-effort; correct the docs on what aborts a reload FileHandler.close() can raise while flushing or closing (a full disk, a stale network mount). In the reload's rollback that raise escaped the except block: the search-alias and utils-globals restores never ran, the original staging failure was replaced, and the watcher exited. The close is now logged and swallowed like the commit path's close of the replaced handler, so the restores that follow it always run. usage.md said "unreachable output destinations" abort a reload. That is true of PostgreSQL, Kafka, and TCP/TLS syslog, whose clients connect in their constructors, but not of Elasticsearch or OpenSearch: set_hosts() builds the client without connecting, and migrate_indexes() catches every cluster call and warns, so a reload pointing at an unreachable search cluster still commits and fails on the first save. The list now names what actually raises during staging and calls out the exception. Both found by Copilot's second review of #910. Co-Authored-By: Claude Fable 5.1 --------- Co-authored-by: Claude Fable 5.1 --- CHANGELOG.md | 1 + docs/source/usage.md | 39 +- parsedmarc/cli.py | 373 ++++++++++++++--- tests/test_cli.py | 926 +++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 1283 insertions(+), 56 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index fd5cbe61..23461479 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,7 @@ - **`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 caller-supplied file-like object's closing behavior changed again later in this same Unreleased version — see the *Changes* section above.) - **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. +- **A SIGHUP configuration reload that fails after the replacement output clients are built now really does keep the previous configuration, instead of leaving the new clients live and the old ones closed.** The reload built the replacement clients, then immediately closed the old ones and swapped them in — and only afterwards reloaded the reverse DNS map and the PSL overrides, reloaded the IP database, re-applied the IPinfo API token, copied the new values onto `opts`, rebuilt the `ParserConfig`, and converted the watch parameters. A failure in any of those (a typo in `local_reverse_dns_map_path` or `local_psl_overrides_path`, an unreadable map or overrides file, an IP database that cannot be resolved) was caught and logged as `Config reload failed, continuing with previous config`, which was not true: the new output clients were already receiving every report, the old clients were closed and could not be brought back, `opts` could be half-applied, and `load_reverse_dns_map()`/`load_psl_overrides()` — both of which empty their target before reading anything — had left one of the two empty. A bad map path emptied the reverse DNS map, so reverse DNS lookups stopped resolving to known services until some later uncached lookup happened to refill the map from the *previous* configuration's paths (`get_service_from_reverse_dns_base_domain()` and `get_ip_address_info()` both reload a map they find empty). A bad PSL overrides path emptied the override list without ever touching the map, so nothing triggered that lazy reload and base domains were folded without the overrides until parsedmarc was restarted. The whole reload is now staged before anything live is touched: the map is loaded into a fresh dict, the `parsedmarc.utils` globals the loaders assign are snapshotted, the replacement `log_file` is opened during staging, and the commit is a run of plain assignments followed by the logging refresh, whose one remaining failure point — closing the replaced log file — is caught and logged rather than allowed to abort a reload that is already live. A `log_file` that cannot be opened now aborts the reload with the previous log file still attached and still receiving logs; previously the old handler had already been removed, the failed open was only a warning, file logging stayed dead, and a later reload of the same path was a no-op that never retried it. Relatedly, a `log_file` that could not be opened at startup — a warning there, since there is no previous configuration to keep — is now retried by the next reload even when the setting is unchanged; startup used to record the unopened file as the active one, so the reload saw nothing to do. On failure the replacement clients are closed, the log file opened for the reload is closed (best-effort, so a close error cannot skip the restores that follow it), the search backends' `default` connection alias is handed back to the old client, and the `utils` globals are restored, so the log message is accurate. The clients the reload replaces are now closed last, after the new configuration is live, and a failure to close one is logged without disturbing the reload. [#906](https://github.com/domainaware/parsedmarc/pull/906) fixed the other half of this — building the clients themselves is all-or-nothing. ## 11.0.1 diff --git a/docs/source/usage.md b/docs/source/usage.md index 978dade1..77f23475 100644 --- a/docs/source/usage.md +++ b/docs/source/usage.md @@ -1383,7 +1383,7 @@ for that batch have completed. The following settings are reloaded: - Multi-tenant index prefix domain map (`index_prefix_domain_map` — the referenced YAML file is re-read on reload) - DNS and GeoIP settings (`nameservers`, `dns_timeout`, `ip_db_path`, - `ip_db_url`, `offline`, etc.) + `ipinfo_url`, `offline`, etc.) - Processing flags (`strip_attachment_payloads`, `batch_size`, `check_timeout`, etc.) - Log level (`debug`, `verbose`, `warnings`, `silent`) @@ -1392,16 +1392,37 @@ Mailbox connection settings (IMAP host/credentials, Microsoft Graph, Gmail API, Maildir path) are **not** reloaded — changing those still requires a full restart. -On a **successful** reload, existing output client connections are -closed and new ones are created from the updated configuration. The -service then resumes watching with the new settings. +On a **successful** reload, the output clients for the updated +configuration are created first, and the connections they replace are +closed once the new configuration is live. The service then resumes +watching with the new settings. If the new configuration file contains errors (missing required -settings, unreachable output destinations, etc.), the **entire reload -is aborted** — no output clients are replaced and the previous -configuration remains fully active. This means a typo in one section -will not take down an otherwise working setup. Check the logs for -details: +settings, an output client that cannot be built — a PostgreSQL server, +Kafka broker, or TCP/TLS syslog server that cannot be reached, or an +`[opensearch]` `auth_type` that is not supported — an unreadable reverse +DNS map or PSL overrides file, a `log_file` that cannot be opened for +writing, a non-numeric `batch_size`, etc.), the **entire reload is +aborted** — any output clients built for the new configuration are +closed again, and the previous configuration remains fully active: the +old output clients stay open and connected, and the reverse DNS map, the +PSL overrides, the IP database selection, and every other setting keep +the values they had before the `SIGHUP`. This means a typo in one +section will not take down an otherwise working setup. Unlike startup, +where an unwritable `log_file` is only a warning, on reload it is one of +the errors that abort the reload, so the previous log file keeps +receiving logs — including the one explaining why. A `log_file` that +could not be opened when parsedmarc started (a warning at startup) is +retried by the next reload even if the setting is unchanged. (One thing +an aborted reload does not put back: if it got as far as downloading a +new IP database from a changed `ipinfo_url`, that file stays in the +shared cache directory. Which database parsedmarc *uses* is unchanged, +and the cached file is the same one the next restart would download.) +Elasticsearch and OpenSearch are the exception to "cannot be reached": +their clients connect lazily and the index migration only warns when the +cluster is unreachable, so a reload that points at an unreachable +cluster still commits, and the failure shows up on the first save +instead. Check the logs for details: ```bash journalctl -u parsedmarc.service -r diff --git a/parsedmarc/cli.py b/parsedmarc/cli.py index 1ff91921..968e4168 100644 --- a/parsedmarc/cli.py +++ b/parsedmarc/cli.py @@ -121,8 +121,10 @@ from parsedmarc.mail import ( ) from parsedmarc.parallel import _parse_report_file_job, parallel_map from parsedmarc.types import ParsedReport, ParsingResults +import parsedmarc.utils from parsedmarc.utils import ( InvalidIPinfoAPIKey, + ReverseDNSMap, configure_ipinfo_api, get_base_domain, get_reverse_dns, @@ -1813,6 +1815,60 @@ def _restore_search_aliases(snapshot: list[tuple[Any, Any, Any]]) -> None: module.connections.add_connection("default", previous) +def _utils_globals_snapshot() -> dict[str, Any]: + """Record the ``parsedmarc.utils`` module-level state the config loaders + write, so a failed SIGHUP reload can put it back. + + Three globals, enumerated from the loaders' bodies rather than from + their docstrings: + + * ``psl_overrides`` -- ``load_psl_overrides()`` clears the list and + repopulates it in place. It declares no ``global``: the list object + itself is the shared state, which is why this is snapshotted by value. + ``load_reverse_dns_map()`` calls that loader before it reads anything + of its own, so staging the map into a fresh dict does *not* keep the + overrides list out of the reload's blast radius. + ``get_base_domain()`` reads it on every lookup. + * ``_IP_DB_PATH`` -- the only global ``load_ip_db()`` assigns. + ``_get_ip_database_path()`` falls back to it whenever the caller's + own ``ip_db_path`` is unset or names a file that is not there, which + is every MMDB lookup in the common case of no ``ip_db_path``. + * ``_IPINFO_API_TOKEN`` -- the only global ``configure_ipinfo_api()`` + assigns, and it is assigned *before* the token probe that can fail, + so a rejected probe leaves the new token behind. + + ``_LAST_LOGGED_IP_DB_PATH`` is deliberately not covered: it is assigned + by ``_get_ip_database_path()``, which none of the loaders the reload + calls reach, and it only decides whether the selected database path is + logged again. + + Returns: + dict: The values, to be handed to :func:`_restore_utils_globals`. + """ + return { + "psl_overrides": list(parsedmarc.utils.psl_overrides), + "_IP_DB_PATH": parsedmarc.utils._IP_DB_PATH, + "_IPINFO_API_TOKEN": parsedmarc.utils._IPINFO_API_TOKEN, + } + + +def _restore_utils_globals(snapshot: dict[str, Any]) -> None: + """Put the state in *snapshot* back the way it was when it was taken. + + ``psl_overrides`` is restored in place, not rebound, for the same reason + it is snapshotted by value: ``load_psl_overrides()`` mutates that one + list object, and rebinding ``parsedmarc.utils.psl_overrides`` would + leave any already-bound reference to the old object holding the failed + reload's contents. + + Args: + snapshot (dict): The return value of :func:`_utils_globals_snapshot`. + """ + parsedmarc.utils.psl_overrides[:] = snapshot["psl_overrides"] + parsedmarc.utils._IP_DB_PATH = snapshot["_IP_DB_PATH"] + parsedmarc.utils._IPINFO_API_TOKEN = snapshot["_IPINFO_API_TOKEN"] + + def _build_output_clients(opts, clients, index_prefix_domain_map=None): """Create output clients based on current opts, into *clients*. @@ -3031,6 +3087,12 @@ def _main(): logger.setLevel(logging.INFO) if opts.debug: logger.setLevel(logging.DEBUG) + # The log file currently being written -- what a SIGHUP reload compares + # the new config's log_file against. None when no file is attached, + # including when the configured one could not be opened, so that a + # reload naming the same path tries again once the operator has fixed + # it. + opts.active_log_file = None if opts.log_file: try: fh = logging.FileHandler(opts.log_file, "a") @@ -3039,10 +3101,9 @@ def _main(): ) fh.setFormatter(formatter) logger.addHandler(fh) + opts.active_log_file = opts.log_file except Exception as error: logger.warning(f"Unable to write to log file: {error}") - - opts.active_log_file = opts.log_file _configure_dependency_logging(logger.level) if ( @@ -3634,7 +3695,26 @@ def _main(): logger.info("SIGHUP received, config will reload after current batch") _reload_requested = False logger.info("Reloading configuration...") + # Both snapshots are taken before the try, because the rollback + # in the handler needs them and neither can fail: taking the + # search snapshot only resolves an alias (KeyError caught) and + # reads elastic._SERVERLESS, and taking the utils snapshot only + # copies three module attributes. + previous_search_state = _search_alias_snapshot() + previous_utils_state = _utils_globals_snapshot() + # Also bound before the try: the rollback closes whatever was + # built, and an empty dict makes that a no-op when the failure + # came before (or from inside) _init_output_clients(). + new_clients: dict[str, Any] = {} + # Bound before the try for the same reason: the rollback closes + # the replacement log file handler if phase 1 opened one, and + # ``None`` means there is nothing to close. + staged_log_handler: logging.FileHandler | None = None try: + # Phase 1 -- stage every fallible step off to the side. + # Nothing the running configuration reads is written here, + # so any failure below is recoverable by the handler. + # # Build a fresh opts starting from CLI-only defaults so that # sections removed from the config file actually take effect. new_opts = Namespace(**vars(opts_from_cli)) @@ -3644,17 +3724,78 @@ def _main(): new_opts, index_prefix_domain_map=new_index_prefix_domain_map ) - # All steps succeeded — commit the changes atomically. - _close_output_clients(clients) - clients = new_clients - index_prefix_domain_map = new_index_prefix_domain_map + # Derived values first, because they are the cheapest steps + # to fail and the only ones that write nothing at all: they + # read new_opts and nothing else. int() raises on a + # non-numeric value, and running that before the loaders + # keeps such a failure from reaching load_ip_db(), whose + # download is the one part of phase 1 with a side effect + # outside this process (see the rollback comment). + new_parser_config = _build_parser_config(new_opts) + new_batch_size = ( + int(new_opts.mailbox_batch_size) + if new_opts.mailbox_batch_size is not None + else 10 + ) + new_check_timeout = ( + int(new_opts.mailbox_check_timeout) + if new_opts.mailbox_check_timeout is not None + else 30 + ) + new_max_unsaved_retries = ( + int(new_opts.mailbox_max_unsaved_retries) + if new_opts.mailbox_max_unsaved_retries is not None + else 2 + ) + + # Open the replacement log file, if the config names a + # different one. Opening it is the one thing in the logging + # refresh that can fail on a bad path, and constructing the + # handler opens the file immediately (FileHandler's default + # ``delay=False``), so opening it here is what leaves phase + # 2 with nothing fallible to do but the close. Nothing in + # the running configuration is touched: the handler is not + # attached to the logger until the commit. It sits before + # the loaders because its only side effect outside this + # process is opening the new log file for append, which is + # smaller than load_ip_db()'s download. + # + # ``opts`` is still the pre-reload namespace here -- phase 2 + # is what copies new_opts onto it -- which is deliberate: + # the comparison is between the file being written now and + # the one the new config names. + # + # Note the asymmetry with startup, where an unwritable log + # file is only a warning: there is no previous configuration + # to keep there. Here there is, so it is a reload failure -- + # and the traceback lands in the old log file, which is + # still attached. + old_log_file = getattr(opts, "active_log_file", None) + new_log_file = new_opts.log_file + if old_log_file != new_log_file and new_log_file: + staged_log_handler = logging.FileHandler(new_log_file, "a") + staged_log_handler.setFormatter( + logging.Formatter( + "%(asctime)s - %(levelname)s" + " - [%(filename)s:%(lineno)d] - %(message)s" + ) + ) # Reload the reverse DNS map so changes to the # map path/URL in the config take effect. PSL overrides # are reloaded alongside it so map entries that depend on # a folded base domain keep working. + # + # Into a fresh dict, not REVERSE_DNS_MAP: load_reverse_dns_map() + # clears the dict it is handed before it has read anything, so + # loading straight into the live map would empty it and leave + # it empty if the read then failed. The overrides list has no + # such seam -- load_psl_overrides() clears and repopulates one + # module-level list in place -- which is what the utils + # snapshot above is for. + staged_reverse_dns_map: ReverseDNSMap = {} load_reverse_dns_map( - REVERSE_DNS_MAP, + staged_reverse_dns_map, always_use_local_file=new_opts.always_use_local_files, local_file_path=new_opts.reverse_dns_map_path, url=new_opts.reverse_dns_map_url, @@ -3664,7 +3805,11 @@ def _main(): ) # Reload the IP database so changes to the - # db path/URL in the config take effect. + # db path/URL in the config take effect. Assigns + # parsedmarc.utils._IP_DB_PATH, which the utils snapshot + # covers, and -- when the download succeeds -- rewrites the + # shared cache file, which nothing can cover. Last but one + # for that reason. load_ip_db( always_use_local_file=new_opts.always_use_local_files, local_file_path=new_opts.ip_db_path, @@ -3672,9 +3817,13 @@ def _main(): offline=new_opts.offline, ) - # Re-apply IPinfo API settings. Passing a falsy token disables - # the API; a rotated token picks up here too. An invalid token - # is fatal even on reload — the operator asked for it. + # Re-apply IPinfo API settings, last, because an invalid + # token is fatal even on reload -- the operator asked for it + # -- and nothing should have to be unwound behind it. + # Passing a falsy token disables the API; a rotated token + # picks up here too. The exit is a SystemExit, not an + # Exception: it deliberately skips the rollback below, + # because the process is leaving. try: configure_ipinfo_api( new_opts.ipinfo_api_token if not new_opts.offline else None, @@ -3682,28 +3831,146 @@ def _main(): except InvalidIPinfoAPIKey as e: logger.critical(str(e)) sys.exit(1) + except Exception: + # Rollback. Everything phase 1 wrote in this process is + # covered by the two snapshots -- the search backends' + # ``default`` alias and ``elastic._SERVERLESS`` (both + # through _init_output_clients), and the three + # parsedmarc.utils globals the loaders assign -- and all of + # it is put back here, in that order. + # + # One side effect is outside them and cannot be undone: a + # load_ip_db() whose download succeeds rewrites the shared + # cache file at /parsedmarc/ipinfo_lite.mmdb before it + # assigns _IP_DB_PATH. Phase 1 runs that step last but one + # so that as little as possible can fail behind it, and the + # file it leaves is the same refresh startup performs on + # every launch -- the configured URL's current database -- + # so the cost of not undoing it is bounded to that. + # + # ``Exception``, where _init_output_clients catches + # ``BaseException`` (#906): that function also runs at + # startup, before the signal handlers exist, so a Ctrl-C can + # land inside it. Here it cannot -- watch mode has replaced + # SIGINT with a handler that sets a flag and, on a second + # press, calls os._exit(130) -- and the one BaseException + # that does reach this point, the SystemExit from an invalid + # IPinfo token, must not be rolled back. + # + # The staged log file handler, if one was opened, is closed + # here too, best-effort: it was never attached to the + # logger, so no record ever reached it and there is + # nothing buffered to lose, but the close itself can still + # raise (a full disk, a stale network mount), and that + # error is logged and swallowed so it cannot skip the two + # restores below it. + # + # Close the new clients *before* restoring the alias, never + # after: a fully built _ElasticsearchHandle owns the new + # client and releases the alias as it closes, so closing + # first leaves the alias unset for the restore to re-register + # the old client into. Restoring first would hand the alias + # back and then close the handle, which would be safe only + # because of how _ElasticsearchHandle.close() happens to + # behave today (see _init_output_clients, and #902). + # + # Per-step end state, reading down phase 1 -- in every case + # `clients` is still the old dict with every old client open, + # REVERSE_DNS_MAP still holds the pre-reload entries (it was + # never handed to the loader), and opts, parser_config and + # the three watch parameters are untouched, because all of + # those are written in phase 2: + # + # _load_config / _parse_config: no state written yet. + # new_clients is still {}, so the close is a no-op; the + # alias and the utils globals still hold what the snapshots + # recorded, so both restores are no-ops too. + # _init_output_clients: rolled itself back before re-raising + # (#906), and never assigned new_clients, so this is the + # same no-op pair -- restoring an unchanged snapshot twice + # is harmless. + # _build_parser_config / the int() conversions: the new + # clients hold the alias, so they are closed and the old + # client is registered again. The utils globals were not + # reached, so their restore is still a no-op. + # The log file open: as above, and the handler was never + # attached to the logger, so it has nothing buffered to + # lose -- the logger still holds the old FileHandler, + # which phase 2 never got to remove. The close itself is + # best-effort (logged and swallowed), so a close that + # fails cannot skip the two restores below it. + # load_reverse_dns_map: as above, plus psl_overrides, which + # load_psl_overrides() cleared and may have refilled from + # the new config; restoring it by value is load-bearing + # here, and the staged map dict is simply dropped. + # load_ip_db: as above. Each of its four assignments to + # _IP_DB_PATH is the last thing on its path -- three are + # followed by a return and the fourth, the bundled + # fallback, by a log line and the end of the function -- + # so a raise from this loader leaves it unchanged. + # configure_ipinfo_api: the last step, so a _IP_DB_PATH or + # _IPINFO_API_TOKEN that differs from the snapshot can + # only be rolled back from here. In practice this may be + # unreachable: the documented failure is + # InvalidIPinfoAPIKey, which exits above without rolling + # back, and _ipinfo_api_lookup() swallows every network + # and decoding error. Those two restores are therefore + # defensive -- a rollback's job is to leave state as it + # found it, and whatever step is added after this one + # makes them load-bearing again. _utils_globals_snapshot's + # own tests are what guard them. + _close_output_clients(new_clients) + if staged_log_handler is not None: + try: + staged_log_handler.close() + except Exception as close_error: + logger.warning( + "Unable to close the log file opened for the " + f"reload: {close_error}" + ) + _restore_search_aliases(previous_search_state) + _restore_utils_globals(previous_utils_state) + logger.exception( + "Config reload failed, continuing with previous config" + ) + else: + # Phase 2 -- all steps succeeded; commit the changes + # atomically. Nothing from here to the end of this block may + # raise, since there is no going back once the first + # statement lands: the commit itself is assignments only, and + # the logging refresh that follows it guards the one call + # that can fail, closing the replaced log file -- the + # replacement was opened back in phase 1. That is also why + # this is an ``else`` and not the tail of the ``try`` -- a + # rollback running over committed state would close the + # clients that are now live. + # In place, never rebound. Rebinding would only move this + # module's own name -- cli.py imports REVERSE_DNS_MAP from + # parsedmarc, so `REVERSE_DNS_MAP = staged` here would + # desync it from parsedmarc.REVERSE_DNS_MAP and from + # config.REVERSE_DNS_MAP, which is the object + # ParserConfig.__setstate__ rebinds to when a worker + # unpickles a config. The configs that name this dict are + # built elsewhere and keep whichever object they were given: + # _build_parser_config() and parsedmarc._resolve_config() + # pass it explicitly (the dataclass field's own default is a + # fresh dict, via default_factory), and new_parser_config + # was built back in phase 1 holding the pre-reload object. + REVERSE_DNS_MAP.clear() + REVERSE_DNS_MAP.update(staged_reverse_dns_map) for k, v in vars(new_opts).items(): setattr(opts, k, v) - parser_config = _build_parser_config(opts) + old_clients = clients + clients = new_clients + index_prefix_domain_map = new_index_prefix_domain_map + parser_config = new_parser_config # Update watch parameters from reloaded config - mailbox_batch_size_value = ( - int(opts.mailbox_batch_size) - if opts.mailbox_batch_size is not None - else 10 - ) - mailbox_check_timeout_value = ( - int(opts.mailbox_check_timeout) - if opts.mailbox_check_timeout is not None - else 30 - ) - mailbox_max_unsaved_retries_value = ( - int(opts.mailbox_max_unsaved_retries) - if opts.mailbox_max_unsaved_retries is not None - else 2 - ) + mailbox_batch_size_value = new_batch_size + mailbox_check_timeout_value = new_check_timeout + mailbox_max_unsaved_retries_value = new_max_unsaved_retries # Update log level logger.setLevel(logging.ERROR) @@ -3714,36 +3981,48 @@ def _main(): if opts.debug: logger.setLevel(logging.DEBUG) - # Refresh FileHandler if log_file changed - old_log_file = getattr(opts, "active_log_file", None) - new_log_file = opts.log_file + # Refresh FileHandler if log_file changed. Both names were + # computed in phase 1, against the configuration that was + # running then. if old_log_file != new_log_file: # Remove old FileHandlers for h in list(logger.handlers): if isinstance(h, logging.FileHandler): - h.close() + close_error = None + try: + h.close() + except Exception as error_: + # Flushing the replaced log file can fail -- + # a full disk, a stale network mount. The + # reloaded configuration is already live by + # this point, so raising would kill the + # watcher over a log file, and rolling back + # to report a failed reload would be a lie. + close_error = error_ + # Detach first, then report: the handler that + # just failed to close is no place to send the + # record describing that failure. logger.removeHandler(h) - # Add new FileHandler if configured - if new_log_file: - try: - fh = logging.FileHandler(new_log_file, "a") - file_formatter = logging.Formatter( - "%(asctime)s - %(levelname)s" - " - [%(filename)s:%(lineno)d] - %(message)s" - ) - fh.setFormatter(file_formatter) - logger.addHandler(fh) - except Exception as log_error: - logger.warning(f"Unable to write to log file: {log_error}") + if close_error is not None: + logger.warning( + f"Unable to close the log file: {close_error}" + ) + # Attach the handler phase 1 opened, if the new + # configuration names a log file at all. + if staged_log_handler is not None: + logger.addHandler(staged_log_handler) opts.active_log_file = new_log_file _configure_dependency_logging(logger.level) + # Phase 3 -- the new configuration is live and correct, so + # the clients it replaced can go. Last, and best effort: + # _close_output_clients() logs close errors and swallows + # them, so a dying connection on the way out cannot undo + # the commit above. + _close_output_clients(old_clients) + logger.info("Configuration reloaded successfully") - except Exception: - logger.exception( - "Config reload failed, continuing with previous config" - ) # Close output clients on the success path (one-shot or graceful # watch-loop exit). atexit-registered above is the safety net for diff --git a/tests/test_cli.py b/tests/test_cli.py index 0317b333..6fb792f2 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -8,6 +8,7 @@ import io import json import logging import os +import shutil import signal import stat import sys @@ -37,6 +38,8 @@ import parsedmarc.elastic import parsedmarc.log import parsedmarc.mail import parsedmarc.opensearch as opensearch_module +import parsedmarc.resources.ipinfo +import parsedmarc.utils from parsedmarc.types import AggregateReport, ParsedReport SAMPLE_AGGREGATE_REPORT_PATH = ( @@ -3898,6 +3901,929 @@ watch = true self.assertEqual(second_config.dns_timeout, 42.0) +class _RaisingCloseClient: + """An output client whose close() fails, e.g. a Kafka producer whose + broker connection is already gone. Unlike the search handles, a plain + client like this lets the error reach _close_output_clients(), which is + where the SIGHUP reload closes the configuration it just replaced.""" + + def __init__(self): + self.close_count = 0 + + def close(self): + self.close_count += 1 + raise RuntimeError("broker connection already gone") + + +class _RaisingCloseFileHandler(logging.FileHandler): + """A log FileHandler whose close() fails, the way flushing to a full + disk or a stale network mount does. Counts the attempts so a test can + tell "close() was tried and failed" from "close() was never called".""" + + def __init__(self, filename): + super().__init__(filename, "a") + self.close_attempts = 0 + + def close(self): + self.close_attempts += 1 + raise OSError("No space left on device") + + +@unittest.skipUnless( + hasattr(signal, "SIGHUP"), + "SIGHUP not available on this platform", +) +class TestSighupReloadAtomicity(unittest.TestCase): + """A SIGHUP reload is all-or-nothing. + + The reload builds the replacement output clients first and only then + re-reads the reverse DNS map, the PSL overrides and the IP database, + re-applies opts, and rebuilds the ParserConfig. Committing the clients + before those steps ran meant a failure in any of them logged "Config + reload failed, continuing with previous config" while the *new* clients + were already live, the old ones were closed and unrecoverable, and the + reverse DNS map had been emptied by the loader that then failed to + refill it. #906 made _init_output_clients() itself all-or-nothing; these + tests cover the same guarantee for the whole reload. + + Each test drives the real _main() through one SIGHUP and reads the state + the *next* watch_inbox() call sees, which is the first point at which the + running configuration is observable from outside. + """ + + def setUp(self): + self._stdout_patch = patch("sys.stdout", new_callable=io.StringIO) + self._stderr_patch = patch("sys.stderr", new_callable=io.StringIO) + self._stdout_patch.start() + self._stderr_patch.start() + self.addCleanup(self._stderr_patch.stop) + self.addCleanup(self._stdout_patch.stop) + + # _main() sets the parsedmarc logger's level and swaps its + # FileHandlers, and _configure_dependency_logging() copies its + # handler list onto every dependency logger -- including the + # temporary handler assertLogs installs. Snapshot all of them here, + # before anything runs, so no test leaves a dead handler behind for + # the next one. + loggers = [parsedmarc.log.logger] + [ + logging.getLogger(name) for name in parsedmarc.cli._DEPENDENCY_LOGGERS + ] + saved_logging = [ + (lg, lg.level, list(lg.handlers), lg.propagate, lg.disabled) + for lg in loggers + ] + + def _restore_logging(): + for lg, level, handlers, propagate, disabled in saved_logging: + for stray in lg.handlers: + # A reload that changes log_file leaves an open + # FileHandler behind; close it rather than waiting for + # the garbage collector to release the descriptor. + if isinstance(stray, logging.FileHandler) and ( + stray not in handlers + ): + stray.close() + lg.setLevel(level) + lg.handlers[:] = handlers + lg.propagate = propagate + lg.disabled = disabled + + self.addCleanup(_restore_logging) + parsedmarc.log.logger.disabled = False + + # _main() installs SIGHUP/SIGTERM/SIGINT handlers that would + # otherwise outlive the test and answer signals raised by later + # ones. Registered before the first _main() call installs them. + for sig in (signal.SIGHUP, signal.SIGTERM, signal.SIGINT): + self.addCleanup(signal.signal, sig, signal.getsignal(sig)) + + # Process-wide state a reload writes. Restored explicitly rather + # than through _utils_globals_snapshot()/_restore_utils_globals(), + # which are themselves under test here. + original_map = dict(parsedmarc.REVERSE_DNS_MAP) + + def _restore_map(): + parsedmarc.REVERSE_DNS_MAP.clear() + parsedmarc.REVERSE_DNS_MAP.update(original_map) + + self.addCleanup(_restore_map) + + saved_overrides = list(parsedmarc.utils.psl_overrides) + saved_ip_db_path = parsedmarc.utils._IP_DB_PATH + saved_ipinfo_token = parsedmarc.utils._IPINFO_API_TOKEN + + def _restore_utils(): + parsedmarc.utils.psl_overrides[:] = saved_overrides + parsedmarc.utils._IP_DB_PATH = saved_ip_db_path + parsedmarc.utils._IPINFO_API_TOKEN = saved_ipinfo_token + + self.addCleanup(_restore_utils) + + # A distinctive map entry, so "the map still holds what it held + # before the reload" is an assertion about this dict and not about + # whatever the bundled CSV happens to contain. + parsedmarc.REVERSE_DNS_MAP.clear() + parsedmarc.REVERSE_DNS_MAP["before.example.com"] = { + "name": "Before Reload", + "type": "test", + } + + def _write_file(self, contents, suffix): + with tempfile.NamedTemporaryFile("w", suffix=suffix, delete=False) as f: + f.write(contents) + path = f.name + self.addCleanup(lambda: os.path.exists(path) and os.remove(path)) + return path + + def _fake_search_clients(self, extra_old_clients=None): + """Stand in for _init_output_clients() against the *real* search + connection registry. + + The real thing calls ``elastic.set_hosts()``, which registers the + client it builds under the process-wide ``default`` alias the moment + it is constructed, and stores it in the returned dict wrapped in an + _ElasticsearchHandle. That registry and that handle are the seam the + reload's rollback is written against, so both are real here and only + the cluster connection is a stand-in. + + Returns: + tuple: the side effect to give the _init_output_clients patch, + the old connection, and the new one. + """ + connections = parsedmarc.elastic.connections + _isolate_default_alias(self, connections) + old_conn = _FakeSearchConnection("old") + new_conn = _FakeSearchConnection("new") + self._search_conns = (old_conn, new_conn) + calls = [] + + def init_clients(opts, index_prefix_domain_map=None): + conn = old_conn if not calls else new_conn + calls.append(conn) + connections.add_connection("default", cast(Elasticsearch, conn)) + clients = { + "elasticsearch": parsedmarc.cli._ElasticsearchHandle(conn), + } + if conn is old_conn and extra_old_clients: + clients.update(extra_old_clients) + return clients + + return init_clients, old_conn, new_conn + + def _current_alias(self): + try: + return parsedmarc.elastic.connections.get_connection("default") + except KeyError: + return None + + def _capture_state(self): + """Everything the reload can write, as the running configuration + sees it. Read at the moment the reload is over rather than after + _main() returns, because the trailing _close_output_clients() would + otherwise blur which clients the reload itself closed.""" + old_conn, new_conn = self._search_conns + return SimpleNamespace( + reverse_dns_map=dict(parsedmarc.REVERSE_DNS_MAP), + psl_overrides=list(parsedmarc.utils.psl_overrides), + ip_db_path=parsedmarc.utils._IP_DB_PATH, + ipinfo_api_token=parsedmarc.utils._IPINFO_API_TOKEN, + alias=self._current_alias(), + old_closed=old_conn.close_count, + new_closed=new_conn.close_count, + # Read here rather than after _main() returns: assertLogs() + # restores the logger's handler list on the way out, so the + # handlers a reload attached or detached are only visible from + # inside the run. + log_handlers=list(parsedmarc.log.logger.handlers), + ) + + def _drive_reload( + self, initial_config, reload_config, init_clients, before_signal=None + ): + """Run _main() through exactly one SIGHUP reload. + + The first watch_inbox() call rewrites the config file the way an + operator would, runs *before_signal* if one was given -- the seam + for anything that has to be set up inside the run, after + assertLogs() has replaced the logger's handlers -- and signals + SIGHUP. The second records the state the reload left behind and + signals SIGTERM, so the watch loop breaks and _main() returns + normally -- which runs its trailing + _close_output_clients(clients). That trailing close is what makes + "``clients`` still names the old dict" observable: the clients it + closes are, by definition, the ones the run ended with. + + Returns: + tuple: the captured state, the watch_inbox mock, and the + captured log output. + """ + cfg_path = self._write_file(initial_config, ".ini") + captured = {} + watch_calls = [] + + def watch_side_effect(*args, **kwargs): + watch_calls.append(kwargs) + if len(watch_calls) == 1: + with open(cfg_path, "w") as f: + f.write(reload_config) + if before_signal is not None: + before_signal() + os.kill(os.getpid(), signal.SIGHUP) + return + captured["state"] = self._capture_state() + os.kill(os.getpid(), signal.SIGTERM) + + with ( + patch("parsedmarc.cli.IMAPConnection", return_value=object()), + patch( + "parsedmarc.cli.get_dmarc_reports_from_mailbox", + return_value={ + "aggregate_reports": [], + "failure_reports": [], + "smtp_tls_reports": [], + }, + ), + patch( + "parsedmarc.cli.watch_inbox", side_effect=watch_side_effect + ) as mock_watch, + patch("parsedmarc.cli._init_output_clients", side_effect=init_clients), + patch.object(sys, "argv", ["parsedmarc", "-c", cfg_path]), + self.assertLogs(parsedmarc.log.logger, level="INFO") as logs, + ): + parsedmarc.cli._main() + + self.assertEqual(mock_watch.call_count, 2) + self.assertIn("state", captured) + return captured["state"], mock_watch, logs.output + + _CONFIG_TEMPLATE = """[general] +debug = true +offline = true +dns_timeout = {dns_timeout} +{general_extra} +[imap] +host = imap.example.com +user = user +password = pass + +[mailbox] +watch = true +batch_size = {batch_size} +""" + + def _initial_config(self, general_extra=""): + """The configuration the run starts with: batch_size 3 and a 5 + second DNS timeout.""" + return self._CONFIG_TEMPLATE.format( + dns_timeout="5.0", general_extra=general_extra, batch_size=3 + ) + + def _reload_config(self, general_extra=""): + """What the operator edits the file to before signalling: every + value here differs from the initial config, so committing half of + them would show.""" + return self._CONFIG_TEMPLATE.format( + dns_timeout="42.0", general_extra=general_extra, batch_size=9 + ) + + def _assert_previous_config_still_running(self, state, mock_watch, logs): + """The claim "continuing with previous config" makes, item by + item.""" + self.assertTrue( + any("continuing with previous config" in line for line in logs), + logs, + ) + self.assertEqual( + state.reverse_dns_map, + {"before.example.com": {"name": "Before Reload", "type": "test"}}, + ) + second_call = mock_watch.call_args_list[1].kwargs + self.assertEqual(second_call["batch_size"], 3) + self.assertEqual(second_call["config"].dns_timeout, 5.0) + + def testFailedReloadKeepsOldClientsWhenMapFileIsMissing(self): + """A reverse DNS map path that does not exist fails the reload after + the replacement clients were built: the new clients must be closed + and the old ones left running and still registered under the + ``default`` alias. + + open() on the missing file is the injected failure -- the loader + catches httpx and CSV errors but not a missing local file, which is + exactly what a typo in ``reverse_dns_map_path`` produces. + """ + init_clients, old_conn, new_conn = self._fake_search_clients() + + state, mock_watch, logs = self._drive_reload( + self._initial_config(), + self._reload_config( + general_extra="local_reverse_dns_map_path = /nonexistent/map.csv\n" + ), + init_clients, + ) + + # The new clients were built, then closed and stripped of the alias. + self.assertEqual(state.new_closed, 1) + # The old ones are untouched and still hold the alias, so every save + # between now and the next reload reaches the cluster opts describes. + self.assertEqual(state.old_closed, 0) + self.assertIs(state.alias, old_conn) + self._assert_previous_config_still_running(state, mock_watch, logs) + # _main()'s trailing close ran against the old clients, which is + # only possible if `clients` still named them -- and it did not + # close the discarded clients a second time. + self.assertEqual(old_conn.close_count, 1) + self.assertEqual(new_conn.close_count, 1) + + def testFailedReloadRestoresPslOverrides(self): + """load_psl_overrides() empties the module-level list before it + reads anything, so a missing overrides file leaves it empty -- and + load_reverse_dns_map() calls it first thing, which is why staging + the map into a fresh dict is not enough on its own. + + The rollback must put the list back by value, or every later + get_base_domain() call would fold domains without the overrides. + """ + init_clients, old_conn, new_conn = self._fake_search_clients() + overrides_before = list(parsedmarc.utils.psl_overrides) + self.assertGreater(len(overrides_before), 0) + + state, mock_watch, logs = self._drive_reload( + self._initial_config(), + self._reload_config( + general_extra="local_psl_overrides_path = /nonexistent/psl_overrides.txt\n" + ), + init_clients, + ) + + self.assertEqual(state.psl_overrides, overrides_before) + self.assertEqual(state.new_closed, 1) + self.assertEqual(state.old_closed, 0) + self.assertIs(state.alias, old_conn) + self._assert_previous_config_still_running(state, mock_watch, logs) + + def testFailedReloadKeepsOldClientsWhenIpDatabaseIsUnavailable(self): + """load_ip_db() falls back to the bundled MMDB when the configured + file, the download and the cache are all unavailable; an install + without that bundled resource makes the fallback itself raise. + + This is the deepest point a reload can fail and still be rolled + back: only configure_ipinfo_api() runs after it, and its documented + failure (an invalid token) exits the process rather than returning + to the watch loop. By the time it fails, the replacement clients + hold the ``default`` alias, the reverse DNS map has been staged from + a different file, and load_psl_overrides() has already replaced the + live overrides list -- so this is also where the most restores are + load-bearing at once. + """ + init_clients, old_conn, new_conn = self._fake_search_clients() + overrides_before = list(parsedmarc.utils.psl_overrides) + self.assertNotIn("reload-only.example", overrides_before) + # Files the reload names and the run must end up not using. + reload_map_path = self._write_file( + "base_reverse_dns,name,type\nafter.example.com,After Reload,test\n", + ".csv", + ) + reload_overrides_path = self._write_file("reload-only.example\n", ".txt") + + # The IP database the run starts with, and keeps. + ip_db_path = self._write_file("not really an mmdb", ".mmdb") + # An empty cache directory, so load_ip_db() cannot answer from a + # copy some earlier run downloaded and must reach the bundled file. + cache_dir = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, cache_dir, True) + + real_files = parsedmarc.utils.files + + def files_without_bundled_mmdb(package): + # The importlib.resources boundary: the maps package (the + # reverse DNS map and the PSL overrides, both loaded earlier in + # the same reload) still resolves normally. + if package is parsedmarc.resources.ipinfo: + raise ModuleNotFoundError("parsedmarc.resources.ipinfo") + return real_files(package) + + with ( + patch("parsedmarc.utils.files", side_effect=files_without_bundled_mmdb), + patch("parsedmarc.utils.tempfile.gettempdir", return_value=cache_dir), + ): + state, mock_watch, logs = self._drive_reload( + # Startup resolves the local file directly and never reaches + # the bundled fallback, so only the reload -- which drops + # ip_db_path -- fails. + self._initial_config(f"ip_db_path = {ip_db_path}\n"), + self._reload_config( + general_extra=( + f"local_reverse_dns_map_path = {reload_map_path}\n" + f"local_psl_overrides_path = {reload_overrides_path}\n" + ) + ), + init_clients, + ) + + # The two restores that carry weight at this depth: the overrides + # list load_psl_overrides() had already replaced, and the map, which + # was staged into a dict that is simply dropped -- the live one + # still holds the pre-reload entry rather than the reload's file. + self.assertEqual(state.psl_overrides, overrides_before) + self.assertNotIn("after.example.com", state.reverse_dns_map) + # load_ip_db() assigns _IP_DB_PATH only as the last thing on each + # of its paths, so a raise from it leaves the previous selection in + # place rather than a half-applied one; that the snapshot can put a + # *changed* value back is covered by TestUtilsGlobalsSnapshot. + self.assertEqual(state.ip_db_path, ip_db_path) + self.assertEqual(state.new_closed, 1) + self.assertEqual(state.old_closed, 0) + self.assertIs(state.alias, old_conn) + self._assert_previous_config_still_running(state, mock_watch, logs) + + def testSuccessfulReloadCommitsEveryStagedValue(self): + """The other half: when every step succeeds, the staged state is + what the run continues with -- new clients live and holding the + alias, old clients closed, the reverse DNS map replaced by the file + the new config names, and the new watch parameters and ParserConfig + in use. + + (Passes against the pre-fix code too, necessarily: a reload with + nothing to roll back has to end in the same place either way. It is + here so that the failure tests above cannot be satisfied by a + reload that quietly stopped applying anything.)""" + init_clients, old_conn, new_conn = self._fake_search_clients() + map_path = self._write_file( + "base_reverse_dns,name,type\nafter.example.com,After Reload,test\n", + ".csv", + ) + + state, mock_watch, logs = self._drive_reload( + self._initial_config(), + self._reload_config( + general_extra=f"local_reverse_dns_map_path = {map_path}\n" + ), + init_clients, + ) + + self.assertTrue( + any("Configuration reloaded successfully" in line for line in logs), + logs, + ) + self.assertEqual( + state.reverse_dns_map, + {"after.example.com": {"name": "After Reload", "type": "test"}}, + ) + self.assertIs(state.alias, new_conn) + self.assertEqual(state.old_closed, 1) + self.assertEqual(state.new_closed, 0) + second_call = mock_watch.call_args_list[1].kwargs + self.assertEqual(second_call["batch_size"], 9) + self.assertEqual(second_call["config"].dns_timeout, 42.0) + # The ParserConfig keeps the live map object, so the in-place + # refresh above has to reach the parser through it. + self.assertIs(second_call["config"].reverse_dns_map, parsedmarc.REVERSE_DNS_MAP) + # _main()'s trailing close ran against the new clients. + self.assertEqual(new_conn.close_count, 1) + + def testOldClientCloseFailureDoesNotUndoTheReload(self): + """Closing the replaced clients is the last step and is best effort: + a client whose close() raises is logged and the reloaded + configuration stays live. + + (This one passes against the pre-fix code too -- there the same + close ran before the swap, and _close_output_clients() swallowed the + error either way. It guards the move: closing the old clients last + must not put the commit at risk.) + """ + raising_client = _RaisingCloseClient() + init_clients, old_conn, new_conn = self._fake_search_clients( + extra_old_clients={"kafka": raising_client} + ) + map_path = self._write_file( + "base_reverse_dns,name,type\nafter.example.com,After Reload,test\n", + ".csv", + ) + + state, mock_watch, logs = self._drive_reload( + self._initial_config(), + self._reload_config( + general_extra=f"local_reverse_dns_map_path = {map_path}\n" + ), + init_clients, + ) + + self.assertEqual(raising_client.close_count, 1) + self.assertTrue( + any("Error closing kafka" in line for line in logs), + logs, + ) + self.assertTrue( + any("Configuration reloaded successfully" in line for line in logs), + logs, + ) + self.assertIs(state.alias, new_conn) + self.assertEqual(state.new_closed, 0) + self.assertEqual(state.old_closed, 1) + self.assertEqual( + state.reverse_dns_map, + {"after.example.com": {"name": "After Reload", "type": "test"}}, + ) + self.assertEqual(mock_watch.call_args_list[1].kwargs["batch_size"], 9) + + def testLogFileCloseFailureDoesNotUndoTheReload(self): + """Swapping the log file is the last thing the commit does, and + closing the replaced FileHandler can fail on a full disk or a stale + mount. Opening the replacement happens back in phase 1, so the close + is the only statement left in the commit that can raise, and it must + not take the reload -- which is already live by then -- with it.""" + init_clients, old_conn, new_conn = self._fake_search_clients() + map_path = self._write_file( + "base_reverse_dns,name,type\nafter.example.com,After Reload,test\n", + ".csv", + ) + log_path = self._write_file("", ".log") + stale_handler = _RaisingCloseFileHandler(self._write_file("", ".log")) + # The subclass's close() raises; the unbound FileHandler.close() + # is what actually releases the descriptor. + self.addCleanup(logging.FileHandler.close, stale_handler) + + state, mock_watch, logs = self._drive_reload( + self._initial_config(), + self._reload_config( + general_extra=( + f"local_reverse_dns_map_path = {map_path}\nlog_file = {log_path}\n" + ) + ), + init_clients, + # Inside the run: assertLogs() replaces the logger's handler + # list on the way in, so a handler attached before _main() + # started would not be there for the reload to close. + before_signal=lambda: parsedmarc.log.logger.addHandler(stale_handler), + ) + + self.assertEqual(stale_handler.close_attempts, 1) + self.assertTrue( + any("Unable to close the log file" in line for line in logs), logs + ) + self.assertNotIn(stale_handler, state.log_handlers) + # The commit stands: new clients live and holding the alias, old + # ones closed, map replaced, new watch parameters in use. + self.assertTrue( + any("Configuration reloaded successfully" in line for line in logs), logs + ) + self.assertIs(state.alias, new_conn) + self.assertEqual(state.new_closed, 0) + self.assertEqual(state.old_closed, 1) + self.assertEqual( + state.reverse_dns_map, + {"after.example.com": {"name": "After Reload", "type": "test"}}, + ) + self.assertEqual(mock_watch.call_args_list[1].kwargs["batch_size"], 9) + + def testUnwritableLogFileAbortsTheReload(self): + """The replacement log file is opened in phase 1, with the previous + FileHandler still attached, so a ``log_file`` that cannot be opened + is a reload failure rather than a warning logged after the old + handler is already gone. + + Everything else in this reload would have succeeded -- it names a + readable map file and a new batch size -- so the previous + configuration still running afterwards is the log file open's doing. + The old handler is still attached and still open, which is what + makes the "Config reload failed" record reach the log file the + operator is tailing, and no second FileHandler was left behind by + the open that failed. + """ + init_clients, old_conn, new_conn = self._fake_search_clients() + map_path = self._write_file( + "base_reverse_dns,name,type\nafter.example.com,After Reload,test\n", + ".csv", + ) + # /proc has no subdirectories that can be created, so opening a file + # under a nonexistent one fails the way a bad ``log_file`` path + # does. Same trick as + # test_unwritable_log_file_logs_warning_does_not_raise. + unwritable_log = "/proc/nonexistent/parsedmarc-reload.log" + live_handler = logging.FileHandler(self._write_file("", ".log"), "a") + self.addCleanup(live_handler.close) + self.addCleanup(parsedmarc.log.logger.removeHandler, live_handler) + + state, mock_watch, logs = self._drive_reload( + self._initial_config(), + self._reload_config( + general_extra=( + f"local_reverse_dns_map_path = {map_path}\n" + f"log_file = {unwritable_log}\n" + ) + ), + init_clients, + # Inside the run, for the same reason as the close-failure test: + # assertLogs() replaces the logger's handler list on the way in. + before_signal=lambda: parsedmarc.log.logger.addHandler(live_handler), + ) + + self.assertTrue( + any( + "Config reload failed, continuing with previous config" in line + for line in logs + ), + logs, + ) + self.assertFalse( + any("Configuration reloaded successfully" in line for line in logs), + logs, + ) + # The log file the operator is tailing is still attached and still + # open, so the failure was reported into it -- and it is the only + # FileHandler on the logger, so nothing leaked from the failed open. + self.assertEqual( + [h for h in state.log_handlers if isinstance(h, logging.FileHandler)], + [live_handler], + ) + self.assertTrue( + live_handler.stream is not None and not live_handler.stream.closed, + "the previous log file handler was closed", + ) + # The rest of the reload was rolled back: the replacement clients + # are closed, the old ones still hold the alias, the map still holds + # its pre-reload entry, and the watch parameters are the originals. + self.assertEqual(state.new_closed, 1) + self.assertEqual(state.old_closed, 0) + self.assertIs(state.alias, old_conn) + self.assertNotIn("after.example.com", state.reverse_dns_map) + self._assert_previous_config_still_running(state, mock_watch, logs) + + def testFailedReloadClosesTheStagedLogFileHandler(self): + """A reload that opens the replacement log file and then fails on a + later step has to close the file it opened: the handler was never + attached to the logger, so closing it is the whole undo, and leaving + it open would leak a descriptor on every such reload. + + The missing map path is the injected failure, as in + testFailedReloadKeepsOldClientsWhenMapFileIsMissing; this reload also + names a writable log file, so the staged handler exists by the time + that failure lands. Capturing it needs a real FileHandler subclass + rather than a mock, because the code the reload runs afterwards + checks handlers with isinstance(). + """ + init_clients, old_conn, new_conn = self._fake_search_clients() + log_path = self._write_file("", ".log") + opened: list[logging.FileHandler] = [] + + class _CapturingFileHandler(logging.FileHandler): + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + opened.append(self) + + with patch("logging.FileHandler", _CapturingFileHandler): + state, mock_watch, logs = self._drive_reload( + self._initial_config(), + self._reload_config( + general_extra=( + "local_reverse_dns_map_path = /nonexistent/map.csv\n" + f"log_file = {log_path}\n" + ) + ), + init_clients, + ) + + # The initial config names no log file, so the staging step is the + # only thing in the run that opens one. + self.assertEqual(len(opened), 1) + staged = opened[0] + self.addCleanup(staged.close) + self.assertEqual(staged.baseFilename, os.path.abspath(log_path)) + # FileHandler.close() drops the stream it owns, so a None stream is + # the rollback having closed this handler. + self.assertIsNone(staged.stream) + # It was never attached, so there was nothing else to undo -- and + # the logger is still writing wherever it was before the SIGHUP. + self.assertNotIn(staged, state.log_handlers) + self.assertEqual(state.new_closed, 1) + self.assertEqual(state.old_closed, 0) + self.assertIs(state.alias, old_conn) + self._assert_previous_config_still_running(state, mock_watch, logs) + + def testFailedReloadSurvivesAStagedLogFileHandlerThatCannotClose(self): + """A rollback whose staged-handler close fails must still restore + the search alias and finish as "continuing with previous config", + instead of the close error escaping the handler and taking the + watcher down with the alias left unrestored. + + Same missing-map injected failure as + testFailedReloadClosesTheStagedLogFileHandler, but the staged + handler here is a _RaisingCloseFileHandler, so the rollback's own + ``staged_log_handler.close()`` call raises. Before the fix, that + raise propagated out of the ``except Exception:`` block, so + _restore_search_aliases() and _restore_utils_globals() below it + never ran and the exception itself replaced the "Config reload + failed" log message. + """ + init_clients, old_conn, new_conn = self._fake_search_clients() + log_path = self._write_file("", ".log") + opened: list[_RaisingCloseFileHandler] = [] + + class _CapturingRaisingCloseFileHandler(_RaisingCloseFileHandler): + # cli.py constructs the handler as FileHandler(path, "a"), and + # _RaisingCloseFileHandler always opens in append mode, so the + # mode argument only needs accepting. + def __init__(self, filename, mode="a"): + super().__init__(filename) + opened.append(self) + + with patch("logging.FileHandler", _CapturingRaisingCloseFileHandler): + state, mock_watch, logs = self._drive_reload( + self._initial_config(), + self._reload_config( + general_extra=( + "local_reverse_dns_map_path = /nonexistent/map.csv\n" + f"log_file = {log_path}\n" + ) + ), + init_clients, + ) + + # The initial config names no log file, so the staging step is the + # only thing in the run that opens one. + self.assertEqual(len(opened), 1) + self.addCleanup(logging.FileHandler.close, opened[0]) + self.assertEqual(opened[0].close_attempts, 1) + self.assertTrue( + any( + "Unable to close the log file opened for the reload" in line + for line in logs + ), + logs, + ) + # The close error was logged and swallowed rather than escaping, so + # the rollback still finished and reported itself normally. + self._assert_previous_config_still_running(state, mock_watch, logs) + # ... and the two restores that come after the close in source order + # still ran: the new clients are closed, the old ones still hold the + # alias, and the staged handler was never attached to the logger. + self.assertEqual(state.new_closed, 1) + self.assertEqual(state.old_closed, 0) + self.assertIs(state.alias, old_conn) + self.assertNotIn(opened[0], state.log_handlers) + + def testLogFileUnopenableAtStartupIsRetriedOnReload(self): + """A ``log_file`` that could not be opened when parsedmarc started is + only a warning there -- startup has no previous configuration to + keep, so it logs "Unable to write to log file" and carries on. The + next reload has to retry that open even though the operator never + changed the setting, once the directory that was missing exists. + + Before this fix, startup recorded the unopened path onto + ``opts.active_log_file`` anyway, so the reload's + ``old_log_file != new_log_file`` comparison saw the same path on + both sides and treated the log file as nothing to refresh -- file + logging stayed dead until the process was restarted, even after the + directory was created. Recording ``None`` when the open fails makes + the comparison true, so the reload opens it for real. + """ + tmpdir = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, tmpdir, True) + log_path = os.path.join(tmpdir, "later", "parsedmarc.log") + + init_clients, old_conn, new_conn = self._fake_search_clients() + + state, mock_watch, logs = self._drive_reload( + self._initial_config(general_extra=f"log_file = {log_path}\n"), + self._reload_config(general_extra=f"log_file = {log_path}\n"), + init_clients, + # The operator fixes the missing directory before the SIGHUP + # that _drive_reload sends right after this runs. + before_signal=lambda: os.makedirs(os.path.dirname(log_path)), + ) + + self.assertTrue( + any("Unable to write to log file" in line for line in logs), logs + ) + self.assertTrue( + any("Configuration reloaded successfully" in line for line in logs), + logs, + ) + file_handlers = [ + h for h in state.log_handlers if isinstance(h, logging.FileHandler) + ] + self.assertEqual(len(file_handlers), 1) + handler = file_handlers[0] + # assertLogs() already restored the logger's own handler list by + # the time this runs, so this handler is not attached to anything + # -- closing it is enough, with no removeHandler needed alongside + # it (contrast testUnwritableLogFileAbortsTheReload's live_handler, + # which stays attached and needs both). + self.addCleanup(handler.close) + self.assertEqual(handler.baseFilename, os.path.abspath(log_path)) + self.assertEqual(state.old_closed, 1) + self.assertEqual(state.new_closed, 0) + self.assertIs(state.alias, new_conn) + self.assertTrue(os.path.exists(log_path)) + + def testLogFileOpenedAtStartupIsNotReopenedByAnUnchangedReload(self): + """The other half of testLogFileUnopenableAtStartupIsRetriedOnReload: + a ``log_file`` startup *can* open gets recorded as the active one, so + a reload that leaves the setting unchanged does not reopen it. + + This is what guards the ``opts.active_log_file = opts.log_file`` + assignment moving inside the ``try`` -- if startup stopped recording + a successful open, ``old_log_file`` would stay ``None`` and every + reload would treat an unchanged, already-open ``log_file`` as a + replacement to stage, opening it a second time for no reason. + """ + log_path = self._write_file("", ".log") + opened: list[logging.FileHandler] = [] + + class _CapturingFileHandler(logging.FileHandler): + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + opened.append(self) + + init_clients, old_conn, new_conn = self._fake_search_clients() + + with patch("logging.FileHandler", _CapturingFileHandler): + state, mock_watch, logs = self._drive_reload( + self._initial_config(general_extra=f"log_file = {log_path}\n"), + self._reload_config(general_extra=f"log_file = {log_path}\n"), + init_clients, + ) + + self.assertFalse( + any("Unable to write to log file" in line for line in logs), logs + ) + self.assertTrue( + any("Configuration reloaded successfully" in line for line in logs), + logs, + ) + # Startup opened the file once; the reload, seeing the same path as + # its own active_log_file, did not stage a second FileHandler. + self.assertEqual(len(opened), 1) + file_handlers = [ + h for h in state.log_handlers if isinstance(h, logging.FileHandler) + ] + self.assertEqual(len(file_handlers), 1) + self.assertIs(file_handlers[0], opened[0]) + self.assertEqual(opened[0].baseFilename, os.path.abspath(log_path)) + # assertLogs() already restored the logger's own handler list by the + # time this runs, so this handler is not attached to anything -- + # closing it is enough, with no removeHandler needed alongside it. + self.addCleanup(opened[0].close) + self.assertEqual(state.old_closed, 1) + self.assertEqual(state.new_closed, 0) + self.assertIs(state.alias, new_conn) + + +class TestUtilsGlobalsSnapshot(unittest.TestCase): + """_utils_globals_snapshot() / _restore_utils_globals() cover every + module global the config loaders the SIGHUP reload calls assign: + ``psl_overrides`` (load_psl_overrides), ``_IP_DB_PATH`` (load_ip_db) and + ``_IPINFO_API_TOKEN`` (configure_ipinfo_api).""" + + def setUp(self): + saved_overrides = list(parsedmarc.utils.psl_overrides) + saved_ip_db_path = parsedmarc.utils._IP_DB_PATH + saved_token = parsedmarc.utils._IPINFO_API_TOKEN + + def _restore(): + parsedmarc.utils.psl_overrides[:] = saved_overrides + parsedmarc.utils._IP_DB_PATH = saved_ip_db_path + parsedmarc.utils._IPINFO_API_TOKEN = saved_token + + self.addCleanup(_restore) + + def testRestoresEveryLoaderGlobalAfterTheLoadersRanAgain(self): + parsedmarc.utils.psl_overrides[:] = ["example.co.uk"] + parsedmarc.utils._IP_DB_PATH = "/before/ipinfo_lite.mmdb" + parsedmarc.utils._IPINFO_API_TOKEN = "before-token" + overrides_object = parsedmarc.utils.psl_overrides + + snapshot = parsedmarc.cli._utils_globals_snapshot() + + # What the loaders do: the overrides list is cleared and refilled in + # place, and the other two are rebound. + parsedmarc.utils.psl_overrides.clear() + parsedmarc.utils.psl_overrides.append("example.com.au") + parsedmarc.utils._IP_DB_PATH = "/after/ipinfo_lite.mmdb" + parsedmarc.utils._IPINFO_API_TOKEN = "after-token" + + parsedmarc.cli._restore_utils_globals(snapshot) + + self.assertEqual(parsedmarc.utils.psl_overrides, ["example.co.uk"]) + self.assertEqual(parsedmarc.utils._IP_DB_PATH, "/before/ipinfo_lite.mmdb") + self.assertEqual(parsedmarc.utils._IPINFO_API_TOKEN, "before-token") + # In place, not rebound: get_base_domain() reads the list through + # the module global, but the snapshot must not hand back a + # different object to anything holding a reference to this one. + self.assertIs(parsedmarc.utils.psl_overrides, overrides_object) + + def testSnapshotIsNotAliasedToTheLiveOverridesList(self): + """The overrides list is snapshotted by value: load_psl_overrides() + mutates the live object, so a snapshot that aliased it would record + the failed reload's contents instead of the previous ones.""" + parsedmarc.utils.psl_overrides[:] = ["example.co.uk"] + + snapshot = parsedmarc.cli._utils_globals_snapshot() + parsedmarc.utils.psl_overrides.clear() + + self.assertEqual(snapshot["psl_overrides"], ["example.co.uk"]) + + class TestSigtermShutdown(unittest.TestCase): """Tests for graceful SIGTERM/SIGINT shutdown."""