computer-playbook/plugins/action/package_install.py

#!/usr/bin/env python3
"""Install a package by its logical id from the meta/packages.yml registry.

The registry owns every distro mapping and every acquisition path, so a
role states what it needs and never how to get it::

    - name: "📦 Install the userspace NFS server"
      package_install:
        id: nfs-ganesha

    - name: "📦 Install the storage stack"
      package_install:
        id:
          - nfs-ganesha
          - libntirpc

Distribution repositories, the AUR, COPR, PPA and source builds all run
through here, including the unprivileged build environment the AUR needs.
:mod:`utils.packages.plan` decides which module calls that takes; this
plugin only executes them. Ids whose plans only call the host's package
manager share one call, so a long list costs one transaction, and
:mod:`utils.packages.installed` skips that call when the host's package
database already shows every name in the requested state.

Args:
    id: logical package id, or a list of ids, each declared either in the
        calling role's own meta/packages.yml or in the shared root
        meta/packages.yml.
    state: ``present`` (default) or ``absent``.
"""

from __future__ import annotations

import time
from typing import Any

from ansible.errors import AnsibleActionFail, AnsibleError
from ansible.plugins.action import ActionBase

from utils.packages.calls import (
    GENERIC_PACKAGE,
    MANAGER_ARGS,
    STATE_PRESENT,
    STATES,
    ModuleCall,
)
from utils.packages.facts import distribution_and_family, read_fact
from utils.packages.installed import (
    deb_pin,
    deb_pin_argv,
    installed_version,
    installed_version_argv,
    minimum_argv,
    minimum_reached,
    query_argv,
    state_holds,
)
from utils.packages.outcome import aggregate
from utils.packages.plan import build_plan, merge_plans
from utils.packages.registry import build_registry, project_root_from_env, resolve
from utils.packages.schema import ROLE_FILE_META_PACKAGES, PackagesShapeError


class ActionModule(ActionBase):
    def run(
        self, tmp: Any = None, task_vars: dict[str, Any] | None = None
    ) -> dict[str, Any]:
        super().run(tmp, task_vars)
        task_vars = task_vars or {}

        state = self._state()
        facts = self._target_facts(task_vars)
        distribution, os_family = distribution_and_family(facts)
        if not distribution:
            raise AnsibleActionFail(
                "package_install needs gathered facts; ansible_facts.distribution "
                "and ansible_facts.os_family must both be set."
            )

        package_ids = self._package_ids()
        registry = self._registry()
        role = self._role()

        plans: list[list[ModuleCall]] = []
        skipped: list[str] = []

        for package_id in package_ids:
            spec = self._spec_for(registry, package_id, distribution, os_family, role)
            plan = self._plan_for(spec, state)
            if not plan:
                skipped.append(package_id)
                continue
            plans.append(plan)

        results: list[dict[str, Any]] = []
        for plan in merge_plans(plans):
            results.extend(self._run_plan(plan, task_vars, facts))

        return aggregate(results, skipped, distribution)

    def _plan_for(self, spec, state) -> list[ModuleCall]:
        try:
            return build_plan(spec, state)
        except ValueError as exc:
            raise AnsibleActionFail(str(exc)) from exc

    def _run_plan(
        self,
        plan: list[ModuleCall],
        task_vars: dict[str, Any],
        facts: dict[str, Any],
    ) -> list[dict[str, Any]]:
        results: list[dict[str, Any]] = []
        for call in plan:
            result = self._attempt(call, task_vars, facts)
            results.append(result)
            if result.get("failed"):
                break
        return results

    def _attempt(
        self, call: ModuleCall, task_vars: dict[str, Any], facts: dict[str, Any]
    ) -> dict[str, Any]:
        """Run *call* once, or up to its retry policy while it keeps
        failing. Returns the last result, tagged with the module that
        produced it. A call whose state the host already holds does not
        run at all."""
        if self._state_holds(call, task_vars, facts):
            return {
                "changed": False,
                "msg": f"{', '.join(call.args['name'])} already {call.args['state']}",
                "module": call.module,
            }
        reached = self._pin_reached(call, task_vars)
        if reached is not None:
            return {"changed": False, "msg": reached, "module": call.module}
        if call.retry is None:
            return self._tagged(call, task_vars, facts)
        result: dict[str, Any] = {}
        for attempt in range(1, call.retry.attempts + 1):
            result = self._tagged(call, task_vars, facts)
            if not result.get("failed"):
                return result
            if attempt < call.retry.attempts:
                time.sleep(call.retry.seconds)
        return result

    def _state_holds(
        self, call: ModuleCall, task_vars: dict[str, Any], facts: dict[str, Any]
    ) -> bool:
        pkg_mgr = read_fact(facts, "pkg_mgr")
        argv = query_argv(call, pkg_mgr)
        if argv is None:
            return False
        return state_holds(call, pkg_mgr, self._query(argv, task_vars))

    def _pin_reached(self, call: ModuleCall, task_vars: dict[str, Any]) -> str | None:
        argv = deb_pin_argv(call)
        if argv is None:
            return None
        pin = deb_pin(self._query(argv, task_vars))
        if pin is None:
            return None
        package, pinned = pin
        installed = installed_version(
            self._query(installed_version_argv(package), task_vars)
        )
        if installed is None:
            return None
        if not minimum_reached(self._query(minimum_argv(installed, pinned), task_vars)):
            return None
        return (
            f"{package} {installed} already installed, at or above the pinned {pinned}"
        )

    def _query(self, argv: list[str], task_vars: dict[str, Any]) -> dict[str, Any]:
        try:
            return self._execute_module(
                module_name="ansible.builtin.command",
                module_args={"argv": argv, "expand_argument_vars": False},
                task_vars=task_vars,
            )
        except AnsibleError:
            return {}

    def _tagged(
        self, call: ModuleCall, task_vars: dict[str, Any], facts: dict[str, Any]
    ) -> dict[str, Any]:
        result = self._execute(call, task_vars, facts)
        result["module"] = call.module
        return result

    def _execute(
        self, call: ModuleCall, task_vars: dict[str, Any], facts: dict[str, Any]
    ) -> dict[str, Any]:
        args = {k: v for k, v in call.args.items() if v is not None}
        module = self._module_name(call, facts)
        if call.module == GENERIC_PACKAGE:
            args.update(MANAGER_ARGS.get(module, {}))
        if not call.become_user:
            return self._execute_module(
                module_name=module, module_args=args, task_vars=task_vars
            )

        become = self._connection.become
        if become is None:
            raise AnsibleActionFail(
                f"package_install must escalate to '{call.become_user}' to build "
                "this package, but the task runs without become."
            )
        previous = become.get_option("become_user")
        become.set_option("become_user", call.become_user)
        try:
            return self._execute_module(
                module_name=module, module_args=args, task_vars=task_vars
            )
        finally:
            become.set_option("become_user", previous)

    def _module_name(self, call: ModuleCall, facts: dict[str, Any]) -> str:
        if call.module != GENERIC_PACKAGE:
            return call.module
        pkg_mgr = read_fact(facts, "pkg_mgr")
        if not pkg_mgr:
            raise AnsibleActionFail(
                "package_install needs ansible_facts.pkg_mgr to choose a package "
                "manager module; gather facts before installing."
            )
        return pkg_mgr

    def _target_facts(self, task_vars: dict[str, Any]) -> dict[str, Any]:
        """Facts of the host the package modules will actually run on.

        Under ``delegate_to`` task_vars still carries the INVENTORY host's
        facts, so a fedora node delegating to a Debian controller resolves the
        RedHat mapping and then runs ``dnf`` on a machine that has none. The
        delegation target is asked directly instead.
        """
        delegate = getattr(self._task, "delegate_to", None)
        if not delegate or delegate == task_vars.get("inventory_hostname"):
            return task_vars.get("ansible_facts") or {}

        result = self._execute_module(
            module_name="ansible.builtin.setup",
            module_args={"gather_subset": ["!all", "!min", "distribution", "pkg_mgr"]},
            task_vars=task_vars,
        )
        if result.get("failed"):
            raise AnsibleActionFail(
                f"package_install could not gather facts from the delegation "
                f"target '{delegate}': {result.get('msg', 'setup failed')}"
            )
        return result.get("ansible_facts") or {}

    def _role(self) -> str | None:
        role = getattr(self._task, "_role", None)
        return str(role.get_name(include_role_fqcn=False)) if role is not None else None

    def _registry(self):
        try:
            return build_registry(project_root_from_env())
        except PackagesShapeError as exc:
            raise AnsibleActionFail(str(exc)) from exc

    def _state(self) -> str:
        state = str(self._task.args.get("state", STATE_PRESENT)).strip()
        if state not in STATES:
            raise AnsibleActionFail(
                f"package_install state {state!r} is not one of {STATES}."
            )
        return state

    def _package_ids(self) -> list[str]:
        raw = self._task.args.get("id")
        if isinstance(raw, str) and raw.strip():
            return [raw.strip()]
        if isinstance(raw, list) and raw:
            return [str(item).strip() for item in raw if str(item).strip()]
        raise AnsibleActionFail(
            "package_install requires 'id' as a package id or a list of ids."
        )

    def _spec_for(
        self,
        registry,
        package_id: str,
        distribution: str,
        os_family: str,
        role: str | None,
    ):
        declaration = registry.get(package_id)
        if declaration is None:
            raise AnsibleActionFail(
                f"Package id '{package_id}' is declared in no meta/packages.yml. "
                f"Declare it in the owning role before installing it."
            )
        if role and not declaration.shared and declaration.role != role:
            raise AnsibleActionFail(
                f"Role '{role}' installs '{package_id}', which "
                f"'{declaration.role}' declares. A role may only install ids "
                f"from its own {ROLE_FILE_META_PACKAGES} or from the shared "
                f"root {ROLE_FILE_META_PACKAGES}. Move the package to the root "
                f"file when more than one role needs it."
            )
        try:
            spec = resolve(declaration, distribution, os_family)
        except PackagesShapeError as exc:
            raise AnsibleActionFail(str(exc)) from exc
        if spec is None:
            raise AnsibleActionFail(
                f"Package '{package_id}' has no mapping for distribution "
                f"'{distribution}' (os_family '{os_family}') in {declaration.path}."
            )
        return spec