Superuser route to flush the metrics collector now

POST /observability/flush handles what the collector has queued and writes
it, so the soak harness no longer sleeps FLUSH_INTERVAL_S before clearing
its own rows.

backend/fluksio/api/routes/observability.py

@@ -19,12 +19,17 @@ from sqlalchemy import select as sa_select
 from sqlmodel import col, select
 
 from fluksio import __version__
-from fluksio.api.deps import FlowControllerDep, SessionDep, get_current_user
+from fluksio.api.deps import (
+    FlowControllerDep,
+    SessionDep,
+    get_current_active_superuser,
+    get_current_user,
+)
 from fluksio.api.routes.runs import elapsed_ms
 from fluksio.core.config import settings
 from fluksio.flow.controller import NodeStatus
 from fluksio.flow.pipeline import ADVISORY_ISSUES
-from fluksio.models import EngineEvent, FlowRun, MetricBucket
+from fluksio.models import EngineEvent, FlowRun, Message, MetricBucket
 
 router = APIRouter(
     prefix="/observability",
@@ -508,3 +513,18 @@ async def read_dead_letters(
     if controller.execution is None:
         return []
     return await run_in_threadpool(controller.execution.queue.dead_letters, limit)
+
+
+@router.post(
+    "/flush",
+    response_model=Message,
+    dependencies=[Depends(get_current_active_superuser)],
+)
+async def flush_metrics(request: Request) -> Any:
+    """Write what the metrics collector holds now, not at its next interval.
+
+    For tooling that reads or clears these tables right after causing events,
+    such as the soak harness clearing its own rows.
+    """
+    await request.app.state.metrics_collector.flush_now()
+    return Message(message="Metrics flushed")

backend/fluksio/flow/metrics.py

@@ -103,6 +103,7 @@ class MetricsCollector:
         # held with the time it arrived so an unpaired one does not stay put.
         self._tracebacks: dict[tuple[str, str], tuple[float, str]] = {}
         self._last_prune = 0.0
+        self._queue: asyncio.Queue[dict[str, Any]] | None = None
 
     # -------------------------------------------------------------------------
     # The loop
@@ -112,6 +113,7 @@ class MetricsCollector:
         """Consume the bus until cancelled, flushing on a fixed interval."""
         last = time.monotonic()
         async with self._events.subscribe() as queue:
+            self._queue = queue
             while True:
                 # A busy bus never idles, so the flush is on a deadline rather
                 # than on the timeout alone.
@@ -129,6 +131,21 @@ class MetricsCollector:
                     await self.flush()
                     last = time.monotonic()
 
+    async def flush_now(self) -> None:
+        """Write what has been heard so far, without waiting for the interval.
+
+        Events the bus has handed over and the loop has not reached yet are
+        taken first, so whoever caused an event and then asks for a flush finds
+        it written.
+        """
+        while self._queue is not None and not self._queue.empty():
+            event = self._queue.get_nowait()
+            try:
+                self.handle(event)
+            except Exception:
+                logger.exception("Could not record %s", event.get("type"))
+        await self.flush()
+
     # -------------------------------------------------------------------------
     # Accumulating
     # -------------------------------------------------------------------------

backend/fluksio/main.py

@@ -384,7 +384,9 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
 
         _background(watchdog.run(), "loop-watchdog")
         _background(alerts.run(), "alert-manager")
-        _background(MetricsCollector(event_bus).run(), "metrics-collector")
+        # On the app so a superuser can ask it to flush now.
+        app.state.metrics_collector = MetricsCollector(event_bus)
+        _background(app.state.metrics_collector.run(), "metrics-collector")
         _background(_sweep_artifacts(artifacts, controller), "artifact-gc")
         _background(_trim_volatile(volatile), "artifact-ring")
         await controller.start()

backend/scripts/soak.py

@@ -36,7 +36,6 @@ from typing import Any
 import httpx
 
 from fluksio.core.config import settings
-from fluksio.flow.metrics import FLUSH_INTERVAL_S
 
 log = logging.getLogger("soak")
 
@@ -374,6 +373,14 @@ def create_flow(
     log.info("created flow '%s'", name)
 
 
+def flush_metrics(client: httpx.Client) -> None:
+    """Have the metrics collector write what it holds now."""
+    try:
+        client.post("/api/v1/observability/flush").raise_for_status()
+    except httpx.HTTPError as exc:
+        log.error("could not flush the metrics collector: %s", exc)
+
+
 def delete_flows(client: httpx.Client) -> None:
     for name in FLOWS:
         try:
@@ -887,11 +894,11 @@ def cleanup(client: httpx.Client | None) -> None:
         delete_flows(client)
 
     clear_redis()
-    # Deleting a flow publishes an audit event of its own, and the metrics
-    # collector buffers what it hears for up to FLUSH_INTERVAL_S before writing
-    # it. Clearing the tables any sooner races that flush and leaves the rows
-    # this very cleanup caused.
-    time.sleep(FLUSH_INTERVAL_S + 5.0)
+    # Deleting a flow publishes an audit event of its own, which the metrics
+    # collector holds until its next flush. Clearing the tables before it is
+    # written would leave the rows this very cleanup caused.
+    if client is not None:
+        flush_metrics(client)
     clear_observability()
 
 

backend/tests/api/routes/test_observability.py

@@ -1,9 +1,11 @@
+import time
 from datetime import UTC, datetime, timedelta
 
 from fastapi.testclient import TestClient
-from sqlmodel import Session
+from sqlmodel import Session, select
 
 from fluksio.core.config import settings
+from fluksio.flow.events import event_bus
 from fluksio.models import EngineEvent, FlowRun, MetricBucket
 
 PREFIX = f"{settings.API_V1_STR}/observability"
@@ -482,3 +484,31 @@ def test_a_node_failing_on_every_call_makes_the_summary_degraded(
     assert body["nodes"]["failing"] == 1
     (problem,) = [p for p in body["problems"] if "failing" in p]
     assert "house.calc" in problem and "house.old" not in problem
+
+
+def test_a_superuser_can_flush_the_collector_now(
+    client: TestClient,
+    superuser_token_headers: dict[str, str],
+    normal_user_token_headers: dict[str, str],
+    db: Session,
+) -> None:
+    """Whoever reads or clears the tables need not wait out the interval."""
+    event_bus.publish(
+        {
+            "type": "audit",
+            "action": "flushed",
+            "flow": FLOW,
+            "user": "a@example.com",
+            "ts": time.time(),
+        }
+    )
+    url = f"{PREFIX}/flush"
+    assert client.post(url, headers=normal_user_token_headers).status_code == 403
+    assert client.post(url, headers=superuser_token_headers).status_code == 200
+
+    written = db.exec(
+        select(EngineEvent).where(
+            EngineEvent.flow == FLOW, EngineEvent.detail == "flushed"
+        )
+    ).all()
+    assert len(written) == 1

docs/code/api.md

@@ -208,6 +208,7 @@ Connector state keys can also be listed and cleared by a superuser; see
 | `GET` | `/observability/summary` | health and engine version; always returns `200` |
 | `GET` | `/observability/timeseries`, `/observability/flows` | execution metrics or flow rollups |
 | `GET` | `/observability/runs`, `/observability/events`, `/observability/dead-letter` | cascades, events, or abandoned work |
+| `POST` | `/observability/flush` | write buffered metrics and events now; superuser only |
 | `GET` | `/utils/health/` | deep health check; fails on loop or state-backend problems |
 | `GET` | `/workers`, `/workers/resources` | attached workers and resource availability |
 | `GET`, `POST` | `/workers/tokens` | list or mint worker credentials; superuser only |

frontend/src/client/types.gen.ts

@@ -2001,6 +2001,8 @@ export type ObservabilityReadDeadLettersData = {
 
 export type ObservabilityReadDeadLettersResponse = (Array<DeadLetter>);
 
+export type ObservabilityFlushMetricsResponse = (Message);
+
 export type PanelsReadPanelsResponse = (PanelsPublic);
 
 export type PanelsSavePanelsData = {