feat(host_agent): a host says what updates it has waiting and whether it needs a reboot (#5765)

Step 1 of milestone 514 "Updates and reboots on a cadence, carried out by
each host's agent…": #5765 "The agent reports pending updates and whether
a reboot is required".

- Agent 1.16.0: an hourly background check (apt-check, apt list, or dnf -C
  with updateinfo and needs-restarting), each command bounded at 120 s, no
  root and no network. Debian's reboot-required marker and its packages
  are read every sample.
- Ingest records updates_pending, updates_security and reboot_required for
  rules, and keeps the newest on the registration (host_agent_005_updates).
- The Agent card leads with "Reboot required since …", then the updates
  waiting or "Up to date", with when the check ran. list_agents carries
  them too.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

plugins/host_agent/agent.py

@@ -22,7 +22,7 @@ from collections import deque
 from concurrent.futures import ThreadPoolExecutor, wait as wait_futures
 from datetime import datetime, timezone
 
-AGENT_VERSION = "1.15.0"
+AGENT_VERSION = "1.16.0"
 
 # Default path to the local Docker Engine socket. Overridable via the
 # `docker_socket` config key; collection is silently skipped if it's absent or
@@ -1303,6 +1303,144 @@ class RingBuffer:
         return len(self._dq)
 
 
+# ─── pending updates and reboots (#5765) ─────────────────────────────────────
+#
+# What the package manager has waiting, read once an hour in a background
+# thread so a slow apt never holds up a sample, and whether the host needs a
+# reboot, read every sample (a file test). Nothing here needs root or the
+# network: apt-check and `apt list` read the lists apt-daily already fetched,
+# and dnf runs from its cache (-C).
+
+UPDATES_INTERVAL_SECONDS = 3600.0
+UPDATES_DEADLINE_SECONDS = 120.0
+REBOOT_REQUIRED = "/var/run/reboot-required"
+REBOOT_PACKAGES_MAX = 20
+
+
+def _run(cmd: list, deadline: float = UPDATES_DEADLINE_SECONDS):
+    """(returncode, stdout, stderr) of `cmd`, or None when it isn't installed.
+    Raises subprocess.TimeoutExpired past the deadline (the child is killed)."""
+    if shutil.which(cmd[0]) is None and not os.path.exists(cmd[0]):
+        return None
+    env = dict(os.environ, LANG="C", LC_ALL="C")
+    done = subprocess.run(cmd, capture_output=True, text=True, timeout=deadline, env=env)
+    return done.returncode, done.stdout, done.stderr
+
+
+def parse_apt_check(stderr: str):
+    """apt-check writes "<pending>;<security>" to stderr; (pending, security) or None."""
+    m = re.search(r"(\d+);(\d+)", stderr or "")
+    return (int(m.group(1)), int(m.group(2))) if m else None
+
+
+def parse_apt_list(stdout: str) -> tuple:
+    """(pending, security) from `apt list --upgradable`: one line per package,
+    "name/suite[,suite…] version arch [upgradable from: …]"; a security update
+    comes from a "-security" suite."""
+    pending = security = 0
+    for line in (stdout or "").splitlines():
+        if "upgradable from" not in line:
+            continue
+        pending += 1
+        suites = line.split("/", 1)[1].split(" ", 1)[0] if "/" in line else ""
+        if "-security" in suites:
+            security += 1
+    return pending, security
+
+
+def parse_dnf_check_update(stdout: str) -> int:
+    """Packages `dnf check-update` lists, up to its "Obsoleting Packages" part."""
+    count = 0
+    for line in (stdout or "").splitlines():
+        if line.startswith("Obsoleting"):
+            break
+        if line.strip() and not line.startswith((" ", "Last metadata")):
+            count += 1
+    return count
+
+
+def collect_updates() -> dict:
+    """{"manager", "pending", "security", "checked_at"} (security None when the
+    manager can't say), {"error"} when the check failed, or {} on a host with
+    no package manager we know. dnf hosts also say whether a reboot is
+    required, from needs-restarting."""
+    at = _now_iso()
+    try:
+        got = _run(["/usr/lib/update-notifier/apt-check"])
+        if got is not None:
+            counts = parse_apt_check(got[2])
+            if counts:
+                return {"manager": "apt", "pending": counts[0], "security": counts[1],
+                        "checked_at": at}
+        got = _run(["apt", "list", "--upgradable"])
+        if got is not None and got[0] == 0:
+            pending, security = parse_apt_list(got[1])
+            return {"manager": "apt", "pending": pending, "security": security, "checked_at": at}
+        got = _run(["dnf", "-q", "-C", "check-update"])
+        if got is not None:
+            if got[0] not in (0, 100):        # 100: updates are waiting
+                return {"manager": "dnf", "error": (got[2] or "dnf check-update failed").strip()[:200],
+                        "checked_at": at}
+            out = {"manager": "dnf", "pending": parse_dnf_check_update(got[1]),
+                   "security": None, "checked_at": at}
+            sec = _run(["dnf", "-q", "-C", "updateinfo", "list", "--security"])
+            if sec is not None and sec[0] == 0:
+                out["security"] = len({ln.split()[-1] for ln in sec[1].splitlines() if ln.strip()})
+            restart = _run(["needs-restarting", "-r"])
+            if restart is not None and restart[0] in (0, 1):
+                out["reboot_required"] = restart[0] == 1
+            return out
+    except subprocess.TimeoutExpired:
+        return {"error": "the package manager didn't answer in %.0f s" % UPDATES_DEADLINE_SECONDS,
+                "checked_at": at}
+    except OSError as exc:
+        return {"error": (str(exc) or type(exc).__name__)[:200], "checked_at": at}
+    return {}
+
+
+def refresh_updates(state: dict, now_mono: float | None = None) -> None:
+    """Start an update check in the background when one is due (hourly, and at
+    start). Its answer lands in state["updates"] for the samples that follow;
+    one check runs at a time. state["updates_due"] = True asks for one now."""
+    now_mono = time.monotonic() if now_mono is None else now_mono
+    if state.get("updates_running"):
+        return
+    last = state.get("updates_mono")
+    if not state.pop("updates_due", False) and last is not None \
+            and now_mono - last < UPDATES_INTERVAL_SECONDS:
+        return
+    state["updates_running"], state["updates_mono"] = True, now_mono
+
+    def work():
+        try:
+            state["updates"] = collect_updates()
+        except Exception as exc:              # never take the thread down silently
+            state["updates"] = {"error": (str(exc) or type(exc).__name__)[:200],
+                                "checked_at": _now_iso()}
+        finally:
+            state["updates_running"] = False
+
+    threading.Thread(target=work, name="updates", daemon=True).start()
+
+
+def collect_reboot(path: str = REBOOT_REQUIRED):
+    """{"required", "since", "packages"} from Debian's reboot-required marker,
+    or None where there is no such marker system (no apt) to ask."""
+    try:
+        st = os.stat(path)
+    except FileNotFoundError:
+        return {"required": False} if shutil.which("apt-get") else None
+    packages: list = []
+    try:
+        with open(path + ".pkgs") as fh:
+            packages = sorted({ln.strip() for ln in fh if ln.strip()})[:REBOOT_PACKAGES_MAX]
+    except OSError:
+        pass
+    return {"required": True,
+            "since": datetime.fromtimestamp(st.st_mtime, timezone.utc).isoformat(),
+            "packages": packages}
+
+
 def collect_self(state: dict) -> dict:
     """The agent's own footprint: resident memory and CPU % of one core since
     the last sample (#5764). A 1.13.0 agent once grew to 2.1 GB and spun for a
@@ -1436,6 +1574,17 @@ def build_sample(mounts: list[str], state: dict,
         if logs:
             sample["docker_logs"] = logs
 
+    # Pending updates: the last hourly answer rides every sample (#5765).
+    try:
+        refresh_updates(state)
+        reboot = collect_reboot()
+    except Exception:
+        reboot = None
+    if state.get("updates"):
+        sample["updates"] = state["updates"]
+    if reboot is not None:
+        sample["reboot"] = reboot
+
     # Last, so its CPU covers the whole of this sample's work.
     try:
         sample["agent"] = collect_self(state)