diff --git a/docs/plugins/writing-a-plugin.md b/docs/plugins/writing-a-plugin.md index 1999302..a0c0499 100644 --- a/docs/plugins/writing-a-plugin.md +++ b/docs/plugins/writing-a-plugin.md @@ -403,7 +403,7 @@ async def mcp_events(db, host_id, host_name, since, until) -> list[dict]: for r in await _jobs_on(db, host_id, since, until)] ``` -- Return what happened between `since` and `until`, oldest first: something that lasted gives `until`. Keep it to tens of rows; the caller keeps the first 40. +- Return what happened between `since` and `until`, oldest first: something that lasted gives `until`. Bound your query (a few hundred rows); Steward keeps the 40 nearest the occurrence's start and tells the caller how many it left out (`around_left_out`). - It is called only while the plugin is enabled, with its own session. If it raises, the anomaly is still returned without the plugin's events. The host agent (maintenance jobs, reboots) and Docker (container starts, deaths, OOM kills) are the examples to read. A resource with no host (a Traefik router, say) has no host to read events by. A plugin that knows what happened to one of its own resources exports `mcp_resource_events(db, source, resource, since, until)` instead, same return shape, and returns `[]` for a source that isn't its own. Traefik's is the example: it gives a router's swarm service's container events, on whichever hosts they ran. diff --git a/plugins/docker/mcp_tools.py b/plugins/docker/mcp_tools.py index ec87c00..5a3b5f5 100644 --- a/plugins/docker/mcp_tools.py +++ b/plugins/docker/mcp_tools.py @@ -758,7 +758,7 @@ async def mcp_events(db, host_id: str, host_name: str, since, until) -> list[dic from steward.mcp.tools._events import event rows = (await db.execute(select(DockerEvent).where( DockerEvent.host_id == host_id, DockerEvent.at >= since, DockerEvent.at <= until) - .order_by(DockerEvent.at).limit(40))).scalars() + .order_by(DockerEvent.at).limit(500))).scalars() return [event(e.at, "container", f"{e.container_name} {e.event}" + (f" ({e.detail})" if e.detail else ""), subject=f"{host_name}/{e.container_name}") for e in rows] diff --git a/plugins/host_agent/mcp_tools.py b/plugins/host_agent/mcp_tools.py index cc2879a..3058fe2 100644 --- a/plugins/host_agent/mcp_tools.py +++ b/plugins/host_agent/mcp_tools.py @@ -303,7 +303,7 @@ async def mcp_events(db, host_id: str, host_name: str, since, until) -> list[dic 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(20))).scalars() + .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})) @@ -311,7 +311,7 @@ async def mcp_events(db, host_id: str, host_name: str, since, until) -> list[dic 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(20))).scalars() + .order_by(HostDowntime.started_at).limit(200))).scalars() for row in rows: seen = downtime.describe(row, now) out.append(event(row.started_at, "downtime", diff --git a/plugins/traefik/mcp_tools.py b/plugins/traefik/mcp_tools.py index 41714ba..0c25d71 100644 --- a/plugins/traefik/mcp_tools.py +++ b/plugins/traefik/mcp_tools.py @@ -323,7 +323,7 @@ async def service_events(db, router: str, since, until) -> list[dict]: host: a router anomaly has no host of its own to read events by (#5862). [] without the Docker plugin, or when no single service matches.""" from steward.core.plugin_manager import loaded_plugin_submodule - from steward.mcp.tools._events import PER_SOURCE, event + from steward.mcp.tools._events import event from steward.models.hosts import Host models = loaded_plugin_submodule("docker", "models") if models is None: @@ -335,7 +335,7 @@ async def service_events(db, router: str, since, until) -> list[dict]: rows = (await db.execute( select(e, Host.name).join(Host, Host.id == e.host_id) .where(e.container_name.startswith(f"{service}.", autoescape=True), e.at >= since, e.at <= until) - .order_by(e.at).limit(PER_SOURCE))).all() + .order_by(e.at).limit(500))).all() return [event(ev.at, "container", f"{ev.container_name} {ev.event}" + (f" ({ev.detail})" if ev.detail else ""), subject=f"{host}/{ev.container_name}") for ev, host in rows] diff --git a/steward/mcp/tools/_events.py b/steward/mcp/tools/_events.py index 1d6c53b..b5827dc 100644 --- a/steward/mcp/tools/_events.py +++ b/steward/mcp/tools/_events.py @@ -34,7 +34,8 @@ from ._shape import db_session, iso logger = logging.getLogger(__name__) -PER_SOURCE = 40 # events any one source contributes to one window +PER_SOURCE = 40 # events any one source contributes to one window: the + # nearest the occurrence's start, the rest counted (#5861) # Steward samples its own health once a minute (core/self_monitor.py); a # longer break between two samples is a stretch it wasn't recording. STEWARD_GAP_S = 90.0 @@ -98,7 +99,7 @@ async def alert_events(db, place: Place, since: datetime, until: datetime) -> li note = f" ({ev.note})" if ev.note else "" out.append(event(ev.transitioned_at, "alert", f"{rule.name}: {ev.from_state} → {ev.to_state}{note}", subject=rule.resource_name, ids={"rule_id": rule.id})) - return out[:PER_SOURCE] + return out async def ansible_runs(db, place: Place, since: datetime, until: datetime) -> list[dict]: @@ -119,7 +120,7 @@ async def ansible_runs(db, place: Place, since: datetime, until: datetime) -> li out.append(event(r.started_at, "ansible_run", f"Ansible {r.playbook_path}{check}: {r.status.value}", until=r.finished_at, ids={"run_id": r.id})) - return out[:PER_SOURCE] + return out def _same_host(source: str, resource: str, place: Place) -> bool: @@ -148,7 +149,7 @@ async def other_anomalies(db, place: Place, anomaly_id: str, since: datetime, until=span.ended_at, subject=a.resource_name, ids={"anomaly_id": a.id})) else: elsewhere += 1 - return here[:PER_SOURCE], elsewhere + return here, elsewhere def gaps_in(times: list[datetime], since: datetime, until: datetime) -> list[dict]: @@ -179,9 +180,10 @@ async def steward_gaps(db, since: datetime, until: datetime) -> list[dict]: return gaps_in(list(rows), since, until) -async def plugin_events(place: Place, since: datetime, until: datetime) -> list[dict]: - """What enabled plugins know happened on the host (`mcp_events`), each - in its own session so one that fails can't spoil the others' reads.""" +async def plugin_events(place: Place, since: datetime, until: datetime) -> list[list[dict]]: + """What enabled plugins know happened on the host (`mcp_events`), one + list per plugin, each read in its own session so one that fails can't + spoil the others' reads.""" server = getattr(mcp_app(), "mcp_server", None) if server is None or not place.host_id: return [] @@ -191,15 +193,16 @@ async def plugin_events(place: Place, since: datetime, until: datetime) -> list[ continue try: async with db_session() as db: - out.extend((await source(db, place.host_id, place.host_name, since, until))[:PER_SOURCE]) + out.append(await source(db, place.host_id, place.host_name, since, until)) except Exception: logger.exception("MCP events source of plugin %r failed", plugin) return out -async def resource_events(place: Place, since: datetime, until: datetime) -> list[dict]: +async def resource_events(place: Place, since: datetime, until: datetime) -> list[list[dict]]: """What enabled plugins know happened to the resource itself - (`mcp_resource_events`), for one with no host to read events by.""" + (`mcp_resource_events`), for one with no host to read events by: one + list per plugin.""" server = getattr(mcp_app(), "mcp_server", None) if server is None: return [] @@ -209,18 +212,37 @@ async def resource_events(place: Place, since: datetime, until: datetime) -> lis continue try: async with db_session() as db: - out.extend((await source(db, place.source, place.resource, since, until))[:PER_SOURCE]) + out.append(await source(db, place.source, place.resource, since, until)) except Exception: logger.exception("MCP resource events source of plugin %r failed", plugin) return out -async def around(db, place: Place, anomaly_id: str, since: datetime, - until: datetime) -> tuple[list[dict], int]: - """Everything above for one window, oldest first, and the count of - anomalies elsewhere.""" +def _at(e: dict) -> datetime | None: + return datetime.fromisoformat(e["at"].replace("Z", "+00:00")) if e.get("at") else None + + +def nearest(events: list[dict], focus: datetime, keep: int = PER_SOURCE) -> tuple[list[dict], int]: + """The `keep` events nearest `focus` (an occurrence's start), and how + many were left out: a busy host's window is cut where it matters least.""" + if len(events) <= keep: + return events, 0 + by_distance = sorted(events, key=lambda e: abs((_at(e) - focus).total_seconds()) if _at(e) else float("inf")) + return by_distance[:keep], len(events) - keep + + +async def around(db, place: Place, anomaly_id: str, since: datetime, until: datetime, + focus: datetime) -> tuple[list[dict], int, int]: + """Everything above for one window, oldest first; the count of anomalies + elsewhere; and how many events were left out, each source keeping its + PER_SOURCE nearest `focus`.""" anomalies, elsewhere = await other_anomalies(db, place, anomaly_id, since, until) - found = (await alert_events(db, place, since, until) + await ansible_runs(db, place, since, until) - + anomalies + await plugin_events(place, since, until) - + await resource_events(place, since, until) + await steward_gaps(db, since, until)) - return sorted(found, key=lambda e: e["at"] or ""), elsewhere + groups = [await alert_events(db, place, since, until), await ansible_runs(db, place, since, until), + anomalies, *await plugin_events(place, since, until), + *await resource_events(place, since, until), await steward_gaps(db, since, until)] + found, left_out = [], 0 + for group in groups: + kept, cut = nearest(group, focus) + found += kept + left_out += cut + return sorted(found, key=lambda e: e["at"] or ""), elsewhere, left_out diff --git a/steward/mcp/tools/anomalies.py b/steward/mcp/tools/anomalies.py index d4660c1..3651447 100644 --- a/steward/mcp/tools/anomalies.py +++ b/steward/mcp/tools/anomalies.py @@ -163,7 +163,9 @@ async def get_anomaly(anomaly_id: str, occurrences: int = 5) -> dict: stops; for a Traefik router, its swarm service's containers on any host), and kind "steward": a stretch Steward itself wasn't recording (a restart or deploy), around which any reading can be wrong. From 30 - minutes before to 30 minutes after; and + minutes before to 30 minutes after. Each kind keeps the 40 events + nearest the occurrence's start; `around_left_out` counts the rest + (0 when nothing was cut). And `anomalies_elsewhere`, how many anomalies on other hosts overlapped it. Many at once points at something shared (network, storage, Steward). - `series`: the newest shown occurrence with time either side (as long @@ -194,8 +196,8 @@ async def get_anomaly(anomaly_id: str, occurrences: int = 5) -> dict: shown = [] for span in spans[:limit]: usual = _usual_at(kind, slots, span.started_at) - near, elsewhere = await events.around(db, place, a.id, span.started_at - AROUND, - (span.ended_at or now) + AROUND) + near, elsewhere, left_out = await events.around(db, place, a.id, span.started_at - AROUND, + (span.ended_at or now) + AROUND, span.started_at) shown.append({ "started_at": iso(span.started_at), "ended_at": iso(span.ended_at), "ongoing": span.ended_at is None, @@ -206,6 +208,7 @@ async def get_anomaly(anomaly_id: str, occurrences: int = 5) -> dict: "flag_above": rnd(flag_above(usual.usual_hold, floor), 3) if floor is not None else None} if usual else None), "around": near, + "around_left_out": left_out, "anomalies_elsewhere": elsewhere, }) series = None diff --git a/tests/mcp/test_anomaly_review.py b/tests/mcp/test_anomaly_review.py index 60b45dd..8858274 100644 --- a/tests/mcp/test_anomaly_review.py +++ b/tests/mcp/test_anomaly_review.py @@ -119,6 +119,16 @@ def test_a_plugin_adds_a_hostless_resources_events_through_its_hook(app): mcp.resource_event_sources.pop("events_test", None) +def test_a_busy_window_keeps_the_events_nearest_the_occurrence_and_counts_the_rest(): + """#5861: 50 container events over an hour; the occurrence starts at :45.""" + evs = [_events.event(NOW + timedelta(minutes=m), "container", f"c{m} start") for m in range(50)] + kept, cut = _events.nearest(evs, NOW + timedelta(minutes=45)) + assert len(kept) == 40 and cut == 10 + assert {e["what"] for e in kept} >= {"c45 start", "c49 start", "c20 start"} + assert "c0 start" not in {e["what"] for e in kept} + assert _events.nearest(evs[:5], NOW) == (evs[:5], 0) + + def test_steward_not_recording_is_an_event_beside_every_anomaly(): """A Steward restart between two of its once-a-minute samples (#5862).""" ticks = [NOW + timedelta(minutes=m) for m in (0, 1, 2)] + [NOW + timedelta(minutes=4, seconds=10)]