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>
854 lines
39 KiB
Python
854 lines
39 KiB
Python
"""Integration: per-host Docker schema + host-scoped persistence.
|
|
|
|
Validates the docker_002 migration (host_id columns + composite key) applied and
|
|
that persist_host_docker scopes containers/metrics by host so same-named
|
|
containers on different hosts no longer collide. Requires STEWARD_DATABASE_URL.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import os
|
|
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)
|
|
|
|
|
|
@_NEEDS_DB
|
|
def test_docker_tables_are_host_scoped(app):
|
|
from sqlalchemy import text
|
|
|
|
async def _go():
|
|
async with app.db_sessionmaker() as s:
|
|
# docker_containers carries a host_id column after docker_002.
|
|
for tbl in ("docker_containers",):
|
|
cols = {r[0] for r in (await s.execute(text(
|
|
"SELECT column_name FROM information_schema.columns "
|
|
"WHERE table_name = :t"), {"t": tbl})).all()}
|
|
assert "host_id" in cols, f"{tbl} missing host_id"
|
|
# docker_containers is keyed by (host_id, name), not name alone.
|
|
pk_cols = {r[0] for r in (await s.execute(text(
|
|
"SELECT a.attname FROM pg_index i "
|
|
"JOIN pg_attribute a ON a.attrelid = i.indrelid "
|
|
" AND a.attnum = ANY(i.indkey) "
|
|
"WHERE i.indrelid = 'docker_containers'::regclass AND i.indisprimary"
|
|
))).all()}
|
|
assert pk_cols == {"host_id", "name"}
|
|
|
|
asyncio.run(_go())
|
|
|
|
|
|
@_NEEDS_DB
|
|
def test_events_and_swarm_tables_exist(app):
|
|
"""docker_004 created the lifecycle + swarm topology tables with host_id FKs."""
|
|
from sqlalchemy import text
|
|
|
|
async def _go():
|
|
async with app.db_sessionmaker() as s:
|
|
for tbl in ("docker_events", "docker_swarm_services", "docker_swarm_nodes"):
|
|
cols = {r[0] for r in (await s.execute(text(
|
|
"SELECT column_name FROM information_schema.columns "
|
|
"WHERE table_name = :t"), {"t": tbl})).all()}
|
|
assert cols, f"{tbl} missing entirely"
|
|
assert "host_id" in cols, f"{tbl} missing host_id"
|
|
# docker_events carries the lifecycle columns the ingest derivation writes.
|
|
ev_cols = {r[0] for r in (await s.execute(text(
|
|
"SELECT column_name FROM information_schema.columns "
|
|
"WHERE table_name = 'docker_events'"))).all()}
|
|
assert {"event", "container_name", "at", "detail"} <= ev_cols
|
|
|
|
asyncio.run(_go())
|
|
|
|
|
|
@_NEEDS_DB
|
|
def test_docker_logs_table_shape(app):
|
|
"""docker_009 created docker_logs: host-scoped, with the twin indexes the
|
|
viewer (host, container, ts) and the age-cutoff prune (ts) rely on."""
|
|
from sqlalchemy import text
|
|
|
|
async def _go():
|
|
async with app.db_sessionmaker() as s:
|
|
cols = {r[0] for r in (await s.execute(text(
|
|
"SELECT column_name FROM information_schema.columns "
|
|
"WHERE table_name = 'docker_logs'"))).all()}
|
|
assert cols, "docker_logs missing entirely"
|
|
assert {"id", "host_id", "container_name", "ts", "stream", "line"} <= cols
|
|
idx = {r[0] for r in (await s.execute(text(
|
|
"SELECT indexname FROM pg_indexes "
|
|
"WHERE tablename = 'docker_logs'"))).all()}
|
|
assert "ix_docker_logs_host_container_time" in idx
|
|
assert "ix_docker_logs_ts" in idx
|
|
|
|
asyncio.run(_go())
|
|
|
|
|
|
def _persist_fn(app):
|
|
"""Resolve persist_host_docker via the registered capability if the docker
|
|
plugin is loaded, else import it directly (the import is safe only when the
|
|
plugin is NOT loaded — otherwise its models are already mapped)."""
|
|
from steward.core.capabilities import has_capability, get_capability
|
|
if has_capability("docker.persist_host_samples"):
|
|
return get_capability("docker.persist_host_samples").fn
|
|
from plugins.docker.ingest import persist_host_docker
|
|
return persist_host_docker
|
|
|
|
|
|
def _retention_fn(app):
|
|
"""Resolve run_docker_retention via capability (or direct import if unloaded)."""
|
|
from steward.core.capabilities import has_capability, get_capability
|
|
if has_capability("docker.run_retention"):
|
|
return get_capability("docker.run_retention").fn
|
|
from plugins.docker.retention import run_docker_retention
|
|
return run_docker_retention
|
|
|
|
|
|
def _prune_logs_fn(app):
|
|
"""Resolve prune_docker_logs via capability (or direct import if unloaded)."""
|
|
from steward.core.capabilities import has_capability, get_capability
|
|
if has_capability("docker.prune_logs"):
|
|
return get_capability("docker.prune_logs").fn
|
|
from plugins.docker.retention import prune_docker_logs
|
|
return prune_docker_logs
|
|
|
|
|
|
@_NEEDS_DB
|
|
def test_persist_scopes_containers_by_host(app):
|
|
from sqlalchemy import text
|
|
from steward.models.hosts import Host
|
|
|
|
persist = _persist_fn(app)
|
|
now = datetime.now(timezone.utc)
|
|
snapshot = [(now, [{
|
|
"name": "web", "container_id": "abc123", "image": "nginx",
|
|
"status": "running", "cpu_pct": 12.5, "mem_used_pct": 30.0,
|
|
"mem_used_bytes": 100, "mem_limit_bytes": 200, "mem_cache_bytes": 900,
|
|
"ports": [], "started_at": None,
|
|
# enrichment (agent ≥ 1.4.0)
|
|
"health": "healthy", "restart_count": 2, "exit_code": 0, "oom_killed": False,
|
|
"compose_project": "stack", "service_name": "web",
|
|
"net_rx_bytes": 1000, "net_tx_bytes": 2000,
|
|
"blk_read_bytes": 4096, "blk_write_bytes": 8192,
|
|
}])]
|
|
|
|
async def _go():
|
|
async with app.db_sessionmaker() as s:
|
|
async with s.begin():
|
|
await s.execute(text("DELETE FROM docker_containers"))
|
|
await s.execute(text(
|
|
"DELETE FROM plugin_metrics WHERE source_module = 'docker'"))
|
|
h1 = Host(id=str(uuid.uuid4()), name="alpha", address="10.0.0.1")
|
|
h2 = Host(id=str(uuid.uuid4()), name="beta", address="10.0.0.2")
|
|
s.add_all([h1, h2])
|
|
await s.flush()
|
|
# Same container name on two hosts — must not collide now.
|
|
await persist(s, h1, snapshot)
|
|
await persist(s, h2, snapshot)
|
|
|
|
web_rows = (await s.execute(text(
|
|
"SELECT COUNT(*) FROM docker_containers WHERE name = 'web'"))).scalar()
|
|
# The series went into the one metric store, host-scoped (#5752),
|
|
# at the snapshot's time.
|
|
metric_hosts = (await s.execute(text(
|
|
"SELECT COUNT(DISTINCT resource_name) FROM plugin_metrics "
|
|
"WHERE source_module = 'docker' AND metric_name = 'mem_used_bytes' "
|
|
"AND recorded_at = :at"), {"at": now})).scalar()
|
|
resources = {r[0] for r in (await s.execute(text(
|
|
"SELECT DISTINCT resource_name FROM plugin_metrics "
|
|
"WHERE source_module = 'docker'"))).all()}
|
|
# Enrichment columns persisted (check one host's row).
|
|
enrich = (await s.execute(text(
|
|
"SELECT health, restart_count, service_name, net_rx_bytes "
|
|
"FROM docker_containers WHERE name = 'web' LIMIT 1"))).first()
|
|
# File cache is kept beside the memory figure, not in it (#6030).
|
|
memory = (await s.execute(text(
|
|
"SELECT mem_used_bytes, mem_used_pct, mem_cache_bytes "
|
|
"FROM docker_containers WHERE name = 'web' LIMIT 1"))).first()
|
|
cache_series = (await s.execute(text(
|
|
"SELECT COUNT(*) FROM plugin_metrics WHERE source_module = 'docker' "
|
|
"AND metric_name = 'mem_cache_bytes' AND value = 900"))).scalar()
|
|
return web_rows, metric_hosts, resources, enrich, memory, cache_series
|
|
|
|
web_rows, metric_hosts, resources, enrich, memory, cache_series = asyncio.run(_go())
|
|
assert web_rows == 2 # one 'web' per host, keyed by (host_id, name)
|
|
assert metric_hosts == 2 # time-series rows scoped per host
|
|
assert resources == {"alpha/web", "beta/web"}
|
|
assert enrich == ("healthy", 2, "web", 1000) # enrichment round-trips
|
|
assert memory == (100, 30.0, 900)
|
|
assert cache_series == 2
|
|
|
|
|
|
@_NEEDS_DB
|
|
def test_persist_logs_stores_lines(app):
|
|
"""Pushed container logs land in docker_logs, host-scoped, one row per line,
|
|
ordered by their own ts; the agent's _steward truncation marker is dropped."""
|
|
from sqlalchemy import text
|
|
from steward.models.hosts import Host
|
|
|
|
persist = _persist_fn(app)
|
|
now = datetime.now(timezone.utc)
|
|
log_batches = [(now, [
|
|
{"container": "web", "stream": "stdout",
|
|
"ts": "2026-01-01T00:00:01+00:00", "line": "started"},
|
|
{"container": "web", "stream": "stderr",
|
|
"ts": "2026-01-01T00:00:02+00:00", "line": "warn: x"},
|
|
{"container": "_steward", "stream": "stderr", "ts": None, "line": "truncated"},
|
|
])]
|
|
|
|
async def _go():
|
|
async with app.db_sessionmaker() as s:
|
|
async with s.begin():
|
|
await s.execute(text("DELETE FROM docker_logs"))
|
|
h = Host(id=str(uuid.uuid4()), name="loghost", address="10.0.0.9")
|
|
s.add(h)
|
|
await s.flush()
|
|
hid = h.id
|
|
# snapshots empty; logs passed as the 6th positional arg.
|
|
await persist(s, h, [], None, None, log_batches)
|
|
rows = (await s.execute(text(
|
|
"SELECT container_name, stream, line FROM docker_logs "
|
|
"WHERE host_id = :h ORDER BY ts"), {"h": hid})).all()
|
|
return [tuple(r) for r in rows]
|
|
|
|
assert asyncio.run(_go()) == [
|
|
("web", "stdout", "started"),
|
|
("web", "stderr", "warn: x"),
|
|
]
|
|
|
|
|
|
@_NEEDS_DB
|
|
def test_persist_logs_skips_lines_a_restarted_agent_sends_again(app):
|
|
"""#5809: a restarted agent has lost its cursors and re-sends each
|
|
container's last lines. Those are stored once; a new line, or another line
|
|
at the same instant, is still kept."""
|
|
from types import SimpleNamespace
|
|
from sqlalchemy import text
|
|
from steward.models.hosts import Host
|
|
|
|
persist = _persist_fn(app)
|
|
now = datetime.now(timezone.utc)
|
|
|
|
def line(container, at, text_, stream="stdout"):
|
|
return {"container": container, "stream": stream,
|
|
"ts": f"2026-01-01T00:00:0{at}+00:00", "line": text_}
|
|
|
|
first = [(now, [line("web", 1, "started"), line("web", 2, "listening"),
|
|
line("db", 1, "ready")])]
|
|
again = [(now, [line("web", 1, "started"), line("web", 2, "listening"),
|
|
line("web", 2, "worker 2 up"), line("web", 3, "GET /"),
|
|
line("db", 1, "ready"), line("db", 1, "ready", stream="stderr")])]
|
|
|
|
hid = str(uuid.uuid4())
|
|
host = SimpleNamespace(id=hid, name="loghost-again") # as below: no expiry across commits
|
|
|
|
async def _go():
|
|
async with app.db_sessionmaker() as s:
|
|
async with s.begin():
|
|
s.add(Host(id=hid, name="loghost-again", address="10.0.0.19"))
|
|
await s.flush()
|
|
await persist(s, host, [], None, None, first)
|
|
async with s.begin():
|
|
await persist(s, host, [], None, None, again)
|
|
rows = (await s.execute(text(
|
|
"SELECT container_name, stream, line FROM docker_logs "
|
|
"WHERE host_id = :h ORDER BY container_name, ts, stream, line"),
|
|
{"h": hid})).all()
|
|
return [tuple(r) for r in rows]
|
|
|
|
assert asyncio.run(_go()) == [
|
|
("db", "stderr", "ready"),
|
|
("db", "stdout", "ready"),
|
|
("web", "stdout", "started"),
|
|
("web", "stdout", "listening"),
|
|
("web", "stdout", "worker 2 up"),
|
|
("web", "stdout", "GET /"),
|
|
]
|
|
|
|
|
|
@_NEEDS_DB
|
|
def test_persist_logs_respects_toggle_and_exclude(app):
|
|
"""Server-side controls (Settings): an excluded container's lines are dropped
|
|
while others persist; the global toggle off stores nothing new."""
|
|
from types import SimpleNamespace
|
|
from sqlalchemy import text
|
|
from steward.models.hosts import Host
|
|
from steward.core.settings import set_setting
|
|
|
|
persist = _persist_fn(app)
|
|
now = datetime.now(timezone.utc)
|
|
batch = [(now, [
|
|
{"container": "web", "stream": "stdout", "ts": None, "line": "keep-me"},
|
|
{"container": "noisy", "stream": "stdout", "ts": None, "line": "drop-me"},
|
|
])]
|
|
|
|
hid = str(uuid.uuid4())
|
|
# Stand-in host: the logs-only persist path reads only host.id, so a plain
|
|
# object sidesteps ORM attribute-expiry across the commit boundaries below.
|
|
host = SimpleNamespace(id=hid, name="loghost3")
|
|
|
|
async def _go():
|
|
async with app.db_sessionmaker() as s:
|
|
# Back-to-back begin blocks with NO read in between — a SELECT between
|
|
# them would autobegin a txn and make the next begin() collide.
|
|
async with s.begin():
|
|
await s.execute(text("DELETE FROM docker_logs"))
|
|
await set_setting(s, "docker.logs.enabled", True)
|
|
await set_setting(s, "docker.logs.exclude", ["noisy"])
|
|
s.add(Host(id=hid, name="loghost3", address="10.7.7.10"))
|
|
await s.flush() # host row before its log rows (FK order)
|
|
await persist(s, host, [], None, None, batch) # web kept, noisy excluded
|
|
async with s.begin(): # global kill-switch off → nothing stored
|
|
await set_setting(s, "docker.logs.enabled", False)
|
|
await persist(s, host, [], None, None, batch)
|
|
async with s.begin(): # restore defaults for other tests
|
|
await set_setting(s, "docker.logs.enabled", True)
|
|
await set_setting(s, "docker.logs.exclude", [])
|
|
# Reads only after the last begin block (they autobegin, but nothing
|
|
# opens a begin() after them). Toggle-off added nothing, so the state
|
|
# here still reflects the exclude phase.
|
|
kept = {r[0] for r in (await s.execute(text(
|
|
"SELECT DISTINCT container_name FROM docker_logs WHERE host_id=:h"),
|
|
{"h": hid})).all()}
|
|
after_off = (await s.execute(text(
|
|
"SELECT COUNT(*) FROM docker_logs WHERE host_id=:h"), {"h": hid})).scalar()
|
|
return kept, after_off
|
|
|
|
kept, after_off = asyncio.run(_go())
|
|
assert kept == {"web"} # excluded 'noisy' dropped on ingest
|
|
assert after_off == 1 # toggle off added nothing → still just the one 'web' line
|
|
|
|
|
|
@_NEEDS_DB
|
|
def test_large_memory_values_persist_as_bigint(app):
|
|
"""A container using >2^31 bytes of RAM must persist. Regression: mem_used_bytes
|
|
/ mem_limit_bytes were int4 and overflowed (asyncpg 'value out of int32 range'),
|
|
failing the whole ingest batch for any host with a >2.1 GB container."""
|
|
from sqlalchemy import text
|
|
from steward.models.hosts import Host
|
|
|
|
persist = _persist_fn(app)
|
|
now = datetime.now(timezone.utc)
|
|
big_usage = 8_185_077_760 # ~8.18 GB — well past int4 max (2_147_483_647)
|
|
big_limit = 17_179_869_184 # 16 GB
|
|
snapshot = [(now, [{
|
|
"name": "bigmem", "container_id": "deadbeef", "image": "postgres",
|
|
"status": "running", "cpu_pct": 1.0, "mem_used_pct": 50.0,
|
|
"mem_used_bytes": big_usage, "mem_limit_bytes": big_limit,
|
|
"ports": [], "started_at": None,
|
|
}])]
|
|
|
|
async def _go():
|
|
async with app.db_sessionmaker() as s:
|
|
async with s.begin():
|
|
await s.execute(text("DELETE FROM docker_containers"))
|
|
h = Host(id=str(uuid.uuid4()), name="bigmemhost", address="10.0.0.9")
|
|
s.add(h)
|
|
await s.flush()
|
|
await persist(s, h, snapshot)
|
|
row = (await s.execute(text(
|
|
"SELECT mem_used_bytes, mem_limit_bytes FROM docker_containers "
|
|
"WHERE name = 'bigmem'"))).first()
|
|
metric = (await s.execute(text(
|
|
"SELECT value FROM plugin_metrics WHERE source_module = 'docker' "
|
|
"AND resource_name = 'bigmemhost/bigmem' "
|
|
"AND metric_name = 'mem_used_bytes'"))).scalar()
|
|
return row, metric
|
|
|
|
row, metric = asyncio.run(_go())
|
|
assert row == (big_usage, big_limit) # current-state row round-trips
|
|
assert metric == big_usage # time-series row round-trips
|
|
|
|
|
|
def test_lifecycle_events_derived_across_snapshots(app):
|
|
from sqlalchemy import text
|
|
from steward.models.hosts import Host
|
|
|
|
persist = _persist_fn(app)
|
|
t1 = datetime(2026, 6, 19, 12, 0, 0, tzinfo=timezone.utc)
|
|
t2 = datetime(2026, 6, 19, 12, 0, 30, tzinfo=timezone.utc)
|
|
running = {"name": "web", "container_id": "abc", "image": "nginx",
|
|
"status": "running", "cpu_pct": 5.0, "mem_used_pct": 10.0,
|
|
"restart_count": 0, "exit_code": None, "oom_killed": False,
|
|
"health": "healthy"}
|
|
died = {**running, "status": "exited", "cpu_pct": None,
|
|
"exit_code": 137, "oom_killed": True, "health": None}
|
|
|
|
async def _go():
|
|
async with app.db_sessionmaker() as s:
|
|
async with s.begin():
|
|
await s.execute(text("DELETE FROM docker_events"))
|
|
await s.execute(text("DELETE FROM docker_containers"))
|
|
h = Host(id=str(uuid.uuid4()), name="evt", address="10.9.9.9")
|
|
s.add(h)
|
|
await s.flush()
|
|
# Baseline snapshot establishes state — must NOT emit a start.
|
|
await persist(s, h, [(t1, [running])])
|
|
# Second snapshot: the container OOM-died.
|
|
await persist(s, h, [(t2, [died])])
|
|
hid = h.id
|
|
rows = (await s.execute(text(
|
|
"SELECT event, detail FROM docker_events WHERE host_id = :h "
|
|
"ORDER BY event"), {"h": hid})).all()
|
|
return [(r[0], r[1]) for r in rows]
|
|
|
|
events = asyncio.run(_go())
|
|
# baseline emitted nothing; the death emitted both die (exit 137) and oom.
|
|
assert ("die", "exit 137") in events
|
|
assert ("oom", None) in events
|
|
assert not any(e[0] == "start" for e in events)
|
|
|
|
|
|
@_NEEDS_DB
|
|
def test_vanished_container_is_reaped(app):
|
|
"""A container absent from the latest snapshot is deleted, not left behind as
|
|
a permanent stopped row. Regression: removed one-shot containers (CI jobs,
|
|
buildkit builders) accumulated forever and inflated the dashboard counts."""
|
|
from sqlalchemy import text
|
|
from steward.models.hosts import Host
|
|
|
|
persist = _persist_fn(app)
|
|
t1 = datetime(2026, 6, 19, 12, 0, 0, tzinfo=timezone.utc)
|
|
t2 = datetime(2026, 6, 19, 12, 0, 30, tzinfo=timezone.utc)
|
|
base = {"status": "running", "cpu_pct": 1.0, "mem_used_pct": 1.0,
|
|
"restart_count": 0, "exit_code": None, "oom_killed": False, "health": None}
|
|
keep = {**base, "name": "keep"}
|
|
gone = {**base, "name": "ci-job-123"}
|
|
|
|
async def _go():
|
|
async with app.db_sessionmaker() as s:
|
|
async with s.begin():
|
|
await s.execute(text("DELETE FROM docker_containers"))
|
|
h = Host(id=str(uuid.uuid4()), name="reap", address="10.9.9.8")
|
|
s.add(h)
|
|
await s.flush()
|
|
await persist(s, h, [(t1, [keep, gone])]) # both present
|
|
before = [r[0] for r in (await s.execute(text(
|
|
"SELECT name FROM docker_containers WHERE host_id = :h ORDER BY name"),
|
|
{"h": h.id})).all()]
|
|
await persist(s, h, [(t2, [keep])]) # ci-job-123 vanished
|
|
after = [r[0] for r in (await s.execute(text(
|
|
"SELECT name FROM docker_containers WHERE host_id = :h ORDER BY name"),
|
|
{"h": h.id})).all()]
|
|
return before, after
|
|
|
|
before, after = asyncio.run(_go())
|
|
assert before == ["ci-job-123", "keep"]
|
|
assert after == ["keep"] # the vanished container was reaped
|
|
|
|
|
|
@_NEEDS_DB
|
|
def test_persist_marks_each_part_for_the_slow_ingest_line(app):
|
|
"""Each part of the docker persist is marked, so the slow-ingest line
|
|
names the part its time went to (#5181)."""
|
|
from steward.models.hosts import Host
|
|
|
|
class _Marks:
|
|
def __init__(self):
|
|
self.names = []
|
|
|
|
def mark(self, name):
|
|
self.names.append(name)
|
|
|
|
persist = _persist_fn(app)
|
|
at = datetime(2026, 10, 6, 23, 0, 0, tzinfo=timezone.utc)
|
|
c = {"name": "web", "status": "running", "cpu_pct": 1.0, "mem_used_pct": 1.0,
|
|
"restart_count": 0, "exit_code": None, "oom_killed": False, "health": None}
|
|
logs = [(at, [{"container": "web", "stream": "stdout", "ts": None, "line": "hello"}])]
|
|
marks = _Marks()
|
|
|
|
async def _go():
|
|
async with app.db_sessionmaker() as s:
|
|
async with s.begin():
|
|
h = Host(id=str(uuid.uuid4()), name="marks", address="10.9.9.7")
|
|
s.add(h)
|
|
await s.flush()
|
|
await persist(s, h, [(at, [c])], None, None, logs,
|
|
{"status": "ok", "detail": None, "at": at, "swarm_manager": None},
|
|
phases=marks)
|
|
|
|
asyncio.run(_go())
|
|
assert marks.names == ["docker.status", "docker.logs", "docker.metrics", "docker.events",
|
|
"docker.containers", "docker.alerts", "docker.reap"]
|
|
|
|
|
|
@_NEEDS_DB
|
|
def test_an_empty_snapshot_from_a_docker_that_answered_clears_the_host(app):
|
|
"""Docker answering with no containers removes the host's rows (#5152); a
|
|
sample with no Docker answer at all removes nothing."""
|
|
from sqlalchemy import text
|
|
from steward.models.hosts import Host
|
|
|
|
persist = _persist_fn(app)
|
|
t1 = datetime(2026, 10, 6, 12, 0, 0, tzinfo=timezone.utc)
|
|
t2 = datetime(2026, 10, 6, 12, 0, 30, tzinfo=timezone.utc)
|
|
web = {"name": "web", "status": "running", "cpu_pct": 1.0, "mem_used_pct": 1.0,
|
|
"restart_count": 0, "exit_code": None, "oom_killed": False, "health": None}
|
|
|
|
async def _go():
|
|
async with app.db_sessionmaker() as s:
|
|
async with s.begin():
|
|
h = Host(id=str(uuid.uuid4()), name=f"empty-{uuid.uuid4().hex[:6]}", address="10.9.9.7")
|
|
s.add(h)
|
|
await s.flush()
|
|
|
|
async def names():
|
|
return [r[0] for r in (await s.execute(text(
|
|
"SELECT name FROM docker_containers WHERE host_id = :h"), {"h": h.id})).all()]
|
|
|
|
await persist(s, h, [(t1, [web])], status={"status": "ok", "detail": None, "at": t1})
|
|
first = await names()
|
|
await persist(s, h, [], status={"status": "error", "detail": "timed out", "at": t2})
|
|
after_error = await names()
|
|
await persist(s, h, [(t2, [])], status={"status": "ok", "detail": None, "at": t2})
|
|
return first, after_error, await names()
|
|
|
|
first, after_error, after_empty = asyncio.run(_go())
|
|
assert first == ["web"]
|
|
assert after_error == ["web"] # couldn't ask: keep what Docker last said
|
|
assert after_empty == [] # asked, and there are none
|
|
|
|
|
|
@_NEEDS_DB
|
|
def test_docker_status_is_recorded_once_a_host_has_had_docker(app):
|
|
from sqlalchemy import text
|
|
from steward.models.hosts import Host
|
|
|
|
persist = _persist_fn(app)
|
|
t1 = datetime(2026, 10, 6, 12, 0, 0, tzinfo=timezone.utc)
|
|
t2 = t1 + timedelta(seconds=30)
|
|
t3 = t1 + timedelta(seconds=60)
|
|
|
|
async def _go():
|
|
seen = []
|
|
async with app.db_sessionmaker() as s:
|
|
async with s.begin():
|
|
never = Host(id=str(uuid.uuid4()), name=f"plain-{uuid.uuid4().hex[:6]}", address="10.9.9.5")
|
|
had = Host(id=str(uuid.uuid4()), name=f"had-{uuid.uuid4().hex[:6]}", address="10.9.9.6")
|
|
s.add_all([never, had])
|
|
await s.flush()
|
|
gone = {"status": "unavailable", "detail": "no Docker socket at /var/run/docker.sock"}
|
|
await persist(s, never, [], status=dict(gone, at=t1))
|
|
await s.flush()
|
|
|
|
async def row(host):
|
|
return (await s.execute(text(
|
|
"SELECT status, detail, since, reported_at FROM docker_host_status "
|
|
"WHERE host_id = :h"), {"h": host.id})).first()
|
|
|
|
seen.append(await row(never))
|
|
for status, at in (({"status": "ok", "detail": None}, t1),
|
|
(gone, t2), (gone, t3),
|
|
({"status": "error", "detail": "permission denied on /var/run/docker.sock"}, t3)):
|
|
await persist(s, had, [], status=dict(status, at=at))
|
|
await s.flush()
|
|
seen.append(tuple(await row(had)))
|
|
return seen
|
|
|
|
never, ok, gone1, gone2, err = asyncio.run(_go())
|
|
assert never is None # never had Docker: nothing to say
|
|
assert ok == ("ok", None, t1, t1)
|
|
assert gone1 == ("unavailable", "no Docker socket at /var/run/docker.sock", t2, t2)
|
|
assert gone2[2] == t2 and gone2[3] == t3 # still gone: since holds, the report moves on
|
|
assert err[:3] == ("error", "permission denied on /var/run/docker.sock", t3)
|
|
|
|
|
|
@_NEEDS_DB
|
|
def test_a_node_that_is_no_longer_a_manager_loses_its_swarm_rows(app):
|
|
from sqlalchemy import text
|
|
from steward.models.hosts import Host
|
|
|
|
persist = _persist_fn(app)
|
|
t1 = datetime(2026, 10, 6, 12, 0, 0, tzinfo=timezone.utc)
|
|
swarm = {"services": [{"service_name": "web", "mode": "replicated", "desired": 1, "running": 1}],
|
|
"nodes": [{"node_id": "n1", "hostname": "mgr", "role": "manager"}]}
|
|
|
|
async def _go():
|
|
async with app.db_sessionmaker() as s:
|
|
async with s.begin():
|
|
h = Host(id=str(uuid.uuid4()), name=f"demoted-{uuid.uuid4().hex[:6]}", address="10.9.9.4")
|
|
s.add(h)
|
|
await s.flush()
|
|
|
|
async def counts():
|
|
out = []
|
|
for t in ("docker_swarm_services", "docker_swarm_nodes"):
|
|
out.append((await s.execute(text(
|
|
f"SELECT count(*) FROM {t} WHERE host_id = :h"), {"h": h.id})).scalar())
|
|
return tuple(out)
|
|
|
|
await persist(s, h, [], swarm, status={"status": "ok", "detail": None, "at": t1})
|
|
manager = await counts()
|
|
# /info failed: not knowing is not "no longer a manager".
|
|
await persist(s, h, [], status={"status": "error", "detail": "x", "at": t1,
|
|
"swarm_manager": None})
|
|
unknown = await counts()
|
|
await persist(s, h, [], status={"status": "ok", "detail": None, "at": t1,
|
|
"swarm_manager": False})
|
|
return manager, unknown, await counts()
|
|
|
|
manager, unknown, demoted = asyncio.run(_go())
|
|
assert manager == (1, 1)
|
|
assert unknown == (1, 1)
|
|
assert demoted == (0, 0)
|
|
|
|
|
|
@_NEEDS_DB
|
|
def test_swarm_topology_persisted(app):
|
|
from sqlalchemy import text
|
|
from steward.models.hosts import Host
|
|
|
|
persist = _persist_fn(app)
|
|
swarm = {
|
|
"services": [{"service_name": "web", "mode": "replicated",
|
|
"desired": 3, "running": 2, "image": "nginx",
|
|
"placement": [{"node_id": "n1", "running": 2}],
|
|
"tasks": [{"task_id": "t1", "slot": 3, "node_id": "n1",
|
|
"desired_state": "shutdown", "state": "failed",
|
|
"message": "started", "error": "task: non-zero exit (1)",
|
|
"exit_code": 1, "at": "2026-10-06T13:19:51Z"}]}],
|
|
"nodes": [{"node_id": "n1", "hostname": "mgr", "role": "manager",
|
|
"availability": "active", "status": "ready", "leader": True}],
|
|
}
|
|
|
|
async def _go():
|
|
async with app.db_sessionmaker() as s:
|
|
async with s.begin():
|
|
await s.execute(text("DELETE FROM docker_swarm_services"))
|
|
await s.execute(text("DELETE FROM docker_swarm_nodes"))
|
|
h = Host(id=str(uuid.uuid4()), name="mgr", address="10.8.8.8")
|
|
s.add(h)
|
|
await s.flush()
|
|
# No containers in this sample — swarm must still persist.
|
|
from steward.core.plugin_manager import loaded_plugin_submodule
|
|
loaded_plugin_submodule("docker", "swarm_health")._last_recorded = None
|
|
since = datetime.now(timezone.utc)
|
|
await persist(s, h, [], swarm)
|
|
hid = h.id
|
|
# The swarm-wide figures alert rules watch (#5788): web is 2 of 3.
|
|
figures = dict((await s.execute(text(
|
|
"SELECT metric_name, value FROM plugin_metrics WHERE source_module = 'docker' "
|
|
"AND resource_name = 'swarm' AND recorded_at >= :t"), {"t": since})).all())
|
|
svc = (await s.execute(text(
|
|
"SELECT mode, desired, running, image, placement_json, tasks_json "
|
|
"FROM docker_swarm_services WHERE host_id = :h"), {"h": hid})).first()
|
|
node = (await s.execute(text(
|
|
"SELECT role, availability, status, leader "
|
|
"FROM docker_swarm_nodes WHERE host_id = :h"), {"h": hid})).first()
|
|
return svc, node, figures
|
|
|
|
svc, node, figures = asyncio.run(_go())
|
|
assert figures == {"swarm_services_short": 1.0, "swarm_nodes_down": 0.0,
|
|
"swarm_managers_unreachable": 0.0}
|
|
assert svc[0] == "replicated" and svc[1] == 3 and svc[2] == 2 and svc[3] == "nginx"
|
|
assert "n1" in svc[4] # placement JSON carries the node id
|
|
(task,) = json.loads(svc[5]) # docker_010: the manager's task records survive
|
|
assert task["error"] == "task: non-zero exit (1)" and task["slot"] == 3
|
|
assert node[0] == "manager" and node[2] == "ready" and node[3] is True
|
|
|
|
|
|
@_NEEDS_DB
|
|
def test_disk_usage_persisted(app):
|
|
"""persist_host_docker stores the /system/df summary + image rows, and
|
|
replaces the image set on the next report (stale images pruned)."""
|
|
from sqlalchemy import text
|
|
from steward.models.hosts import Host
|
|
|
|
persist = _persist_fn(app)
|
|
disk1 = {
|
|
"layers_size": 1000, "images_total": 2, "images_active": 1,
|
|
"images_size": 1000, "images_reclaimable": 700,
|
|
"containers_count": 2, "containers_size": 100,
|
|
"volumes_count": 1, "volumes_size": 40, "build_cache_size": 15,
|
|
"images": [
|
|
{"image_id": "aaaaaaaaaaaa", "repo_tag": "app:1", "size": 300,
|
|
"shared_size": 50, "containers": 1},
|
|
{"image_id": "bbbbbbbbbbbb", "repo_tag": "<none>", "size": 700,
|
|
"shared_size": 0, "containers": 0},
|
|
],
|
|
}
|
|
disk2 = {**disk1, "images_reclaimable": 0, "images": [disk1["images"][0]]}
|
|
|
|
async def _go():
|
|
async with app.db_sessionmaker() as s:
|
|
async with s.begin():
|
|
await s.execute(text("DELETE FROM docker_images"))
|
|
await s.execute(text("DELETE FROM docker_disk_usage"))
|
|
hid = str(uuid.uuid4())
|
|
h = Host(id=hid, name=f"dsk-{hid[:8]}", address="10.6.6.6")
|
|
s.add(h)
|
|
await s.flush()
|
|
hid = h.id
|
|
await persist(s, h, [], None, disk1)
|
|
summary = (await s.execute(text(
|
|
"SELECT images_reclaimable, images_total, build_cache_size "
|
|
"FROM docker_disk_usage WHERE host_id=:h"), {"h": hid})).first()
|
|
img_count = (await s.execute(text(
|
|
"SELECT COUNT(*) FROM docker_images WHERE host_id=:h"), {"h": hid})).scalar()
|
|
# End the autobegun read transaction before opening the next write one.
|
|
await s.rollback()
|
|
# Re-report with one image dropped — the stale row must be pruned.
|
|
async with s.begin():
|
|
await persist(s, await s.get(Host, hid), [], None, disk2)
|
|
after = (await s.execute(text(
|
|
"SELECT image_id FROM docker_images WHERE host_id=:h"), {"h": hid})).all()
|
|
reclaimable2 = (await s.execute(text(
|
|
"SELECT images_reclaimable FROM docker_disk_usage WHERE host_id=:h"),
|
|
{"h": hid})).scalar()
|
|
# Each report is recorded as metrics too, resource = the host (#5304).
|
|
metrics = (await s.execute(text(
|
|
"SELECT metric_name, value FROM plugin_metrics WHERE source_module = 'docker' "
|
|
"AND resource_name = :r"), {"r": f"dsk-{hid[:8]}"})).all()
|
|
return summary, img_count, [r[0] for r in after], reclaimable2, metrics
|
|
|
|
summary, img_count, after_ids, reclaimable2, metrics = asyncio.run(_go())
|
|
assert {v for m, v in metrics if m == "disk_total_bytes"} == {1155.0} # 1000 + 100 + 40 + 15
|
|
assert sorted(v for m, v in metrics if m == "disk_reclaimable_bytes") == [0.0, 700.0]
|
|
assert {m for m, _ in metrics} == {"disk_images_bytes", "disk_containers_bytes", "disk_volumes_bytes",
|
|
"disk_build_cache_bytes", "disk_total_bytes", "disk_reclaimable_bytes"}
|
|
assert summary[0] == 700 and summary[1] == 2 and summary[2] == 15
|
|
assert img_count == 2
|
|
assert after_ids == ["aaaaaaaaaaaa"] # the unreferenced image pruned on re-report
|
|
assert reclaimable2 == 0 # summary upserted in place
|
|
|
|
|
|
@_NEEDS_DB
|
|
def test_retention_prunes_events_past_their_window(app):
|
|
"""Docker retention prunes lifecycle events past the events window; the
|
|
container series roll up with the one metric store (#5752)."""
|
|
from sqlalchemy import text
|
|
from steward.models.hosts import Host
|
|
|
|
run_retention = _retention_fn(app)
|
|
now = datetime(2026, 6, 19, 12, 0, 0, tzinfo=timezone.utc)
|
|
old_event = datetime(2026, 5, 1, 0, 0, 0, tzinfo=timezone.utc) # > events window
|
|
new_event = datetime(2026, 6, 18, 0, 0, 0, tzinfo=timezone.utc) # inside events window
|
|
|
|
async def _go():
|
|
async with app.db_sessionmaker() as s:
|
|
async with s.begin():
|
|
await s.execute(text("DELETE FROM docker_events"))
|
|
h = Host(id=str(uuid.uuid4()), name="ret", address="10.7.7.7")
|
|
s.add(h)
|
|
await s.flush()
|
|
hid = h.id
|
|
for ev_at, ev in [(old_event, "stop"), (new_event, "start")]:
|
|
await s.execute(text(
|
|
"INSERT INTO docker_events (id, host_id, container_name, event, at) "
|
|
"VALUES (:id,:h,'web',:ev,:at)"),
|
|
{"id": str(uuid.uuid4()), "h": hid, "ev": ev, "at": ev_at})
|
|
async with s.begin():
|
|
counts = await run_retention(s, events_days=30, now=now)
|
|
events_left = (await s.execute(text(
|
|
"SELECT COUNT(*) FROM docker_events WHERE host_id=:h"), {"h": hid})).scalar()
|
|
tables = {r[0] for r in (await s.execute(text(
|
|
"SELECT table_name FROM information_schema.tables "
|
|
"WHERE table_name LIKE 'docker_metrics%'"))).all()}
|
|
return counts, events_left, tables
|
|
|
|
counts, events_left, tables = asyncio.run(_go())
|
|
assert events_left == 1 # only the in-window event survives
|
|
assert counts == {"events_pruned": 1}
|
|
assert tables == set() # docker_012 dropped both tables
|
|
|
|
|
|
@_NEEDS_DB
|
|
def test_retention_logs_ring(app):
|
|
"""docker_logs ring (m79): lines past the age window are pruned; within it,
|
|
each container keeps only the newest ~cap bytes; containers are independent,
|
|
including a same-named container on another host (#5449 ranks only the
|
|
containers over the cap, joined on host and name)."""
|
|
from datetime import timedelta
|
|
from sqlalchemy import text
|
|
from steward.models.hosts import Host
|
|
|
|
prune_logs = _prune_logs_fn(app)
|
|
now = datetime(2026, 6, 19, 12, 0, 0, tzinfo=timezone.utc)
|
|
line40 = "x" * 40 # 40 chars/line; cap=100 keeps 3 lines (excl-prefix 0/40/80)
|
|
|
|
async def _go():
|
|
async with app.db_sessionmaker() as s:
|
|
async with s.begin():
|
|
await s.execute(text("DELETE FROM docker_logs"))
|
|
h = Host(id=str(uuid.uuid4()), name="loghost2", address="10.7.7.8")
|
|
h2 = Host(id=str(uuid.uuid4()), name="loghost3", address="10.7.7.9")
|
|
s.add_all([h, h2])
|
|
await s.flush()
|
|
hid, hid2 = h.id, h2.id
|
|
|
|
def ins(cn, ts, line, host=None):
|
|
return s.execute(text(
|
|
"INSERT INTO docker_logs "
|
|
"(id, host_id, container_name, ts, stream, line) "
|
|
"VALUES (:id,:h,:cn,:ts,'stdout',:line)"),
|
|
{"id": str(uuid.uuid4()), "h": host or hid, "cn": cn,
|
|
"ts": ts, "line": line})
|
|
|
|
# Past the 3-day age window → age-pruned.
|
|
await ins("old", now - timedelta(days=10), "z")
|
|
# 5 recent 'web' lines (40 bytes each = 200 > cap 100) → keep 3.
|
|
for i in range(1, 6):
|
|
await ins("web", now - timedelta(minutes=6 - i), line40)
|
|
# 2 recent 'db' lines (80 bytes ≤ cap) → both survive (isolation).
|
|
await ins("db", now - timedelta(minutes=2), line40)
|
|
await ins("db", now - timedelta(minutes=1), line40)
|
|
# 'web' on the other host: 2 lines, under the cap → both survive.
|
|
await ins("web", now - timedelta(minutes=2), line40, host=hid2)
|
|
await ins("web", now - timedelta(minutes=1), line40, host=hid2)
|
|
|
|
async with s.begin():
|
|
counts = await prune_logs(
|
|
s, retention_days=3, max_bytes_per_container=100, now=now)
|
|
|
|
def n(cn, host=None):
|
|
return s.execute(text(
|
|
"SELECT COUNT(*) FROM docker_logs WHERE host_id=:h AND container_name=:cn"),
|
|
{"h": host or hid, "cn": cn})
|
|
web = (await n("web")).scalar()
|
|
db = (await n("db")).scalar()
|
|
old = (await n("old")).scalar()
|
|
web2 = (await n("web", hid2)).scalar()
|
|
return counts, web, db, old, web2
|
|
|
|
counts, web, db, old, web2 = asyncio.run(_go())
|
|
assert old == 0 # age window
|
|
assert web == 3 # newest ~100 bytes kept, oldest 2 dropped
|
|
assert db == 2 # under cap → untouched (per-container)
|
|
assert web2 == 2 # same name, other host → its own ring
|
|
assert counts["logs_age_pruned"] == 1
|
|
assert counts["logs_size_pruned"] == 2
|
|
|
|
|
|
def test_widget_dedup_collapses_cross_manager_duplicates():
|
|
"""The same swarm task is reported by every manager (identical container_id);
|
|
the dashboard widget must count it once. Older agents send no container_id,
|
|
so those rows can't be matched and are all kept."""
|
|
from types import SimpleNamespace
|
|
from plugins.docker.dedup import dedup_by_container_id
|
|
|
|
def c(cid, name, status="running"):
|
|
return SimpleNamespace(container_id=cid, name=name, status=status)
|
|
|
|
rows = [c("abc", "web@m1"), c("abc", "web@m2"), # same task, two managers
|
|
c("def", "db"), c("", "noid-1"), c("", "noid-2")]
|
|
out = dedup_by_container_id(rows)
|
|
|
|
assert len(out) == 4 # one "abc" dropped
|
|
abc = [r for r in out if r.container_id == "abc"]
|
|
assert len(abc) == 1 and abc[0].name == "web@m1" # first occurrence wins
|
|
assert sum(1 for r in out if r.container_id == "") == 2 # id-less rows all kept
|