141 lines
7.5 KiB
Python
141 lines
7.5 KiB
Python
|
"""Bounded, privacy preserving performance telemetry.
|
||
|
|
|
||
|
|
Metrics are deliberately process local. RAG workers return one small envelope
|
||
|
|
to the API process, rather than attempting to share an in-memory deque.
|
||
|
|
"""
|
||
|
|
from __future__ import annotations
|
||
|
|
from contextvars import ContextVar
|
||
|
|
from collections import deque
|
||
|
|
import hashlib, threading, time
|
||
|
|
|
||
|
|
_current: ContextVar[object | None] = ContextVar("perf_metrics", default=None)
|
||
|
|
_lock = threading.Lock()
|
||
|
|
_http = deque(maxlen=512)
|
||
|
|
_sql = deque(maxlen=512)
|
||
|
|
_rag = deque(maxlen=512)
|
||
|
|
|
||
|
|
def _pct(values, p):
|
||
|
|
if not values: return 0.0
|
||
|
|
vals = sorted(float(v) for v in values)
|
||
|
|
return round(vals[min(len(vals)-1, int((len(vals)-1)*p))], 2)
|
||
|
|
|
||
|
|
class RequestMetrics:
|
||
|
|
def __init__(self, trace_id=""):
|
||
|
|
self.trace_id = str(trace_id or "")[:64]; self.started = time.perf_counter()
|
||
|
|
self.sql_count = self.sql_ms = self.sql_lock_errors = 0
|
||
|
|
def sql(self, elapsed, error=None):
|
||
|
|
self.sql_count += 1; self.sql_ms += float(elapsed)
|
||
|
|
if error and "locked" in str(error).lower(): self.sql_lock_errors += 1
|
||
|
|
def finish(self, method, path, status):
|
||
|
|
elapsed = (time.perf_counter()-self.started)*1000
|
||
|
|
# route templates are supplied by the caller; raw paths are never stored
|
||
|
|
row={"endpoint": f"{method} {path}", "status": int(status),
|
||
|
|
"elapsed_ms": round(elapsed,2), "sql_count": self.sql_count,
|
||
|
|
"sql_ms": round(self.sql_ms,2), "sql_lock_errors": self.sql_lock_errors,
|
||
|
|
"at": time.time()}
|
||
|
|
with _lock: _http.append(row)
|
||
|
|
return row
|
||
|
|
|
||
|
|
class request_scope:
|
||
|
|
def __init__(self, trace_id=""): self.metrics=RequestMetrics(trace_id); self.token=None
|
||
|
|
def __enter__(self): self.token=_current.set(self.metrics); return self.metrics
|
||
|
|
def __exit__(self, *exc): _current.reset(self.token)
|
||
|
|
|
||
|
|
def current(): return _current.get()
|
||
|
|
def record_sql(elapsed, error=None):
|
||
|
|
m=current()
|
||
|
|
if m: m.sql(elapsed,error)
|
||
|
|
|
||
|
|
def record_rag(row):
|
||
|
|
if not isinstance(row,dict): return
|
||
|
|
# Keep phase-level embedding diagnostics while excluding query text and
|
||
|
|
# filesystem details. These fields let operators distinguish cold model
|
||
|
|
# load, tokenizer, inference, serialization and IPC overhead.
|
||
|
|
allowed = ("total_ms","validation_ms","embedding_ms","recall_ms","postprocess_ms",
|
||
|
|
"cache_hit","result_count","error","embedding_total_ms",
|
||
|
|
"embedding_process_start_ms","embedding_ipc_ms","embedding_model_load_ms",
|
||
|
|
"embedding_tokenizer_ms","embedding_encode_ms","embedding_postprocess_ms",
|
||
|
|
"embedding_source","embedding_count","embedding_model","embedding_version",
|
||
|
|
"embedding_dim","embedding_dimension","embedding_pid","embedding_cache_hit",
|
||
|
|
"embedding_cache_size","embedding_cache_hits","embedding_cache_misses","embedding_cache_hit_rate","worker_pid",
|
||
|
|
"worker_reused","worker_start_ms","serialization_ms","merge_ms","dedup_ms","rrf_ms",
|
||
|
|
"retrieval_scopes","warm_state","queue_wait_ms","transport_total_ms")
|
||
|
|
clean={k: row.get(k) for k in allowed if k in row}
|
||
|
|
with _lock: _rag.append(clean)
|
||
|
|
|
||
|
|
def _summary(rows, rag=False):
|
||
|
|
if not rows: return {"samples":0,"p50_ms":0.0,"p95_ms":0.0,"errors":0}
|
||
|
|
total=[r.get("total_ms",r.get("elapsed_ms",0)) for r in rows]
|
||
|
|
out={"samples":len(rows),"p50_ms":_pct(total,.5),"p95_ms":_pct(total,.95),
|
||
|
|
"errors":sum(1 for r in rows if int(r.get("status",200))>=400 or r.get("error"))}
|
||
|
|
if rag: out["cache_hit_rate"]=round(sum(bool(r.get("cache_hit")) for r in rows)/len(rows),4)
|
||
|
|
return out
|
||
|
|
|
||
|
|
def snapshot():
|
||
|
|
with _lock: h=list(_http); s=list(_sql); r=list(_rag)
|
||
|
|
sql_ms=[x.get("sql_ms",0) for x in h]
|
||
|
|
slowest = sorted(h, key=lambda row: float(row.get("elapsed_ms", 0)), reverse=True)[:10]
|
||
|
|
status_counts = {}
|
||
|
|
for row in h:
|
||
|
|
group = f"{int(row.get('status', 200)) // 100}xx"
|
||
|
|
status_counts[group] = status_counts.get(group, 0) + 1
|
||
|
|
return {"scope":"current_process","window_samples":len(h),"api":{**_summary(h),"p99_ms":_pct([x.get("elapsed_ms",0) for x in h],.99),"sql_count":sum(x.get("sql_count",0) for x in h),"sql_ms":round(sum(sql_ms),2),"sql_lock_errors":sum(x.get("sql_lock_errors",0) for x in h),"slowest":slowest,"status_counts":status_counts},"rag":{**_summary(r,True),"phases":{}},"resets_on_restart":True}
|
||
|
|
|
||
|
|
def rag_snapshot():
|
||
|
|
with _lock: rows=list(_rag)
|
||
|
|
out=_summary(rows,True)
|
||
|
|
for phase in ("validation_ms","embedding_ms","recall_ms","postprocess_ms"):
|
||
|
|
vals=[r.get(phase,0) for r in rows]; out.setdefault("phases",{})[phase]={"p50_ms":_pct(vals,.5),"p95_ms":_pct(vals,.95)}
|
||
|
|
for phase in ("serialization_ms", "merge_ms", "dedup_ms", "rrf_ms", "queue_wait_ms", "transport_total_ms"):
|
||
|
|
if any(phase in r for r in rows):
|
||
|
|
vals = [r.get(phase, 0) for r in rows]
|
||
|
|
out.setdefault("phases", {})[phase] = {
|
||
|
|
"p50_ms": _pct(vals, .5), "p95_ms": _pct(vals, .95)}
|
||
|
|
embedding_phases = ("embedding_total_ms", "embedding_process_start_ms", "embedding_ipc_ms",
|
||
|
|
"embedding_model_load_ms", "embedding_tokenizer_ms",
|
||
|
|
"embedding_encode_ms", "embedding_postprocess_ms")
|
||
|
|
out["embedding"] = {
|
||
|
|
phase: {"p50_ms": _pct([r.get(phase, 0) for r in rows], .5),
|
||
|
|
"p95_ms": _pct([r.get(phase, 0) for r in rows], .95)}
|
||
|
|
for phase in embedding_phases if any(phase in r for r in rows)
|
||
|
|
}
|
||
|
|
if rows:
|
||
|
|
scoped = [r.get("retrieval_scopes") for r in rows if isinstance(r.get("retrieval_scopes"), list)]
|
||
|
|
if scoped:
|
||
|
|
latest = scoped[-1]
|
||
|
|
out["scopes"] = latest[:12]
|
||
|
|
warm_states = [r.get("warm_state") for r in rows if isinstance(r.get("warm_state"), dict)]
|
||
|
|
if warm_states:
|
||
|
|
out["warm_state"] = warm_states[-1]
|
||
|
|
out["embedding"]["cache_hit_rate"] = round(
|
||
|
|
sum(bool(r.get("embedding_cache_hit")) for r in rows if "embedding_cache_hit" in r)
|
||
|
|
/ max(1, sum(1 for r in rows if "embedding_cache_hit" in r)), 4)
|
||
|
|
out["embedding"]["sources"] = sorted({str(r.get("embedding_source")) for r in rows if r.get("embedding_source")})
|
||
|
|
out["embedding"]["models"] = sorted({str(r.get("embedding_model")) for r in rows if r.get("embedding_model")})
|
||
|
|
out["embedding"]["pids"] = sorted({int(r["embedding_pid"]) for r in rows if str(r.get("embedding_pid", "")).isdigit()})
|
||
|
|
counts = [int(r.get("embedding_count", 0) or 0) for r in rows if "embedding_count" in r]
|
||
|
|
if counts:
|
||
|
|
out["embedding"]["count_p50"] = _pct(counts, .5)
|
||
|
|
out["embedding"]["count_max"] = max(counts)
|
||
|
|
out["embedding"]["batch_warning"] = max(counts) > 1
|
||
|
|
return out
|
||
|
|
|
||
|
|
def ingest_http(row):
|
||
|
|
if isinstance(row,dict):
|
||
|
|
with _lock: _http.append(dict(row))
|
||
|
|
def ingest_rag(row): record_rag(row)
|
||
|
|
|
||
|
|
def install_sql_hooks(engine):
|
||
|
|
try:
|
||
|
|
from sqlalchemy import event
|
||
|
|
from sqlalchemy.engine import Engine
|
||
|
|
if getattr(engine,"_perf_hooks_installed",False): return
|
||
|
|
@event.listens_for(engine,"before_cursor_execute")
|
||
|
|
def before(conn,cursor,statement,parameters,context,executemany): context._perf_started=time.perf_counter()
|
||
|
|
@event.listens_for(engine,"after_cursor_execute")
|
||
|
|
def after(conn,cursor,statement,parameters,context,executemany): record_sql((time.perf_counter()-getattr(context,"_perf_started",time.perf_counter()))*1000)
|
||
|
|
@event.listens_for(engine,"handle_error")
|
||
|
|
def error(exception_context): record_sql(0,exception_context.original_exception)
|
||
|
|
engine._perf_hooks_installed=True
|
||
|
|
except Exception: pass
|