diff --git a/README.md b/README.md index df35d58..54c0ba3 100644 --- a/README.md +++ b/README.md @@ -242,6 +242,7 @@ Base URL: `http://localhost:8000` | `GET` | `/v1/datasets/{id}/compare` | Case-by-case diff (`?base=&candidate=`) | | `GET` | `/v1/datasets/runs/{id}` | One run with every case result | | `DELETE` | `/v1/datasets/runs/{id}` | Delete a run | +| `GET` | `/v1/ingest/health` | Last trace and counts per project | | `GET` | `/v1/evaluations/by-model` | Turn health by model and provider | | `POST` | `/v1/analytics/query` | Flexible metrics query | | `GET` | `/v1/analytics/query/schema` | Views, measures and dimensions | @@ -311,6 +312,18 @@ failed traces with reasons, one trace's spans, tools and provider chain, alerts, comparisons and metric queries. Read-only, and trace input/output are never returned unless `OBSERVER_MCP_CONTENT=1`. Setup and tools: [docs/mcp.md](docs/mcp.md). +### Ingestion reliability + +The Python SDK exporter retries only what can succeed later (network errors, 408/425/429/5xx) with +exponential backoff and jitter, honours `Retry-After`, and counts permanent 4xx answers as +rejected instead of retrying them. Pass `spool_dir` to keep batches that still fail on disk (mode +0600, unencrypted, bounded by `spool_max_bytes`) and resend them oldest first after the backend +returns or the process restarts; `exporter.stats()` reports `exported, retried, rejected, dropped, +spooled, recovered, spool_evicted, pending`. `/v1/traces/batch` is idempotent (a retried batch does +not fail on spans already stored) and rate-limited by `RATE_LIMIT_BATCH` (default `120/minute`). +`GET /v1/ingest/health` and the Overview card show when each project last sent a trace; the alert +metric **Minutes without traces** (`ingest_silence_minutes`) notifies when one goes quiet. + ### Test cases (datasets) On a trace page, **Save as case** stores a failure as a reusable test case in a dataset. A case is diff --git a/backend/app/alerts.py b/backend/app/alerts.py index 810a1ff..4040f2a 100644 --- a/backend/app/alerts.py +++ b/backend/app/alerts.py @@ -191,7 +191,14 @@ async def evaluate_rules(session: AsyncSession, *, now: datetime | None = None) if spec is None or rule.comparator not in COMPARATORS: continue details: dict[str, Any] = {} - if spec["kind"] == "quota": + if spec["kind"] == "ingest": + from app.api.ingest_health import silence_minutes + + value = await silence_minutes(session, rule.project_id, rule.environment, now) + if value is None: + continue # the project never sent anything: nothing to be silent about + samples, details = 1, {"project_id": rule.project_id} + elif spec["kind"] == "quota": value, samples, details = _quota_value(rule.scope or "") if value is None: continue # no trustworthy snapshot: neither fire nor resolve @@ -207,7 +214,10 @@ async def evaluate_rules(session: AsyncSession, *, now: datetime | None = None) ) value, samples = compute_metric(rule.metric, cache[key], scope=rule.scope or "") result.evaluated += 1 - enough = value is not None and samples >= int(rule.min_samples) + # Snapshot-style metrics (quota, ingest silence) have one reading, not a sample count. + enough = value is not None and ( + spec["kind"] in {"quota", "ingest"} or samples >= int(rule.min_samples) + ) open_event = ( ( await session.execute( diff --git a/backend/app/api/ingest_health.py b/backend/app/api/ingest_health.py new file mode 100644 index 0000000..2fdf567 --- /dev/null +++ b/backend/app/api/ingest_health.py @@ -0,0 +1,71 @@ +"""Is telemetry still arriving? (``GET /v1/ingest/health``) + +Per project: when the last trace was stored and how many arrived in the last 5 minutes, hour and +day. A sender that stops (crashed, blocked, spool full) shows up as growing silence, which the +``ingest_silence_minutes`` alert metric turns into a notification. +""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +from typing import Any + +from fastapi import APIRouter, Depends +from sqlalchemy import func, select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.db import get_session +from app.models.trace import Trace + +router = APIRouter() + + +def _aware(value: datetime | None) -> datetime | None: + if value is not None and value.tzinfo is None: + return value.replace(tzinfo=timezone.utc) + return value + + +async def silence_minutes( + session: AsyncSession, project_id: str, environment: str | None, now: datetime +) -> float | None: + """Minutes since the project's last stored trace; ``None`` if it never sent one.""" + query = select(func.max(Trace.created_at)).where(Trace.project_id == project_id) + if environment: + query = query.where(Trace.environment == environment) + last = _aware((await session.execute(query)).scalar_one_or_none()) + return None if last is None else max(0.0, (now - last).total_seconds() / 60.0) + + +@router.get("/health") +async def ingest_health(session: AsyncSession = Depends(get_session)) -> dict[str, Any]: + now = datetime.now(timezone.utc) + t5, t60, t24 = (now - timedelta(minutes=5), now - timedelta(hours=1), now - timedelta(days=1)) + rows = ( + await session.execute( + select( + Trace.project_id, + func.max(Trace.created_at), + func.count().filter(Trace.created_at >= t5), + func.count().filter(Trace.created_at >= t60), + func.count().filter(Trace.created_at >= t24), + ) + .where(Trace.project_id.is_not(None)) + .group_by(Trace.project_id) + ) + ).all() + projects = [] + for project_id, last, n5, n60, n24 in rows: + last = _aware(last) + projects.append( + { + "project_id": project_id, + "last_trace_at": last.isoformat() if last else None, + "silence_minutes": round((now - last).total_seconds() / 60.0, 1) if last else None, + "traces_5m": n5, + "traces_1h": n60, + "traces_24h": n24, + } + ) + projects.sort(key=lambda p: p["silence_minutes"] if p["silence_minutes"] is not None else 1e9) + return {"checked_at": now.isoformat(), "projects": projects} diff --git a/backend/app/api/traces.py b/backend/app/api/traces.py index f959286..7570f08 100644 --- a/backend/app/api/traces.py +++ b/backend/app/api/traces.py @@ -2,6 +2,7 @@ from __future__ import annotations +import os import uuid from datetime import datetime, timezone @@ -19,6 +20,9 @@ router = APIRouter() +# The SDK flushes every few seconds, so the old fixed 10/minute rejected healthy senders. +_BATCH_LIMIT = os.getenv("RATE_LIMIT_BATCH", "120/minute") + def _safe_uuid(value: str | None) -> str | None: """Validate and return a UUID string, or None if invalid.""" @@ -249,7 +253,7 @@ async def create_trace( @router.post("/batch", response_model=list[TraceResponse]) -@limiter.limit("10/minute") +@limiter.limit(_BATCH_LIMIT) async def create_traces_batch( request: Request, batch_data: BatchSpansCreate, @@ -270,6 +274,13 @@ async def create_traces_batch( created_traces: list[Trace] = [] + # A retried batch (the first answer was lost) must not fail on spans already stored. + wanted = [s.id for spans in spans_by_trace.values() for s in spans if s.id] + stored_ids: set[str] = set() + if wanted: + rows = await session.execute(select(Span.id).where(Span.id.in_(wanted))) + stored_ids = {str(i) for i in rows.scalars()} + for trace_id, spans in spans_by_trace.items(): # Check if trace already exists query = select(Trace).where(Trace.id == trace_id) @@ -295,6 +306,8 @@ async def create_traces_batch( # Add spans mode = await capture_mode(session, trace.project_id, "full") for span_data in spans: + if span_data.id and span_data.id in stored_ids: + continue captured = apply_capture( mode, inp=span_data.input, diff --git a/backend/app/main.py b/backend/app/main.py index fd48166..cdbb21c 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -15,6 +15,7 @@ from app.api import analytics, evaluations, manitos_ingest, providers, traces from app.api import dataset_runs as dataset_runs_api from app.api import datasets as datasets_api +from app.api import ingest_health as ingest_health_api from app.api import otlp as otlp_api from app.api import privacy as privacy_api from app.auth import require_api_key @@ -121,6 +122,12 @@ def _rate_limit_handler(request: Request, exc: RateLimitExceeded) -> JSONRespons tags=["privacy"], dependencies=[_app_auth], ) +app.include_router( + ingest_health_api.router, + prefix="/v1/ingest", + tags=["ingest"], + dependencies=[_app_auth], +) app.include_router( otlp_api.router, prefix="/v1/otlp", diff --git a/backend/app/metrics_core.py b/backend/app/metrics_core.py index db92f38..25da2cd 100644 --- a/backend/app/metrics_core.py +++ b/backend/app/metrics_core.py @@ -36,6 +36,11 @@ "unit": "ratio", "kind": "provider", }, + "ingest_silence_minutes": { + "label": "Minutos sin recibir trazas", + "unit": "count", + "kind": "ingest", + }, "provider_quota_out": { "label": "Proveedores sin cuota", "unit": "count", diff --git a/backend/tests/test_ingest_reliability.py b/backend/tests/test_ingest_reliability.py new file mode 100644 index 0000000..ab2dddf --- /dev/null +++ b/backend/tests/test_ingest_reliability.py @@ -0,0 +1,119 @@ +import uuid +from datetime import datetime, timedelta, timezone + +import pytest +from sqlalchemy import func, select + +from app.alerts import evaluate_rules +from app.models import AlertEvent, AlertRule, Span, Trace + + +def _batch(span_id, trace_id): + return { + "spans": [ + { + "id": span_id, + "trace_id": trace_id, + "name": "s", + "span_type": "llm", + "start_time": 1_750_000_000.0, + "end_time": 1_750_000_001.0, + "status": "ok", + } + ] + } + + +@pytest.mark.anyio +async def test_a_retried_batch_does_not_fail_or_duplicate(client, db_session): + span_id, trace_id = str(uuid.uuid4()), str(uuid.uuid4()) + first = await client.post("/v1/traces/batch", json=_batch(span_id, trace_id)) + assert first.status_code == 200 + again = await client.post("/v1/traces/batch", json=_batch(span_id, trace_id)) + assert again.status_code == 200 # the lost-answer retry used to be a 500 + count = (await db_session.execute(select(func.count()).select_from(Span))).scalar() + assert count == 1 + + +@pytest.mark.anyio +async def test_a_batch_with_old_and_new_spans_stores_only_the_new(client, db_session): + trace_id, old, new = str(uuid.uuid4()), str(uuid.uuid4()), str(uuid.uuid4()) + await client.post("/v1/traces/batch", json=_batch(old, trace_id)) + both = {"spans": _batch(old, trace_id)["spans"] + _batch(new, trace_id)["spans"]} + assert (await client.post("/v1/traces/batch", json=both)).status_code == 200 + ids = set((await db_session.execute(select(Span.id))).scalars()) + assert ids == {old, new} + + +async def _trace(db, project, minutes_ago): + created = datetime.now(timezone.utc) - timedelta(minutes=minutes_ago) + db.add( + Trace( + name="t", + start_time=created, + status="ok", + project_id=project, + environment="e", + created_at=created, + ) + ) + await db.commit() + + +@pytest.mark.anyio +async def test_ingest_health_reports_last_trace_and_counts(client, db_session): + await _trace(db_session, "alive", 2) + await _trace(db_session, "alive", 30) + await _trace(db_session, "quiet", 600) + data = (await client.get("/v1/ingest/health")).json() + rows = {p["project_id"]: p for p in data["projects"]} + assert rows["alive"]["traces_5m"] == 1 and rows["alive"]["traces_1h"] == 2 + assert rows["alive"]["silence_minutes"] < 5 + assert rows["quiet"]["traces_24h"] == 1 and rows["quiet"]["silence_minutes"] > 500 + assert data["projects"][0]["project_id"] == "alive" # most recently heard first + + +@pytest.mark.anyio +async def test_silence_alert_fires_then_resolves_when_traces_return(db_session): + await _trace(db_session, "svc", 90) + rule = AlertRule( + name="svc silent", + metric="ingest_silence_minutes", + comparator=">", + threshold=30.0, + window_minutes=60, + min_samples=10, # min_samples is ignored + project_id="svc", + severity="warning", + cooldown_minutes=0, + ) + db_session.add(rule) + await db_session.commit() + result = await evaluate_rules(db_session) + assert result.opened == 1 + event = (await db_session.execute(select(AlertEvent))).scalars().one() + assert event.state == "open" and event.value > 80 + await _trace(db_session, "svc", 0) # the sender is back + again = await evaluate_rules(db_session) + assert again.resolved == 1 + await db_session.refresh(event) + assert event.state == "resolved" + + +@pytest.mark.anyio +async def test_a_project_that_never_sent_anything_is_not_silent(db_session): + db_session.add( + AlertRule( + name="ghost", + metric="ingest_silence_minutes", + comparator=">", + threshold=1.0, + window_minutes=60, + min_samples=1, + project_id="never", + severity="info", + cooldown_minutes=0, + ) + ) + await db_session.commit() + assert (await evaluate_rules(db_session)).opened == 0 diff --git a/dashboard/src/api.ts b/dashboard/src/api.ts index d67b227..2138ac9 100644 --- a/dashboard/src/api.ts +++ b/dashboard/src/api.ts @@ -26,6 +26,7 @@ import type { RunComparison, CaptureMode, PrivacyOverview, + IngestHealth, } from './types'; const BASE_URL = '/api'; @@ -176,6 +177,11 @@ export const api = { return request(`/v1/privacy/projects/${encodeURIComponent(project)}/enforce`, { method: 'POST', body: JSON.stringify({ confirm: true }) }); }, + // Is telemetry arriving? + async getIngestHealth(): Promise { + return request('/v1/ingest/health'); + }, + // Alerts async getAlertMetrics(): Promise { return request('/v1/alerts/metrics'); diff --git a/dashboard/src/components/IngestHealth.tsx b/dashboard/src/components/IngestHealth.tsx new file mode 100644 index 0000000..f01e96f --- /dev/null +++ b/dashboard/src/components/IngestHealth.tsx @@ -0,0 +1,40 @@ +import { useEffect, useState } from 'react'; +import { api } from '../api'; +import type { IngestHealth as Health } from '../types'; +import { Card, EmptyState } from './ui'; +import { silenceText, silenceTone } from '../helpers/ingest'; +import { fmtInt } from '../helpers/format'; +import { t } from '../i18n'; + +const DOT: Record = { ok: 'bg-ok', warn: 'bg-warn', bad: 'bg-bad', neutral: 'bg-faint' }; + +/** Is telemetry still arriving? Fails silently: it is a hint, never a reason to show an error. */ +export default function IngestHealth() { + const [health, setHealth] = useState(null); + + useEffect(() => { + let live = true; + const load = () => api.getIngestHealth().then((h) => { if (live) setHealth(h); }).catch(() => undefined); + load(); + const id = setInterval(load, 30_000); + return () => { live = false; clearInterval(id); }; + }, []); + + if (!health) return null; + return ( + + {health.projects.length === 0 ? : ( +
    + {health.projects.map((p) => ( +
  • + + {p.project_id} + {t('hace {x}', { x: silenceText(p.silence_minutes) })} + {t('{a} en 5 min · {b} en 1 h · {c} en 24 h', { a: fmtInt(p.traces_5m), b: fmtInt(p.traces_1h), c: fmtInt(p.traces_24h) })} +
    • + ))} +
+ )} + + ); +} diff --git a/dashboard/src/helpers/ingest.ts b/dashboard/src/helpers/ingest.ts new file mode 100644 index 0000000..e3298c4 --- /dev/null +++ b/dashboard/src/helpers/ingest.ts @@ -0,0 +1,22 @@ +/** Pure helpers for the ingestion-health card (no React imports, so Node can test them). */ + +export type IngestTone = 'ok' | 'warn' | 'bad' | 'neutral'; + +/** How long a project has been silent, in words: "ahora", "12 min", "3 h", "2 d". */ +export function silenceText(minutes: number | null): string { + if (minutes == null) return '–'; + if (minutes < 1) return '<1 min'; + if (minutes < 60) return `${Math.round(minutes)} min`; + if (minutes < 1440) return `${Math.round(minutes / 60)} h`; + return `${Math.round(minutes / 1440)} d`; +} + +/** Tone of a project's silence given how often it normally sends (traces in the last hour). */ +export function silenceTone(minutes: number | null, tracesLastHour: number): IngestTone { + if (minutes == null) return 'neutral'; + if (minutes <= 10) return 'ok'; + // A project that sent several traces an hour ago and is now quiet is worth a look; one that + // was already idle is just idle. + if (tracesLastHour === 0 && minutes > 60) return 'neutral'; + return minutes > 60 ? 'bad' : 'warn'; +} diff --git a/dashboard/src/i18n-en.ts b/dashboard/src/i18n-en.ts index 2c8353c..1b1a829 100644 --- a/dashboard/src/i18n-en.ts +++ b/dashboard/src/i18n-en.ts @@ -304,6 +304,12 @@ export const EN: Record = { 'Modo de {p}': 'Mode of {p}', Predeterminado: 'Default', 'Aplicar a lo guardado': 'Apply to stored data', + 'Recepción de telemetría': 'Telemetry intake', + 'Cuándo llegó la última traza de cada proyecto': 'When the last trace of each project arrived', + 'Aún no llega telemetría': 'No telemetry yet', + 'hace {x}': '{x} ago', + '{a} en 5 min · {b} en 1 h · {c} en 24 h': '{a} in 5 min · {b} in 1 h · {c} in 24 h', + 'Minutos sin recibir trazas': 'Minutes without traces', 'Salud por modelo': 'Health by model', 'Salud operativa media de las evaluaciones automáticas por reglas · {n} turnos · la corrección de las respuestas aún no se mide': 'Average operational health of the automatic rule-based evaluations · {n} turns · answer correctness is not measured yet', diff --git a/dashboard/src/pages/Alerts.tsx b/dashboard/src/pages/Alerts.tsx index acd641a..e37a6b6 100644 --- a/dashboard/src/pages/Alerts.tsx +++ b/dashboard/src/pages/Alerts.tsx @@ -189,7 +189,7 @@ function Alerts() { onChange={(e) => setDraft({ ...draft, threshold: e.target.value })} /> - {draftMetric?.kind !== 'quota' && ( + {draftMetric?.kind !== 'quota' && draftMetric?.kind !== 'ingest' && ( <>