From 7161a6ac7d453d4c53e831d2e7d179325539f166 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 4 Jul 2026 02:17:17 +0000 Subject: [PATCH] perf: single-pass streaming audit metrics; skip redundant persist on rate-limit rejects MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit metrics.compute_metrics made ~5 separate passes over a fully materialized audit.jsonl (an append-only, unbounded file), and GET /audit/summary loaded the whole file into memory on every call. compute_metrics now aggregates in one pass over any iterable; the new iter_events()/summarize_audit() stream the file line-by-line, so the endpoint and the cyclaw-metrics CLI run at O(1) memory regardless of audit history. load_events stays as the materialized form for existing callers. Summary shape and semantics are byte-identical. RateLimiter.allow() fired a sqlite/Postgres upsert on every REJECTED request even when the stored window was unchanged (no timestamp appended, nothing pruned) — write amplification precisely under the hammering condition the limiter exists to cheapen. The reject path now persists only when expiry pruning actually changed the stored state. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01MLvAGGjgyVbJVT7tXcp9UA --- gate.py | 7 ++- metrics.py | 151 ++++++++++++++++++++++++++------------------- utils/ratelimit.py | 10 ++- 3 files changed, 100 insertions(+), 68 deletions(-) diff --git a/gate.py b/gate.py index 8c47414d..2ed9199f 100644 --- a/gate.py +++ b/gate.py @@ -94,7 +94,7 @@ # subprocess wrapper ONLY — it never imports sync/ or agentic/, so gate.py's # out-of-band isolation invariant is preserved (see utils/ops_runner.py). from utils.ops_runner import run_sync_op, run_agentic_op, run_fsconnect_op, run_sqlconnect_op, OpsError -from metrics import load_events, compute_metrics +from metrics import summarize_audit _bearer_scheme = HTTPBearer(auto_error=False) @@ -586,8 +586,9 @@ async def audit_summary(): SOC 2) without creating any new data egress. """ audit_file = cfg.get("logging", {}).get("audit_file", "audit.jsonl") - events = await asyncio.to_thread(load_events, audit_file) - return await asyncio.to_thread(compute_metrics, events) + # Single off-loop pass: summarize_audit streams the JSONL through + # compute_metrics without materializing the (unbounded) file in memory. + return await asyncio.to_thread(summarize_audit, audit_file) # ============================================================================= diff --git a/metrics.py b/metrics.py index 542dd0bc..c5c97fd6 100644 --- a/metrics.py +++ b/metrics.py @@ -11,94 +11,117 @@ import yaml -def load_events(audit_file: str): - events = [] +def iter_events(audit_file: str): + """Yield parsed audit events one line at a time (constant memory). + + ``audit.jsonl`` is append-only and unbounded; streaming keeps + ``GET /audit/summary`` and the ``cyclaw-metrics`` CLI at O(1) memory as + history grows instead of materializing the whole file. + """ if not Path(audit_file).exists(): - return events + return with open(audit_file, encoding="utf-8") as f: for line in f: try: - events.append(json.loads(line)) + yield json.loads(line) except json.JSONDecodeError: pass - return events -def compute_metrics(events: list) -> dict: + +def load_events(audit_file: str): + """Materialized list form of :func:`iter_events` (kept for existing callers).""" + return list(iter_events(audit_file)) + +def compute_metrics(events) -> dict: """Aggregate audit events into a JSON-serializable summary. + Accepts any iterable of event dicts (list or generator) and aggregates in a + single pass — previously this made ~5 separate passes over a fully + materialized list, so cost and memory grew with audit history. + Returns aggregates only — never raw query text. The audit log stores SHA-256 query hashes (not plaintext) by design, so this summary is safe to expose over the API-key-gated ``GET /audit/summary`` endpoint for regulated SMBs that need audit evidence (query volume, external-LLM usage, score distribution) without leaking the underlying queries. """ - summary = { - "total_events": len(events), - "event_breakdown": {}, - "rag_query_count": 0, - "scores": {"avg": None, "min": None, "max": None}, - "retrieval_modes": {}, - "model_used": {}, - "online_escalated": 0, + total = 0 + event_counts: Counter = Counter() + rag_query_count = 0 + score_sum = 0.0 + score_n = 0 + score_min: float | None = None + score_max: float | None = None + mode_counts: Counter = Counter() + model_counts: Counter = Counter() + online_escalated = 0 + + for e in events: + total += 1 + event_counts[e.get("event", "unknown")] += 1 + + if e.get("event") in ("rag_query", "mcp_rag_query"): + rag_query_count += 1 + if "top_score" in e: + s = e["top_score"] + score_sum += s + score_n += 1 + score_min = s if score_min is None or s < score_min else score_min + score_max = s if score_max is None or s > score_max else score_max + # The graph audit path records the retrieval mode under "retrieval_mode"; + # the MCP server (mcp_hybrid_server._handle_search) records it under "mode". + # Reading only "retrieval_mode" silently bucketed every mcp_rag_query as + # "unknown" even though its mode was right there under the other key. + mode_counts[e.get("retrieval_mode") or e.get("mode") or "unknown"] += 1 + # model_used is only meaningful for answered queries. Scope it to rag + # queries so non-answer events — notably the graph audit node's + # "user_gate_pause", which is still stamped model_used="unknown" + # (graph.audit_logger_node) — don't pollute the model-usage breakdown + # shown at GET /audit/summary with a bogus "unknown" bucket. + if e.get("model_used"): + model_counts[e["model_used"]] += 1 + + # An escalation to the external LLM. Prefer the explicit boolean the graph + # audit node already records (audit_logger_node sets + # online_escalated = answer_model == "grok") as the source of truth; fall back + # to user_confirmed_online / the model-name heuristic for older or MCP events + # that predate the explicit field. Relying on user_confirmed_online alone + # undercounted real escalations because the graph never writes that key. + if ( + e.get("online_escalated") is True + or e.get("user_confirmed_online") is True + or str(e.get("model_used", "")).lower().startswith("grok") + ): + online_escalated += 1 + + return { + "total_events": total, + "event_breakdown": dict(event_counts.most_common()), + "rag_query_count": rag_query_count, + "scores": ( + {"avg": score_sum / score_n, "min": score_min, "max": score_max} + if score_n + else {"avg": None, "min": None, "max": None} + ), + "retrieval_modes": dict(mode_counts.most_common()), + "model_used": dict(model_counts.most_common()), + "online_escalated": online_escalated, } - if not events: - return summary - - event_counts = Counter(e.get("event", "unknown") for e in events) - summary["event_breakdown"] = dict(event_counts.most_common()) - - rag_queries = [e for e in events if e.get("event") in ("rag_query", "mcp_rag_query")] - summary["rag_query_count"] = len(rag_queries) - if rag_queries: - scores = [e["top_score"] for e in rag_queries if "top_score" in e] - if scores: - summary["scores"] = { - "avg": sum(scores) / len(scores), - "min": min(scores), - "max": max(scores), - } - # The graph audit path records the retrieval mode under "retrieval_mode"; - # the MCP server (mcp_hybrid_server._handle_search) records it under "mode". - # Reading only "retrieval_mode" silently bucketed every mcp_rag_query as - # "unknown" even though its mode was right there under the other key. - mode_counts = Counter( - e.get("retrieval_mode") or e.get("mode") or "unknown" for e in rag_queries - ) - summary["retrieval_modes"] = dict(mode_counts.most_common()) - - # model_used is only meaningful for answered queries. Scope it to rag_queries - # (mirroring scores/modes above) so non-answer events — notably the graph - # audit node's "user_gate_pause", which is still stamped model_used="unknown" - # (graph.audit_logger_node) — don't pollute the model-usage breakdown shown at - # GET /audit/summary with a bogus "unknown" bucket. - model_counts = Counter(e["model_used"] for e in rag_queries if e.get("model_used")) - summary["model_used"] = dict(model_counts.most_common()) - - # An escalation to the external LLM. Prefer the explicit boolean the graph - # audit node already records (audit_logger_node sets - # online_escalated = answer_model == "grok") as the source of truth; fall back - # to user_confirmed_online / the model-name heuristic for older or MCP events - # that predate the explicit field. Relying on user_confirmed_online alone - # undercounted real escalations because the graph never writes that key. - summary["online_escalated"] = sum( - 1 - for e in events - if e.get("online_escalated") is True - or e.get("user_confirmed_online") is True - or str(e.get("model_used", "")).lower().startswith("grok") - ) - return summary + + +def summarize_audit(audit_file: str) -> dict: + """Stream the audit log through :func:`compute_metrics` in one bounded pass.""" + return compute_metrics(iter_events(audit_file)) def print_metrics(config_path: str = "config.yaml"): with open(config_path, encoding="utf-8") as f: cfg = yaml.safe_load(f) audit_file = cfg["logging"]["audit_file"] - events = load_events(audit_file) - if not events: + summary = summarize_audit(audit_file) + if not summary["total_events"]: print("No audit events found.") return - summary = compute_metrics(events) print(f"Total events: {summary['total_events']}") print("\nEvent breakdown:") for event, count in summary["event_breakdown"].items(): diff --git a/utils/ratelimit.py b/utils/ratelimit.py index 58f762c6..9df2261a 100644 --- a/utils/ratelimit.py +++ b/utils/ratelimit.py @@ -218,10 +218,18 @@ def allow(self, client_ip: str) -> bool: now = self._clock() with self._lock: self._sweep(now) + prior_len = len(self._hits[client_ip]) recent = [t for t in self._hits[client_ip] if now - t < self.window_seconds] if len(recent) >= self.max_requests: self._hits[client_ip] = recent - self._persist(client_ip, now) + # Persist only when expiry pruning actually changed the stored + # window. A rejected request appends no timestamp, so when + # nothing was pruned the backend already holds exactly this + # state — and skipping the redundant upsert removes a per-request + # DB write under precisely the hammering/abuse condition the + # limiter exists to make cheap. + if len(recent) != prior_len: + self._persist(client_ip, now) return False recent.append(now) self._hits[client_ip] = recent