bvandeusenandClaude Opus 5.5
5a0e150789
#5752 "Container CPU and memory history lives in the one metric store; docker_metrics and docker_metrics_hourly are retired". Container cpu_pct and mem_pct already went to plugin_metrics for alerting; now mem_usage_bytes does too, and every snapshot in an ingest (backlog included) is recorded at its own time. The container history chart, Busiest Containers, a service's history and the MCP container tools read plugin_metrics under resource "<host>/<container>", rolled up and pruned by the shared metric retention. docker_012_drop_metrics drops both tables. Their history is not copied (lesson #5221 "A data migration's cost is set by the live table"); the store already holds 7 days of container CPU and memory %. Docker retention now prunes only lifecycle events, and Settings -> Thresholds has one "Metric retention" card for every source. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
325 lines
20 KiB
Python
325 lines
20 KiB
Python
"""Integration: the docker and host_agent plugins' MCP tools against real Postgres.
|
|
|
|
Rows are seeded with plain SQL so this file never imports a plugin's models
|
|
(a second import of them after the app loaded the plugin raises "Table already
|
|
defined"). The tool modules come from the loaded plugin when there is one,
|
|
and from the source tree when the plugin is not loaded.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import importlib
|
|
import json
|
|
import os
|
|
import time
|
|
import uuid
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
_NEEDS_DB = pytest.mark.skipif(
|
|
not os.environ.get("STEWARD_DATABASE_URL"),
|
|
reason="integration test needs a live Postgres (STEWARD_DATABASE_URL)",
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def app():
|
|
if not os.environ.get("STEWARD_DATABASE_URL"):
|
|
pytest.skip("needs Postgres")
|
|
from steward.app import create_app
|
|
return create_app(testing=False)
|
|
|
|
|
|
def _plugin(name):
|
|
from steward.core.plugin_manager import get_loaded_plugin_modules
|
|
loaded = get_loaded_plugin_modules().get(name)
|
|
return loaded if loaded is not None else importlib.import_module(f"plugins.{name}")
|
|
|
|
|
|
def _sub(plugin_module, sub):
|
|
return importlib.import_module(f"{plugin_module.__name__}.{sub}")
|
|
|
|
|
|
@_NEEDS_DB
|
|
def test_docker_and_agent_tools_over_seeded_rows(app):
|
|
from sqlalchemy import text
|
|
from steward.core.api_tokens import Caller
|
|
from steward.mcp import _context
|
|
from steward.models.hosts import Host
|
|
from steward.models.users import UserRole
|
|
|
|
docker, agent = _plugin("docker"), _plugin("host_agent")
|
|
dtools, atools = _sub(docker, "mcp_tools"), _sub(agent, "mcp_tools")
|
|
|
|
now = datetime.now(timezone.utc)
|
|
tag = uuid.uuid4().hex[:8]
|
|
hid, hid2 = str(uuid.uuid4()), str(uuid.uuid4())
|
|
host_name, quiet_host = f"dock-{tag}", f"quiet-{tag}"
|
|
web, db_, job, svc = f"web-{tag}", f"db-{tag}", f"job-{tag}", f"svc-{tag}"
|
|
replaced = f"{svc}.1.old"
|
|
# A service across a deploy (#5468): its live task, and the task the deploy replaced.
|
|
api = f"api-{tag}"
|
|
api_live, api_gone = f"{api}.1.{tag}{'a' * 17}", f"{api}.1.{tag}{'b' * 17}"
|
|
removed = f"{svc}.2.gone{tag}" # a dead task Docker removed: logs and events only
|
|
|
|
def _ago(minutes):
|
|
return (now - timedelta(minutes=minutes)).isoformat().replace("+00:00", "Z")
|
|
|
|
container_sql = text("""
|
|
INSERT INTO docker_containers
|
|
(host_id, name, container_id, image, status, cpu_pct, mem_pct, mem_usage_bytes,
|
|
mem_limit_bytes, restart_count, ports_json, started_at, scraped_at, health,
|
|
exit_code, oom_killed, compose_project, service_name, task_id)
|
|
VALUES (:h, :n, :cid, 'img:1', :st, 5.0, 10.0, 1000, 10000, :rc, '[]', :started, :now,
|
|
:health, :exit, false, :proj, :svc, :task)
|
|
""")
|
|
|
|
async def _seed():
|
|
async with app.db_sessionmaker() as s:
|
|
async with s.begin():
|
|
s.add(Host(id=hid, name=host_name, address="10.8.8.8"))
|
|
s.add(Host(id=hid2, name=quiet_host, address="10.8.8.9"))
|
|
async with s.begin():
|
|
for name, st, health, exit_code, proj, service, task in (
|
|
(web, "running", "healthy", None, f"stack-{tag}", None, None),
|
|
(db_, "running", "unhealthy", None, f"stack-{tag}", None, None),
|
|
(job, "exited", None, 1, None, None, None),
|
|
(f"{svc}.1.abc", "running", None, None, None, svc, f"abc-{tag}"),
|
|
# The task a rolling update replaced: killed (137), not crashed.
|
|
(replaced, "exited", None, 137, None, svc, f"old-{tag}"),
|
|
):
|
|
await s.execute(container_sql, {
|
|
"h": hid, "n": name, "cid": uuid.uuid4().hex, "st": st, "rc": 0,
|
|
"started": now - timedelta(hours=2), "now": now, "health": health,
|
|
"exit": exit_code, "proj": proj, "svc": service, "task": task})
|
|
for name in (job, replaced):
|
|
await s.execute(text(
|
|
"INSERT INTO docker_events (id, host_id, container_name, event, detail, at) "
|
|
"VALUES (:id, :h, :n, 'die', 'exit 1', :at)"),
|
|
{"id": str(uuid.uuid4()), "h": hid, "n": name, "at": now - timedelta(minutes=10)})
|
|
await s.execute(text(
|
|
"INSERT INTO docker_events (id, host_id, container_name, event, detail, at) "
|
|
"VALUES (:id, :h, :n, 'die', 'exit 1', :at)"),
|
|
{"id": str(uuid.uuid4()), "h": hid, "n": removed, "at": now - timedelta(minutes=12)})
|
|
await s.execute(text(
|
|
"INSERT INTO docker_logs (id, host_id, container_name, ts, stream, line) "
|
|
"VALUES (:id, :h, :n, :ts, 'stderr', 'FATAL: cannot open database')"),
|
|
{"id": str(uuid.uuid4()), "h": hid, "n": removed, "ts": now - timedelta(minutes=13)})
|
|
await s.execute(container_sql, {
|
|
"h": hid, "n": api_live, "cid": uuid.uuid4().hex, "st": "running", "rc": 0,
|
|
"started": now - timedelta(minutes=3), "now": now, "health": None,
|
|
"exit": None, "proj": None, "svc": api, "task": f"{tag}{'a' * 17}"})
|
|
await s.execute(text(
|
|
"INSERT INTO docker_events (id, host_id, container_name, event, detail, at) "
|
|
"VALUES (:id, :h, :n, 'start', '', :at)"),
|
|
{"id": str(uuid.uuid4()), "h": hid, "n": api_gone, "at": now - timedelta(minutes=30)})
|
|
for name, mins, line in ((api_gone, 20, "listening (old image)"),
|
|
(api_live, 2, "listening (new image)")):
|
|
await s.execute(text(
|
|
"INSERT INTO docker_logs (id, host_id, container_name, ts, stream, line) "
|
|
"VALUES (:id, :h, :n, :ts, 'stdout', :line)"),
|
|
{"id": str(uuid.uuid4()), "h": hid, "n": name,
|
|
"ts": now - timedelta(minutes=mins), "line": line})
|
|
for i, (stream, line) in enumerate((
|
|
("stdout", "starting up"), ("stdout", "connected to db"),
|
|
("stderr", "ERROR: 100% of disk_used"), ("stdout", "retrying"),
|
|
("stdout", "x" * 5000))):
|
|
await s.execute(text(
|
|
"INSERT INTO docker_logs (id, host_id, container_name, ts, stream, line) "
|
|
"VALUES (:id, :h, :n, :ts, :st, :line)"),
|
|
{"id": str(uuid.uuid4()), "h": hid, "n": web,
|
|
"ts": now - timedelta(minutes=5 - i), "st": stream, "line": line})
|
|
# Container series live in the one metric store (#5752).
|
|
for i, cpu in enumerate((10.0, 30.0, 50.0)):
|
|
for metric, value in (("cpu_pct", cpu), ("mem_pct", 20.0),
|
|
("mem_usage_bytes", 1000.0)):
|
|
await s.execute(text(
|
|
"INSERT INTO plugin_metrics (id, source_module, resource_name, "
|
|
"metric_name, value, recorded_at) "
|
|
"VALUES (:id, 'docker', :r, :m, :v, :t)"),
|
|
{"id": str(uuid.uuid4()), "r": f"{host_name}/{web}", "m": metric,
|
|
"v": value, "t": now - timedelta(minutes=30 - i * 10)})
|
|
await s.execute(text(
|
|
"INSERT INTO docker_swarm_services (host_id, service_name, mode, desired, running, "
|
|
"placement_json, tasks_json, updated_at) "
|
|
"VALUES (:h, :n, 'replicated', 3, 1, '[]', :tasks, :now)"),
|
|
{"h": hid, "n": svc, "now": now, "tasks": json.dumps([
|
|
{"task_id": f"abc-{tag}", "slot": 1, "node_id": f"node-{tag}",
|
|
"desired_state": "running", "state": "running", "message": "started",
|
|
"error": "", "exit_code": None, "at": _ago(9)},
|
|
{"task_id": f"bad-{tag}", "slot": 2, "node_id": f"node-{tag}",
|
|
"desired_state": "shutdown", "state": "failed", "message": "started",
|
|
"error": "task: non-zero exit (1)", "exit_code": 1, "at": _ago(10)},
|
|
{"task_id": f"old-{tag}", "slot": 1, "node_id": f"node-{tag}",
|
|
"desired_state": "shutdown", "state": "shutdown", "message": "shutdown",
|
|
"error": "", "exit_code": 137, "at": _ago(11)}])})
|
|
await s.execute(text(
|
|
"INSERT INTO docker_swarm_nodes (host_id, node_id, hostname, role, availability, "
|
|
"status, leader, updated_at) VALUES (:h, :nid, :hn, 'manager', 'active', 'ready', "
|
|
"true, :now)"), {"h": hid, "nid": f"node-{tag}", "hn": host_name, "now": now})
|
|
await s.execute(text(
|
|
"INSERT INTO docker_disk_usage (host_id, layers_size, images_size, images_reclaimable, "
|
|
"containers_size, volumes_size, build_cache_size, images_total, images_active, "
|
|
"containers_count, volumes_count, updated_at) "
|
|
"VALUES (:h, 1, 5000, 3000, 10, 20, 30, 4, 1, 4, 2, :now)"), {"h": hid, "now": now})
|
|
await s.execute(text(
|
|
"INSERT INTO host_agent_registrations (id, host_id, token_hash, token_created_at, "
|
|
"agent_version, last_seen_at, created_at, updated_at) "
|
|
"VALUES (:id, :h, :th, :t, '1.5.0', :seen, :t, :t)"),
|
|
{"id": str(uuid.uuid4()), "h": hid, "th": uuid.uuid4().hex + uuid.uuid4().hex,
|
|
"t": now - timedelta(days=1), "seen": now - timedelta(seconds=20)})
|
|
await s.execute(text(
|
|
"INSERT INTO host_agent_registrations (id, host_id, token_hash, token_created_at, "
|
|
"created_at, updated_at) VALUES (:id, :h, :th, :t, :t, :t)"),
|
|
{"id": str(uuid.uuid4()), "h": hid2, "th": uuid.uuid4().hex + uuid.uuid4().hex,
|
|
"t": now - timedelta(hours=2)})
|
|
|
|
async def _go():
|
|
await _seed()
|
|
_context._app_ctx.set(app)
|
|
_context._caller_ctx.set(Caller(user_id="u", username="viewer", role=UserRole.viewer, token_id=0))
|
|
|
|
# ── list_containers: problems first, filters ───────────────────────
|
|
mine = [c for c in (await dtools.list_containers(host=host_name, limit=500))["containers"]]
|
|
names = [c["container_name"] for c in mine]
|
|
assert set(names) >= {web, db_, job}
|
|
problems = {c["container_name"]: c["problems"] for c in mine}
|
|
assert problems[db_] == ["unhealthy"] and problems[job] == ["exited_nonzero"]
|
|
assert problems[web] == []
|
|
assert names.index(db_) < names.index(web) and names.index(job) < names.index(web)
|
|
only = (await dtools.list_containers(host=hid, problem="unhealthy"))["containers"]
|
|
assert [c["container_name"] for c in only] == [db_]
|
|
proj = (await dtools.list_containers(compose_project=f"stack-{tag}"))["containers"]
|
|
assert {c["container_name"] for c in proj} == {web, db_}
|
|
with pytest.raises(ValueError, match="is not 'any' or one of"):
|
|
await dtools.list_containers(problem="sad")
|
|
|
|
# ── get_container ──────────────────────────────────────────────────
|
|
c = await dtools.get_container(web, range="1h")
|
|
assert c["host_name"] == host_name and c["status"] == "running"
|
|
assert c["cpu_pct_history"]["max"] == 50.0 and c["cpu_pct_history"]["points"] == 3
|
|
crashed = await dtools.get_container(job, host=host_name)
|
|
assert [e["event"] for e in crashed["events"]] == ["die"]
|
|
# A unique part of a long swarm task name is enough.
|
|
assert (await dtools.get_container(f"{svc}.1.ab"))["swarm_service"] == svc
|
|
with pytest.raises(ValueError, match="list_containers"):
|
|
await dtools.get_container(f"nope-{tag}")
|
|
|
|
# ── search_container_logs ──────────────────────────────────────────
|
|
hit = await dtools.search_container_logs(container=web, text="100%", context=1)
|
|
assert [m["line"] for m in hit["matches"]] == ["ERROR: 100% of disk_used"]
|
|
assert hit["matches"][0]["before"] == ["connected to db"]
|
|
assert hit["matches"][0]["after"] == ["retrying"]
|
|
# % and _ match literally, not as wildcards.
|
|
assert (await dtools.search_container_logs(container=web, text="o_n"))["matches"] == []
|
|
page1 = await dtools.search_container_logs(container=web, limit=2)
|
|
assert page1["truncated"] and len(page1["matches"]) == 2
|
|
assert "characters omitted" in page1["matches"][0]["line"]
|
|
assert len(page1["matches"][0]["line"]) < 2100
|
|
page2 = await dtools.search_container_logs(container=web, limit=2, before=page1["next_before"])
|
|
assert [m["line"] for m in page2["matches"]] == ["ERROR: 100% of disk_used", "connected to db"]
|
|
errs = await dtools.search_container_logs(host=host_name, stream="stderr")
|
|
# Host-wide search reaches a removed container's lines too.
|
|
assert [m["container_name"] for m in errs["matches"]] == [web, removed]
|
|
with pytest.raises(ValueError, match="context needs `container`"):
|
|
await dtools.search_container_logs(text="x", context=2)
|
|
|
|
# ── a container Docker removed is still reachable by name ─────────
|
|
gone = await dtools.get_container(removed)
|
|
assert gone["removed"] is True and gone["host_name"] == host_name
|
|
assert [e["event"] for e in gone["events"]] == ["die"] and "no longer exists" in gone["note"]
|
|
assert (await dtools.get_container(web))["removed"] is False
|
|
lines = await dtools.search_container_logs(container=removed, range="1h")
|
|
assert lines["container_removed"] is True
|
|
assert [m["line"] for m in lines["matches"]] == ["FATAL: cannot open database"]
|
|
# A unique part of its name finds it through its events.
|
|
assert (await dtools.get_container(f"2.gone{tag}"))["container_name"] == removed
|
|
with pytest.raises(ValueError, match="may have been removed"):
|
|
await dtools.get_container(f"nothing-{tag}")
|
|
|
|
# ── a service's name searches every task it has run (#5468) ─────
|
|
both = await dtools.search_container_logs(container=api, text="listening")
|
|
assert [m["line"] for m in both["matches"]] == ["listening (new image)", "listening (old image)"]
|
|
assert [(m["slot"], m["task_id"]) for m in both["matches"]] == [
|
|
("1", f"{tag}{'a' * 17}"), ("1", f"{tag}{'b' * 17}")]
|
|
assert (both["service"], both["tasks_searched"]) == (api, 2)
|
|
assert (await dtools.search_container_logs(container=api, host=quiet_host))["matches"] == []
|
|
# A container's own name still reads as that one container.
|
|
one = await dtools.search_container_logs(container=api_live)
|
|
assert [m["line"] for m in one["matches"]] == ["listening (new image)"] and "service" not in one
|
|
|
|
# ── swarm, disk ────────────────────────────────────────────────────
|
|
sw = await dtools.list_swarm_services(limit=500)
|
|
s = next(x for x in sw["services"] if x["service_name"] == svc)
|
|
assert (s["state"], s["running"], s["desired"]) == ("degraded", 1, 3)
|
|
# Swarm's task records: the replaced task is history, the failed one counts.
|
|
assert [r["container_name"] for r in s["replicas"]] == [f"{svc}.1.abc"]
|
|
assert s["previous_tasks"] == 1 and s["tasks_known"] is True
|
|
assert (s["failures_1h"], s["failures_24h"], s["crash_looping"]) == (1, 1, False)
|
|
assert s["last_error"]["error"] == "task: non-zero exit (1)"
|
|
assert s["last_error"]["node"] == host_name and s["last_error"]["age_s"] >= 600
|
|
# Short of replicas since slot 2's task failed, ten minutes ago.
|
|
assert 600 <= s["trouble_age_s"] < 900
|
|
old = (await dtools.list_containers(host=host_name, status="exited", limit=500))["containers"]
|
|
old = next(c for c in old if c["container_name"] == replaced)
|
|
assert old["swarm_replaced"] is True and old["problems"] == []
|
|
assert any(n["node_id"] == f"node-{tag}" and n["leader"] for n in sw["nodes"])
|
|
disk = (await dtools.get_docker_disk_usage(host=host_name))["hosts"]
|
|
assert disk[0]["images_reclaimable_bytes"] == 3000
|
|
|
|
# ── overview contributions ─────────────────────────────────────────
|
|
async with app.db_sessionmaker() as db:
|
|
findings = await dtools.mcp_overview(db)
|
|
found = {(f["kind"], f["subject"]) for f in findings}
|
|
agent_found = {(f["kind"], f["subject"]) for f in await atools.mcp_overview(db)}
|
|
assert ("container_unhealthy", db_) in found
|
|
assert ("container_exited_nonzero", job) in found
|
|
assert ("swarm_service_degraded", svc) in found
|
|
# Each says how long it has been wrong: the death, the failed task.
|
|
since = {f["subject"]: f["age_s"] for f in findings}
|
|
assert 600 <= since[job] < 900 and 600 <= since[svc] < 900
|
|
assert not any(subject == replaced for _, subject in found)
|
|
assert not any(subject == web for _, subject in found)
|
|
assert ("agent_never_reported", quiet_host) in agent_found
|
|
assert not any(subject == host_name for _, subject in agent_found)
|
|
|
|
# ── the web pages say the same ─────────────────────────────────────
|
|
# A real viewer: every request re-reads the session's user (refresh_session_user).
|
|
from steward.models.users import User
|
|
viewer_id = str(uuid.uuid4())
|
|
async with app.db_sessionmaker() as s, s.begin():
|
|
s.add(User(id=viewer_id, username=f"viewer-{viewer_id[:8]}", email=f"{viewer_id}@test",
|
|
password_hash="x", role=UserRole.viewer, is_active=True))
|
|
client = app.test_client()
|
|
async with client.session_transaction() as sess:
|
|
sess["user_id"], sess["user_role"], sess["username"] = viewer_id, "viewer", "viewer"
|
|
sess["login_at"] = time.time()
|
|
topo = await (await client.get("/plugins/docker/swarm")).get_data(as_text=True)
|
|
assert f'href="/plugins/docker/service/{svc}"' in topo # short of replicas: named first
|
|
assert "task: non-zero exit (1)" in topo
|
|
cards = await (await client.get("/plugins/docker/rows")).get_data(as_text=True)
|
|
assert f'href="/plugins/docker/service/{svc}"' in cards and replaced not in cards
|
|
assert "task: non-zero exit (1)" in cards
|
|
# The overview names the service; its own page names its live replica.
|
|
page = await (await client.get(f"/plugins/docker/service/{svc}")).get_data(as_text=True)
|
|
assert f"{svc}.1.abc" in page and replaced not in page
|
|
assert (await client.get("/plugins/docker/widget")).status_code == 200
|
|
|
|
# ── host agents ────────────────────────────────────────────────────
|
|
agents = {a["host_name"]: a for a in (await atools.list_agents())["agents"]}
|
|
assert agents[host_name]["state"] == "reporting" and agents[host_name]["agent_version"] == "1.5.0"
|
|
assert agents[quiet_host]["state"] == "pending"
|
|
|
|
# ── registration through the plugin hook, gated on enablement ─────
|
|
mcp = app.mcp_server
|
|
mcp.register_plugin("docker", docker)
|
|
mcp.register_plugin("host_agent", agent)
|
|
assert {"list_containers", "search_container_logs", "list_agents"} <= set(mcp.tool_roles)
|
|
assert mcp.tool_plugins["list_containers"] == "docker"
|
|
assert set(mcp.overview_sources) >= {"docker", "host_agent"}
|
|
|
|
asyncio.run(_go())
|