Skip to content

rag-observability.md

Latest commit

 

History

History
152 lines (116 loc) · 6.26 KB

rag-observability.md

File metadata and controls

152 lines (116 loc) · 6.26 KB
Raw

Reference — rag-observability

rag-observability owns the structured-logging configuration, the event registry, the set_log_context() ContextVar manager, the AsyncTelemetrySink (Step 1.1e) — a bounded, drop-on-overflow buffer for telemetry export so that exporter back-pressure never reaches the request path — and (Step 3.11) the status read-side: MetricsCollector (a pull-able metrics snapshot complementing OTel) and LogTail (an in-memory log ring buffer the gateway tails). The latter two back the operator console's Status & Metrics GUI; see status-api.md and architecture/status-metrics-gui.md.

Public surface

Name Purpose
configure_logging(...) Configure the 7-field JSON logger; idempotent. Call once at startup.
get_logger(name) Get a logger under the rag.* namespace (RAG001-compliant).
set_log_context(...) ContextVar-scoped trace/tenant/span enrichment.
reset_logging() Test-only reset of logger state.
AsyncTelemetrySink(export_fn, *, max_buffer, drain_timeout_s) Bounded buffer + background drainer for any async exporter.
SinkStats.from_sink(sink) Snapshot of buffer size + per-kind drop and exporter-error counters.
Event constants (EVT_*) + Pydantic event types The typed event registry consumed by structured log emission.
MetricsCollector / get_metrics_collector() In-memory counter/gauge/histogram registry with a pull-able snapshot() (Step 3.11).
LogTail / RingBufferLogHandler / attach_log_tail() Bounded log ring buffer + tailing handler, polled by seq cursor (Step 3.11).

AsyncTelemetrySink

Usage

from rag_observability import AsyncTelemetrySink

async def export(record):
    await otel_exporter.send(record)

sink = AsyncTelemetrySink(export, max_buffer=10_000)
sink.start()

# In the request path (hot, must not block):
sink.submit({"span": "vector.retrieve", "ms": 12}, kind="spans")

# At shutdown:
await sink.stop()

submit() returns True on enqueue, False if the queue was full and the record was dropped. The drop counter for that kind is incremented; read it via sink.dropped_total("spans") or all-kinds with sink.dropped_total().

Semantics

  • Non-blocking submit. Microsecond latency regardless of exporter state. Stalls in the exporter are absorbed by the queue, then by drops once the queue is full.
  • Single drainer task. Started by start(), cancelled by stop(). Records are exported FIFO.
  • Exporter exceptions swallowed and counted. A failing export_fn never crashes the drainer or surfaces to the caller; the failure is counted via sink.export_errors_total(kind).
  • Cooperative shutdown. stop() awaits queue.join() up to drain_timeout_s (default 5 s) before cancelling the drainer. Idempotent.
  • Default capacity. 10,000 records, matching the value published in docs/architecture/performance.md.

When to wire one in

Any code path that fires telemetry from inside the request path:

  • OTel exporter (logs / traces / metrics)
  • StageEvent stream to the Eval Recorder
  • Audit log shipper if the underlying AuditStore does network I/O
  • Any future log-to-Loki / event-to-Kafka pipeline

Counters and alerting

Surface the per-kind counters in your Telemetry SPI of choice:

metrics.gauge("telemetry.dropped_total", sink.dropped_total("spans"), labels={"kind": "spans"})
metrics.gauge("telemetry.export_errors_total", sink.export_errors_total("spans"), labels={"kind": "spans"})

Alert on either being non-zero. A drop count > 0 means the request path saved itself by sacrificing telemetry — investigate the exporter, not the request path.

Status read-side (Step 3.11)

Two in-process primitives give the operator console a pull-able view of state that OTel (push-only) and stdout logs don't expose. Both are dependency-free and thread-safe.

MetricsCollector

from rag_observability import get_metrics_collector

m = get_metrics_collector()                 # process-wide default
m.incr("gateway.requests_total", route="/v1/query", method="POST", status="2xx")
m.adjust_gauge("gateway.requests_in_flight", 1)   # atomic +/- for in-flight counts
m.observe("gateway.request_duration_ms", 12.4, route="/v1/query")
snap = m.snapshot()                         # sorted, JSON-ready MetricsSnapshot
payload = snap.as_dict()                    # for a JSON response
  • Exact aggregates, estimated percentiles. count/sum/min/max are running totals; p50/p95/p99 are nearest-rank over a bounded per-series reservoir (default 2048). Use bounded label values — route templates, status classes — to keep cardinality sane.
  • Complements OTel. SpiMetrics stays the export path; this is the read side the GUI polls. reset_metrics_collector() resets the default (test-only).

LogTail + RingBufferLogHandler

from rag_observability import attach_log_tail, get_log_tail

attach_log_tail()                            # mirror all rag.* records (idempotent)
tail = get_log_tail()
recent = tail.recent(limit=100, level="WARN", event="gateway.")
batch, cursor = tail.since(cursor, tenant="acme")   # poll for new records
  • The handler mirrors rag.* records into a bounded ring (default 2000), matching the _JsonFormatter field extraction (WARNING→WARN, rag. prefix stripped, context-var trace/span/tenant attached, rag_* extras carried).
  • Polled, not pushed. Each record gets a monotonic seq; streamers keep a cursor and call since(). This keeps emit free of event-loop coupling (it runs on arbitrary threads). attach_log_tail re-points the live handler at the current tail, so resetting the process-wide tail never orphans it.

Related

  • docs/architecture/performance.md — hot-path discipline + reviewer checklist; defines the published default capacity and alerting expectation.
  • docs/guides/logging-policy.md — RAG001 policy (no raw logging.getLogger) enforced for everything that emits log records.
  • TRACKER.md Step 1.1e (this work), Step 0.7b (structured logging foundation), Step 0.7 (OTel tracing + SPI metrics).