bvandeusenandClaude Opus 5.5
56063f27a4
The agent sent Docker's working set (usage less inactive_file), which still counts recently used file cache. A file-heavy container with no limit read 191 GB while its processes held 2.2 GB, and the host's own figure, which already leaves cache out, disagreed with it. Agent 1.23.0 sends mem_used_bytes / mem_used_pct (anon + shmem, so Postgres's shared buffers count; v1: rss + shmem) and mem_cache_bytes apart. The new names keep the two meanings apart in history and in learned normals: docker_013 renames the columns, clears the old values and adds mem_cache_bytes. The Docker page, swarm by-node sums, Busiest Containers, container detail, normals and the MCP tools all read the new figure. File cache is shown only on the container's page and in get_container. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
329 lines
20 KiB
Python
329 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_used_pct, mem_used_bytes,
|
|
mem_limit_bytes, mem_cache_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, 5000, :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_used_pct", 20.0),
|
|
("mem_used_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
|
|
assert c["mem_used_pct_history"]["max"] == 20.0
|
|
# File cache is in the one container's detail, not in the list rows (#6030).
|
|
assert c["mem_used_bytes"] == 1000 and c["mem_cache_bytes"] == 5000
|
|
assert "mem_cache_bytes" not in proj[0]
|
|
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())
|