path: root/prometheus_pacman_exporter/__init__.py

#!/usr/bin/env python3

# SPDX-FileCopyrightText: 2026 Dennis Fink <me+coding@dennisfink.me>
#
# SPDX-License-Identifier: BSD-3-Clause

import os
import shlex
import subprocess
from pathlib import Path
from typing import NamedTuple, cast

import click_extra as click
from cloup.constraints import If, require_all
from prometheus_client import Gauge, write_to_textfile

from . import metrics
from .cli import FlexibleColorOption, debug, error, msg, verbose

VERSION = "1.1.0"
DESCRIPTION = "Collect pacman package metrics and export them for Prometheus."
DATE_OF_CREATION = "2021-05-15"
DATE_OF_REVISION = "2026-09-27"
AUTHOR = "Dennis Fink <me+coding@dennisfink.me>"
LICENSE = "BSD-3-Clause"


class PackageQuery(NamedTuple):
    command: tuple[str, ...]
    description: str
    metric: Gauge
    accepted_exit_codes: tuple[int, ...] = (0,)
    empty_exit_codes: tuple[int, ...] = ()


PACMAN_BASE_COMMAND = ("pacman", "--query")
PACMAN_EMPTY_EXIT_CODES = (1,)
BASE_QUERIES: tuple[PackageQuery, ...] = (
    PackageQuery(PACMAN_BASE_COMMAND, "installed packages", metrics.INSTALLED_PACKAGES),
    PackageQuery(
        (*PACMAN_BASE_COMMAND, "--explicit"),
        "explicitly installed packages",
        metrics.EXPLICIT_PACKAGES,
        empty_exit_codes=PACMAN_EMPTY_EXIT_CODES,
    ),
    PackageQuery(
        (*PACMAN_BASE_COMMAND, "--deps"),
        "packages installed as dependencies",
        metrics.DEPENDS_PACKAGES,
        empty_exit_codes=PACMAN_EMPTY_EXIT_CODES,
    ),
    PackageQuery(
        (*PACMAN_BASE_COMMAND, "--unrequired"),
        "unrequired packages",
        metrics.UNREQUIRED_PACKAGES,
        empty_exit_codes=PACMAN_EMPTY_EXIT_CODES,
    ),
    PackageQuery(
        (*PACMAN_BASE_COMMAND, "--foreign"),
        "foreign packages",
        metrics.FOREIGN_PACKAGES,
        empty_exit_codes=PACMAN_EMPTY_EXIT_CODES,
    ),
    PackageQuery(
        (*PACMAN_BASE_COMMAND, "--native"),
        "native packages",
        metrics.NATIVE_PACKAGES,
        empty_exit_codes=PACMAN_EMPTY_EXIT_CODES,
    ),
    PackageQuery(
        (*PACMAN_BASE_COMMAND, "--deps", "--unrequired"),
        "orphan packages",
        metrics.ORPHAN_PACKAGES,
        (0, 1),
    ),
)


def print_version(
    ctx: click.Context, param: click.Parameter | None, value: bool
) -> None:
    """Print script metadata (name, version, description, author, dates) and exit.

    :param ctx: Click context injected by :func:`click.pass_context`.
    """
    if not value or ctx.resilient_parsing:
        return

    click.echo(click.style("Scriptname:", fg="red", bold=True) + f" {ctx.info_name}")
    click.echo(click.style("Version:", fg="green", bold=True) + f" {VERSION}")
    click.echo(click.style("Description:", fg="yellow", bold=True) + f" {DESCRIPTION}")
    click.echo(click.style("Author:", fg="blue", bold=True) + f" {AUTHOR}")
    click.echo(
        click.style("Date of creation:", fg="magenta", bold=True)
        + f" {DATE_OF_CREATION}"
    )
    click.echo(
        click.style("Date of revision:", fg="cyan", bold=True) + f" {DATE_OF_REVISION}"
    )
    click.echo(click.style("License:", fg="red", bold=True) + f" {LICENSE}")

    click.echo("""Copyright (c) 2026 Dennis Fink <me+coding@dennisfink.me>.

Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions are met:

1. Redistributions of source code must retain the above copyright notice, this
list of conditions and the following disclaimer.

2. Redistributions in binary form must reproduce the above copyright notice,
this list of conditions and the following disclaimer in the documentation
and/or other materials provided with the distribution.

3. Neither the name of the copyright holder nor the names of its contributors
may be used to endorse or promote products derived from this software without
specific prior written permission.

THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS \"AS IS\" AND
ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.""")

    ctx.exit()


@click.pass_context
def run_command(
    ctx: click.Context,
    command: tuple[str, ...],
    accepted_exit_codes: tuple[int, ...] = (0,),
    empty_exit_codes: tuple[int, ...] = (),
) -> subprocess.CompletedProcess[str]:
    """Run a command and return the result.

    Logs the command at debug level before execution. Exits the application
    with the command's return code if it fails.

    :param ctx: Click context injected by :func:`click.pass_context`.
    :param command: Command and its arguments.
    :param accepted_exit_codes: Exit codes considered successful.
    :param empty_exit_codes: Exit codes considered an empty result when both
        stdout and stderr are empty.
    :return: Completed process with stdout and stderr captured.
    """
    command_string = shlex.join(command)
    debug("Running command:", command_string, err=True)
    result = subprocess.run(command, capture_output=True, encoding="utf-8", check=False)

    if result.returncode in accepted_exit_codes:
        return result

    if (
        result.returncode in empty_exit_codes
        and not result.stdout.strip()
        and not result.stderr.strip()
    ):
        debug("Command returned an empty result:", command_string, err=True)
        return result

    error(f"Command ({command_string}) failed with exit code:", str(result.returncode))
    if result.stderr:
        error("stderr:", result.stderr.strip())
    ctx.exit(1)

    return result


@click.command(
    context_settings={"help_option_names": ("-h", "--help", "-?")}, params=[]
)
@click.option(
    "--textfile-collector-dir",
    type=click.Path(
        exists=True,
        readable=True,
        writable=True,
        file_okay=False,
        allow_dash=False,
        path_type=Path,
    ),
    default="/var/lib/prometheus/node-exporter",
    help="Directory where the Prometheus collector textfile will be written.",
)
@click.option(
    "--use-checkupdates",
    "use_checkupdates_flag",
    is_flag=True,
    default=False,
    help="Use checkupdates instead of pacman to check for available package updates.",
)
@click.option(
    "--aur", "aur_flag", is_flag=True, default=False, help="Include AUR updates."
)
@click.option("--aur-command", help="Command used to check for AUR updates.")
@click.constraint(
    If("aur_flag", then=require_all).hidden(), ["aur_flag", "aur_command"]
)
@click.option("--color", cls=FlexibleColorOption)
@click.no_color_option()
@click.option(
    "-q",
    "--quiet",
    "quiet_flag",
    is_flag=True,
    default=False,
    help="Suppress all non-error output, including verbose output.",
)
@click.option(
    "-v",
    "--verbose",
    "verbose_flag",
    is_flag=True,
    default=False,
    help="Enable verbose output.",
)
@click.option(
    "--version",
    callback=print_version,
    default=False,
    expose_value=False,
    help="Show version information and exit.",
    is_eager=True,
    is_flag=True,
)
@click.pass_context
def prometheus_pacman_exporter(
    ctx: click.Context,
    textfile_collector_dir: Path,
    use_checkupdates_flag: bool,
    aur_flag: bool,
    aur_command: str | None,
    quiet_flag: bool,
    verbose_flag: bool,
) -> None:
    """Collect pacman package metrics and export them for Prometheus.

    Queries pacman for installed package statistics and available updates, then
    writes them to a ``.prom`` file for consumption by the Prometheus node
    exporter textfile collector.

    By default, updateable packages are determined using ``pacman --query
    --upgrades``. Use ``--use-checkupdates`` to query updates with
    ``checkupdates`` from ``pacman-contrib`` instead, without modifying the
    system package databases.

    AUR update metrics can optionally be collected by providing ``--aur``
    together with ``--aur-command``.
    """

    ctx.ensure_object(dict)
    ctx.obj["debug"] = os.environ.get("DEBUG", "").lower() in ("1", "true", "yes")
    ctx.obj["quiet"] = quiet_flag
    ctx.obj["verbose"] = verbose_flag

    debug("Effective configuration:", err=True)
    debug("  textfile_collector_dir:", str(textfile_collector_dir), err=True)
    debug("  use_checkupdates:", str(use_checkupdates_flag), err=True)
    debug("  aur:", str(aur_flag), err=True)
    debug("  aur_command:", str(aur_command), err=True)
    debug("  verbose:", str(verbose_flag), err=True)
    debug("  quiet:", str(quiet_flag), err=True)

    queries: list[PackageQuery] = [*BASE_QUERIES]

    if use_checkupdates_flag:
        queries.append(
            PackageQuery(
                ("checkupdates",),
                "updateable packages",
                metrics.UPDATEABLE_PACKAGES,
                (0, 2),
            )
        )
    else:
        queries.append(
            PackageQuery(
                (*PACMAN_BASE_COMMAND, "--upgrades"),
                "updateable packages",
                metrics.UPDATEABLE_PACKAGES,
                empty_exit_codes=PACMAN_EMPTY_EXIT_CODES,
            )
        )

    if aur_flag:
        queries.append(
            PackageQuery(
                tuple(shlex.split(cast(str, aur_command))),
                "AUR updateable packages",
                metrics.AUR_UPDATEABLE_PACKAGES,
            )
        )

    for query in queries:
        output = run_command(
            query.command, query.accepted_exit_codes, query.empty_exit_codes
        )
        value = len(output.stdout.splitlines())
        verbose(f"Setting {query.description}:", str(value), err=True)
        query.metric.set(value)

    metrics_file = textfile_collector_dir / "pacman.prom"
    write_to_textfile(str(metrics_file.absolute()), metrics.REGISTRY)
    msg("Metrics written to:", str(metrics_file.absolute()), err=True)


if __name__ == "__main__":
    prometheus_pacman_exporter()