Files
computer-playbook/utils/install/system_pkg.py
T
kevinveenbirkenbachandClaude Opus 5 56036be2a1 fix(install): retry transient download failures, keep 4xx fatal
A single anonymous GET for the pinned hadolint binary completed its TLS
handshake and its request, then got a clean FIN with no status line after
3.6s. download_file was a bare urlopen with no retry, and make install-lint
is a prerequisite of autoformat, lint and every lint-* target -- so one shed
socket took down the whole quality gate. The same asset downloaded fine five
times elsewhere in that run, 15s after the failure on another distro.

Retry four times with exponential backoff, catching (OSError, HTTPException).
Two names, not the obvious four: ssl.SSLError is an OSError but NOT a
URLError, so the narrow tuple would have missed it, while OSError subsumes
URLError, ConnectionError, TimeoutError and gaierror in one token.
HTTPException is needed separately because IncompleteRead is not an OSError.

HTTPError is caught FIRST and re-raised immediately for any status outside
{408,429,500,502,503,504}. That ordering is load-bearing -- HTTPError is
itself an OSError, so a broader catch first would swallow it -- and it is
what keeps this robustness rather than masking: a 404 is exactly what a
stale version pin looks like after roles/sys-lint is bumped, and it still
fails on the first attempt with the URL. The final attempt re-raises the
original with its traceback, and write_bytes stays in the else branch so a
full disk is never retried.

apt was the other uncached, unretried input in the same target (its default
Acquire::Retries is 0), so give both invocations -o Acquire::Retries=3,
matching what eight Dockerfiles in this repo already do.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-10 10:12:20 +02:00

132 lines
4.2 KiB
Python

"""System-package-manager dispatch (pacman / apt-get / dnf / yum / brew)."""
from __future__ import annotations
import contextlib
import shutil
import subprocess
from utils.install.primitives import log, run_privileged, warn
_SUPPORTED = ("pacman", "apt-get", "dnf", "yum", "brew")
def detect_package_manager() -> str:
for manager in _SUPPORTED:
if shutil.which(manager) is not None:
return manager
raise RuntimeError("No supported package manager found")
def _prepare_manager(manager: str) -> None:
if manager == "apt-get":
run_privileged(
[
"apt-get",
"-o",
"DPkg::Lock::Timeout=600",
"-o",
"Acquire::Retries=3",
"update",
]
)
elif manager == "dnf":
with contextlib.suppress(subprocess.CalledProcessError):
run_privileged(["dnf", "-y", "install", "dnf-plugins-core"])
with contextlib.suppress(subprocess.CalledProcessError):
run_privileged(["dnf", "-y", "install", "epel-release"])
elif manager == "yum":
with contextlib.suppress(subprocess.CalledProcessError):
run_privileged(["yum", "-y", "install", "yum-utils"])
with contextlib.suppress(subprocess.CalledProcessError):
run_privileged(["yum", "-y", "install", "epel-release"])
def _install_one(manager: str, package: str) -> bool:
log(f"Installing package '{package}' via {manager}")
try:
if manager == "pacman":
run_privileged(["pacman", "-Syu", "--noconfirm", "--needed", package])
elif manager == "apt-get":
run_privileged(
[
"apt-get",
"-o",
"DPkg::Lock::Timeout=600",
"-o",
"Acquire::Retries=3",
"install",
"-y",
"--no-install-recommends",
package,
]
)
elif manager == "dnf":
run_privileged(["dnf", "-y", "install", package])
elif manager == "yum":
run_privileged(["yum", "-y", "install", package])
elif manager == "brew":
subprocess.run(["brew", "install", package], check=True)
else:
warn(f"Unsupported package manager: {manager}")
return False
except subprocess.CalledProcessError:
return False
return True
def install_package_candidates(manager: str, packages: list[str]) -> None:
_prepare_manager(manager)
for package in packages:
if _install_one(manager, package):
return
raise RuntimeError(f"All package candidates failed via {manager}: {packages}")
_COMMAND_PACKAGES: dict[str, dict[str, list[str]]] = {
"ansible-playbook": {
"pacman": ["ansible-core", "ansible"],
"apt-get": ["ansible-core", "ansible"],
"dnf": ["ansible-core", "ansible"],
"yum": ["ansible-core", "ansible"],
"brew": ["ansible"],
},
"ansible-galaxy": {
"pacman": ["ansible-core", "ansible"],
"apt-get": ["ansible-core", "ansible"],
"dnf": ["ansible-core", "ansible"],
"yum": ["ansible-core", "ansible"],
"brew": ["ansible"],
},
"ruff": {m: ["ruff"] for m in _SUPPORTED},
"shfmt": {m: ["shfmt"] for m in _SUPPORTED},
"shellcheck": {m: ["shellcheck"] for m in _SUPPORTED},
"unzip": {m: ["unzip"] for m in _SUPPORTED},
"npm": {
"pacman": ["npm", "nodejs"],
"apt-get": ["npm", "nodejs"],
"dnf": ["npm", "nodejs"],
"yum": ["npm", "nodejs"],
"brew": ["node"],
},
}
def install_command_via_pkg(command_name: str) -> None:
manager = detect_package_manager()
mapping = _COMMAND_PACKAGES.get(command_name)
if mapping is None or manager not in mapping:
raise RuntimeError(
f"No installer mapping defined for '{command_name}' on '{manager}'."
)
log(f"Missing command '{command_name}'. Attempting installation via {manager}.")
install_package_candidates(manager, mapping[manager])
__all__ = [
"detect_package_manager",
"install_command_via_pkg",
"install_package_candidates",
]