bvandeusenandClaude Opus 5.5
d102fc3333
On 2026-10-25 the 09:00 job on VDNT-Docker02 drained the node, which ran Traefik and Steward. The agent's result reaches Steward through Traefik, and Traefik took 11.5 minutes to start on Docker03 (its image had to be pulled). The job gave up at the 10-minute drain deadline and finished timed_out where it stood, so the node stayed drained for hours. - A job that fails or times out while its node may be drained (the drain was handed out, or went through and wasn't undone) goes to undraining first. The outcome is kept in the new column ends_as, and the job finishes as that once the node is active again. If the undrain goes wrong too, detail says both. - The drain deadline is 30 minutes instead of 10, so a result held up while the proxy moves still lands. - Migration host_agent_012_ends_as. get_maintenance_job returns ends_as. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
365 lines
18 KiB
Python
365 lines
18 KiB
Python
"""The host_agent plugin's MCP tools and its get_overview findings."""
|
|
from __future__ import annotations
|
|
|
|
from datetime import timedelta
|
|
|
|
from sqlalchemy import select
|
|
|
|
import json
|
|
|
|
from steward.hosts import service as host_svc
|
|
from steward.mcp._context import current_caller, mcp_app
|
|
from steward.mcp.tools._audit import audit_mcp
|
|
from steward.mcp.tools._shape import age_s, clamp_limit, db_session, iso, lookback, now_utc, pick, rnd
|
|
from steward.mcp.tools.overview import finding
|
|
from steward.models.hosts import Host
|
|
from steward.models.users import UserRole
|
|
|
|
from . import cadence, maintenance, senders
|
|
from .models import HostAgentRegistration, HostMaintenanceJob, HostMaintenancePlan, HostMaintenanceRun
|
|
|
|
# A registration this old that has never checked in is a failed install, not one in progress.
|
|
PENDING_GRACE = timedelta(minutes=30)
|
|
|
|
|
|
def agent_state(reg, now, stale_after_s: int) -> str:
|
|
"""The agent's state: reporting, stale (stopped checking in) or pending (never has)."""
|
|
if reg.last_seen_at is None:
|
|
return "pending"
|
|
return "reporting" if (now - reg.last_seen_at).total_seconds() <= stale_after_s else "stale"
|
|
|
|
|
|
async def _agents(db):
|
|
return (await db.execute(
|
|
select(HostAgentRegistration, Host)
|
|
.join(Host, Host.id == HostAgentRegistration.host_id)
|
|
.order_by(Host.name)
|
|
)).all()
|
|
|
|
|
|
def _updates(reg) -> dict | None:
|
|
"""A registration's pending updates and reboot, as list_agents says them (#5765)."""
|
|
if reg.updates_checked_at is None and reg.reboot_required is None:
|
|
return None
|
|
return {
|
|
"manager": reg.updates_manager, "pending": reg.updates_pending,
|
|
"security": reg.updates_security, "checked_at": iso(reg.updates_checked_at),
|
|
"error": reg.updates_error, "reboot_required": reg.reboot_required,
|
|
"reboot_required_since": iso(reg.reboot_required_since),
|
|
"reboot_packages": reg.reboot_packages.split(", ") if reg.reboot_packages else [],
|
|
}
|
|
|
|
|
|
async def list_agents() -> dict:
|
|
"""Every host agent Steward has registered, with its version and whether it is reporting.
|
|
|
|
`state` is "reporting", "stale" (it checked in before but its last
|
|
check-in is older than the stale window, `stale_after_s`) or "pending"
|
|
(registered, never checked in, usually an install that is still running
|
|
or that failed). Each row carries the agent version, kernel, distro, arch
|
|
and the IP the agent reports. get_host shows an agent's latest metrics.
|
|
|
|
`updates` is what the host's package manager has waiting, from the
|
|
agent's hourly check (agent ≥ 1.16.0): manager, pending, security (null
|
|
when the manager can't say), checked_at, error, and reboot_required with
|
|
since when and the packages that asked for it. null before the first check.
|
|
"""
|
|
now = now_utc()
|
|
stale_after = host_svc.agent_stale_after_seconds(mcp_app().config)
|
|
async with db_session() as db:
|
|
rows = await _agents(db)
|
|
agents = [
|
|
{
|
|
"host_id": h.id, "host_name": h.name,
|
|
"state": agent_state(reg, now, stale_after),
|
|
"last_seen_at": iso(reg.last_seen_at), "last_seen_age_s": age_s(reg.last_seen_at, now),
|
|
"agent_version": reg.agent_version, "kernel": reg.kernel, "distro": reg.distro,
|
|
"arch": reg.arch, "reported_ip": reg.host_ip,
|
|
"registered_at": iso(reg.created_at),
|
|
"updates": _updates(reg),
|
|
}
|
|
for reg, h in rows
|
|
]
|
|
counts = {s: sum(1 for a in agents if a["state"] == s) for s in ("reporting", "stale", "pending")}
|
|
return {"stale_after_s": stale_after, "counts": counts, "agents": agents, "total": len(agents)}
|
|
|
|
|
|
# ── maintenance (#5834): plans, runs, jobs, and starting one ─────────────────
|
|
|
|
JOBS_MAX = 50
|
|
# A failed job stays on the overview this long, unless a later one succeeded.
|
|
FAILED_JOB_DAYS = 7
|
|
|
|
|
|
async def _names(db, ids) -> dict[str, str]:
|
|
from steward.models.users import User
|
|
ids = {i for i in ids if i}
|
|
if not ids:
|
|
return {}
|
|
hosts = dict((await db.execute(select(Host.id, Host.name).where(Host.id.in_(ids)))).all())
|
|
users = dict((await db.execute(select(User.id, User.username).where(User.id.in_(ids)))).all())
|
|
return {**users, **hosts}
|
|
|
|
|
|
def _job_row(job, names: dict, now) -> dict:
|
|
deadline = maintenance.DEADLINES.get(job.state)
|
|
started = job.created_at if job.state == "queued" else job.state_at
|
|
return {
|
|
"job_id": job.id,
|
|
"host": names.get(job.host_id, job.host_id),
|
|
"state": job.state,
|
|
"label": maintenance.LABELS.get(job.state, job.state),
|
|
"summary": maintenance.summary(job),
|
|
"active": job.state in maintenance.ACTIVE,
|
|
"reboot_policy": job.reboot_policy,
|
|
"packages_updated": job.packages,
|
|
"reboot_needed": job.reboot_needed,
|
|
"rebooted_at": iso(job.back_at),
|
|
"swarm_node": job.node_id,
|
|
"drained_by": names.get(job.drain_host_id, job.drain_host_id),
|
|
"drained_at": iso(job.drained_at),
|
|
"came_back_at": iso(job.came_back_at),
|
|
"ends_as": job.ends_as,
|
|
"docker_update": job.docker_update,
|
|
"checked": job.checked,
|
|
"detail": job.detail,
|
|
"state_since": iso(job.state_at), "state_age_s": age_s(job.state_at, now),
|
|
"gives_up_at": iso(started + deadline) if deadline and started else None,
|
|
"started_by": names.get(job.created_by) if job.created_by else
|
|
("maintenance plan" if job.run_id else None),
|
|
"run_id": job.run_id,
|
|
"created_at": iso(job.created_at),
|
|
"finished_at": iso(job.finished_at),
|
|
}
|
|
|
|
|
|
async def list_maintenance_plans() -> dict:
|
|
"""Each maintenance plan: which inventory group it updates, when (weekly
|
|
slot, time zone, how long a run may keep starting hosts), its reboot
|
|
policy, whether it is enabled, its next slot, and its latest run.
|
|
|
|
A run updates the group's hosts one at a time; `latest_run` says how far
|
|
it got (`hosts_done` of `hosts`), which hosts it passed over and why
|
|
(`skipped`), and how it ended. `last_slot_at` is the last scheduled slot
|
|
that fired, null until one has. get_maintenance_job shows one host's job.
|
|
Plans are created and edited on the Maintenance page.
|
|
"""
|
|
from steward.models.ansible_inventory import AnsibleGroup
|
|
now = now_utc()
|
|
async with db_session() as db:
|
|
plans = list((await db.execute(select(HostMaintenancePlan)
|
|
.order_by(HostMaintenancePlan.name))).scalars())
|
|
groups = dict((await db.execute(select(AnsibleGroup.id, AnsibleGroup.name))).all())
|
|
latest = {}
|
|
for plan in plans:
|
|
latest[plan.id] = (await db.execute(
|
|
select(HostMaintenanceRun).where(HostMaintenanceRun.plan_id == plan.id)
|
|
.order_by(HostMaintenanceRun.started_at.desc()).limit(1))).scalar_one_or_none()
|
|
names = await _names(db, [h for r in latest.values() if r
|
|
for h in json.loads(r.host_ids or "[]")
|
|
+ [x["host_id"] for x in json.loads(r.skipped or "[]")]])
|
|
out = []
|
|
for plan in plans:
|
|
run = latest[plan.id]
|
|
nxt = cadence.next_slot(plan, now) if plan.enabled else None
|
|
row = {
|
|
"plan_id": plan.id, "name": plan.name,
|
|
"group": groups.get(plan.group_id, plan.group_id),
|
|
"schedule": cadence.describe(plan),
|
|
"weekday": cadence.WEEKDAYS[plan.weekday], "at_time": plan.at_time,
|
|
"time_zone": plan.time_zone, "window_minutes": plan.window_minutes,
|
|
"reboot_policy": plan.reboot_policy, "enabled": plan.enabled,
|
|
"next_slot_at": iso(nxt), "last_slot_at": iso(cadence.fired_slot(plan)),
|
|
"latest_run": None,
|
|
}
|
|
if run:
|
|
hosts = json.loads(run.host_ids or "[]")
|
|
row["latest_run"] = {
|
|
"run_id": run.id, "state": run.state, "manual": run.manual,
|
|
"hosts": [names.get(h, h) for h in hosts],
|
|
"hosts_done": min(run.position, len(hosts)),
|
|
"skipped": [{"host": names.get(x["host_id"], x["host_id"]), "reason": x["reason"]}
|
|
for x in json.loads(run.skipped or "[]")],
|
|
"detail": run.detail,
|
|
"started_at": iso(run.started_at), "finished_at": iso(run.finished_at),
|
|
}
|
|
out.append(row)
|
|
return {"plans": out, "total": len(out)}
|
|
|
|
|
|
async def get_maintenance_job(host: str = "", job: str = "", limit: int = 10) -> dict:
|
|
"""A host's maintenance jobs, newest first (its running one first), or one
|
|
job by id.
|
|
|
|
host: a host's name or id (list_hosts); job: a job id (from here, or a
|
|
plan's run). One of the two.
|
|
limit: how many of the host's jobs (default 10, max 50).
|
|
|
|
A job is one host's update and, only when the update needs it, a reboot,
|
|
with its swarm node drained first and made active again after. It first
|
|
checks what the update would change: `docker_update` is whether it
|
|
upgrades Docker (whose packages restart the daemon), and `checked` holds
|
|
{pending, docker_packages, containers}, the containers then running
|
|
outside the swarm. A swarm node is drained before an update that upgrades
|
|
Docker, else before a reboot; `drained_at` is when, and its tasks get 5
|
|
min to move (settling). `state` moves queued → [checking →] [draining →
|
|
settling →] updating → [draining → settling →] [rebooting →]
|
|
[undraining →] [recovering →] done, or failed / timed_out; a job given
|
|
up while its node may be drained goes through undraining first, and
|
|
`ends_as` is what it then finishes as. recovering
|
|
waits up to 10 min for those containers to run again; `detail` says why
|
|
a job failed, naming any that didn't. A job failed only for not coming
|
|
back into service in time has `came_back_at`, when its host was first
|
|
found back; from then it no longer flags the host. `summary` is the host page's line
|
|
("Done: 12 packages updated, rebooted"). `gives_up_at` is when an active
|
|
job's current state times out.
|
|
"""
|
|
if bool(host.strip()) == bool(job.strip()):
|
|
raise ValueError("pass either `host` (a name or id) or `job` (a job id).")
|
|
now = now_utc()
|
|
limit = clamp_limit(limit, 10, JOBS_MAX)
|
|
async with db_session() as db:
|
|
if job.strip():
|
|
row = await db.get(HostMaintenanceJob, job.strip())
|
|
if row is None:
|
|
raise ValueError(f"no maintenance job {job.strip()!r}; "
|
|
"get_maintenance_job(host=...) lists a host's jobs.")
|
|
jobs = [row]
|
|
else:
|
|
h = pick(host, (await db.execute(select(Host))).scalars(),
|
|
kind="host", list_tool="list_hosts")
|
|
jobs = list((await db.execute(
|
|
select(HostMaintenanceJob).where(HostMaintenanceJob.host_id == h.id)
|
|
.order_by(HostMaintenanceJob.state.in_(maintenance.ACTIVE).desc(),
|
|
HostMaintenanceJob.created_at.desc())
|
|
.limit(limit + 1))).scalars())
|
|
names = await _names(db, [j.host_id for j in jobs] + [j.drain_host_id for j in jobs]
|
|
+ [j.created_by for j in jobs])
|
|
rows = [_job_row(j, names, now) for j in jobs[:limit]]
|
|
return {"jobs": rows, "total": len(rows), "truncated": len(jobs) > limit}
|
|
|
|
|
|
async def get_host_senders(host: str, range: str = "1h", direction: str = "tx", limit: int = 10) -> dict:
|
|
"""Who sent (or received) a host's network traffic: its containers and NFS
|
|
mounts, ranked against the host's own total.
|
|
|
|
host: name or id (list_hosts). range: "1h" (default), "6h", "24h" or "7d"
|
|
(at most the raw window; the rollup keeps no per-sender detail).
|
|
direction: "tx" (default, what the host sent) or "rx" (what it received).
|
|
limit: senders listed (default 10, max 50).
|
|
|
|
The window is cut into about 120 buckets (`bucket_s` wide). `host` is the
|
|
host's physical interfaces: average, peak and the bucket it peaked in.
|
|
Each sender is a container (a swarm service's tasks added up, by service
|
|
name) or an NFS mount, with its average over the window, its own peak
|
|
and when, and `at_host_peak_bps`, what it sent in the host's peak bucket;
|
|
biggest there first. `unaccounted_*` is what the host carried beyond them
|
|
all: the host itself (a download, a backup run on the host, swarm ingress
|
|
passing traffic to another node). A container's traffic to another on
|
|
the same host counts for it without crossing an interface, so senders
|
|
can add up to more than the host. Per-container rates need Steward after
|
|
#6115; NFS bytes need agent 1.25.0. Bytes a second throughout.
|
|
"""
|
|
since, key = lookback(range)
|
|
limit = clamp_limit(limit, 10, 50)
|
|
async with db_session() as db:
|
|
h = pick(host, (await db.execute(select(Host))).scalars(), kind="host", list_tool="list_hosts")
|
|
got = await senders.senders(db, h.name, since, direction=direction, limit=limit)
|
|
if got["host"] is None:
|
|
raise ValueError(f"no net_{direction}_bps recorded for host {h.name!r} in the last {key}; "
|
|
"is its agent reporting? list_agents says.")
|
|
return {
|
|
"host_id": h.id, "host_name": h.name, "range": key, "direction": direction,
|
|
"since": iso(got["since"]), "bucket_s": got["bucket_s"],
|
|
"host": {"avg_bps": rnd(got["host"]["avg_bps"], 1), "peak_bps": rnd(got["host"]["peak_bps"], 1),
|
|
"peak_at": iso(got["host"]["peak_at"])},
|
|
"senders": [{"kind": r["kind"], "name": r["name"], "at_host_peak_bps": rnd(r["at_host_peak_bps"], 1),
|
|
"avg_bps": rnd(r["avg_bps"], 1), "peak_bps": rnd(r["peak_bps"], 1),
|
|
"peak_at": iso(r["peak_at"])} for r in got["senders"]],
|
|
"more": got["more"],
|
|
"unaccounted_avg_bps": rnd(got["unaccounted_avg_bps"], 1),
|
|
"unaccounted_at_peak_bps": rnd(got["unaccounted_at_peak_bps"], 1),
|
|
}
|
|
|
|
|
|
async def run_maintenance(host: str, reboot: str = "if_required") -> dict:
|
|
"""Update a host now, as the host page's Update now does: the job starts
|
|
with the agent's next post and get_maintenance_job follows it.
|
|
|
|
host: the host's name or id.
|
|
reboot: "if_required" (default) reboots only when the update says the
|
|
host needs it, draining its swarm node first and making it active after;
|
|
"never" updates and leaves any reboot for later.
|
|
|
|
Refused while the host already has a job running, or when its agent has
|
|
no maintenance helper yet (Update agent installs it). Admin only, as on
|
|
the page. Recorded in the audit log.
|
|
"""
|
|
if reboot not in ("if_required", "never"):
|
|
raise ValueError('reboot must be "if_required" or "never". Nothing was started.')
|
|
caller = current_caller()
|
|
async with db_session() as db, db.begin():
|
|
h = pick(host, (await db.execute(select(Host))).scalars(),
|
|
kind="host", list_tool="list_hosts")
|
|
reg = (await db.execute(select(HostAgentRegistration).where(
|
|
HostAgentRegistration.host_id == h.id))).scalar_one_or_none()
|
|
try:
|
|
created = await maintenance.create_job(db, h, reg, caller.user_id, reboot)
|
|
except maintenance.CannotStart as exc:
|
|
raise ValueError(f"{h.name} can't be updated now: {exc}. Nothing was started.") from None
|
|
job_id, host_id, name = created.id, h.id, h.name
|
|
await audit_mcp("host_agent.maintenance", "host", host_id,
|
|
{"job_id": job_id, "reboot_policy": reboot})
|
|
return {"job_id": job_id, "host": name, "state": "queued", "reboot_policy": reboot,
|
|
"next": f'get_maintenance_job(job="{job_id}") follows it; it starts with the '
|
|
"agent's next post."}
|
|
|
|
|
|
async def _failed_jobs(db, now) -> list:
|
|
"""Each host's latest finished job, when it failed or timed out within
|
|
FAILED_JOB_DAYS and still flags the host (a later success clears it, and
|
|
so does a late return once the host is back, #6175)."""
|
|
rows = (await db.execute(
|
|
select(HostMaintenanceJob, Host.name).join(Host, Host.id == HostMaintenanceJob.host_id)
|
|
.where(HostMaintenanceJob.state.in_(maintenance.FINISHED),
|
|
HostMaintenanceJob.finished_at >= now - timedelta(days=FAILED_JOB_DAYS))
|
|
.order_by(HostMaintenanceJob.finished_at.desc()))).all()
|
|
latest = {}
|
|
for j, name in rows:
|
|
latest.setdefault(j.host_id, (j, name))
|
|
return [(j, name) for j, name in latest.values() if maintenance.still_failed(j)]
|
|
|
|
|
|
async def mcp_overview(db) -> list[dict]:
|
|
"""Agents registered more than PENDING_GRACE ago that have never checked
|
|
in, and hosts whose latest maintenance job failed or timed out.
|
|
|
|
Agents that stopped reporting are already flagged by the core overview
|
|
from their metrics, so they are not repeated here.
|
|
"""
|
|
now = now_utc()
|
|
out = [
|
|
finding("warning", "agent_never_reported", h.name,
|
|
"host agent was registered but has never checked in; the install may have failed",
|
|
since=reg.created_at, ids={"host_id": h.id}, next="list_agents()")
|
|
for reg, h in await _agents(db)
|
|
if reg.last_seen_at is None and now - reg.created_at > PENDING_GRACE
|
|
]
|
|
out += [
|
|
finding("warning", "maintenance_failed", name,
|
|
f"maintenance {maintenance.LABELS[j.state].lower()}"
|
|
+ (f": {j.detail}" if j.detail else ""),
|
|
since=j.finished_at, ids={"host_id": j.host_id, "job_id": j.id},
|
|
next=f'get_maintenance_job(job="{j.id}")')
|
|
for j, name in await _failed_jobs(db, now)
|
|
]
|
|
return out
|
|
|
|
|
|
def register_tools(register) -> None:
|
|
register(list_agents, role=UserRole.viewer, read_only=True)
|
|
register(list_maintenance_plans, role=UserRole.viewer, read_only=True)
|
|
register(get_maintenance_job, role=UserRole.viewer, read_only=True)
|
|
register(get_host_senders, role=UserRole.viewer, read_only=True)
|
|
register(run_maintenance, role=UserRole.admin, read_only=False)
|