bvandeusenandClaude Opus 5.5
6a68bd75d9
#5766 "A root helper the agent can start, for a fixed set of maintenance actions and nothing else", milestone 514. The agent runs unprivileged under NoNewPrivileges, so it can't update or reboot its host. It now leaves a request in /var/lib/steward-agent/requests, the one path its unit lets it write; steward-maint.path starts maint.py as root, which reads it without following links, accepts only update, reboot, drain or undrain, deletes it, and writes the outcome to /var/lib/steward-maint/results. The agent (1.17.0) reports each new result state once as a `maintenance` event, re-checks pending updates when an update finishes, and says in its metadata whether the helper is installed. The units and the helper are served from one copy at /plugins/host_agent/files/<name> (a fixed list), and install.sh plus the install, provision and update playbooks all fetch them, replacing three inline copies of the unit. Update agent installs the helper on existing hosts; the Agent panel says under Manage agent whether it is there. Steward handing out jobs is #5767; the cadence is #5768. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1292 lines
57 KiB
Python
1292 lines
57 KiB
Python
# plugins/host_agent/routes.py
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import hashlib
|
||
import re
|
||
import time
|
||
from datetime import datetime, timezone
|
||
from typing import Any
|
||
|
||
from pathlib import Path
|
||
|
||
import secrets
|
||
from quart import Blueprint, current_app, jsonify, request, Response, render_template, redirect, url_for, session
|
||
from steward.auth.middleware import public_route, require_role
|
||
from steward.models.users import UserRole
|
||
from sqlalchemy import select, or_, and_
|
||
from datetime import timedelta
|
||
|
||
from steward.core.formatting import worst_reading
|
||
from steward.core.health import ingest as ingest_tally
|
||
from steward.core.metric_series import points as metric_points
|
||
from steward.core.settings import get_setting, public_base_url
|
||
from steward.core.time_range import parse_range, RANGE_OPTIONS
|
||
from steward.hosts import service as host_svc
|
||
from steward.models.hosts import Host
|
||
from steward.models.metrics import PluginMetric
|
||
from .downtime import recent as recent_downtime, record_events
|
||
from .models import HostAgentRegistration
|
||
from .scheduler import arrivals
|
||
# Query helpers live in a model-free module so integration tests can import them
|
||
# without the plugin-loader double-registration guard tripping (see metrics_query).
|
||
from .metrics_query import (
|
||
SOURCE_MODULE,
|
||
_history_for_host,
|
||
_latest_metrics_for_host,
|
||
)
|
||
|
||
host_agent_bp = Blueprint("host_agent", __name__, template_folder="templates")
|
||
|
||
|
||
def _hash_token(raw: str) -> str:
|
||
return hashlib.sha256(raw.encode("utf-8")).hexdigest()
|
||
|
||
|
||
def _parse_ts(ts: str) -> datetime:
|
||
if ts.endswith("Z"):
|
||
ts = ts[:-1] + "+00:00"
|
||
return datetime.fromisoformat(ts)
|
||
|
||
|
||
def clock_offset(sent_at, received_at: datetime) -> float | None:
|
||
"""Seconds this host's clock is ahead of Steward's (negative: behind),
|
||
from the agent's sent_at and the moment Steward received the post. It
|
||
carries the request's own latency, well under a second on a LAN. None
|
||
when the post has no usable sent_at (an agent older than 1.12.0)."""
|
||
if not isinstance(sent_at, str):
|
||
return None
|
||
try:
|
||
sent = _parse_ts(sent_at)
|
||
except ValueError:
|
||
return None
|
||
if sent.tzinfo is None:
|
||
return None
|
||
return round((sent - received_at).total_seconds(), 2)
|
||
|
||
|
||
def clock_words(offset: float | None, thresholds: dict) -> tuple[str, str]:
|
||
"""('clock 0.3s ahead', 'ok') or ('clock 2m 10s behind', 'warn'): how far
|
||
a host's clock is from Steward's, graded by Settings → Thresholds' clock
|
||
drift, either way."""
|
||
from steward.core.formatting import human_duration, status_class
|
||
if offset is None:
|
||
return "", "unknown"
|
||
off = abs(offset)
|
||
words = "clock in step" if off < 1 else \
|
||
f"clock {human_duration(off)} {'ahead' if offset > 0 else 'behind'}"
|
||
return words, status_class(off, "clock", thresholds)
|
||
|
||
|
||
def pick_host_address(current: str | None, reported: str | None) -> str | None:
|
||
"""Return the address to store on the Host, or None for no change.
|
||
|
||
The agent-reported IP fills Host.address only when the current value is
|
||
blank — an admin-typed address (DNS name, management IP) is never
|
||
overwritten. The reported IP is always kept on the registration regardless,
|
||
so admins can still see drift.
|
||
"""
|
||
if reported and not (current or "").strip():
|
||
return reported
|
||
return None
|
||
|
||
|
||
# A board's Super-I/O inputs (Nuvoton/ITE: SYSTIN, CPUTIN, AUXTIN0-4, PCH_*)
|
||
# are thermistor pins that are often left unwired, and a floating one reads a
|
||
# junk constant. Wired, they read the board, which runs far below this.
|
||
_BOARD_INPUT = re.compile(r"^(SYSTIN|CPUTIN|AUXTIN\d*|PCH_)", re.IGNORECASE)
|
||
BOARD_INPUT_CEILING_C = 100
|
||
|
||
|
||
def counts_as_hottest(label: str, celsius: float) -> bool:
|
||
"""Whether a sensor's reading may set the host's temperature (#5759).
|
||
|
||
VDNT-Util01's SYSTIN read 114 °C and AUXTIN0-4 105-112, flat at every
|
||
load, while its CPU package moved 49-87 and CPUTIN read 49: unwired
|
||
inputs. A board input at or past 100 °C is one of those; below it, it
|
||
counts. No running host's sensor reads 0 or below, and chips shut down
|
||
before 125. Each sensor's own row is kept either way."""
|
||
if not 0 < celsius < 125:
|
||
return False
|
||
return not (_BOARD_INPUT.match(label) and celsius >= BOARD_INPUT_CEILING_C)
|
||
|
||
|
||
def _ts_or(value, fallback: datetime) -> datetime:
|
||
"""An agent-sent ISO time, or `fallback` when it isn't one."""
|
||
try:
|
||
got = _parse_ts(value)
|
||
except (AttributeError, TypeError, ValueError):
|
||
return fallback
|
||
return got if got.tzinfo else got.replace(tzinfo=timezone.utc)
|
||
|
||
|
||
def reboot_required(sample: dict):
|
||
"""Whether the sample says its host needs a reboot: from the reboot-required
|
||
marker (apt hosts) or the hourly update check (dnf's needs-restarting);
|
||
None when it can't say."""
|
||
reboot = sample.get("reboot") if isinstance(sample.get("reboot"), dict) else {}
|
||
if isinstance(reboot.get("required"), bool):
|
||
return reboot["required"]
|
||
upd = sample.get("updates") if isinstance(sample.get("updates"), dict) else {}
|
||
return upd["reboot_required"] if isinstance(upd.get("reboot_required"), bool) else None
|
||
|
||
|
||
def apply_updates(reg, sample: dict, now: datetime) -> bool:
|
||
"""Keep the registration's view of the host's updates and reboot current
|
||
from its newest sample (#5765); True when anything changed."""
|
||
before = (reg.updates_manager, reg.updates_pending, reg.updates_security,
|
||
reg.updates_checked_at, reg.updates_error, reg.reboot_required,
|
||
reg.reboot_required_since, reg.reboot_packages)
|
||
upd = sample.get("updates") if isinstance(sample.get("updates"), dict) else None
|
||
if upd:
|
||
reg.updates_checked_at = _ts_or(upd.get("checked_at"), now)
|
||
if upd.get("manager") in ("apt", "dnf"):
|
||
reg.updates_manager = upd["manager"]
|
||
if upd.get("error"):
|
||
reg.updates_error = str(upd["error"])[:200]
|
||
elif isinstance(upd.get("pending"), int):
|
||
reg.updates_pending = upd["pending"]
|
||
reg.updates_security = upd["security"] if isinstance(upd.get("security"), int) else None
|
||
reg.updates_error = None
|
||
required = reboot_required(sample)
|
||
if required is not None:
|
||
reg.reboot_required = required
|
||
if not required:
|
||
reg.reboot_required_since, reg.reboot_packages = None, None
|
||
else:
|
||
reboot = sample.get("reboot") if isinstance(sample.get("reboot"), dict) else {}
|
||
if reboot.get("since") is not None:
|
||
reg.reboot_required_since = _ts_or(reboot["since"], now)
|
||
elif reg.reboot_required_since is None:
|
||
reg.reboot_required_since = now
|
||
packages = reboot.get("packages")
|
||
if isinstance(packages, list):
|
||
reg.reboot_packages = ", ".join(str(p) for p in packages[:20])[:2000] or None
|
||
return before != (reg.updates_manager, reg.updates_pending, reg.updates_security,
|
||
reg.updates_checked_at, reg.updates_error, reg.reboot_required,
|
||
reg.reboot_required_since, reg.reboot_packages)
|
||
|
||
|
||
def _expand_sample_to_metrics(
|
||
sample: dict, host_name: str, recorded_at: datetime
|
||
) -> list[PluginMetric]:
|
||
rows: list[PluginMetric] = []
|
||
|
||
def row(metric: str, resource: str, value: float) -> None:
|
||
rows.append(PluginMetric(
|
||
source_module=SOURCE_MODULE,
|
||
resource_name=resource,
|
||
metric_name=metric,
|
||
value=float(value),
|
||
recorded_at=recorded_at,
|
||
))
|
||
|
||
if sample.get("cpu_pct") is not None:
|
||
row("cpu_pct", host_name, sample["cpu_pct"])
|
||
|
||
mem = sample.get("mem") or {}
|
||
if mem.get("total_bytes"):
|
||
total = mem["total_bytes"]
|
||
available = mem.get("available_bytes", 0)
|
||
used_pct = 100.0 * (total - available) / total if total else 0.0
|
||
row("mem_used_pct", host_name, used_pct)
|
||
row("mem_available_bytes", host_name, available)
|
||
row("swap_used_bytes", host_name, mem.get("swap_used_bytes", 0))
|
||
|
||
load = sample.get("load") or {}
|
||
for key in ("1m", "5m", "15m"):
|
||
if key in load:
|
||
row(f"load_{key}", host_name, load[key])
|
||
# Load per core as a %, the figure the pages grade, so it can learn a
|
||
# normal (#5742): a load of 6 is idle on 32 cores and busy on 4.
|
||
if "1m" in load and sample.get("cpu_cores"):
|
||
row("load_pct", host_name, 100.0 * load["1m"] / len(sample["cpu_cores"]))
|
||
|
||
if sample.get("uptime_secs") is not None:
|
||
row("uptime_secs", host_name, sample["uptime_secs"])
|
||
|
||
# What the host has waiting (#5765): the agent's hourly answer rides
|
||
# every sample, so a rule on it reads like any other host metric.
|
||
upd = sample.get("updates") if isinstance(sample.get("updates"), dict) else {}
|
||
if isinstance(upd.get("pending"), int):
|
||
row("updates_pending", host_name, upd["pending"])
|
||
if isinstance(upd.get("security"), int):
|
||
row("updates_security", host_name, upd["security"])
|
||
required = reboot_required(sample)
|
||
if required is not None:
|
||
row("reboot_required", host_name, 1 if required else 0)
|
||
|
||
# The agent's own footprint (#5764), so a runaway agent shows before it
|
||
# goes silent.
|
||
me = sample.get("agent") or {}
|
||
if me.get("rss_bytes") is not None:
|
||
row("agent_rss_bytes", host_name, me["rss_bytes"])
|
||
if me.get("cpu_pct") is not None:
|
||
row("agent_cpu_pct", host_name, me["cpu_pct"])
|
||
|
||
# The fullest disk is the fullest by space or by inodes (#5745): out of
|
||
# inodes, every file create fails with space to spare, so the disk rules
|
||
# and the overview, which read disk_used_pct_worst, see either.
|
||
worst_pct = 0.0
|
||
stalled = 0
|
||
for disk in sample.get("storage") or []:
|
||
# A mount that didn't answer in time has no figures (#5760); how many
|
||
# did is its own metric, so a rule can say "an NFS share stopped".
|
||
if disk.get("stalled"):
|
||
stalled += 1
|
||
continue
|
||
total = disk.get("total_bytes", 0)
|
||
used = disk.get("used_bytes", 0)
|
||
pct = 100.0 * used / total if total else 0.0
|
||
mount_res = f"{host_name}:{disk['mount']}"
|
||
row("disk_used_pct", mount_res, pct)
|
||
row("disk_used_bytes", mount_res, used)
|
||
row("disk_total_bytes", mount_res, total)
|
||
if disk.get("inodes_total"):
|
||
inode_pct = 100.0 * disk.get("inodes_used", 0) / disk["inodes_total"]
|
||
row("inodes_used_pct", mount_res, inode_pct)
|
||
pct = max(pct, inode_pct)
|
||
if pct > worst_pct:
|
||
worst_pct = pct
|
||
if sample.get("storage"):
|
||
# Every mount stalled says nothing about how full they are.
|
||
if stalled < len(sample["storage"]):
|
||
row("disk_used_pct_worst", host_name, worst_pct)
|
||
row("mounts_stalled", host_name, stalled)
|
||
|
||
# Per-core CPU. Sub-resources carry a ':' so the fleet widget filters them out.
|
||
for i, core_pct in enumerate(sample.get("cpu_cores") or []):
|
||
if core_pct is not None:
|
||
row("cpu_pct", f"{host_name}:core{i}", core_pct)
|
||
|
||
# Richer memory breakdown.
|
||
if mem.get("cached_bytes") is not None:
|
||
row("mem_cached_bytes", host_name, mem["cached_bytes"])
|
||
if mem.get("buffers_bytes") is not None:
|
||
row("mem_buffers_bytes", host_name, mem["buffers_bytes"])
|
||
|
||
# Network throughput — per interface plus a host-level total for the fleet view.
|
||
net_rx_total = net_tx_total = 0.0
|
||
for iface in sample.get("net") or []:
|
||
rx, tx = iface.get("rx_bps", 0.0), iface.get("tx_bps", 0.0)
|
||
res = f"{host_name}:net:{iface['iface']}"
|
||
row("net_rx_bps", res, rx)
|
||
row("net_tx_bps", res, tx)
|
||
net_rx_total += rx
|
||
net_tx_total += tx
|
||
if sample.get("net"):
|
||
row("net_rx_bps", host_name, net_rx_total)
|
||
row("net_tx_bps", host_name, net_tx_total)
|
||
|
||
# Disk I/O — per device plus host-level total.
|
||
dr_total = dw_total = 0.0
|
||
for dev in sample.get("diskio") or []:
|
||
rd, wr = dev.get("read_bps", 0.0), dev.get("write_bps", 0.0)
|
||
res = f"{host_name}:diskio:{dev['device']}"
|
||
row("disk_read_bps", res, rd)
|
||
row("disk_write_bps", res, wr)
|
||
dr_total += rd
|
||
dw_total += wr
|
||
if sample.get("diskio"):
|
||
row("disk_read_bps", host_name, dr_total)
|
||
row("disk_write_bps", host_name, dw_total)
|
||
|
||
# Temperatures — per sensor plus the host max for at-a-glance. The max
|
||
# counts only readings that can be real (#5759).
|
||
max_temp: float | None = None
|
||
for t in sample.get("temps") or []:
|
||
c = t.get("celsius")
|
||
if c is None:
|
||
continue
|
||
row("temp_c", f"{host_name}:temp:{t['label']}", c)
|
||
if counts_as_hottest(str(t["label"]), c) and (max_temp is None or c > max_temp):
|
||
max_temp = c
|
||
if max_temp is not None:
|
||
row("temp_c_max", host_name, max_temp)
|
||
|
||
# Pressure stall information (PSI) — host-level gauges (mem/cpu/io some|full).
|
||
for k, v in (sample.get("psi") or {}).items():
|
||
if isinstance(v, (int, float)):
|
||
row(f"psi_{k}", host_name, v)
|
||
|
||
return rows
|
||
|
||
|
||
def _error(status: int, code: str, detail: str | None = None) -> tuple[Any, int]:
|
||
body: dict = {"ok": False, "error": code}
|
||
if detail:
|
||
body["detail"] = detail
|
||
return jsonify(body), status
|
||
|
||
|
||
@host_agent_bp.post("/ingest")
|
||
@public_route("the agent authenticates with its bearer token")
|
||
async def ingest():
|
||
# Taken first, so the clock offset carries as little of this handler as it can.
|
||
received_at = datetime.now(timezone.utc)
|
||
auth = request.headers.get("Authorization", "")
|
||
if not auth.startswith("Bearer "):
|
||
return _error(401, "invalid_token")
|
||
raw = auth[len("Bearer "):].strip()
|
||
if not raw:
|
||
return _error(401, "invalid_token")
|
||
|
||
try:
|
||
payload = await request.get_json(force=True)
|
||
except Exception:
|
||
return _error(400, "malformed_payload", "invalid JSON")
|
||
if not isinstance(payload, dict) or "samples" not in payload:
|
||
return _error(400, "malformed_payload", "missing 'samples'")
|
||
|
||
samples = payload.get("samples") or []
|
||
# What happened to the host (agent ≥ 1.14.0, #5756). A stopping agent's last
|
||
# post may carry its events alone, with no sample left to send.
|
||
events = payload.get("events") if isinstance(payload.get("events"), list) else []
|
||
if not isinstance(samples, list) or not (samples or events):
|
||
return _error(400, "malformed_payload", "'samples' must be a non-empty list")
|
||
|
||
token_hash = _hash_token(raw)
|
||
phases = _Phases()
|
||
samples, dropped = newest_samples(samples, MAX_SAMPLES_PER_INGEST)
|
||
if dropped:
|
||
current_app.logger.warning(
|
||
"host_agent ingest: kept the newest %d of %d samples for token=%s…",
|
||
len(samples), len(samples) + dropped, token_hash[:8])
|
||
try:
|
||
async with current_app.db_sessionmaker() as session:
|
||
reg = (await session.execute(
|
||
select(HostAgentRegistration).where(
|
||
HostAgentRegistration.token_hash == token_hash)
|
||
)).scalar_one_or_none()
|
||
if reg is None:
|
||
return _error(401, "invalid_token")
|
||
host = (await session.execute(
|
||
select(Host).where(Host.id == reg.host_id)
|
||
)).scalar_one_or_none()
|
||
if host is None:
|
||
return _error(401, "invalid_token")
|
||
# The agent reached us: that holds even if storing its post fails (#5514).
|
||
arrivals(current_app)[reg.host_id] = received_at
|
||
|
||
accepted = 0
|
||
metric_rows: list[PluginMetric] = []
|
||
latest_ts: datetime | None = None
|
||
docker_snapshots: list[tuple[datetime, list]] = []
|
||
docker_log_batches: list[tuple[datetime, list]] = []
|
||
latest_swarm: dict | None = None
|
||
latest_swarm_ts: datetime | None = None
|
||
latest_disk: dict | None = None
|
||
latest_disk_ts: datetime | None = None
|
||
# Whether Docker answered, from the newest sample that says (agent ≥ 1.9.0).
|
||
docker_status: dict | None = None
|
||
newest_sample: dict | None = None
|
||
for sample in samples:
|
||
try:
|
||
recorded_at = _parse_ts(sample["ts"])
|
||
except (KeyError, ValueError):
|
||
continue
|
||
metric_rows.extend(_expand_sample_to_metrics(sample, host.name, recorded_at))
|
||
# An empty list is a real answer: Docker is up with no containers.
|
||
docker = sample.get("docker")
|
||
if isinstance(docker, list):
|
||
docker_snapshots.append((recorded_at, docker))
|
||
status = sample.get("docker_status")
|
||
if status in ("ok", "unavailable", "error") and (
|
||
docker_status is None or recorded_at > docker_status["at"]):
|
||
docker_status = {
|
||
"status": status, "detail": sample.get("docker_detail"),
|
||
"at": recorded_at, "swarm_manager": sample.get("swarm_manager")}
|
||
# Container logs are time-series (append every line, not newest-only);
|
||
# each record carries its own Docker ts, recorded_at is the fallback.
|
||
docker_logs = sample.get("docker_logs")
|
||
if isinstance(docker_logs, list) and docker_logs:
|
||
docker_log_batches.append((recorded_at, docker_logs))
|
||
# Swarm is current-state, not time-series — keep only the newest
|
||
# sample's topology (a manager re-reports it every interval).
|
||
swarm = sample.get("swarm")
|
||
if isinstance(swarm, dict) and (
|
||
latest_swarm_ts is None or recorded_at > latest_swarm_ts):
|
||
latest_swarm, latest_swarm_ts = swarm, recorded_at
|
||
# Disk usage is current-state too — keep only the newest sample's.
|
||
disk = sample.get("docker_disk")
|
||
if isinstance(disk, dict) and (
|
||
latest_disk_ts is None or recorded_at > latest_disk_ts):
|
||
latest_disk, latest_disk_ts = disk, recorded_at
|
||
accepted += 1
|
||
if latest_ts is None or recorded_at > latest_ts:
|
||
latest_ts, newest_sample = recorded_at, sample
|
||
|
||
if accepted == 0 and not events:
|
||
await session.rollback()
|
||
return _error(400, "malformed_payload", "no valid samples")
|
||
# How far this host's clock is from Steward's, either way, so an
|
||
# alert rule reads "> 30" (milestone 490).
|
||
offset = clock_offset(payload.get("sent_at"), received_at)
|
||
if offset is not None:
|
||
metric_rows.append(PluginMetric(
|
||
source_module=SOURCE_MODULE, resource_name=host.name,
|
||
metric_name="clock_off_secs", value=abs(offset), recorded_at=received_at))
|
||
phases.mark("parse")
|
||
|
||
# Through the alert pipeline, so rules on host metrics can fire.
|
||
from steward.core.alerts import record_metrics
|
||
phases.size = f"samples={accepted} metric_rows={len(metric_rows)}"
|
||
await record_metrics(session, metric_rows, phases)
|
||
|
||
# Hand host-scoped container data to the docker plugin if it's enabled
|
||
# (opportunistic synergy via the capability registry — no hard import,
|
||
# no-op when docker is disabled). A failure here must never sink the
|
||
# whole ingest, so the metrics above still land.
|
||
if (docker_snapshots or latest_swarm is not None or docker_status is not None
|
||
or latest_disk is not None or docker_log_batches):
|
||
from steward.core.capabilities import has_capability, invoke_capability
|
||
if has_capability("docker.persist_host_samples"):
|
||
try:
|
||
# SAVEPOINT so a docker-side failure rolls back only the
|
||
# docker writes, leaving the host metrics intact to commit.
|
||
async with session.begin_nested():
|
||
await invoke_capability(
|
||
"docker.persist_host_samples", UserRole.admin,
|
||
session, host, docker_snapshots, latest_swarm,
|
||
latest_disk, docker_log_batches, docker_status,
|
||
phases=phases,
|
||
)
|
||
except Exception:
|
||
current_app.logger.exception(
|
||
"host_agent ingest: docker persist failed for host=%s",
|
||
host.name)
|
||
phases.mark("docker")
|
||
|
||
# Before last_seen_at moves: a boot after it that nobody announced
|
||
# was a restart without a shutdown.
|
||
if events:
|
||
await record_events(session, host.id, events, reg.last_seen_at)
|
||
|
||
md = payload.get("metadata") or {}
|
||
changed = False
|
||
if latest_ts and (reg.last_seen_at is None or latest_ts > reg.last_seen_at):
|
||
reg.last_seen_at = latest_ts
|
||
changed = True
|
||
version = payload.get("agent_version")
|
||
if version and reg.agent_version != version:
|
||
reg.agent_version = version
|
||
changed = True
|
||
for field in ("kernel", "distro", "arch"):
|
||
if field in md and getattr(reg, field) != md[field]:
|
||
setattr(reg, field, md[field])
|
||
changed = True
|
||
# Whether the maintenance helper is installed (agent ≥ 1.17.0, #5766).
|
||
if isinstance(md.get("maint_helper"), bool) and reg.maint_helper != md["maint_helper"]:
|
||
reg.maint_helper = md["maint_helper"]
|
||
changed = True
|
||
host_ip = md.get("host_ip")
|
||
if host_ip and reg.host_ip != host_ip:
|
||
reg.host_ip = host_ip
|
||
changed = True
|
||
if newest_sample is not None and apply_updates(reg, newest_sample, received_at):
|
||
changed = True
|
||
# Mirror the reported IP into Host.address only when it's blank.
|
||
new_address = pick_host_address(host.address, host_ip)
|
||
if new_address is not None:
|
||
host.address = new_address
|
||
if changed:
|
||
reg.updated_at = datetime.now(timezone.utc)
|
||
|
||
if offset is not None:
|
||
reg.clock_offset_s, reg.clock_checked_at = offset, received_at
|
||
clock_th = current_app.config.get("THRESHOLDS", {}).get("clock") or {"warn": 30}
|
||
if abs(offset) > clock_th["warn"]:
|
||
current_app.logger.warning(
|
||
"host_agent ingest: host=%s clock is %.1fs %s Steward's",
|
||
host.name, abs(offset), "ahead of" if offset > 0 else "behind")
|
||
|
||
await session.commit()
|
||
phases.mark("commit")
|
||
except asyncio.CancelledError:
|
||
# The agent hung up (its 10 s timeout) and the server dropped the
|
||
# request: say how far it got before re-raising.
|
||
phases.cancelled(token_hash[:8])
|
||
ingest_tally.done(phases.elapsed_ms(), ok=False)
|
||
raise
|
||
except Exception:
|
||
ingest_tally.done(phases.elapsed_ms(), ok=False)
|
||
raise
|
||
# Counted for Steward's own health (#5467). The early returns above (a
|
||
# bad token, no valid samples) answer the agent and aren't counted.
|
||
ingest_tally.done(phases.elapsed_ms())
|
||
|
||
phases.report(host.name, accepted, len(metric_rows),
|
||
sum(len(c) for _, c in docker_snapshots),
|
||
sum(len(lines) for _, lines in docker_log_batches))
|
||
return jsonify({"ok": True, "samples_accepted": accepted, "samples_dropped": dropped}), 200
|
||
|
||
|
||
# The agent gives up on a post after 10 s (agent.py); an ingest past this
|
||
# says where its time went, so a slow phase shows in the logs (#5149).
|
||
SLOW_INGEST_SECONDS = 2.0
|
||
|
||
# How many samples one ingest writes (#5149). An agent that could not post
|
||
# sends its whole buffer (up to 21 samples, ~4,000 metric rows); on the live
|
||
# instance that did not finish inside the agent's 10 s, the rollback left the
|
||
# buffer full, and every later post failed the same way. The newest samples
|
||
# are kept: the older ones would be lost to the agent's ring buffer anyway.
|
||
# Five matches the agent's MAX_SAMPLES_PER_POST (1.10+), so its backlog posts
|
||
# land whole: ~1,000 metric rows at the 1,300-4,000 rows/s measured once
|
||
# plugin_metrics got time-ordered ids, with posts answering in 0.4-1.3 s (#5181).
|
||
MAX_SAMPLES_PER_INGEST = 5
|
||
|
||
|
||
def newest_samples(samples: list, limit: int) -> tuple[list, int]:
|
||
"""(the newest `limit` samples, oldest first; how many were left out).
|
||
|
||
Ordered by their ISO-8601 `ts` (the agent writes them all in one form);
|
||
one without a ts sorts oldest and is rejected later as malformed."""
|
||
if len(samples) <= limit:
|
||
return samples, 0
|
||
ordered = sorted(samples, key=lambda s: str(s.get("ts") or "") if isinstance(s, dict) else "")
|
||
return ordered[-limit:], len(samples) - limit
|
||
|
||
|
||
class _Phases:
|
||
"""Wall-clock per phase of one ingest request."""
|
||
|
||
def __init__(self):
|
||
self.start = self.last = time.monotonic()
|
||
self.took: dict[str, float] = {}
|
||
self.size = ""
|
||
|
||
def elapsed_ms(self) -> float:
|
||
return (time.monotonic() - self.start) * 1000.0
|
||
|
||
def mark(self, name: str) -> None:
|
||
now = time.monotonic()
|
||
self.took[name] = self.took.get(name, 0.0) + (now - self.last)
|
||
self.last = now
|
||
|
||
def cancelled(self, who: str) -> None:
|
||
current_app.logger.warning(
|
||
"host_agent ingest: dropped after %.1fs for token=%s… (completed: %s) %s",
|
||
time.monotonic() - self.start, who,
|
||
" ".join(f"{k}={v:.1f}s" for k, v in self.took.items()) or "nothing", self.size)
|
||
|
||
def report(self, host: str, samples: int, metrics: int, containers: int, log_lines: int) -> None:
|
||
total = time.monotonic() - self.start
|
||
if total < SLOW_INGEST_SECONDS:
|
||
return
|
||
current_app.logger.warning(
|
||
"host_agent ingest: slow %.1fs for host=%s (%s) samples=%d metric_rows=%d "
|
||
"containers=%d log_lines=%d", total, host,
|
||
" ".join(f"{k}={v:.1f}s" for k, v in self.took.items()),
|
||
samples, metrics, containers, log_lines)
|
||
|
||
|
||
AGENT_SOURCE_PATH = Path(__file__).parent / "agent.py"
|
||
|
||
|
||
def _agent_version() -> str:
|
||
try:
|
||
for line in AGENT_SOURCE_PATH.read_text().splitlines():
|
||
if line.startswith("AGENT_VERSION"):
|
||
return line.split("=", 1)[1].strip().strip('"').strip("'")
|
||
except OSError:
|
||
pass
|
||
return "unknown"
|
||
|
||
|
||
@host_agent_bp.get("/install.sh")
|
||
@public_route("fetched by curl on the host being enrolled; carries an enrolment token")
|
||
async def install_script():
|
||
token = request.args.get("token", "")
|
||
if not token:
|
||
return _error(401, "invalid_token")
|
||
async with current_app.db_sessionmaker() as session:
|
||
reg = (await session.execute(
|
||
select(HostAgentRegistration).where(
|
||
HostAgentRegistration.token_hash == _hash_token(token))
|
||
)).scalar_one_or_none()
|
||
if reg is None:
|
||
return _error(401, "invalid_token")
|
||
host = (await session.execute(
|
||
select(Host).where(Host.id == reg.host_id)
|
||
)).scalar_one_or_none()
|
||
if host is None:
|
||
return _error(401, "invalid_token")
|
||
|
||
url = public_base_url(request)
|
||
|
||
rendered = await render_template(
|
||
"install.sh.j2",
|
||
url=url,
|
||
token=token,
|
||
agent_version=_agent_version(),
|
||
host_name=host.name,
|
||
host_address=host.address,
|
||
)
|
||
return Response(rendered, mimetype="text/plain")
|
||
|
||
|
||
@host_agent_bp.get("/agent.py")
|
||
@public_route("fetched by the installer and by agent self-update; holds no secrets")
|
||
async def agent_source():
|
||
try:
|
||
body = AGENT_SOURCE_PATH.read_text(encoding="utf-8")
|
||
except OSError:
|
||
return _error(500, "agent_missing")
|
||
return Response(body, mimetype="text/x-python")
|
||
|
||
|
||
# What a host installs besides agent.py: the units and the maintenance helper
|
||
# (#5766), served from one copy so install.sh and every playbook agree.
|
||
AGENT_FILES_DIR = Path(__file__).parent / "files"
|
||
AGENT_FILES = {
|
||
"maint.py": "text/x-python",
|
||
"steward-agent.service": "text/plain",
|
||
"steward-maint.service": "text/plain",
|
||
"steward-maint.path": "text/plain",
|
||
}
|
||
|
||
|
||
@host_agent_bp.get("/files/<name>")
|
||
@public_route("fetched by the installer and the agent playbooks; holds no secrets")
|
||
async def agent_file(name: str):
|
||
if name not in AGENT_FILES:
|
||
return _error(404, "not_found")
|
||
try:
|
||
body = (AGENT_FILES_DIR / name).read_text(encoding="utf-8")
|
||
except OSError:
|
||
return _error(500, "agent_missing")
|
||
return Response(body, mimetype=AGENT_FILES[name])
|
||
|
||
|
||
_FLEET_ORDER = {"stale": 0, "pending": 1, "up": 2}
|
||
|
||
|
||
def fleet_state(reg, now: datetime, stale_after: int) -> str:
|
||
"""An agent as the fleet table shows it: "up" while it reports, "stale"
|
||
once its last report is older than stale_after, "pending" if it never has."""
|
||
if reg.last_seen_at is None:
|
||
return "pending"
|
||
return "stale" if (now - reg.last_seen_at).total_seconds() > stale_after else "up"
|
||
|
||
|
||
def fleet_rows(regs, hosts_by_id: dict, now: datetime, stale_after: int) -> list[dict]:
|
||
"""The fleet table: silent agents first, then never-reported, then live,
|
||
each by host name."""
|
||
rows = []
|
||
for r in regs:
|
||
host = hosts_by_id.get(r.host_id)
|
||
rows.append({"reg": r, "host": host, "name": host.name if host else r.host_id,
|
||
"state": fleet_state(r, now, stale_after)})
|
||
return sorted(rows, key=lambda x: (_FLEET_ORDER[x["state"]], x["name"].lower()))
|
||
|
||
|
||
def _stale_after_seconds() -> int:
|
||
return int(
|
||
current_app.config.get("PLUGINS", {})
|
||
.get(SOURCE_MODULE, {})
|
||
.get("stale_after_seconds", 180)
|
||
)
|
||
|
||
|
||
@host_agent_bp.get("/widget/history")
|
||
@require_role(UserRole.viewer)
|
||
async def widget_history():
|
||
host_id = request.args.get("host_id", "")
|
||
try:
|
||
hours = max(1, min(168, int(request.args.get("hours", "6"))))
|
||
except ValueError:
|
||
hours = 6
|
||
# wid (the dashboard widget row id) keeps the <canvas> id unique when several
|
||
# history widgets share a page; fall back to the host id.
|
||
wid = request.args.get("wid", "") or host_id
|
||
cutoff = datetime.now(timezone.utc) - timedelta(hours=hours)
|
||
|
||
series: dict[str, list[list]] = {"cpu_pct": [], "mem_used_pct": [], "disk_root": []}
|
||
host = None
|
||
if host_id:
|
||
async with current_app.db_sessionmaker() as session:
|
||
host = (await session.execute(
|
||
select(Host).where(Host.id == host_id))).scalar_one_or_none()
|
||
if host is not None:
|
||
points = (await session.execute(
|
||
select(PluginMetric).where(
|
||
PluginMetric.source_module == SOURCE_MODULE,
|
||
PluginMetric.recorded_at >= cutoff,
|
||
or_(
|
||
and_(PluginMetric.resource_name == host.name,
|
||
PluginMetric.metric_name.in_(("cpu_pct", "mem_used_pct"))),
|
||
and_(PluginMetric.resource_name == host.name + ":/",
|
||
PluginMetric.metric_name == "disk_used_pct"),
|
||
),
|
||
).order_by(PluginMetric.recorded_at)
|
||
)).scalars().all()
|
||
# Disk = root (/), consistent with the host panel — not "worst".
|
||
# Epoch-ms x values let the chart use a plain linear axis (no
|
||
# Chart.js date adapter needed), matching the host-detail charts.
|
||
for p in points:
|
||
key = "disk_root" if p.resource_name != host.name else p.metric_name
|
||
series[key].append([int(p.recorded_at.timestamp() * 1000), round(p.value, 2)])
|
||
|
||
# Downsample: the agent reports every few seconds, so a multi-hour window is
|
||
# hundreds–thousands of points — a noisy, unreadable line. Bucket-average to a
|
||
# readable point count, keeping the shape.
|
||
series = {k: _downsample(v) for k, v in series.items()}
|
||
|
||
# Stable y-axis ceiling. All three series are percentages, so snap the top up
|
||
# to the next 10% band (floor 20, cap 100). Auto-scaling to the exact peak made
|
||
# the chart "zoom" jump on every poll as the peak drifted a few %; snapping to
|
||
# a band keeps the axis steady across refreshes while still filling the panel.
|
||
peak = max((p[1] for s in series.values() for p in s), default=0.0)
|
||
y_max = min(100, max(20, ((int(peak) // 10) + 1) * 10))
|
||
return await render_template(
|
||
"widget_history.html", host=host, series=series, hours=hours, wid=wid,
|
||
y_max=y_max,
|
||
)
|
||
|
||
|
||
def _downsample(series: list[list], target: int = 120) -> list[list]:
|
||
"""Bucket-average a time series [[epoch_ms, value], …] down to ~target points.
|
||
|
||
Consecutive samples are grouped into equal buckets and averaged (x = bucket
|
||
midpoint). Cheap, order-preserving, and smooths the high-frequency noise that
|
||
makes dense agent series hard to read.
|
||
"""
|
||
n = len(series)
|
||
if n <= target:
|
||
return series
|
||
bucket = (n + target - 1) // target
|
||
out: list[list] = []
|
||
for i in range(0, n, bucket):
|
||
chunk = series[i:i + bucket]
|
||
mid = chunk[len(chunk) // 2][0]
|
||
avg = sum(p[1] for p in chunk) / len(chunk)
|
||
out.append([mid, round(avg, 2)])
|
||
return out
|
||
|
||
|
||
# Full-metrics refresh cadences. The agent pushes every ~30s by default, so
|
||
# polling the current snapshot at 15s keeps it ≤ one cadence stale; the history
|
||
# charts move slowly over a multi-hour/day window, so they refresh less often.
|
||
_DETAIL_POLL_SECONDS = 15
|
||
_CHART_POLL_SECONDS = 60
|
||
|
||
|
||
def _split_host_metrics(host_name: str, latest: dict) -> tuple:
|
||
"""Split a host's latest snapshot into
|
||
(hostlvl, cores, nets, disks_io, temps, mounts) for the current-state fragment."""
|
||
hostlvl = latest.get(host_name, {})
|
||
cores: list = []
|
||
nets: dict = {}
|
||
disks_io: dict = {}
|
||
temps: dict = {}
|
||
mounts: dict = {}
|
||
plen = len(host_name) + 1
|
||
for res, metrics in latest.items():
|
||
if res == host_name:
|
||
continue
|
||
suffix = res[plen:]
|
||
if suffix.startswith("core"):
|
||
try:
|
||
idx = int(suffix[4:])
|
||
except ValueError:
|
||
idx = 0
|
||
cores.append((idx, metrics.get("cpu_pct")))
|
||
elif suffix.startswith("net:"):
|
||
nets[suffix[len("net:"):]] = metrics
|
||
elif suffix.startswith("diskio:"):
|
||
disks_io[suffix[len("diskio:"):]] = metrics
|
||
elif suffix.startswith("temp:"):
|
||
temps[suffix[len("temp:"):]] = metrics.get("temp_c")
|
||
else:
|
||
mounts[suffix] = metrics
|
||
cores.sort(key=lambda c: c[0])
|
||
return hostlvl, cores, nets, disks_io, temps, mounts
|
||
|
||
|
||
@host_agent_bp.get("/<host_id>/")
|
||
@require_role(UserRole.viewer)
|
||
async def host_detail(host_id: str):
|
||
"""Full-metrics shell: header + range toggle only. The current-state and
|
||
history sections stream in via HTMX (host_detail_metrics / host_detail_charts)
|
||
so the heavy history query never blocks first paint and the data live-updates."""
|
||
_, range_key = parse_range(request.args.get("range"))
|
||
async with current_app.db_sessionmaker() as session:
|
||
host = (await session.execute(
|
||
select(Host).where(Host.id == host_id))).scalar_one_or_none()
|
||
if host is None:
|
||
return _error(404, "not_found")
|
||
return await render_template(
|
||
"host_detail.html",
|
||
host=host, current_range=range_key, range_options=RANGE_OPTIONS,
|
||
poll_seconds=_DETAIL_POLL_SECONDS, chart_poll_seconds=_CHART_POLL_SECONDS,
|
||
)
|
||
|
||
|
||
@host_agent_bp.get("/<host_id>/metrics")
|
||
@require_role(UserRole.viewer)
|
||
async def host_detail_metrics(host_id: str):
|
||
"""Current-state fragment (gauges/cores/filesystems/IO/temps) — polled live so
|
||
the page shows the latest snapshot the server holds without a full reload."""
|
||
async with current_app.db_sessionmaker() as session:
|
||
host = (await session.execute(
|
||
select(Host).where(Host.id == host_id))).scalar_one_or_none()
|
||
if host is None:
|
||
return "", 404
|
||
reg = (await session.execute(select(HostAgentRegistration).where(
|
||
HostAgentRegistration.host_id == host_id))).scalar_one_or_none()
|
||
latest = await _latest_metrics_for_host(session, host.name)
|
||
hostlvl, cores, nets, disks_io, temps, mounts = _split_host_metrics(host.name, latest)
|
||
ls = reg.last_seen_at if reg else None
|
||
stale = ls is None or (datetime.now(timezone.utc) - ls).total_seconds() > _stale_after_seconds()
|
||
return await render_template(
|
||
"_host_metrics.html",
|
||
host=host, reg=reg, stale=stale, hostlvl=hostlvl,
|
||
cores=cores, nets=nets, disks_io=disks_io, temps=temps, mounts=mounts,
|
||
)
|
||
|
||
|
||
@host_agent_bp.get("/<host_id>/charts")
|
||
@require_role(UserRole.viewer)
|
||
async def host_detail_charts(host_id: str):
|
||
"""History-charts fragment — lazy-loaded so the date_bin query never blocks
|
||
the page shell; refreshed on a slow cadence and on range change."""
|
||
since, range_key = parse_range(request.args.get("range"))
|
||
async with current_app.db_sessionmaker() as session:
|
||
host = (await session.execute(
|
||
select(Host).where(Host.id == host_id))).scalar_one_or_none()
|
||
if host is None:
|
||
return "", 404
|
||
raw_days = int(await get_setting(session, "metrics.retention.raw_days") or 7)
|
||
series = await _history_for_host(session, host.name, since, raw_days=raw_days)
|
||
return await render_template("_host_charts.html", series=series, range_key=range_key)
|
||
|
||
|
||
def _new_token_pair() -> tuple[str, str]:
|
||
raw = secrets.token_urlsafe(32)
|
||
return raw, _hash_token(raw)
|
||
|
||
|
||
@host_agent_bp.get("/vitals/<host_id>")
|
||
@require_role(UserRole.viewer)
|
||
async def host_vitals(host_id: str):
|
||
"""Compact live vitals strip (CPU/MEM/DISK/LOAD + pressure) for the Hosts hub.
|
||
|
||
Self-contained fragment, polled by hosts/detail.html. Renders nothing until
|
||
the agent reports — the Agent panel below then carries provisioning.
|
||
"""
|
||
async with current_app.db_sessionmaker() as db:
|
||
host = (await db.execute(
|
||
select(Host).where(Host.id == host_id))).scalar_one_or_none()
|
||
if host is None:
|
||
return "", 404
|
||
reg = (await db.execute(select(HostAgentRegistration).where(
|
||
HostAgentRegistration.host_id == host_id))).scalar_one_or_none()
|
||
latest = await _latest_metrics_for_host(db, host.name) if reg else {}
|
||
norms = await host_svc.host_norms(db, host.name) if reg else {}
|
||
|
||
# The last 6 h for the at-a-glance sparklines (cpu/mem/load host-level,
|
||
# disk from the root mount), averaged into ~120 points in SQL, so the
|
||
# line spans the whole window and only what is drawn is fetched.
|
||
series_raw: dict[str, list[float]] = {"cpu": [], "mem": [], "disk": [], "load": []}
|
||
if reg:
|
||
since = datetime.now(timezone.utc) - timedelta(hours=6)
|
||
for key, resource, metric in (("cpu", host.name, "cpu_pct"),
|
||
("mem", host.name, "mem_used_pct"),
|
||
("load", host.name, "load_1m"),
|
||
("disk", host.name + ":/", "disk_used_pct")):
|
||
series_raw[key] = [v for _t, v in await metric_points(
|
||
db, SOURCE_MODULE, resource, metric, since, target=120)]
|
||
|
||
from steward.core.status import sparkline_svg
|
||
hostlvl = latest.get(host.name, {})
|
||
# At-a-glance disk = the root filesystem (what people actually care about),
|
||
# not the "worst" mount which is opaque. Worst stays on the full-metrics page.
|
||
disk_root = (latest.get(host.name + ":/", {}) or {}).get("disk_used_pct")
|
||
# Out of inodes, "/" is full with space to spare (#5745): named when worse.
|
||
inodes_root = (latest.get(host.name + ":/", {}) or {}).get("inodes_used_pct")
|
||
sparks = {k: sparkline_svg(v, height=24, fill_width=True) for k, v in series_raw.items()}
|
||
# Normalize load by core count so it's comparable across hardware (load/core
|
||
# as %). Cores derived from the per-core CPU sub-resources already collected.
|
||
cores = sum(1 for k in latest if k.startswith(host.name + ":core"))
|
||
load1 = hostlvl.get("load_1m")
|
||
load_per_core = round(load1 / cores * 100) if (cores and load1 is not None) else None
|
||
psi = {
|
||
"cpu": hostlvl.get("psi_cpu_some_avg10"),
|
||
"mem": hostlvl.get("psi_mem_some_avg10"),
|
||
"io": hostlvl.get("psi_io_some_avg10"),
|
||
}
|
||
ls = reg.last_seen_at if reg else None
|
||
reporting = bool(reg) and ls is not None
|
||
stale = reporting and (datetime.now(timezone.utc) - ls).total_seconds() > _stale_after_seconds()
|
||
cpu, mem = hostlvl.get("cpu_pct"), hostlvl.get("mem_used_pct")
|
||
thresholds = current_app.config.get("THRESHOLDS", {})
|
||
worst = worst_reading((("CPU", cpu, "cpu"), ("Memory", mem, "mem"), ("Disk /", disk_root, "disk"),
|
||
("Inodes on /", inodes_root, "disk"), ("Load per core", load_per_core, "load")),
|
||
thresholds, host_svc.host_normals(norms, thresholds))
|
||
# Each tile's grade; CPU, memory and load within the host's own normal
|
||
# for the hour read healthy, and say what is usual (#5742).
|
||
grades = host_svc.host_grades({"cpu": cpu, "mem": mem, "disk": host_svc.fuller(disk_root, inodes_root),
|
||
"load": load_per_core}, norms, thresholds)
|
||
usual = [f"{label} {value:.0f}% is {grades[kind][1]}"
|
||
for label, value, kind in (("CPU", cpu, "cpu"), ("Memory", mem, "mem"),
|
||
("load per core", load_per_core, "load"))
|
||
if grades[kind][1]]
|
||
return await render_template(
|
||
"_host_vitals.html",
|
||
reg=reg, cpu=cpu, mem=mem, disk_root=disk_root, load_per_core=load_per_core,
|
||
inodes_root=inodes_root,
|
||
sparks=sparks, psi=psi, reporting=reporting, stale=stale, worst=worst,
|
||
grades=grades, usual=usual,
|
||
clock=clock_words(reg.clock_offset_s if reg else None, current_app.config.get("THRESHOLDS", {})),
|
||
)
|
||
|
||
|
||
@host_agent_bp.get("/panel/<host_id>")
|
||
@require_role(UserRole.viewer)
|
||
async def host_panel(host_id: str):
|
||
"""Per-host agent management panel embedded into the Hosts hub via HTMX.
|
||
|
||
Lifecycle actions (update / rotate / remove / re-provision) when the host
|
||
reports, otherwise provisioning — tied to the host's linked Ansible target.
|
||
The live vitals are a separate strip (host_vitals); this panel is management
|
||
only. Self-contained fragment (no base.html) so it can be hx-swapped in.
|
||
"""
|
||
from steward.core.capabilities import has_capability
|
||
from steward.models.ansible_inventory import AnsibleTarget
|
||
|
||
async with current_app.db_sessionmaker() as db:
|
||
host = (await db.execute(
|
||
select(Host).where(Host.id == host_id))).scalar_one_or_none()
|
||
if host is None:
|
||
return "", 404
|
||
reg = (await db.execute(select(HostAgentRegistration).where(
|
||
HostAgentRegistration.host_id == host_id))).scalar_one_or_none()
|
||
target = (await db.execute(select(AnsibleTarget).where(
|
||
AnsibleTarget.host_id == host_id))).scalar_one_or_none()
|
||
downtime = await recent_downtime(db, host_id, datetime.now(timezone.utc)) if reg else []
|
||
|
||
ls = reg.last_seen_at if reg else None
|
||
# "Deployed" is gated on the agent actually checking in (last_seen_at), NOT on
|
||
# the registration row — that row is minted before the playbook runs, so a
|
||
# failed provision would otherwise look deployed.
|
||
reporting = bool(reg) and ls is not None
|
||
stale = reporting and (datetime.now(timezone.utc) - ls).total_seconds() > _stale_after_seconds()
|
||
|
||
ansible_cfg = current_app.config.get("ANSIBLE", {})
|
||
return await render_template(
|
||
"panel.html",
|
||
host=host, reg=reg, reporting=reporting, stale=stale, target=target, downtime=downtime,
|
||
ansible_available=has_capability("ansible.run_playbook"),
|
||
managed_key_set=bool((ansible_cfg.get("ssh_public_key") or "").strip()),
|
||
)
|
||
|
||
|
||
@host_agent_bp.get("/fleet/")
|
||
@require_role(UserRole.admin)
|
||
async def fleet():
|
||
from steward.core.capabilities import has_capability
|
||
from steward.models.ansible_inventory import AnsibleTarget, AnsibleGroup
|
||
|
||
ansible_available = has_capability("ansible.run_playbook")
|
||
async with current_app.db_sessionmaker() as db:
|
||
regs = (await db.execute(select(HostAgentRegistration))).scalars().all()
|
||
hosts_by_id = {
|
||
h.id: h for h in (await db.execute(
|
||
select(Host).where(Host.id.in_([r.host_id for r in regs])))).scalars().all()
|
||
} if regs else {}
|
||
targets, groups = [], []
|
||
if ansible_available:
|
||
targets = (await db.execute(
|
||
select(AnsibleTarget).order_by(AnsibleTarget.name))).scalars().all()
|
||
groups = (await db.execute(
|
||
select(AnsibleGroup).order_by(AnsibleGroup.name))).scalars().all()
|
||
|
||
new_token = request.args.get("new_token")
|
||
new_host_id = request.args.get("host_id")
|
||
install_url = None
|
||
if new_token and new_host_id:
|
||
install_url = f"{public_base_url(request)}/plugins/host_agent/install.sh?token={new_token}"
|
||
|
||
managed_key_set = bool(
|
||
(current_app.config.get("ANSIBLE", {}).get("ssh_public_key") or "").strip())
|
||
|
||
now, stale_after = datetime.now(timezone.utc), _stale_after_seconds()
|
||
return await render_template(
|
||
"settings_list.html",
|
||
registrations=fleet_rows(regs, hosts_by_id, now, stale_after),
|
||
new_token=new_token,
|
||
install_url=install_url,
|
||
ansible_available=ansible_available,
|
||
deploy_targets=targets,
|
||
deploy_groups=groups,
|
||
managed_key_set=managed_key_set,
|
||
)
|
||
|
||
|
||
@host_agent_bp.post("/fleet/<host_id>/register")
|
||
@require_role(UserRole.admin)
|
||
async def register_host(host_id: str):
|
||
"""Issue an agent token for an existing host, for a manual install: the
|
||
fleet page then shows the one-line install command, once. Hosts are added
|
||
on Hosts → Add host; this is only the agent half (#5282)."""
|
||
async with current_app.db_sessionmaker() as session:
|
||
host = await session.get(Host, host_id)
|
||
if host is None:
|
||
return _error(404, "not_found")
|
||
existing = (await session.execute(
|
||
select(HostAgentRegistration).where(
|
||
HostAgentRegistration.host_id == host.id))).scalar_one_or_none()
|
||
if existing is not None:
|
||
return _error(400, "already_registered")
|
||
raw, hashed = _new_token_pair()
|
||
session.add(HostAgentRegistration(host_id=host.id, token_hash=hashed))
|
||
await session.commit()
|
||
return redirect(url_for("host_agent.fleet") + f"?new_token={raw}&host_id={host_id}")
|
||
|
||
|
||
@host_agent_bp.post("/fleet/<host_id>/rotate-token")
|
||
@require_role(UserRole.admin)
|
||
async def rotate_token(host_id: str):
|
||
async with current_app.db_sessionmaker() as session:
|
||
reg = (await session.execute(
|
||
select(HostAgentRegistration).where(
|
||
HostAgentRegistration.host_id == host_id))).scalar_one_or_none()
|
||
if reg is None:
|
||
return _error(404, "not_found")
|
||
raw, hashed = _new_token_pair()
|
||
reg.token_hash = hashed
|
||
reg.token_created_at = datetime.now(timezone.utc)
|
||
await session.commit()
|
||
return redirect(url_for("host_agent.fleet") + f"?new_token={raw}&host_id={host_id}")
|
||
|
||
|
||
@host_agent_bp.post("/fleet/<host_id>/delete")
|
||
@require_role(UserRole.admin)
|
||
async def delete_registration(host_id: str):
|
||
async with current_app.db_sessionmaker() as session:
|
||
reg = (await session.execute(
|
||
select(HostAgentRegistration).where(
|
||
HostAgentRegistration.host_id == host_id))).scalar_one_or_none()
|
||
if reg is not None:
|
||
await session.delete(reg)
|
||
await session.commit()
|
||
return redirect(url_for("host_agent.fleet"))
|
||
|
||
|
||
# ── Deploy via Ansible (plugin↔core synergy via the capability registry) ──────
|
||
|
||
async def _ensure_host_for_target(db, target) -> "Host":
|
||
"""Find or create the Host that an AnsibleTarget should report under."""
|
||
if getattr(target, "host_id", None):
|
||
h = await db.get(Host, target.host_id)
|
||
if h:
|
||
return h
|
||
h = (await db.execute(
|
||
select(Host).where(Host.name == target.name))).scalar_one_or_none()
|
||
if h is None:
|
||
h = Host(name=target.name, address=getattr(target, "address", "") or "")
|
||
db.add(h)
|
||
await db.flush()
|
||
return h
|
||
|
||
|
||
async def _mint_registration_token(db, host) -> str:
|
||
"""Create or rotate the host's agent registration; return the raw token.
|
||
|
||
Tokens are stored hashed (unrecoverable), so deploying always mints a fresh
|
||
token — the run installs the agent with it. Existing agents get rotated.
|
||
"""
|
||
reg = (await db.execute(select(HostAgentRegistration).where(
|
||
HostAgentRegistration.host_id == host.id))).scalar_one_or_none()
|
||
raw, hashed = _new_token_pair()
|
||
if reg is None:
|
||
db.add(HostAgentRegistration(host_id=host.id, token_hash=hashed))
|
||
else:
|
||
reg.token_hash = hashed
|
||
reg.token_created_at = datetime.now(timezone.utc)
|
||
return raw
|
||
|
||
|
||
@host_agent_bp.post("/deploy")
|
||
@require_role(UserRole.admin)
|
||
async def deploy_via_ansible():
|
||
"""Install/update the agent on Ansible inventory targets via the bundled
|
||
install playbook — the 'curl | sh becomes a button' synergy.
|
||
|
||
Uses the core "ansible.run_playbook" capability (no hard import of the
|
||
runner) and injects a freshly-minted per-host token as an inventory hostvar.
|
||
"""
|
||
from steward.core.capabilities import has_capability, invoke_capability
|
||
from steward.ansible.inventory_gen import (
|
||
fetch_scope_targets, generate_inventory, inventory_to_yaml)
|
||
from steward.ansible.sources import BUILTIN_SOURCE_NAME
|
||
|
||
if not has_capability("ansible.run_playbook"):
|
||
return _error(400, "ansible_unavailable", "Ansible is not available")
|
||
|
||
form = await request.form
|
||
scope = (form.get("inventory_scope", "") or "").strip()
|
||
try:
|
||
interval = max(5, int(form.get("agent_interval", "30")))
|
||
except (TypeError, ValueError):
|
||
interval = 30
|
||
if not (scope.startswith("steward:target:")
|
||
or scope.startswith("steward:group:")
|
||
or scope == "steward:all"):
|
||
return _error(400, "bad_scope", "Choose a target or group")
|
||
|
||
url = public_base_url(request)
|
||
async with current_app.db_sessionmaker() as db:
|
||
targets = await fetch_scope_targets(db, scope)
|
||
if not targets:
|
||
return _error(400, "no_targets", "No Ansible targets in that scope")
|
||
# The read above autobegins the session transaction (SQLAlchemy 2.0), so
|
||
# mint into it directly and commit — a nested db.begin() would double-begin.
|
||
tokens: dict[str, str] = {}
|
||
for t in targets:
|
||
host = await _ensure_host_for_target(db, t)
|
||
tokens[t.name] = await _mint_registration_token(db, host)
|
||
inv = generate_inventory(targets) # built while ORM objects are still live
|
||
# Per-host token as a host var; steward_url goes in as an extra-var below.
|
||
for name, tok in tokens.items():
|
||
inv["_meta"]["hostvars"].setdefault(name, {})["steward_token"] = tok
|
||
inventory_content = inventory_to_yaml(inv)
|
||
await db.commit()
|
||
|
||
actor_role = UserRole(session.get("user_role", "viewer"))
|
||
run, _source, err = await invoke_capability(
|
||
"ansible.run_playbook", actor_role,
|
||
current_app._get_current_object(), # type: ignore[attr-defined]
|
||
source_name=BUILTIN_SOURCE_NAME,
|
||
playbook_path="host_agent/install.yml",
|
||
inventory_content=inventory_content,
|
||
inventory_scope=scope,
|
||
redact=list(tokens.values()),
|
||
# extra-vars outrank the playbook's `vars:` defaults; token stays per-host.
|
||
params={"extra_vars_map": {"steward_url": url, "agent_interval": interval}},
|
||
triggered_by=session.get("user_id"),
|
||
)
|
||
if err:
|
||
return _error(400, "deploy_failed", err)
|
||
return redirect(url_for("ansible.run_detail", run_id=run.id))
|
||
|
||
|
||
@host_agent_bp.post("/provision")
|
||
@require_role(UserRole.admin)
|
||
async def provision_via_ansible():
|
||
"""First-contact provisioning: create the steward account + install the
|
||
managed key + NOPASSWD sudo, then install the agent — over bootstrap
|
||
(password) auth. After this, hosts are reachable with the managed key.
|
||
|
||
The bootstrap password is passed to the runner as a connection override and
|
||
is NOT persisted on the run row.
|
||
"""
|
||
from steward.core.capabilities import has_capability, invoke_capability
|
||
from steward.ansible.inventory_gen import (
|
||
fetch_scope_targets, generate_inventory, inventory_to_yaml)
|
||
from steward.ansible.sources import BUILTIN_SOURCE_NAME
|
||
|
||
if not has_capability("ansible.run_playbook"):
|
||
return _error(400, "ansible_unavailable", "Ansible is not available")
|
||
|
||
ansible_cfg = current_app.config.get("ANSIBLE", {})
|
||
pubkey = (ansible_cfg.get("ssh_public_key") or "").strip()
|
||
if not pubkey:
|
||
return _error(400, "no_managed_key",
|
||
"Generate a managed SSH key in Settings → Ansible first")
|
||
steward_user = (ansible_cfg.get("ssh_user") or "steward").strip() or "steward"
|
||
|
||
form = await request.form
|
||
scope = (form.get("inventory_scope", "") or "").strip()
|
||
boot_user = (form.get("bootstrap_user", "") or "").strip()
|
||
boot_password = form.get("bootstrap_password", "") or ""
|
||
try:
|
||
interval = max(5, int(form.get("agent_interval", "30")))
|
||
except (TypeError, ValueError):
|
||
interval = 30
|
||
if not (scope.startswith("steward:target:")
|
||
or scope.startswith("steward:group:")
|
||
or scope == "steward:all"):
|
||
return _error(400, "bad_scope", "Choose a target or group")
|
||
if not boot_user or not boot_password:
|
||
return _error(400, "missing_bootstrap",
|
||
"Bootstrap user and password are required for first contact")
|
||
|
||
url = public_base_url(request)
|
||
async with current_app.db_sessionmaker() as db:
|
||
targets = await fetch_scope_targets(db, scope)
|
||
if not targets:
|
||
return _error(400, "no_targets", "No Ansible targets in that scope")
|
||
# The read above autobegins the session transaction (SQLAlchemy 2.0), so
|
||
# mint into it directly and commit — a nested db.begin() would double-begin.
|
||
tokens: dict[str, str] = {}
|
||
for t in targets:
|
||
host = await _ensure_host_for_target(db, t)
|
||
tokens[t.name] = await _mint_registration_token(db, host)
|
||
inv = generate_inventory(targets) # built while ORM objects are still live
|
||
# Only the per-host token is a host var; the globals go in as extra-vars
|
||
# below (extra-vars outrank play vars, host vars do not).
|
||
for name, tok in tokens.items():
|
||
inv["_meta"]["hostvars"].setdefault(name, {})["steward_token"] = tok
|
||
inventory_content = inventory_to_yaml(inv)
|
||
await db.commit()
|
||
|
||
actor_role = UserRole(session.get("user_role", "viewer"))
|
||
run, _source, err = await invoke_capability(
|
||
"ansible.run_playbook", actor_role,
|
||
current_app._get_current_object(), # type: ignore[attr-defined]
|
||
source_name=BUILTIN_SOURCE_NAME,
|
||
playbook_path="host_agent/provision.yml",
|
||
inventory_content=inventory_content,
|
||
inventory_scope=scope,
|
||
redact=list(tokens.values()),
|
||
# extra_vars_map → a JSON -e @file: highest precedence (beats play vars)
|
||
# and space-safe (steward_pubkey carries a comment with a space).
|
||
params={"extra_vars_map": {
|
||
"steward_url": url,
|
||
"steward_pubkey": pubkey,
|
||
"steward_user": steward_user,
|
||
"agent_interval": interval,
|
||
}},
|
||
triggered_by=session.get("user_id"),
|
||
connection={"user": boot_user, "password": boot_password},
|
||
)
|
||
if err:
|
||
return _error(400, "provision_failed", err)
|
||
return redirect(url_for("ansible.run_detail", run_id=run.id))
|
||
|
||
|
||
@host_agent_bp.post("/update")
|
||
@require_role(UserRole.admin)
|
||
async def update_via_ansible():
|
||
"""Refresh agent.py on already-installed hosts via the bundled update
|
||
playbook. Connects as the managed steward account — no token rotation, no
|
||
config rewrite (the host keeps its existing identity)."""
|
||
from steward.core.capabilities import has_capability, invoke_capability
|
||
from steward.ansible.sources import BUILTIN_SOURCE_NAME
|
||
|
||
if not has_capability("ansible.run_playbook"):
|
||
return _error(400, "ansible_unavailable", "Ansible is not available")
|
||
|
||
form = await request.form
|
||
scope = (form.get("inventory_scope", "") or "").strip()
|
||
if not (scope.startswith("steward:target:")
|
||
or scope.startswith("steward:group:")
|
||
or scope == "steward:all"):
|
||
return _error(400, "bad_scope", "Choose a target or group")
|
||
|
||
url = public_base_url(request)
|
||
actor_role = UserRole(session.get("user_role", "viewer"))
|
||
run, _source, err = await invoke_capability(
|
||
"ansible.run_playbook", actor_role,
|
||
current_app._get_current_object(), # type: ignore[attr-defined]
|
||
source_name=BUILTIN_SOURCE_NAME,
|
||
playbook_path="host_agent/update.yml",
|
||
inventory_scope=scope,
|
||
params={"extra_vars": [f"steward_url={url}"]},
|
||
triggered_by=session.get("user_id"),
|
||
)
|
||
if err:
|
||
return _error(400, "update_failed", err)
|
||
return redirect(url_for("ansible.run_detail", run_id=run.id))
|