FabledSteward/plugins/docker/rekey.py

# plugins/docker/rekey.py
"""Moves history kept under a swarm task's own name, "<host>/<service>.<slot>.
<task>", onto its service's key, "swarm/<service>": samples, hourly rollups,
anomalies and the alert rules and maintenance windows that name it.

The raw table holds millions of rows, too many to move in a migration that the
app has to wait for. A small task moves what it can in a short pass in the
background instead, newest history first, and does nothing once none are left. Network
rates stay where they are: they are kept per host, and age out with retention.
"""
from __future__ import annotations

import asyncio
import logging
import time
from typing import TYPE_CHECKING

from sqlalchemy import delete, select, text, update

from steward.core.scheduler import ScheduledTask
from steward.models.alerts import AlertRule, MaintenanceWindow
from steward.models.metrics import PluginMetric, PluginMetricHourly, PluginMetricLatest
from steward.models.norms import MetricAnomaly, MetricAnomalySpan
from steward.models.tickets import TicketLink
from steward.models.timeline import TimelineEntry
from steward.timeline.record import anomaly_changed

from .service import container_resource, task_service

if TYPE_CHECKING:
    from quart import Quart

logger = logging.getLogger(__name__)

BATCH = 1000            # raw samples moved in one go: each rewrites its row and every index on it
EVERY_S = 15
PASS_S = 2              # a key with only a few rows takes milliseconds, and there can be many
REST = 4                # after a pass, wait this many times as long as it took, so ingest keeps the disk
TRAFFIC = ("net_rx_bps", "net_tx_bps")
_TASK_KEY = r"^[^/]+/.+\.([0-9]+|[a-z0-9]{25})\.[a-z0-9]{25}$"
_NOT_TRAFFIC = "metric_name NOT IN ('net_rx_bps', 'net_tx_bps')"

# Hours the service has no row for change name; hours it has are combined, as
# two tasks of a service can share an hour.
_RENAME_HOURS = text(f"""
UPDATE plugin_metrics_hourly o SET resource_name = :new
WHERE o.source_module = 'docker' AND o.resource_name = :old AND o.{_NOT_TRAFFIC}
  AND NOT EXISTS (SELECT 1 FROM plugin_metrics_hourly n
                  WHERE n.source_module = 'docker' AND n.resource_name = :new
                    AND n.metric_name = o.metric_name AND n.bucket = o.bucket)
""")
_COMBINE_HOURS = text("""
UPDATE plugin_metrics_hourly n SET
    value_avg = COALESCE((n.value_avg * n.sample_count + o.value_avg * o.sample_count)
                         / NULLIF(n.sample_count + o.sample_count, 0), n.value_avg),
    value_max = GREATEST(n.value_max, o.value_max),
    value_p95 = GREATEST(n.value_p95, o.value_p95),
    sample_count = n.sample_count + o.sample_count
FROM plugin_metrics_hourly o
WHERE o.source_module = 'docker' AND o.resource_name = :old
  AND n.source_module = 'docker' AND n.resource_name = :new
  AND n.metric_name = o.metric_name AND n.bucket = o.bucket
""")
_DROP_HOURS = text(f"""
DELETE FROM plugin_metrics_hourly
WHERE source_module = 'docker' AND resource_name = :old AND {_NOT_TRAFFIC}
""")


def service_key(old: str) -> str:
    """The key a task-named resource belongs under now."""
    return container_resource("", old.split("/", 1)[1])


async def old_keys(session) -> list[str]:
    """Every task-named docker resource still holding something to move, to be
    taken from the end: those with raw samples last, as they are what a chart
    and a normal read first. The latest-value table stands for the raw samples,
    as a key stays in it until its samples have aged out."""
    async def names(column, *where) -> set[str]:
        found = (await session.execute(select(column).distinct().where(*where, column.op("~")(_TASK_KEY)))).scalars()
        return {k for k in found if task_service(k.split("/", 1)[1])}

    recent = await names(PluginMetricLatest.resource_name, PluginMetricLatest.source_module == "docker",
                         PluginMetricLatest.metric_name.notin_(TRAFFIC))
    hourly = await names(PluginMetricHourly.resource_name, PluginMetricHourly.source_module == "docker",
                         PluginMetricHourly.metric_name.notin_(TRAFFIC))
    named = (await names(MetricAnomaly.resource_name, MetricAnomaly.source_module == "docker")
             | await names(AlertRule.resource_name, AlertRule.source_module == "docker")
             | await names(MaintenanceWindow.scope_resource, MaintenanceWindow.scope_source == "docker"))
    logger.info("Swarm history under task names: %d keys with raw samples, %d with hourly rows, "
                "%d named by anomalies, rules or windows", len(recent), len(hourly), len(named))
    return sorted(hourly - recent) + sorted((named - hourly) - recent) + sorted(recent)


async def _merge_anomalies(session, new: str) -> None:
    """One row per metric still to review, as for any key: the rows waiting on
    the service become the newest of them, with all their occurrences."""
    rows = (await session.execute(
        select(MetricAnomaly).where(
            MetricAnomaly.source_module == "docker", MetricAnomaly.resource_name == new,
            MetricAnomaly.review == "to_review")
        .order_by(MetricAnomaly.last_started_at.desc()))).scalars().all()
    keep: dict[str, MetricAnomaly] = {}
    for row in rows:
        into = keep.setdefault(row.metric_name, row)
        if into is row:
            continue
        await session.execute(update(MetricAnomalySpan).where(
            MetricAnomalySpan.anomaly_id == row.id).values(anomaly_id=into.id))
        into.occurrences += row.occurrences
        into.started_at = min(into.started_at, row.started_at)
        if row.ended_at is None:
            into.ended_at = None
        if row.peak_level > into.peak_level:
            into.peak_level, into.peak_at = row.peak_level, row.peak_at
            into.usual_hold, into.usual_peak = row.usual_hold, row.usual_peak
        linked = (await session.execute(select(TicketLink.ticket_id).where(
            TicketLink.ref_type == "metric_anomaly", TicketLink.ref_id == into.id))).scalars().all()
        await session.execute(update(TicketLink).where(
            TicketLink.ref_type == "metric_anomaly", TicketLink.ref_id == row.id,
            TicketLink.ticket_id.notin_(linked)).values(ref_id=into.id))
        await session.execute(delete(TicketLink).where(
            TicketLink.ref_type == "metric_anomaly", TicketLink.ref_id == row.id))
        await session.execute(delete(TimelineEntry).where(
            TimelineEntry.ref_type == "metric_anomaly", TimelineEntry.ref_id == row.id))
        await session.delete(row)


async def move_key(session, old: str) -> bool:
    """Move the next batch of `old`, and the rest of what it holds once its raw
    samples are done. Returns whether it is finished. Runs inside the caller's
    transaction."""
    new = service_key(old)
    batch = select(PluginMetric.id).where(
        PluginMetric.source_module == "docker", PluginMetric.resource_name == old,
        PluginMetric.metric_name.notin_(TRAFFIC)).limit(BATCH)
    moved = await session.execute(
        update(PluginMetric).where(PluginMetric.id.in_(batch)).values(resource_name=new)
        .execution_options(synchronize_session=False))
    if moved.rowcount >= BATCH:
        return False
    keys = {"old": old, "new": new}
    for statement in (_RENAME_HOURS, _COMBINE_HOURS, _DROP_HOURS):
        await session.execute(statement, keys)
    await session.execute(delete(PluginMetricLatest).where(
        PluginMetricLatest.source_module == "docker", PluginMetricLatest.resource_name == old,
        PluginMetricLatest.metric_name.notin_(TRAFFIC)))
    await session.execute(update(AlertRule).where(
        AlertRule.source_module == "docker", AlertRule.resource_name == old).values(resource_name=new))
    await session.execute(update(MaintenanceWindow).where(
        MaintenanceWindow.scope_source == "docker", MaintenanceWindow.scope_resource == old)
        .values(scope_resource=new))
    renamed = await session.execute(update(MetricAnomaly).where(
        MetricAnomaly.source_module == "docker", MetricAnomaly.resource_name == old)
        .values(resource_name=new).execution_options(synchronize_session=False))
    if not renamed.rowcount:
        return True
    session.expire_all()
    await _merge_anomalies(session, new)
    await session.flush()
    for row in (await session.execute(select(MetricAnomaly).where(
            MetricAnomaly.source_module == "docker", MetricAnomaly.resource_name == new))).scalars().all():
        await anomaly_changed(session, row)
    return True


def make_rekey_task(app: "Quart") -> ScheduledTask:
    remaining: list[str] | None = None
    busy = asyncio.Lock()
    resume_at = 0.0

    async def rekey() -> None:
        nonlocal remaining, resume_at
        started = time.monotonic()
        if remaining == [] or busy.locked() or started < resume_at:
            return
        async with busy, app.db_sessionmaker() as session:
            if remaining is None:
                async with session.begin():
                    remaining = await old_keys(session)
            deadline = time.monotonic() + PASS_S
            while remaining and time.monotonic() < deadline:
                async with session.begin():
                    finished = await move_key(session, remaining[-1])
                if not finished:        # a full batch of samples is enough for one pass
                    break
                remaining.pop()
            if not remaining:
                logger.info("Swarm history under task names: all moved")
        now = time.monotonic()
        resume_at = now + REST * (now - started)

    return ScheduledTask(name="docker_rekey", coro_factory=rekey, interval_seconds=EVERY_S)