obs(quotas): measure overshoot cancelled by the enforcement sweep

oddish/src/oddish/core/quota_enforcement.py

@@ -2,7 +2,9 @@
 
 import asyncio
 import logging
-from collections.abc import Awaitable, Callable
+import statistics
+from collections.abc import Awaitable, Callable, Sequence
+from datetime import datetime
 from typing import Any
 
 from sqlalchemy import select, text
@@ -119,6 +121,53 @@ def _active_trial_predicates(
     return predicates
 
 
+def _log_overshoot_cancelled(
+    trials: Sequence[TrialModel],
+    *,
+    scope: str,
+    org_id: str,
+    billed_user_id: str | None,
+    now: datetime,
+) -> None:
+    """Quantify admission overshoot: one line per enforcement action.
+
+    Lock-free admission (#1118) lets concurrent submissions both observe
+    headroom, so the sweep is what claws the overage back. A cancelled trial
+    that is still PENDING/QUEUED seconds after ``created_at`` is the overshoot
+    signature; a long-RUNNING one just crossed the cap mid-flight. Everything
+    here comes from the rows the sweep already loaded — no extra queries.
+    ``reserved_usd`` mirrors the per-trial reservation basis in
+    ``_sum_inflight_reserved_usd``.
+    """
+    queued = sum(
+        1
+        for trial in trials
+        if trial.status in (TrialStatus.PENDING, TrialStatus.QUEUED)
+    )
+    ages = sorted(
+        max((now - trial.created_at).total_seconds(), 0.0) for trial in trials
+    )
+    reserved_usd = sum(
+        max(trial.cost_usd or 0.0, float(settings.pending_trial_reservation_usd))
+        for trial in trials
+    )
+    logger.info(
+        "metric=quota.overshoot_cancelled scope=%s org_id=%s billed_user_id=%s "
+        "trials=%s queued=%s running=%s age_min_s=%.1f age_median_s=%.1f "
+        "age_max_s=%.1f reserved_usd=%.2f",
+        scope,
+        org_id,
+        billed_user_id,
+        len(trials),
+        queued,
+        len(trials) - queued,
+        ages[0],
+        statistics.median(ages),
+        ages[-1],
+        reserved_usd,
+    )
+
+
 async def _cancel_worker_jobs(
     session: AsyncSession,
     *,
@@ -257,6 +306,10 @@ async def cancel_trials_if_quota_reached(
     now = utcnow()
     trial_ids = [trial.id for trial in trials]
     affected_task_ids = sorted({trial.task_id for trial in trials})
+    # Before the mutation loop: it reads each trial's pre-cancellation status.
+    _log_overshoot_cancelled(
+        trials, scope=scope, org_id=org_id, billed_user_id=billed_user_id, now=now
+    )
     for trial in trials:
         trial.status = TrialStatus.FAILED
         trial.error_message = QUOTA_CANCELLED_MESSAGE

oddish/tests/test_quota_enforcement.py

@@ -1,8 +1,9 @@
 from __future__ import annotations
 
+import logging
 import uuid
 from contextlib import asynccontextmanager
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 from decimal import Decimal
 
 import pytest
@@ -395,8 +396,46 @@ async def user_limit(*_args):
     )
 
 
+def test_overshoot_metric_reports_statuses_ages_and_reserved(caplog, monkeypatch):
+    """One quota.overshoot_cancelled line: queued vs running, age spread, USD."""
+    monkeypatch.setattr(settings, "pending_trial_reservation_usd", Decimal("0.25"))
+    now = datetime.now(timezone.utc)
+    trials = []
+    for index, (status, age_s, cost_usd) in enumerate(
+        [
+            (TrialStatus.QUEUED, 5, None),
+            (TrialStatus.PENDING, 30, None),
+            (TrialStatus.RUNNING, 600, 1.75),
+        ]
+    ):
+        trial = _trial(
+            trial_id=f"task-om-{index}",
+            task_id="task-om",
+            experiment_id="exp-om",
+            org_id="org-om",
+            billed_user_id="user-om",
+            status=status,
+            cost_usd=cost_usd,
+        )
+        trial.created_at = now - timedelta(seconds=age_s)
+        trials.append(trial)
+
+    with caplog.at_level(logging.INFO, logger="oddish.core.quota_enforcement"):
+        quota_enforcement._log_overshoot_cancelled(
+            trials, scope="user", org_id="org-om", billed_user_id="user-om", now=now
+        )
+
+    assert (
+        "metric=quota.overshoot_cancelled scope=user org_id=org-om "
+        "billed_user_id=user-om trials=3 queued=2 running=1 "
+        "age_min_s=5.0 age_median_s=30.0 age_max_s=600.0 reserved_usd=2.25"
+    ) in caplog.text
+
+
 @pytest.mark.asyncio
-async def test_org_quota_cancels_every_users_active_trials(session, monkeypatch):
+async def test_org_quota_cancels_every_users_active_trials(
+    session, monkeypatch, caplog
+):
     suffix = uuid.uuid4().hex[:8]
     org_id = f"org-org-qc-{suffix}"
     experiment_id = f"exp-org-qc-{suffix}"
@@ -504,12 +543,17 @@ async def org_limit(*_args):
     session.add_all([preserved_job, cancelled_job])
     await session.flush()
 
-    result = await cancel_trials_if_quota_reached(
-        session, org_id=org_id, billed_user_id="user-a"
-    )
+    with caplog.at_level(logging.INFO, logger="oddish.core.quota_enforcement"):
+        result = await cancel_trials_if_quota_reached(
+            session, org_id=org_id, billed_user_id="user-a"
+        )
 
     assert result["scope"] == "org"
     assert result["trials_cancelled"] == 4
+    assert (
+        f"metric=quota.overshoot_cancelled scope=org org_id={org_id} "
+        "billed_user_id=user-a trials=4 queued=2 running=2 age_min_s="
+    ) in caplog.text
     assert (await session.get(TrialModel, f"{task_id}-1")).status == TrialStatus.FAILED
     assert (await session.get(TrialModel, f"{task_id}-2")).status == TrialStatus.FAILED
     assert (await session.get(TaskModel, task_id)).status == TaskStatus.FAILED