FabledSteward/plugins/host_agent/cadence.py

# plugins/host_agent/cadence.py
"""A group's maintenance cadence (#5768): every week, its hosts updated one at
a time, each rebooting only when its update needs it (or as the plan says).

A plan names an inventory group, a weekday and a time in its time zone, and a
window. At that time a run starts with the group's hosts in a fixed order. Each
host gets a maintenance job (maintenance.py); the next host starts only when
the last one is back in service: its agent posting, its swarm node ready and
active, every replicated service at its replicas. A job that fails, or a host
that doesn't come back into service, stops the run, and the host's
maintenance_failed figure goes to 1 for an alert rule to catch. No host is
started after the window; the run then stops and says which hosts it didn't
reach. A run that drained swarm nodes ends by spreading the swarm's services
back out (rebalance.py), unless its plan says not to.

The decisions are model-free here (plans and runs are anything with the
attributes); the database side is below them, and the plugin's minute tick
drives it. Everything is on the rows, so a restart of Steward, or the reboot of
the host it runs on, picks up where it was.
"""
from __future__ import annotations

import json
from datetime import datetime, timedelta

from steward.core.weekly import WEEKDAYS, describe, fired_slot, is_due, last_slot, next_slot, parse_at  # noqa: F401

from . import maintenance, rebalance, timeline

# How long a host may take to be back in service after its job finished.
SETTLE_LIMIT = timedelta(minutes=30)
# An agent heard from this recently is posting.
POSTING_WITHIN = timedelta(minutes=3)
RUN_STATES = ("running", "done", "failed", "stopped")


def back_in_service(*, last_seen_at, node, services, job_finished_at, now: datetime,
                    expect_active: bool = True) -> str | None:
    """None when the host is back in service, else why not yet. `node` is the
    freshest swarm-node row for the job's node (None outside a swarm, or when
    the job has no node); `services` the freshest row per swarm service. Only
    rows reported after the job finished count, so a stale "all fine" from
    before the reboot never passes. `expect_active` is False for a node the
    job didn't drain (the operator keeps it drained): it need only be ready."""
    if last_seen_at is None or now - last_seen_at > POSTING_WITHIN:
        return "its agent isn't posting yet"
    if node is None:
        return None
    if node.updated_at < job_finished_at:
        return "no swarm report since it came back"
    if node.status != "ready" or (expect_active and node.availability != "active"):
        return f"its swarm node is {node.status}/{node.availability}"
    fresh = [s for s in services if s.updated_at >= job_finished_at]
    if not fresh:
        return "no swarm report since it came back"
    short = sorted(s.service_name for s in fresh
                   if s.mode == "replicated" and s.running < s.desired)
    if short:
        return "services below their replicas: " + ", ".join(short[:5])
    return None


def next_move(run, plan, job, back: str | None, now: datetime) -> tuple[str, str | None]:
    """What a running run does now: ("start", None) the current host's job;
    ("wait", None); ("next", None) on to the next host; ("settling", why);
    ("fail", why); ("stop", why) the window ended; ("done", None)."""
    hosts = json.loads(run.host_ids)
    if run.position >= len(hosts):
        return "done", None
    if job is None:
        if now >= run.started_at + timedelta(minutes=plan.window_minutes):
            left = len(hosts) - run.position
            return "stop", f"the window ended with {left} host{'s' if left != 1 else ''} not started"
        return "start", None
    if job.state in maintenance.ACTIVE:
        return "wait", None
    if job.state != "done":
        return "fail", job.detail or maintenance.LABELS[job.state]
    if back is None:
        return "next", None
    if run.settle_since is not None and now - run.settle_since > SETTLE_LIMIT:
        return "fail", f"{maintenance.NOT_BACK} {int(SETTLE_LIMIT.total_seconds() // 60)} min: {back}"
    return "settling", back


# ── applied to the database: the minute tick, Run now, the page ──────────────

async def group_hosts(session, group_id: str) -> list:
    """The group's hosts (through their linked Ansible targets), by name."""
    from sqlalchemy import select
    from steward.models.ansible_inventory import AnsibleTarget, ansible_target_groups
    from steward.models.hosts import Host
    rows = (await session.execute(
        select(Host).join(AnsibleTarget, AnsibleTarget.host_id == Host.id)
        .join(ansible_target_groups, ansible_target_groups.c.target_id == AnsibleTarget.id)
        .where(ansible_target_groups.c.group_id == group_id))).scalars().all()
    return sorted({h.id: h for h in rows}.values(), key=lambda h: h.name.lower())


async def start_run(session, plan, now: datetime, manual: bool = False):
    """A run of the plan with the group's hosts in order. A host whose agent
    has no maintenance helper is passed over, and the run says so."""
    from sqlalchemy import select
    from .models import HostAgentRegistration, HostMaintenanceRun
    hosts = await group_hosts(session, plan.group_id)
    regs = {r.host_id: r for r in (await session.execute(select(HostAgentRegistration).where(
        HostAgentRegistration.host_id.in_([h.id for h in hosts])))).scalars()} if hosts else {}
    order, skipped = [], []
    for h in hosts:
        reg = regs.get(h.id)
        if reg is None:
            skipped.append({"host_id": h.id, "reason": "no agent"})
        elif not reg.maint_helper:
            skipped.append({"host_id": h.id, "reason": "no maintenance helper: run Update agent"})
        else:
            order.append(h.id)
    run = HostMaintenanceRun(plan_id=plan.id, state="running", host_ids=json.dumps(order),
                             position=0, skipped=json.dumps(skipped), manual=manual,
                             started_at=now)
    session.add(run)
    await session.flush()
    return run


async def _back(session, host, job, reg, now: datetime) -> str | None:
    from sqlalchemy import select
    node = None
    services: list = []
    if job.node_id:
        models = maintenance.docker_models()
        if models is not None:
            DockerSwarmNode, DockerSwarmService = models.DockerSwarmNode, models.DockerSwarmService
            node = (await session.execute(select(DockerSwarmNode).where(
                DockerSwarmNode.node_id == job.node_id)
                .order_by(DockerSwarmNode.updated_at.desc()).limit(1))).scalar_one_or_none()
            freshest: dict = {}
            for s in (await session.execute(select(DockerSwarmService))).scalars():
                if s.service_name not in freshest or s.updated_at > freshest[s.service_name].updated_at:
                    freshest[s.service_name] = s
            services = list(freshest.values())
    return back_in_service(last_seen_at=reg.last_seen_at if reg else None, node=node,
                           services=services, job_finished_at=job.finished_at or now, now=now,
                           expect_active=bool(job.drain_host_id))


async def advance(session, run, plan, now: datetime) -> str:
    """Move one running run on as far as it can go now; returns its state."""
    from sqlalchemy import select
    from steward.models.hosts import Host
    from .models import HostAgentRegistration, HostMaintenanceJob
    for _ in range(64):                             # a host per pass; bounded
        hosts = json.loads(run.host_ids)
        host_id = hosts[run.position] if run.position < len(hosts) else None
        job = host = reg = None
        if host_id is not None:
            host = await session.get(Host, host_id)
            reg = (await session.execute(select(HostAgentRegistration).where(
                HostAgentRegistration.host_id == host_id))).scalar_one_or_none()
            job = (await session.execute(select(HostMaintenanceJob).where(
                HostMaintenanceJob.run_id == run.id, HostMaintenanceJob.host_id == host_id)
                .order_by(HostMaintenanceJob.created_at.desc()).limit(1))).scalar_one_or_none()
        back = None
        if job is not None and job.state == "done":
            back = await _back(session, host, job, reg, now)
        move, why = next_move(run, plan, job, back, now)
        if move == "start":
            if host is None:
                _skip(run, host_id, "the host is gone")
                continue
            try:
                await maintenance.create_job(session, host, reg, plan.created_by,
                                             plan.reboot_policy, now, run_id=run.id)
            except maintenance.CannotStart as exc:
                _skip(run, host_id, str(exc))
                continue
            return run.state
        if move == "next":
            run.position += 1
            run.settle_since = None
            continue
        if move == "settling":
            run.settle_since = run.settle_since or now
            return run.state
        if move == "fail":
            if job is not None and job.state == "done":
                maintenance._finish(job, "failed", now, why)     # the host never came back into service
                await timeline.write(session, f"maintenance job {job.id}",
                                     timeline.job_entry(session, job, host.name if host else None))
            name = host.name if host else host_id
            _end(run, "failed", f"{name}: {why}", now=now)
        elif move == "stop":
            _end(run, "stopped", why, now=now)
        elif move == "done":
            if await rebalance.begin(session, run, plan, now) or \
                    (run.rebalance or {}).get("state") == "handed":
                return run.state            # a manager is spreading services back out
            _end(run, *rebalance.outcome(run), now=now)
        return run.state
    return run.state


def _skip(run, host_id: str, reason: str) -> None:
    skipped = json.loads(run.skipped)
    skipped.append({"host_id": host_id, "reason": reason})
    run.skipped = json.dumps(skipped)
    hosts = json.loads(run.host_ids)
    hosts.pop(run.position)
    run.host_ids = json.dumps(hosts)


def _end(run, state: str, detail: str | None = None, *, now: datetime) -> None:
    run.state, run.finished_at, run.settle_since = state, now, None
    if detail:
        run.detail = detail[:maintenance.DETAIL_MAX]


async def tick(session, now: datetime) -> list[str]:
    """Start due plans and move running runs on; returns lines worth logging."""
    from sqlalchemy import select
    from .models import HostMaintenancePlan, HostMaintenanceRun
    said = []
    running = {r.plan_id: r for r in (await session.execute(select(HostMaintenanceRun).where(
        HostMaintenanceRun.state == "running"))).scalars()}
    for plan in (await session.execute(select(HostMaintenancePlan))).scalars():
        slot = is_due(plan, now)
        if slot is None:
            continue
        plan.last_slot_at = slot
        if plan.id in running:
            said.append(f"plan {plan.name!r}: the {slot:%Y-%m-%d %H:%M}Z slot came with a run still going")
            continue
        running[plan.id] = await start_run(session, plan, now)
        said.append(f"plan {plan.name!r}: run started")
    for run in running.values():
        plan = await session.get(HostMaintenancePlan, run.plan_id)
        if plan is None:
            continue
        if rebalance.check_deadline(run, now):
            await rebalance.finish(session, run, now)
        before = run.state
        after = await advance(session, run, plan, now)
        if after != before:
            said.append(f"plan {plan.name!r}: run {after}" + (f" ({run.detail})" if run.detail else ""))
    return said


async def failed_hosts(session, now: datetime) -> dict[str, float]:
    """{host name: 1 or 0} for every host with an agent: whether its latest
    finished maintenance job failed or timed out, for the maintenance_failed
    figure an alert rule watches. A host with no job yet reads 0, so a rule
    on it reports rather than sitting at "no data". A later job that
    succeeds puts it back to 0, and so does a host that failed only for not
    coming back into service in time, once it is back (#6105). That return
    is kept on the job (came_back_at), so a later wobble in the host's
    services doesn't flag it again (#6175)."""
    from sqlalchemy import select
    from steward.models.hosts import Host
    from .models import HostAgentRegistration, HostMaintenanceJob
    agents = (await session.execute(
        select(Host.name).join(HostAgentRegistration, HostAgentRegistration.host_id == Host.id))).scalars()
    out = {name: 0.0 for name in agents}
    rows = (await session.execute(
        select(Host, HostMaintenanceJob)
        .join(Host, Host.id == HostMaintenanceJob.host_id)
        .where(HostMaintenanceJob.state.in_(maintenance.FINISHED),
               HostMaintenanceJob.finished_at >= now - timedelta(days=30)))).all()
    latest: dict = {}
    for host, job in rows:
        if host.name not in latest or job.finished_at > latest[host.name][1].finished_at:
            latest[host.name] = (host, job)
    for name, (host, job) in latest.items():
        if (job.state == "failed" and job.came_back_at is None
                and (job.detail or "").startswith(maintenance.NOT_BACK)):
            reg = (await session.execute(select(HostAgentRegistration).where(
                HostAgentRegistration.host_id == host.id))).scalar_one_or_none()
            if await _back(session, host, job, reg, now) is None:
                job.came_back_at = now
                await timeline.write(session, f"maintenance job {job.id}",
                                     timeline.job_entry(session, job, host.name))
        out[name] = 1.0 if maintenance.still_failed(job) else 0.0
    return out