diff --git a/plugins/docker/rekey.py b/plugins/docker/rekey.py index 7927b4c..95687fd 100644 --- a/plugins/docker/rekey.py +++ b/plugins/docker/rekey.py @@ -4,14 +4,15 @@ 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 a batch at a time in the background -instead, one key after another, and does nothing once none are left. Network +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 @@ -33,6 +34,7 @@ logger = logging.getLogger(__name__) BATCH = 5000 # raw samples moved in one go EVERY_S = 15 +PASS_S = 2 # a key with only a few rows takes milliseconds, and there can be many 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')" @@ -70,24 +72,24 @@ def service_key(old: str) -> str: async def old_keys(session) -> list[str]: - """Every task-named docker resource still holding something to move. The - latest-value table stands for the raw samples, as a key stays in it until - its samples have aged out.""" - found: set[str] = set() - for model in (PluginMetricLatest, PluginMetricHourly): - found.update((await session.execute( - select(model.resource_name).distinct().where( - model.source_module == "docker", model.metric_name.notin_(TRAFFIC), - model.resource_name.op("~")(_TASK_KEY)))).scalars()) - for model in (MetricAnomaly, AlertRule): - found.update((await session.execute( - select(model.resource_name).distinct().where( - model.source_module == "docker", model.resource_name.op("~")(_TASK_KEY)))).scalars()) - found.update((await session.execute( - select(MaintenanceWindow.scope_resource).distinct().where( - MaintenanceWindow.scope_source == "docker", - MaintenanceWindow.scope_resource.op("~")(_TASK_KEY)))).scalars()) - return sorted(k for k in found if task_service(k.split("/", 1)[1])) + """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: @@ -148,9 +150,11 @@ async def move_key(session, old: str) -> bool: await session.execute(update(MaintenanceWindow).where( MaintenanceWindow.scope_source == "docker", MaintenanceWindow.scope_resource == old) .values(scope_resource=new)) - await session.execute(update(MetricAnomaly).where( + 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() @@ -169,11 +173,17 @@ def make_rekey_task(app: "Quart") -> ScheduledTask: if remaining == [] or busy.locked(): return async with busy, app.db_sessionmaker() as session: - async with session.begin(): - if remaining is None: + if remaining is None: + async with session.begin(): remaining = await old_keys(session) - logger.info("Swarm history under task names: %d keys to move", len(remaining)) - if remaining and await move_key(session, remaining[-1]): - remaining.pop() + 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") return ScheduledTask(name="docker_rekey", coro_factory=rekey, interval_seconds=EVERY_S) diff --git a/tests/integration/test_docker_rekey.py b/tests/integration/test_docker_rekey.py index b16d0a0..c2b23f4 100644 --- a/tests/integration/test_docker_rekey.py +++ b/tests/integration/test_docker_rekey.py @@ -121,6 +121,40 @@ def test_a_tasks_history_moves_onto_its_service_a_batch_at_a_time(app, monkeypat assert [sp.anomaly_id for sp in spans] == [row.id] +def test_keys_with_raw_samples_are_taken_first_and_a_pass_finishes_many_small_keys(app): + """Taken from the end of old_keys: a chart and a normal read recent history, + so a key with raw samples doesn't wait behind thousands that only have old + hours. One pass moves every small key it has time for.""" + from sqlalchemy import select + from steward.models.metrics import PluginMetricHourly, PluginMetricLatest + rekey = _docker("rekey") + + tag = uuid.uuid4().hex[:8] + aged = [f"nas/it{tag}_old.1.{c * 25}" for c in "bcd"] + recent = f"nas/it{tag}_new.1.{TASK}" + + async def _go(): + async with app.db_sessionmaker() as s: + async with s.begin(): + s.add_all([_hour(k, "cpu_pct", NOW, avg=1.0, top=2.0) for k in aged]) + s.add(PluginMetricLatest(source_module="docker", resource_name=recent, metric_name="cpu_pct", + value=1.0, recorded_at=NOW)) + async with s.begin(): + found = await rekey.old_keys(s) + await rekey.make_rekey_task(app).coro_factory() + async with app.db_sessionmaker() as s: + left = (await s.execute(select(PluginMetricHourly.resource_name).where( + PluginMetricHourly.resource_name.in_(aged)))).scalars().all() + moved = (await s.execute(select(PluginMetricHourly.value_avg).where( + PluginMetricHourly.resource_name == f"swarm/it{tag}_old"))).scalars().all() + return found, left, moved + + found, left, moved = asyncio.run(_go()) + assert all(found.index(recent) > found.index(k) for k in aged) + assert left == [] + assert moved == [1.0] + + def test_an_alert_rule_on_a_task_follows_it_to_its_service(app): from sqlalchemy import select from steward.models.alerts import AlertOperator, AlertRule