bvandeusenandClaude Opus 5.5
26bc57ce04
The host page gains a "Who's sending" panel, and the MCP a tool, that put the host's outbound (or inbound) traffic down to its containers and NFS mounts: read in shared time buckets, so the host's busiest bucket says who was sending in it. A swarm service's tasks add up under its name; what none of them accounts for is the host itself. Also fixes CI 9488: per-container net_rx_bps / net_tx_bps join the docker alert catalog, so a rule can catch a container flooding the uplink. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
384 lines
19 KiB
Python
384 lines
19 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, downtime, maintenance, senders
|
|
from .models import (HostAgentRegistration, HostDowntime, 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),
|
|
"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. recovering
|
|
waits up to 10 min for those containers to run again; `detail` says why
|
|
a job failed, naming any that didn't. `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 (a later success clears it)."""
|
|
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 j.state != "done"]
|
|
|
|
|
|
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
|
|
|
|
|
|
async def mcp_events(db, host_id: str, host_name: str, since, until) -> list[dict]:
|
|
"""What the host agent knows happened on a host in a window (get_anomaly):
|
|
maintenance jobs, and downtime (reboots, agent stops, lost contact)."""
|
|
from steward.mcp.tools._events import event
|
|
out = []
|
|
jobs = (await db.execute(select(HostMaintenanceJob).where(
|
|
HostMaintenanceJob.host_id == host_id, HostMaintenanceJob.created_at <= until,
|
|
(HostMaintenanceJob.finished_at.is_(None)) | (HostMaintenanceJob.finished_at >= since))
|
|
.order_by(HostMaintenanceJob.created_at).limit(200))).scalars()
|
|
for job in jobs:
|
|
out.append(event(job.created_at, "maintenance", f"Maintenance: {maintenance.summary(job)}",
|
|
until=job.finished_at, ids={"job_id": job.id}))
|
|
now = now_utc()
|
|
rows = (await db.execute(select(HostDowntime).where(
|
|
HostDowntime.host_id == host_id, HostDowntime.started_at <= until,
|
|
(HostDowntime.ended_at.is_(None)) | (HostDowntime.ended_at >= since))
|
|
.order_by(HostDowntime.started_at).limit(200))).scalars()
|
|
for row in rows:
|
|
seen = downtime.describe(row, now)
|
|
out.append(event(row.started_at, "downtime",
|
|
f"{seen['label']} ({int(seen['seconds'])} s{', still going' if seen['ongoing'] else ''})",
|
|
until=row.ended_at))
|
|
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)
|