diff --git a/docs/plugins/host-agent-design.md b/docs/plugins/host-agent-design.md index b53942c..3487554 100644 --- a/docs/plugins/host-agent-design.md +++ b/docs/plugins/host-agent-design.md @@ -113,9 +113,19 @@ Python 3.8+ stdlib only. `urllib.request` for the POST, plus `json`, `os`, `shut ### Signals and lifecycle - **SIGHUP** → re-read config file without restart (lets admins change interval/mounts cleanly). -- **SIGTERM** (from systemd) → finish current poll cycle → flush buffer one last time → exit. +- **SIGTERM** (from systemd) → finish current poll cycle → flush buffer one last time, with a `stopping` event saying whether the host is going down too (`systemctl is-system-running` says `stopping`) → exit. - **Systemd restart policy** → `Restart=on-failure`, `RestartSec=10`. +### What happened to the host (1.14.0, #5756) + +A post may carry `events`, said once each, with the next post that gets through: + +- `started` — the agent starting, with `boot_at` (from `/proc/uptime`). +- `stopping` — on SIGTERM, with `host_shutdown`: true, false, or null when systemd can't say. +- `lost_contact` — `from` the first post that got no answer (or a proxy's 5xx) `to` the post carrying it. + +Steward turns them into the host's downtime (`host_downtime`): **reboot** (a stop, then a boot after it), **agent_stop** (the host stayed up), **unclean** (a boot after the agent was last heard that nobody announced: power loss or a crash) and **lost_contact**. An announced stop explains the silence for `reboot_grace_seconds` (Settings → Plugins → host_agent, 10 minutes): `seconds_since_seen` reads 0 until then, and the host's other rules go quiet with it, as when it is offline. Past the grace, agent offline fires. The events live in memory, so a span the agent saw is lost if it restarts mid-outage; the boot time still reads the restart as unclean. + ### Logging To stderr only (systemd captures to journal). No file logging, no log rotation. Levels: diff --git a/plugins/host_agent/__init__.py b/plugins/host_agent/__init__.py index fa50043..5892c19 100644 --- a/plugins/host_agent/__init__.py +++ b/plugins/host_agent/__init__.py @@ -25,6 +25,15 @@ def setup(app: "Quart") -> None: from steward.hosts.service import agent_stale_after_seconds agent_heartbeat("host_agent", "seconds_since_seen", lambda: agent_stale_after_seconds(app.config)) agent_fed("host_agent", lambda resource: resource.split(":", 1)[0]) + # A reboot the agent announced is silence with a reason, for a while (#5756). + from datetime import datetime, timezone + from steward.core.alerts import agents_away + from .downtime import hosts_away, restart_grace_seconds + agents_away(lambda session: hosts_away(session, datetime.now(timezone.utc), + restart_grace_seconds(app.config))) + from steward.core.cleanup import prune_by_age + from .models import HostDowntime + prune_by_age(HostDowntime.started_at) def get_scheduled_tasks() -> list: diff --git a/plugins/host_agent/agent.py b/plugins/host_agent/agent.py index 2024604..1237756 100644 --- a/plugins/host_agent/agent.py +++ b/plugins/host_agent/agent.py @@ -12,6 +12,7 @@ import re import shutil import signal import socket +import subprocess import sys import time import urllib.error @@ -20,7 +21,7 @@ from collections import deque from concurrent.futures import ThreadPoolExecutor, wait as wait_futures from datetime import datetime, timezone -AGENT_VERSION = "1.13.0" +AGENT_VERSION = "1.14.0" # Default path to the local Docker Engine socket. Overridable via the # `docker_socket` config key; collection is silently skipped if it's absent or @@ -1330,16 +1331,70 @@ def build_sample(mounts: list[str], state: dict, return sample -def build_payload(samples: list[dict], hostname: str, metadata: dict) -> dict: +def build_payload(samples: list[dict], hostname: str, metadata: dict, + events: list | None = None) -> dict: # sent_at is this host's clock at the moment of sending: Steward compares # it with its own on arrival to say how far apart the two clocks are. - return { + payload = { "agent_version": AGENT_VERSION, "hostname": hostname, "metadata": metadata, "sent_at": datetime.now(timezone.utc).isoformat(), "samples": samples, } + if events: + payload["events"] = events + return payload + + +# ─── what happened to the host (#5756) ─────────────────────────────────────── +# Events ride the next post that gets through, so Steward records a reboot as a +# reboot and a lost connection as the span it lasted, not as the agent gone. + + +def _now_iso() -> str: + return datetime.now(timezone.utc).isoformat() + + +def started_event() -> dict: + """The agent starting, with when the host booted: Steward tells a host + that rebooted from an agent that only restarted by it.""" + event = {"kind": "started", "at": _now_iso()} + try: + booted = time.time() - float(_read_file(UPTIME_PATH).split()[0]) + event["boot_at"] = datetime.fromtimestamp(booted, timezone.utc).isoformat() + except Exception: + pass + return event + + +def host_shutting_down(): + """Whether the host itself is going down, not only this agent: systemd says + "stopping" while it shuts down or reboots. None when it can't tell.""" + try: + out = subprocess.run(["systemctl", "is-system-running"], stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, universal_newlines=True, + timeout=3).stdout.strip() + except (OSError, subprocess.SubprocessError): + return None + return out == "stopping" if out else None + + +def stopping_event() -> dict: + return {"kind": "stopping", "at": _now_iso(), "host_shutdown": host_shutting_down()} + + +def lost_contact_event(since: str) -> dict: + """The span this agent could not reach Steward: from its first failed post + to now, the post that is carrying it.""" + return {"kind": "lost_contact", "from": since, "to": _now_iso()} + + +def lost_contact(status) -> bool: + """Whether a failed post means Steward couldn't be reached: no answer, or + a proxy's 5xx in front of a Steward that is down. A 400 or 401 is Steward + answering.""" + return status is None or status >= 500 # ─── POST + backoff ────────────────────────────────────────────────────────── @@ -1441,6 +1496,9 @@ def main_loop(conf_path: str) -> int: docker_log_exclude = cfg.get("docker_log_exclude") or [] buffer = RingBuffer(maxlen=20) backoff = 0 + # Said once each, with the next post that gets through (#5756). + events: list = [started_event()] + lost_since = None # Carries previous net/disk counters + monotonic ts for rate computation. rate_state: dict = {} @@ -1470,10 +1528,14 @@ def main_loop(conf_path: str) -> int: samples=send + [sample], hostname=hostname, metadata=metadata, + events=events + ([lost_contact_event(lost_since)] if lost_since else []), ) ok, status = post_payload(cfg["url"], cfg["token"], payload) if ok: + if lost_since: + _log("INFO", f"reached Steward again (unreachable since {lost_since})") + events, lost_since = [], None for s in held: buffer.push(s) if send: @@ -1482,6 +1544,8 @@ def main_loop(conf_path: str) -> int: backoff = 0 sleep_for = cfg["interval_seconds"] else: + if lost_since is None and lost_contact(status): + lost_since = sample["ts"] if status == 400: _log("ERROR", "server rejected payload (400) — dropping sample") elif status == 401: @@ -1501,10 +1565,12 @@ def main_loop(conf_path: str) -> int: slept += 1.0 _log("INFO", "SIGTERM — flushing and exiting") - final = buffer.drain() - if final: - post_payload(cfg["url"], cfg["token"], - build_payload(final, hostname, metadata)) + # Say it is stopping, and whether the host is going down with it, so Steward + # reads the silence that follows as a reboot rather than the agent lost. + final_events = events + ([lost_contact_event(lost_since)] if lost_since else []) + final_events.append(stopping_event()) + post_payload(cfg["url"], cfg["token"], + build_payload(buffer.drain(), hostname, metadata, events=final_events)) return 0 diff --git a/plugins/host_agent/downtime.py b/plugins/host_agent/downtime.py new file mode 100644 index 0000000..9c7361e --- /dev/null +++ b/plugins/host_agent/downtime.py @@ -0,0 +1,160 @@ +# plugins/host_agent/downtime.py +"""What a host's agent says happened to it while it was away (#5756). + +An agent ≥ 1.14.0 tells Steward when it stops (and whether its host is going +down with it), when it starts (and when its host booted), and how long it +couldn't reach Steward. Each becomes a HostDowntime row, so a reboot reads as +a reboot with its length, and "agent offline" is left for silence nobody +explained. A post can be retried, so every event is applied idempotently. +""" +from __future__ import annotations + +from datetime import datetime, timedelta, timezone + +from sqlalchemy import select + +from .models import HostDowntime + +# A stop the agent announced: the silence after it is explained, for a while. +ANNOUNCED = ("reboot", "agent_stop") + +LABELS = { + "reboot": "Rebooted", + "agent_stop": "Agent stopped", + "unclean": "Restarted without shutting down", + "lost_contact": "Lost contact", +} + + +def restart_grace_seconds(config) -> int: + """How long an announced stop explains the agent's silence (Settings → + Plugins → host_agent, default 10 minutes): past it, agent offline fires.""" + return int(((config.get("PLUGINS") or {}).get("host_agent") or {}) + .get("reboot_grace_seconds", 600)) + + +def _at(value) -> datetime | None: + if not isinstance(value, str): + return None + try: + got = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + return None + return got if got.tzinfo else got.replace(tzinfo=timezone.utc) + + +def _when(event: dict) -> datetime | None: + return _at(event.get("at") or event.get("from")) + + +async def record_events(session, host_id: str, events, last_seen_at: datetime | None) -> int: + """Apply one post's events to the host's downtime, oldest first; returns + how many rows were added or changed. `last_seen_at` is when the agent was + last heard before this post: a boot after it that nobody announced is a + restart without a shutdown.""" + if not isinstance(events, list): + return 0 + timed = sorted(((t, e) for e in events if isinstance(e, dict) + for t in [_when(e)] if t is not None), key=lambda te: te[0]) + changed = 0 + for at, event in timed: + kind = event.get("kind") + if kind == "stopping": + changed += await _stopping(session, host_id, at, event.get("host_shutdown")) + elif kind == "started": + changed += await _started(session, host_id, at, _at(event.get("boot_at")), last_seen_at) + elif kind == "lost_contact": + changed += await _lost_contact(session, host_id, at, _at(event.get("to"))) + return changed + + +async def _stopping(session, host_id: str, at: datetime, host_shutdown) -> int: + seen = (await session.execute(select(HostDowntime.id).where( + HostDowntime.host_id == host_id, HostDowntime.kind.in_(ANNOUNCED), + HostDowntime.started_at == at))).first() + if seen: + return 0 + session.add(HostDowntime(host_id=host_id, kind="reboot" if host_shutdown else "agent_stop", + started_at=at)) + return 1 + + +async def _started(session, host_id: str, at: datetime, boot_at: datetime | None, + last_seen_at: datetime | None) -> int: + stop = (await session.execute(select(HostDowntime).where( + HostDowntime.host_id == host_id, HostDowntime.kind.in_(ANNOUNCED), + HostDowntime.ended_at.is_(None), HostDowntime.started_at <= at) + .order_by(HostDowntime.started_at.desc()).limit(1))).scalar_one_or_none() + if stop is not None: + stop.ended_at = at + # The boot time settles what the stop was: the host came back from a + # boot after it, or it stayed up while only the agent restarted. + if boot_at is not None: + stop.kind = "reboot" if boot_at > stop.started_at else "agent_stop" + return 1 + if boot_at is None or last_seen_at is None or boot_at <= last_seen_at: + return 0 + seen = (await session.execute(select(HostDowntime.id).where( + HostDowntime.host_id == host_id, HostDowntime.kind == "unclean", + HostDowntime.ended_at == at))).first() + if seen: + return 0 + session.add(HostDowntime(host_id=host_id, kind="unclean", started_at=last_seen_at, ended_at=at)) + return 1 + + +async def _lost_contact(session, host_id: str, since: datetime, until: datetime | None) -> int: + if until is None or until <= since: + return 0 + row = (await session.execute(select(HostDowntime).where( + HostDowntime.host_id == host_id, HostDowntime.kind == "lost_contact", + HostDowntime.started_at == since))).scalar_one_or_none() + if row is None: + session.add(HostDowntime(host_id=host_id, kind="lost_contact", started_at=since, ended_at=until)) + return 1 + if row.ended_at is None or until > row.ended_at: + row.ended_at = until + return 1 + return 0 + + +async def announced_stops(session, now: datetime, grace_seconds: int) -> dict[str, datetime]: + """{host id: when the silence its agent announced stops being explained} + for every announced stop still going and inside its grace.""" + rows = (await session.execute(select(HostDowntime.host_id, HostDowntime.started_at).where( + HostDowntime.kind.in_(ANNOUNCED), HostDowntime.ended_at.is_(None), + HostDowntime.started_at > now - timedelta(seconds=grace_seconds)))).all() + out: dict[str, datetime] = {} + for host_id, started in rows: + until = started + timedelta(seconds=grace_seconds) + out[host_id] = max(out.get(host_id, until), until) + return out + + +async def hosts_away(session, now: datetime, grace_seconds: int) -> set[str]: + """Names of the hosts in announced_stops: their silence has a reason.""" + from steward.models.hosts import Host + ids = list(await announced_stops(session, now, grace_seconds)) + if not ids: + return set() + return set((await session.execute(select(Host.name).where(Host.id.in_(ids)))).scalars()) + + +def describe(row: HostDowntime, now: datetime) -> dict: + """One downtime row as the host page shows it.""" + label = LABELS.get(row.kind, row.kind) + if row.kind == "agent_stop" and row.ended_at is not None: + label = "Agent restarted" + end = row.ended_at or now + return {"kind": row.kind, "label": label, "started_at": row.started_at, + "ended_at": row.ended_at, "ongoing": row.ended_at is None, + "seconds": max(0.0, (end - row.started_at).total_seconds())} + + +async def recent(session, host_id: str, now: datetime, days: int = 7, limit: int = 6) -> list[dict]: + """The host's downtime in the last `days`, newest first.""" + rows = (await session.execute(select(HostDowntime).where( + HostDowntime.host_id == host_id, + HostDowntime.started_at > now - timedelta(days=days)) + .order_by(HostDowntime.started_at.desc()).limit(limit))).scalars().all() + return [describe(r, now) for r in rows] diff --git a/plugins/host_agent/migrations/versions/host_agent_004_downtime.py b/plugins/host_agent/migrations/versions/host_agent_004_downtime.py new file mode 100644 index 0000000..7e2e8ca --- /dev/null +++ b/plugins/host_agent/migrations/versions/host_agent_004_downtime.py @@ -0,0 +1,32 @@ +# plugins/host_agent/migrations/versions/host_agent_004_downtime.py +"""host_agent: each host's downtime and what it was (#5756) + +Revision ID: host_agent_004_downtime +""" +from typing import Sequence, Union +from alembic import op +import sqlalchemy as sa + +revision: str = "host_agent_004_downtime" +down_revision: Union[str, None] = "host_agent_003_clock_offset" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "host_downtime", + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("host_id", sa.String(36), sa.ForeignKey("hosts.id", ondelete="CASCADE"), + nullable=False), + sa.Column("kind", sa.String(16), nullable=False), + sa.Column("started_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("ended_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + ) + op.create_index("ix_host_downtime_host_id", "host_downtime", ["host_id"]) + + +def downgrade() -> None: + op.drop_index("ix_host_downtime_host_id", table_name="host_downtime") + op.drop_table("host_downtime") diff --git a/plugins/host_agent/models.py b/plugins/host_agent/models.py index cc61f09..8612e8c 100644 --- a/plugins/host_agent/models.py +++ b/plugins/host_agent/models.py @@ -37,3 +37,22 @@ class HostAgentRegistration(Base): created_at = Column(DateTime(timezone=True), nullable=False, default=_utcnow) updated_at = Column(DateTime(timezone=True), nullable=False, default=_utcnow, onupdate=_utcnow) + + +class HostDowntime(Base): + """A span a host's agent was away, and what the agent said it was (#5756). + + kind is "reboot" (the host went down and came back), "agent_stop" (the + agent stopped; the host stayed up), "unclean" (the host came back from a + boot nobody announced: power loss or a crash) or "lost_contact" (the agent + ran but couldn't reach Steward). ended_at is None while it is going on. + """ + __tablename__ = "host_downtime" + + id = Column(String(36), primary_key=True, default=_uuid) + host_id = Column(String(36), ForeignKey("hosts.id", ondelete="CASCADE"), + nullable=False, index=True) + kind = Column(String(16), nullable=False) + started_at = Column(DateTime(timezone=True), nullable=False) + ended_at = Column(DateTime(timezone=True), nullable=True) + created_at = Column(DateTime(timezone=True), nullable=False, default=_utcnow) diff --git a/plugins/host_agent/plugin.yaml b/plugins/host_agent/plugin.yaml index ab23525..e96d2b5 100644 --- a/plugins/host_agent/plugin.yaml +++ b/plugins/host_agent/plugin.yaml @@ -16,4 +16,7 @@ tags: config: stale_after_seconds: 180 + # How long a reboot or restart the agent announced explains its silence + # before "agent offline" fires (agent >= 1.14.0, #5756). + reboot_grace_seconds: 600 default_interval_seconds: 30 diff --git a/plugins/host_agent/routes.py b/plugins/host_agent/routes.py index ea693a4..fddd49a 100644 --- a/plugins/host_agent/routes.py +++ b/plugins/host_agent/routes.py @@ -24,6 +24,7 @@ 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 @@ -232,7 +233,10 @@ async def ingest(): return _error(400, "malformed_payload", "missing 'samples'") samples = payload.get("samples") or [] - if not isinstance(samples, list) or not samples: + # 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) @@ -305,7 +309,7 @@ async def ingest(): if latest_ts is None or recorded_at > latest_ts: latest_ts = recorded_at - if accepted == 0: + 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 @@ -346,6 +350,11 @@ async def ingest(): 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): @@ -818,6 +827,7 @@ async def host_panel(host_id: str): 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 @@ -829,7 +839,7 @@ async def host_panel(host_id: str): ansible_cfg = current_app.config.get("ANSIBLE", {}) return await render_template( "panel.html", - host=host, reg=reg, reporting=reporting, stale=stale, target=target, + 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()), ) diff --git a/plugins/host_agent/scheduler.py b/plugins/host_agent/scheduler.py index c9beb1a..c1d8dbb 100644 --- a/plugins/host_agent/scheduler.py +++ b/plugins/host_agent/scheduler.py @@ -12,6 +12,7 @@ from steward.core.watch import watch from steward.hosts.service import agent_stale_after_seconds from steward.models.hosts import Host from steward.models.metrics import PluginMetricLatest +from .downtime import announced_stops, restart_grace_seconds from .models import HostAgentRegistration logger = logging.getLogger(__name__) @@ -36,6 +37,7 @@ def seconds_since_seen(regs: Iterable, host_names: dict[str, str], now: datetime, watching_since: datetime | None = None, arrived: dict[str, datetime] | None = None, already_silent: set[str] | frozenset = frozenset(), + explained_until: dict[str, datetime] | None = None, ) -> list[tuple[str, float]]: """(host name, seconds since its agent last reported) per registration. @@ -52,6 +54,11 @@ def seconds_since_seen(regs: Iterable, host_names: dict[str, str], whole silence counts. Counting them from `watching_since` read them as just seen, which resolved their alert on every restart and fired it again three minutes later (#5719). + + `explained_until` ({host id: time}) is a stop the agent announced, a reboot + or its own restart, plus its grace (#5756): until then the agent reads as + just seen. A host still silent past it reads its whole silence, so agent + offline fires as the grace runs out, for silence nobody explained. """ out = [] for r in regs: @@ -62,6 +69,8 @@ def seconds_since_seen(regs: Iterable, host_names: dict[str, str], continue if watching_since is not None and name not in already_silent: seen = max(seen, watching_since) + if r.host_id in (explained_until or {}): + seen = max(seen, explained_until[r.host_id]) out.append((name, max(0.0, (now - seen).total_seconds()))) return out @@ -88,9 +97,10 @@ async def record_seconds_since_seen(app) -> None: PluginMetricLatest.metric_name == "seconds_since_seen", PluginMetricLatest.resource_name.in_(list(host_names.values())), PluginMetricLatest.value > stale_after))).scalars()) + explained = await announced_stops(session, now, restart_grace_seconds(app.config)) batch = MetricBatch() for name, seconds in seconds_since_seen(regs, host_names, now, watching_since, - arrivals(app), already_silent): + arrivals(app), already_silent, explained): batch.add(source_module="host_agent", resource_name=name, metric_name="seconds_since_seen", value=seconds, recorded_at=now) await batch.record(session) diff --git a/plugins/host_agent/templates/panel.html b/plugins/host_agent/templates/panel.html index 150fc6c..07f0cf2 100644 --- a/plugins/host_agent/templates/panel.html +++ b/plugins/host_agent/templates/panel.html @@ -19,6 +19,20 @@ "Manage agent" (#5278). #} Full metrics → + {# What the agent said happened while it was away, last 7 days (#5756). #} + {% if downtime %} +