bvandeusenandClaude Opus 5.5
1ba4784fa6
On 2026-10-18 an update on VDNT-Docker02 restarted the Docker daemon six seconds in, before the drain that came only with the reboot, and every container on the node stopped. Docker's own packages restart the daemon from their install scripts; needrestart's list-only mode doesn't stop that. - The helper gains a `check` action (agent 1.24.0): a dry run of the update for each package manager, whether Docker is among it, and the running containers outside the swarm with their restart policies. - A swarm node is drained before an update that upgrades Docker, else only before a reboot, and at most once a job. - After a drain the job settles 5 minutes before the update or reboot. - Once the work is over, the containers that ran outside the swarm must be seen running again within 10 minutes, or the job fails naming them. - An older agent gets no check and its job runs as before. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
223 lines
10 KiB
Python
223 lines
10 KiB
Python
"""Integration (#5767): Steward hands a host a maintenance job and follows it
|
|
to the end, through a restart of its own, from the agent's posts alone.
|
|
|
|
Rows are seeded and read with plain SQL: importing the plugin's models again
|
|
after the app loaded the plugin raises "Table already defined" (see
|
|
test_host_agent_alerts)."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import hashlib
|
|
import os
|
|
import uuid
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
|
|
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)
|
|
|
|
|
|
async def _seed(app):
|
|
from sqlalchemy import text
|
|
from steward.models.hosts import Host
|
|
from steward.models.users import User
|
|
tag = uuid.uuid4().hex[:8]
|
|
token = f"tok-{tag}-{uuid.uuid4().hex}"
|
|
job_id = str(uuid.uuid4())
|
|
async with app.db_sessionmaker() as s:
|
|
async with s.begin():
|
|
user = User(username=f"mj-{tag}", email=f"mj-{tag}@test", password_hash="x")
|
|
host = Host(name=f"maint-{tag}", address="10.9.9.7")
|
|
s.add_all([user, host])
|
|
await s.flush()
|
|
await s.execute(text(
|
|
"INSERT INTO host_agent_registrations "
|
|
"(id, host_id, token_hash, token_created_at, last_seen_at, maint_helper, "
|
|
" created_at, updated_at) "
|
|
"VALUES (:id, :host, :hash, now(), now(), true, now(), now())"),
|
|
{"id": str(uuid.uuid4()), "host": host.id,
|
|
"hash": hashlib.sha256(token.encode()).hexdigest()})
|
|
await s.execute(text(
|
|
"INSERT INTO host_maintenance_jobs "
|
|
"(id, host_id, state, state_at, reboot_policy, step_seen, created_by, created_at) "
|
|
"VALUES (:id, :host, 'queued', now(), 'if_required', false, :user, now())"),
|
|
{"id": job_id, "host": host.id, "user": user.id})
|
|
host_id = host.id
|
|
return host_id, token, job_id
|
|
|
|
|
|
async def _post(app, token, events=(), sample=True):
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
resp = await app.test_client().post(
|
|
"/plugins/host_agent/ingest",
|
|
json={"samples": [{"ts": now, "cpu_pct": 2.0}] if sample else [],
|
|
"events": list(events)},
|
|
headers={"Authorization": f"Bearer {token}"})
|
|
assert resp.status_code == 200, await resp.get_data(as_text=True)
|
|
return (await resp.get_json()).get("maintenance", [])
|
|
|
|
|
|
async def _job(app, job_id):
|
|
from sqlalchemy import text
|
|
async with app.db_sessionmaker() as s:
|
|
async with s.begin():
|
|
return (await s.execute(text(
|
|
"SELECT j.state, j.packages, j.reboot_needed, j.back_at, w.end_at "
|
|
"FROM host_maintenance_jobs j LEFT JOIN maintenance_windows w ON w.id = j.window_id "
|
|
"WHERE j.id = :id"), {"id": job_id})).one()
|
|
|
|
|
|
def test_a_job_is_followed_through_a_steward_restart_and_the_host_reboot():
|
|
app = _app()
|
|
|
|
async def _first_half():
|
|
host_id, token, job_id = await _seed(app)
|
|
# The agent's next post starts the job: update first, under a window.
|
|
assert await _post(app, token) == [{"id": f"{job_id}-update", "action": "update"}]
|
|
state, *_, window_end = await _job(app, job_id)
|
|
assert state == "updating" and window_end > datetime.now(timezone.utc)
|
|
# The helper took it: the step is no longer handed out.
|
|
running = {"kind": "maintenance", "id": f"{job_id}-update", "action": "update",
|
|
"state": "running"}
|
|
assert await _post(app, token, [running]) == []
|
|
return token, job_id
|
|
|
|
token, job_id = asyncio.run(_first_half())
|
|
|
|
# Steward restarts: a new process, nothing in memory but the database.
|
|
app = _app()
|
|
|
|
async def _second_half():
|
|
done = {"kind": "maintenance", "id": f"{job_id}-update", "action": "update",
|
|
"state": "done", "ok": True, "packages": 12, "reboot_required": True}
|
|
assert await _post(app, token, [done]) == [{"id": f"{job_id}-reboot", "action": "reboot"}]
|
|
assert (await _job(app, job_id))[:3] == ("rebooting", 12, True)
|
|
# The agent restarting on a host that stayed up doesn't count as the reboot.
|
|
old_boot = datetime.now(timezone.utc) - timedelta(days=3)
|
|
await _post(app, token, [{"kind": "started", "at": datetime.now(timezone.utc).isoformat(),
|
|
"boot_at": old_boot.isoformat()}])
|
|
assert (await _job(app, job_id))[0] == "rebooting"
|
|
# After the reboot: the helper's result and the agent's boot, in one post.
|
|
boot = datetime.now(timezone.utc)
|
|
rebooted = {"kind": "maintenance", "id": f"{job_id}-reboot", "action": "reboot",
|
|
"state": "rebooting", "ok": True}
|
|
started = {"kind": "started", "at": (boot + timedelta(seconds=20)).isoformat(),
|
|
"boot_at": boot.isoformat()}
|
|
assert await _post(app, token, [rebooted, started]) == []
|
|
state, packages, reboot_needed, back_at, window_end = await _job(app, job_id)
|
|
assert state == "done" and back_at is not None
|
|
assert window_end <= datetime.now(timezone.utc)
|
|
|
|
asyncio.run(_second_half())
|
|
|
|
|
|
def test_a_swarm_node_s_job_also_holds_the_swarm_wide_alerts():
|
|
"""Draining and rebooting a node moves the swarm figures; that is the job,
|
|
not news (#5788). The window over them follows the job's own, and ends
|
|
with it."""
|
|
from types import SimpleNamespace
|
|
|
|
from sqlalchemy import text
|
|
from steward.core.plugin_manager import loaded_plugin_submodule
|
|
from steward.models.users import User
|
|
app = _app()
|
|
maintenance = loaded_plugin_submodule("host_agent", "maintenance")
|
|
tag = uuid.uuid4().hex[:8]
|
|
host_name = f"swarm-node-{tag}"
|
|
|
|
async def _go():
|
|
now = datetime.now(timezone.utc)
|
|
async with app.db_sessionmaker() as s:
|
|
async with s.begin():
|
|
user = User(username=f"sw-{tag}", email=f"sw-{tag}@test", password_hash="x")
|
|
s.add(user)
|
|
await s.flush()
|
|
job = SimpleNamespace(window_id=None, created_by=user.id, state="draining",
|
|
node_id="node-1", created_at=now - timedelta(seconds=1))
|
|
await maintenance._window(s, job, host_name, now)
|
|
await maintenance._window(s, job, host_name, now + timedelta(minutes=1)) # the next tick
|
|
job.state = "done"
|
|
await maintenance._window(s, job, host_name, now + timedelta(minutes=2)) # finished
|
|
return (await s.execute(text(
|
|
"SELECT scope_source, scope_resource, end_at FROM maintenance_windows "
|
|
"WHERE name LIKE :n ORDER BY scope_source"), {"n": f"Maintenance: {host_name}%"})).all(), now
|
|
|
|
rows, now = asyncio.run(_go())
|
|
assert [(r[0], r[1]) for r in rows] == [("docker", "swarm"), ("host_agent", host_name)]
|
|
assert all(r[2] == now + timedelta(minutes=2) for r in rows) # ended with the job, not before
|
|
|
|
|
|
def test_the_tick_ends_the_wait_after_a_drain_and_waits_for_the_containers():
|
|
"""#6098: settling and recovering are Steward's own waits. The minute tick
|
|
moves a drained node on after 5 minutes, and finishes a job only once the
|
|
containers that ran outside the swarm before it are seen running again."""
|
|
import json
|
|
from sqlalchemy import text
|
|
from steward.core.plugin_manager import loaded_plugin_submodule
|
|
app = _app()
|
|
m = loaded_plugin_submodule("host_agent", "maintenance")
|
|
|
|
async def _go():
|
|
host_id, _token, job_id = await _seed(app)
|
|
now = datetime.now(timezone.utc)
|
|
|
|
async def _tick(at):
|
|
async with app.db_sessionmaker() as s:
|
|
async with s.begin():
|
|
await m.tick(s, at)
|
|
return (await _job(app, job_id))[0]
|
|
|
|
async def _set(sql, **kw):
|
|
async with app.db_sessionmaker() as s:
|
|
async with s.begin():
|
|
await s.execute(text(sql), {"id": job_id, "h": host_id, **kw})
|
|
|
|
await _set("UPDATE host_maintenance_jobs SET state = 'settling', state_at = :at, "
|
|
"drained_at = :at WHERE id = :id", at=now)
|
|
assert await _tick(now + timedelta(minutes=4)) == "settling"
|
|
assert await _tick(now + timedelta(minutes=5)) == "updating" # not updated yet: update next
|
|
|
|
checked = {"pending": 3, "docker_packages": [],
|
|
"containers": [{"name": "ak-outpost", "restart": "unless-stopped"},
|
|
{"name": "scratch", "restart": "no"}]}
|
|
await _set("UPDATE host_maintenance_jobs SET state = 'recovering', state_at = :at, "
|
|
"checked = CAST(:c AS JSON) WHERE id = :id", at=now, c=json.dumps(checked))
|
|
container = ("INSERT INTO docker_containers (host_id, name, container_id, image, status, "
|
|
"restart_count, ports_json, scraped_at, oom_killed) "
|
|
"VALUES (:h, :n, '', 'img:1', :st, 0, '[]', :at, false)")
|
|
await _set(container, n="ak-outpost", st="running", at=now + timedelta(seconds=30))
|
|
await _set(container, n="scratch", st="running", at=now - timedelta(minutes=20)) # before: stale
|
|
assert await _tick(now + timedelta(minutes=1)) == "recovering"
|
|
await _set("UPDATE docker_containers SET scraped_at = :at WHERE host_id = :h AND name = 'scratch'",
|
|
at=now + timedelta(minutes=2))
|
|
assert await _tick(now + timedelta(minutes=2)) == "done"
|
|
|
|
asyncio.run(_go())
|
|
|
|
|
|
def test_a_new_agent_is_handed_a_check_first():
|
|
app = _app()
|
|
|
|
async def _go():
|
|
from sqlalchemy import text
|
|
host_id, token, job_id = await _seed(app)
|
|
async with app.db_sessionmaker() as s:
|
|
async with s.begin():
|
|
await s.execute(text("UPDATE host_agent_registrations SET agent_version = '1.24.0' "
|
|
"WHERE host_id = :h"), {"h": host_id})
|
|
assert await _post(app, token) == [{"id": f"{job_id}-check", "action": "check"}]
|
|
checked = {"kind": "maintenance", "id": f"{job_id}-check", "action": "check", "state": "done",
|
|
"ok": True, "pending": 2, "docker_update": True, "docker_packages": ["docker-ce"],
|
|
"containers": []}
|
|
# Not a swarm node: nothing to drain, so on to the update.
|
|
assert await _post(app, token, [checked]) == [{"id": f"{job_id}-update", "action": "update"}]
|
|
|
|
asyncio.run(_go())
|