diff options
Diffstat (limited to 'prometheus_borgmatic_exporter')
| -rw-r--r-- | prometheus_borgmatic_exporter/__init__.py | 134 | ||||
| -rw-r--r-- | prometheus_borgmatic_exporter/cli.py | 143 | ||||
| -rw-r--r-- | prometheus_borgmatic_exporter/types.py | 2 |
3 files changed, 154 insertions, 125 deletions
diff --git a/prometheus_borgmatic_exporter/__init__.py b/prometheus_borgmatic_exporter/__init__.py index 2cfcce5..8bfd629 100644 --- a/prometheus_borgmatic_exporter/__init__.py +++ b/prometheus_borgmatic_exporter/__init__.py @@ -89,14 +89,11 @@ def run_command( Logs the command at debug level before execution. Exits the application with the command's return code if it fails. - Args: - ctx: The Click context, injected by @click.pass_context. - command: The command and its arguments as a list of strings. - - Returns: - The completed process result with stdout and stderr captured. + :param ctx: Click context injected by :func:`click.pass_context`. + :param command: Command and its arguments. + :return: Completed process with stdout and stderr captured. """ - debug("Running command", " ".join(command)) + debug("Running command", " ".join(command), err=True) try: return subprocess.run( command, capture_output=True, encoding="utf-8", check=True @@ -112,15 +109,12 @@ def run_command( def load_json(ctx: click.Context, raw: str) -> Any: """Parse a JSON string and return the result. - Exits the application with code 1 if the input is not valid JSON, printing - the parse error via the error() helper. - - Args: - ctx: The Click context, injected by @click.pass_context. - raw: The raw JSON string to parse. + Exits the application with status code 1 if the input is not valid JSON and + reports the parsing error using :func:`error`. - Returns: - The parsed JSON value. + :param ctx: Click context injected by :func:`click.pass_context`. + :param raw: Raw JSON string to parse. + :return: Parsed JSON value. """ try: return json.loads(raw) @@ -132,15 +126,11 @@ def load_json(ctx: click.Context, raw: str) -> Any: def extract_archive_counts( repositories: list[types.BorgInfoReturnType] | list[types.BorgListReturnType], ) -> list[types.ArchiveCount]: - """Extract archive counts and repository metadata from borgmatic info output. + """Extract archive counts and repository metadata from borgmatic output. - Args: - repositories: List of repository objects as returned by borgmatic info - or borgmatic list. - - Returns: - A list of ArchiveCount dicts, one per repository, containing the - repository id, label, location, and total number of archives. + :param repositories: Repository objects returned by ``borgmatic info`` or + ``borgmatic list``. + :return: Archive counts and identifying metadata for each repository. """ return [ { @@ -158,29 +148,24 @@ def fetch_repositories( ) -> tuple[list[types.BorgInfoReturnType], list[types.ArchiveCount]]: """Fetch repository and archive data from borgmatic. - When all_archives is False, runs `borgmatic info --archive latest` and - `borgmatic list` in complementary intent: info provides per-repository - stats and the latest archive metrics, while list provides the total archive - count — since info with --archive latest only returns a single archive, - making len() unreliable for counting. - - When all_archives is True, only `borgmatic info` is run. The full archive - list is available directly from the info output, so no separate list call - is needed and archive counts are derived from it. + When ``all_archives`` is false, runs ``borgmatic info --archive latest`` + and ``borgmatic list``. The former provides per-repository statistics and + metrics for the latest archive, while the latter provides the complete + archive list needed to determine archive counts. - Args: - borgmatic_bin: Path to the borgmatic executable. - borgmatic_config: Path to the borgmatic configuration file. - all_archives: If True, fetch and return metrics for all archives. - This is significantly slower for large repositories. + When ``all_archives`` is true, only ``borgmatic info`` is run. Its output + already contains the complete archive list, so archive counts can be + derived directly from it. - Returns: - A tuple of: - - repositories: Raw borgmatic info output, one entry per repository. - - archive_counts: Archive count and identifying metadata per repository, - always populated regardless of the all_archives flag. + :param borgmatic_bin: Path to the borgmatic executable. + :param borgmatic_configs: Paths to borgmatic configuration files or + directories. + :param all_archives: Whether to fetch metrics for all archives instead of + only the latest archive. + :return: Raw ``borgmatic info`` output and archive counts with identifying + metadata for each repository. """ - verbose("Fetching repositories") + verbose("Fetching repositories", err=True) borgmatic_cmd = [str(borgmatic_bin)] for config in borgmatic_configs: @@ -193,7 +178,7 @@ def fetch_repositories( repositories = cast( list[types.BorgInfoReturnType], load_json(borgmatic_info.stdout) ) - debug(f"Fetched {len(repositories)} repositories") + debug(f"Fetched {len(repositories)} repositories", err=True) return repositories, extract_archive_counts(repositories) info_command += ["--archive", "latest"] @@ -201,7 +186,7 @@ def fetch_repositories( repositories = cast( list[types.BorgInfoReturnType], load_json(borgmatic_info.stdout) ) - debug(f"Fetched {len(repositories)} repositories") + debug(f"Fetched {len(repositories)} repositories", err=True) borgmatic_list = run_command([*borgmatic_cmd, "list", "--json"]) archives = cast(list[types.BorgListReturnType], load_json(borgmatic_list.stdout)) @@ -211,16 +196,14 @@ def fetch_repositories( def collect_repository_metrics( repository: types.BorgInfoReturnType, all_archives: bool ) -> None: - """Collect and set all Prometheus metrics for a single repository. + """Collect Prometheus metrics for a repository. - Sets cache stats, encryption info, last modified timestamp, and latest - archive metrics. When all_archives is True, also collects metrics for every - individual archive by calling _collect_archive_metrics. + Sets cache statistics, encryption information, the last-modified timestamp, + and metrics for the latest archive. When ``all_archives`` is true, also + collects metrics for every individual archive. - Args: - repository: Repository info as returned by borgmatic. - all_archives: If True, collect metrics for all archives, not just - the latest. + :param repository: Repository information returned by borgmatic. + :param all_archives: Whether to collect metrics for every archive. """ repository_metadata = repository["repository"] @@ -229,6 +212,7 @@ def collect_repository_metrics( repository_metadata["label"], f"({repository_metadata['id']})", repository_metadata["location"], + err=True, ) repository_labels: types.RepositoryLabels = { @@ -260,11 +244,15 @@ def collect_repository_metrics( metric.labels(**repository_labels).set(repository_stats[key]) if not repository["archives"]: - warn("No archives found for repository:", repository_metadata["label"]) + warn( + "No archives found for repository:", repository_metadata["label"], err=True + ) return latest_archive = max(repository["archives"], key=lambda archive: archive["start"]) - debug(f"Latest archive {latest_archive['name']}:", latest_archive["start"]) + debug( + f"Latest archive {latest_archive['name']}:", latest_archive["start"], err=True + ) metrics.LATEST_ARCHIVE_DURATION.labels(**repository_labels).set( latest_archive["duration"] @@ -287,14 +275,12 @@ def collect_repository_metrics( def collect_archive_metrics( archive: types.BorgArchive, repository_labels: types.RepositoryLabels ) -> None: - """Collect and set Prometheus metrics for a single archive. + """Collect Prometheus metrics for an archive. - Args: - archive: Archive info as returned by borgmatic. - repository_labels: The parent repository's label dict, used to - populate the shared label dimensions on each metric. + :param archive: Archive information returned by borgmatic. + :param repository_labels: Labels inherited from the parent repository. """ - msg("Processing archive:", archive["name"], f"({archive['id']})") + msg("Processing archive:", archive["name"], f"({archive['id']})", err=True) archive_labels = { **repository_labels, @@ -406,15 +392,12 @@ def prometheus_borgmatic_exporter( """Collect borgmatic repository metrics and export them for Prometheus. Queries borgmatic for repository and archive statistics and writes them to - a .prom file for consumption by the Prometheus node exporter textfile + a ``.prom`` file for consumption by the Prometheus node exporter textfile collector. - By default only the latest archive is queried per repository. Use - --all-archives to collect metrics for every archive, at the cost of a - significantly longer runtime. - - Using --all-archives will create one time series per archive and may - significantly increase Prometheus storage usage. + By default, only the latest archive is queried for each repository. Use + ``--all-archives`` to collect metrics for every archive, at the cost of a + significantly longer runtime and increased Prometheus storage usage. """ ctx.ensure_object(dict) @@ -422,17 +405,18 @@ def prometheus_borgmatic_exporter( ctx.obj["quiet"] = quiet_flag ctx.obj["verbose"] = verbose_flag - debug("Effective configuration:") - debug(" borgmatic_bin:", str(borgmatic_bin)) + debug("Effective configuration:", err=True) + debug(" borgmatic_bin:", str(borgmatic_bin), err=True) debug( " borgmatic_configs:", ", ".join(str(config) for config in borgmatic_configs) or "<borgmatic defaults>", + err=True, ) - debug(" textfile_collector_dir:", str(textfile_collector_dir)) - debug(" all_archives:", str(all_archives_flag)) - debug(" verbose:", str(verbose_flag)) - debug(" quiet:", str(quiet_flag)) + debug(" textfile_collector_dir:", str(textfile_collector_dir), err=True) + debug(" all_archives:", str(all_archives_flag), err=True) + debug(" verbose:", str(verbose_flag), err=True) + debug(" quiet:", str(quiet_flag), err=True) repositories, archive_counts = fetch_repositories( borgmatic_bin, borgmatic_configs, all_archives_flag @@ -450,7 +434,7 @@ def prometheus_borgmatic_exporter( metrics_file = textfile_collector_dir / "borgmatic.prom" write_to_textfile(str(metrics_file.absolute()), metrics.REGISTRY) - msg("Metrics written to:", str(metrics_file.absolute())) + msg("Metrics written to:", str(metrics_file.absolute()), err=True) if __name__ == "__main__": diff --git a/prometheus_borgmatic_exporter/cli.py b/prometheus_borgmatic_exporter/cli.py index cc6f374..5d91b05 100644 --- a/prometheus_borgmatic_exporter/cli.py +++ b/prometheus_borgmatic_exporter/cli.py @@ -2,6 +2,12 @@ # # SPDX-License-Identifier: BSD-3-Clause +"""Define Click parameter types and styled terminal message helpers.""" + +from collections.abc import Callable +from functools import update_wrapper +from typing import Any, Concatenate + import click_extra as click @@ -11,22 +17,40 @@ class FlexibleColorOption(click.ColorOption): _gnu_optional_value = False -def emit(prefix: str, color: str, *message: str, enabled: bool = True) -> None: - """Render a styled prefix and message to stdout. +def pass_obj_silent[T, **P, R]( + f: Callable[Concatenate[T | None, P], R], +) -> Callable[P, R]: + """Pass the current context object to a callback, if available. + + The current Click context is retrieved silently. If no context exists or + its object is not a dictionary, ``None`` is passed instead. + + The context object is inserted as the first positional argument to *f*. + """ - Internal helper used by error(), msg(), warn(), verbose(), and debug() - to avoid duplicating the Click styling and echo logic. The first message - argument is rendered in bold, subsequent arguments are appended unstyled. + def new_func(*args: P.args, **kwargs: P.kwargs) -> R: + ctx = click.get_current_context(silent=True) + obj = ctx.obj if ctx is not None else None + return f(obj, *args, **kwargs) - Args: - prefix: The prefix string shown before the message (e.g. "==>", "==> ERROR:"). - color: A Click-compatible color name (e.g. "red", "green") applied to the prefix. - *message: One or more message parts to display. At least one is required. - enabled: If False, the function returns immediately without printing. - Defaults to True. + return update_wrapper(new_func, f) - Raises: - TypeError: If no message arguments are provided. + +def emit( + prefix: str, color: str, *message: str, enabled: bool = True, err: bool = False +) -> None: + """Render a styled prefix and message to stdout. + + This helper is used by :func:`error`, :func:`msg`, :func:`warn`, + :func:`verbose`, and :func:`debug`. The first message argument is rendered in + bold and subsequent arguments are appended unstyled. + + :param prefix: Prefix displayed before the message. + :param color: Click-compatible color name applied to the prefix. + :param message: One or more message parts to display. + :param enabled: Whether the message should be emitted. + :param err: Write to `stderr` instead of `stdout`. + :raises TypeError: If no message arguments are provided. """ if not message: raise TypeError("emit() missing 1 required positional argument: 'message'") @@ -41,76 +65,97 @@ def emit(prefix: str, color: str, *message: str, enabled: bool = True) -> None: click.style(message[0], bold=True), *message[1:], ] - ) + ), + err=err, ) def error(*message: str) -> None: """Print a formatted error message to stdout in red. - The first argument is rendered in bold, subsequent arguments are appended - unstyled. Always prints regardless of quiet or verbose flags. + The first argument is rendered in bold and subsequent arguments are appended + unstyled. Error messages are always printed regardless of quiet or verbose + flags. - Args: - *message: One or more message parts to display. + :param message: One or more message parts to display. """ - emit("==> ERROR:", "red", *message) + emit("==> ERROR:", "red", *message, err=True) -@click.pass_obj -def msg(obj: dict[str, bool], *message: str) -> None: +@pass_obj_silent +def msg(obj: dict[str, Any] | None, *message: str, err: bool = False) -> None: """Print a formatted informational message to stdout in green. - Suppressed when the quiet flag is set. The first argument is rendered - in bold, subsequent arguments are appended unstyled. + The message is suppressed when the quiet flag is set. When called outside a + Click context, the message is emitted normally. The first argument is rendered + in bold and subsequent arguments are appended unstyled. - Args: - *message: One or more message parts to display. + :param message: One or more message parts to display. """ - emit("==>", "green", *message, enabled=not obj.get("quiet", False)) + emit( + "==>", + "green", + *message, + enabled=obj is None or not obj.get("quiet", False), + err=err, + ) -@click.pass_obj -def warn(obj: dict[str, bool], *message: str) -> None: +@pass_obj_silent +def warn(obj: dict[str, Any] | None, *message: str, err: bool = False) -> None: """Print a formatted warning message to stdout in yellow. - Suppressed when the quiet flag is set. The first argument is rendered - in bold, subsequent arguments are appended unstyled. + The message is suppressed when the quiet flag is set. When called outside a + Click context, the message is emitted normally. The first argument is rendered + in bold and subsequent arguments are appended unstyled. - Args: - *message: One or more message parts to display. + :param message: One or more message parts to display. """ - emit("==>", "yellow", *message, enabled=not obj.get("quiet", False)) + emit( + "==>", + "yellow", + *message, + enabled=obj is None or not obj.get("quiet", False), + err=err, + ) -@click.pass_obj -def verbose(obj: dict[str, bool], *message: str) -> None: +@pass_obj_silent +def verbose(obj: dict[str, Any] | None, *message: str, err: bool = False) -> None: """Print a formatted verbose message to stdout in blue. - Only prints when the verbose flag is set and the quiet flag is not. - The first argument is rendered in bold, subsequent arguments are - appended unstyled. + The message is printed only when called from a Click context with the verbose + flag set and the quiet flag unset. Calls made outside a Click context are a + no-op. - Args: - *message: One or more message parts to display. + :param message: One or more message parts to display. """ emit( "==>", "blue", *message, - enabled=obj.get("verbose", False) and not obj.get("quiet", False), + enabled=( + obj is not None + and obj.get("verbose", False) + and not obj.get("quiet", False) + ), + err=err, ) -@click.pass_obj -def debug(obj: dict[str, bool], *message: str) -> None: +@pass_obj_silent +def debug(obj: dict[str, Any] | None, *message: str, err: bool = False) -> None: """Print a formatted debug message to stdout in magenta. - Only prints when the DEBUG environment variable is set to a truthy - value (1, true, yes). The first argument is rendered in bold, - subsequent arguments are appended unstyled. + The message is printed only when called from a Click context with the debug + flag enabled. Calls made outside a Click context are a no-op. - Args: - *message: One or more message parts to display. + :param message: One or more message parts to display. """ - emit("==>", "magenta", *message, enabled=obj.get("debug", False)) + emit( + "==>", + "magenta", + *message, + enabled=obj is not None and obj.get("debug", False), + err=err, + ) diff --git a/prometheus_borgmatic_exporter/types.py b/prometheus_borgmatic_exporter/types.py index 043bf25..8b82fbe 100644 --- a/prometheus_borgmatic_exporter/types.py +++ b/prometheus_borgmatic_exporter/types.py @@ -1,7 +1,7 @@ -"""TypedDict definitions for borgmatic JSON output structures.""" # SPDX-FileCopyrightText: 2026 Dennis Fink <me+coding@dennisfink.me> # # SPDX-License-Identifier: BSD-3-Clause +"""TypedDict definitions for borgmatic JSON output structures.""" from typing import Any, TypedDict |