diff --git a/README.md b/README.md index ba1c9db..a81d0e8 100644 --- a/README.md +++ b/README.md @@ -224,6 +224,7 @@ Base URL: `http://localhost:8000` | `GET` | `/v1/alerts/events` | Alert events (`?state=open`) | | `POST` | `/v1/alerts/events/{id}/ack` | Mark an alert as seen | | `POST` | `/v1/alerts/evaluate` | Evaluate all rules now | +| `POST` | `/v1/otlp/v1/traces` | Standard OTLP/HTTP trace receiver | | `GET` | `/v1/datasets` | List test-case datasets | | `POST` | `/v1/datasets` | Create a dataset | | `GET` | `/v1/datasets/{id}` | A dataset with its cases | @@ -280,6 +281,13 @@ Scores of the previous rubric are replaced when the turn is re-scored. A backgro backfills history. Evaluations are idempotent (the id derives from the trace and the rubric). The Evaluations page shows health by model, both axes, the grade and the per-criterion breakdown. +### Standard OpenTelemetry (OTLP) ingestion + +Point any stock OpenTelemetry exporter at `OTEL_EXPORTER_OTLP_ENDPOINT=http://127.0.0.1:8100/v1/otlp` +(protobuf or JSON, optionally gzipped) and its traces appear without the Observer SDK. GenAI +conventions map to Observer span types, tokens and tools. Content-like attributes are dropped by +default (`OBSERVATORY_OTLP_CAPTURE=metadata`). Mapping, privacy and limits: [docs/otlp.md](docs/otlp.md). + ### Read-only MCP server `observer-mcp` lets Claude Code, Codex or any MCP client investigate Observer without screenshots: diff --git a/backend/app/api/manitos_ingest.py b/backend/app/api/manitos_ingest.py index 2c61e37..9169ad2 100644 --- a/backend/app/api/manitos_ingest.py +++ b/backend/app/api/manitos_ingest.py @@ -312,6 +312,14 @@ async def ingest_manitos_trace( """Upsert a complete integration trace with retry-safe semantics.""" del request # Required by SlowAPI's decorator contract. + return await apply_envelope(session, payload) + + +async def apply_envelope( + session: AsyncSession, payload: ManitOSIngestRequest +) -> ManitOSIngestResponse: + """The ingestion itself (also used by the OTLP receiver, which builds envelopes).""" + digest = _payload_hash(payload) receipt = await _existing_receipt( session, diff --git a/backend/app/api/otlp.py b/backend/app/api/otlp.py new file mode 100644 index 0000000..f2df6e9 --- /dev/null +++ b/backend/app/api/otlp.py @@ -0,0 +1,109 @@ +"""OTLP/HTTP trace receiver (``POST /v1/otlp/v1/traces``). + +Point a stock OpenTelemetry exporter at it: +``OTEL_EXPORTER_OTLP_ENDPOINT=http://127.0.0.1:8100/v1/otlp`` (the exporter appends ``/v1/traces``), +plus ``OTEL_EXPORTER_OTLP_HEADERS=Authorization=Bearer `` when a key is required. +""" + +from __future__ import annotations + +import os +from datetime import timezone +from typing import Any + +from fastapi import APIRouter, Depends, Request, Response +from pydantic import ValidationError +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.api.manitos_ingest import ManitOSIngestRequest, apply_envelope +from app.db import get_session +from app.models.trace import Span, Trace +from app.otlp import ( + MAX_BODY_BYTES, + OtlpError, + decompress, + request_from_json, + request_from_protobuf, + response_body, + translate, +) +from app.ratelimit import limiter + +router = APIRouter() + +_OTLP_LIMIT = os.getenv("RATE_LIMIT_OTLP", "600/minute") + + +def _plain(status: int, message: str) -> Response: + return Response(content=message, status_code=status, media_type="text/plain") + + +async def _link_and_widen(session: AsyncSession, envelope: dict[str, Any]) -> None: + """Spans of one trace can arrive in several batches: keep the trace's full time range and link + a span to a parent stored by an earlier batch (instead of failing or orphaning it).""" + trace_id = envelope["trace"]["id"] + external = {s["_external_parent"] for s in envelope["spans"] if s.get("_external_parent")} + stored: set[str] = set() + if external: + rows = await session.execute( + select(Span.id).where(Span.id.in_(external), Span.trace_id == trace_id) + ) + stored = {str(i) for i in rows.scalars()} + for span in envelope["spans"]: + parent = span.pop("_external_parent", None) + if parent in stored: + span["parent_id"] = parent + existing = await session.get(Trace, trace_id) + if existing is not None and existing.start_time is not None: + trace = envelope["trace"] + start = existing.start_time + if start.tzinfo is None: + start = start.replace(tzinfo=timezone.utc) + trace["start_time"] = min(trace["start_time"], start.timestamp()) + ends = [trace["end_time"]] if trace.get("end_time") else [] + if existing.end_time is not None: + end = existing.end_time + if end.tzinfo is None: + end = end.replace(tzinfo=timezone.utc) + ends.append(end.timestamp()) + trace["end_time"] = max(ends) if ends else None + trace["name"] = existing.name if existing.name else trace["name"] + + +@router.post("/v1/traces") +@limiter.limit(_OTLP_LIMIT) +async def receive_traces( + request: Request, session: AsyncSession = Depends(get_session) +) -> Response: + declared = request.headers.get("content-length") + if declared and declared.isdigit() and int(declared) > MAX_BODY_BYTES: + return _plain(413, "request too large") + content_type = request.headers.get("content-type", "").split(";")[0].strip().lower() + protobuf = content_type == "application/x-protobuf" + if not protobuf and content_type != "application/json": + return _plain(415, "use application/x-protobuf or application/json") + try: + body = decompress(await request.body(), request.headers.get("content-encoding")) + data = request_from_protobuf(body) if protobuf else request_from_json(body) + translated = translate(data, project_id=request.headers.get("x-observer-project")) + except OtlpError as exc: + return _plain(exc.status, str(exc)) + + rejected = translated.rejected + errors = list(translated.errors) + for envelope in translated.envelopes: + count = len(envelope["spans"]) + try: + await _link_and_widen(session, envelope) + payload = ManitOSIngestRequest.model_validate(envelope) + async with session.begin_nested(): # a rejected trace leaves no receipt behind + await apply_envelope(session, payload) + except ValidationError as exc: + rejected += count + errors.append(f"trace rejected: {exc.errors()[0]['msg']}"[:200]) + except Exception as exc: # HTTPException from the shared ingestion: conflicts + rejected += count + errors.append(f"trace rejected: {getattr(exc, 'detail', exc.__class__.__name__)}"[:200]) + content, media = response_body(rejected, "; ".join(errors[:3]), protobuf=protobuf) + return Response(content=content, media_type=media) diff --git a/backend/app/main.py b/backend/app/main.py index a8e365d..88a6e8c 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 otlp as otlp_api from app.auth import require_api_key from app.db.postgres import engine from app.ratelimit import limiter @@ -113,6 +114,12 @@ def _rate_limit_handler(request: Request, exc: RateLimitExceeded) -> JSONRespons tags=["datasets"], dependencies=[_app_auth], ) +app.include_router( + otlp_api.router, + prefix="/v1/otlp", + tags=["otlp"], + dependencies=[_app_auth], +) app.include_router( datasets_api.router, prefix="/v1/datasets", diff --git a/backend/app/otlp.py b/backend/app/otlp.py new file mode 100644 index 0000000..89bafcf --- /dev/null +++ b/backend/app/otlp.py @@ -0,0 +1,397 @@ +"""OTLP/HTTP trace receiver: translate standard OpenTelemetry spans into Observer envelopes. + +Any app instrumented with a stock OpenTelemetry exporter can send here (JSON or protobuf, +optionally gzipped) without the Observer SDK. The translation produces the same +``manitos.telemetry.v1`` envelopes the Manitos exporter sends, so ingestion, idempotency and +the dashboard are shared. Mapping reference: ``docs/otlp.md``. + +Privacy: Observer is metadata-only by default. In ``metadata`` capture mode (the default, +``OBSERVATORY_OTLP_CAPTURE``) attributes that can carry prompts, replies or other content are +dropped before storage; ``full`` keeps them. +""" + +from __future__ import annotations + +import base64 +import gzip +import hashlib +import json +import os +import re +import uuid +import zlib +from dataclasses import dataclass +from typing import Any + +MAX_BODY_BYTES = 16 * 1024 * 1024 +_NAMESPACE = uuid.UUID("0b3f6d1e-6a54-4c63-9a52-7d2f5d9c1a77") +_MAX_ATTR_BYTES = 200 +_MAX_ATTRS = 64 +_SPANS_PER_ENVELOPE = 500 + +# Standard attributes kept in metadata mode even though their name looks content-like. +_ALLOWED = ( + "gen_ai.usage.", + "gen_ai.operation.name", + "gen_ai.system", + "gen_ai.provider.name", + "gen_ai.request.model", + "gen_ai.response.model", + "gen_ai.response.finish_reasons", + "gen_ai.tool.name", + "gen_ai.agent.name", + "gen_ai.request.temperature", + "gen_ai.request.max_tokens", +) +_DENY = re.compile( + r"prompt|completion|content|message|input|output|query|statement|body|text|payload|" + r"password|passwd|token|secret|credential|authorization|cookie|api[._-]?key|email|" + r"phone|address|url|path|file|argument|param", + re.IGNORECASE, +) + +_GEN_AI_OPERATION_TYPE = { + "chat": "llm", + "text_completion": "llm", + "generate_content": "llm", + "completion": "llm", + "embeddings": "embedding", + "execute_tool": "tool", + "invoke_agent": "agent", + "create_agent": "agent", +} +_OPENINFERENCE_KIND = { + "LLM": "llm", + "TOOL": "tool", + "AGENT": "agent", + "CHAIN": "chain", + "RETRIEVER": "retriever", + "EMBEDDING": "embedding", + "RERANKER": "reranker", +} + + +class OtlpError(ValueError): + """The request cannot be decoded (answered 400/413/415).""" + + def __init__(self, message: str, status: int = 400) -> None: + super().__init__(message) + self.status = status + + +def capture_mode() -> str: + return ( + "full" + if os.getenv("OBSERVATORY_OTLP_CAPTURE", "metadata").strip().lower() == "full" + else "metadata" + ) + + +def decompress(body: bytes, encoding: str | None) -> bytes: + """Undo ``Content-Encoding`` with a cap on the decompressed size (zip-bomb guard).""" + encoding = (encoding or "").strip().lower() + if encoding in {"", "identity"}: + data = body + elif encoding in {"gzip", "x-gzip"}: + decoder = zlib.decompressobj(16 + zlib.MAX_WBITS) + data = _bounded_decompress(decoder, body) + elif encoding == "deflate": + data = _bounded_decompress(zlib.decompressobj(), body) + else: + raise OtlpError(f"unsupported Content-Encoding: {encoding}", 415) + if len(data) > MAX_BODY_BYTES: + raise OtlpError("request too large", 413) + return data + + +def _bounded_decompress(decoder: Any, body: bytes) -> bytes: + try: + data = decoder.decompress(body, MAX_BODY_BYTES + 1) + except (zlib.error, EOFError, gzip.BadGzipFile) as exc: + raise OtlpError("invalid compressed body") from exc + if decoder.unconsumed_tail: + raise OtlpError("request too large", 413) + return data + + +def _hex_id(value: Any) -> str: + """OTLP/JSON ids are hex strings; protobuf ids are bytes.""" + if isinstance(value, (bytes, bytearray)): + return bytes(value).hex() + return str(value or "").strip().lower() + + +def trace_uuid(trace_id_hex: str) -> str: + """The 16-byte trace id as a UUID (a bijection, so a trace keeps one identity).""" + return str(uuid.UUID(hex=trace_id_hex.rjust(32, "0")[-32:])) + + +def span_uuid(trace_id_hex: str, span_id_hex: str) -> str: + """Span ids are 8 bytes: a stable derived UUID, so a re-sent span upserts, not duplicates.""" + return str(uuid.uuid5(_NAMESPACE, f"{trace_id_hex}:{span_id_hex}")) + + +def _any_value(value: dict[str, Any]) -> Any: + for key in ("stringValue", "boolValue", "doubleValue"): + if key in value: + return value[key] + if "intValue" in value: + try: + return int(value["intValue"]) + except (TypeError, ValueError): + return None + if "arrayValue" in value: + items = [_any_value(v) for v in (value["arrayValue"] or {}).get("values", [])] + return items if all(isinstance(i, (str, int, float, bool)) for i in items) else None + return None + + +def _attrs(items: list[dict[str, Any]] | None) -> dict[str, Any]: + out: dict[str, Any] = {} + for item in items or []: + key = str(item.get("key") or "") + if key: + value = _any_value(item.get("value") or {}) + if value is not None: + out[key] = value + return out + + +def _safe_attributes(raw: dict[str, Any], mode: str) -> dict[str, Any]: + out: dict[str, Any] = {} + for key, value in raw.items(): + if len(out) >= _MAX_ATTRS: + break + if mode != "full" and not key.startswith(_ALLOWED) and _DENY.search(key): + continue + if isinstance(value, str): + value = value[:_MAX_ATTR_BYTES] if mode != "full" else value[:4000] + elif isinstance(value, list): + value = [v[:_MAX_ATTR_BYTES] if isinstance(v, str) else v for v in value[:16]] + out[key[:128]] = value + return out + + +def _span_type(attrs: dict[str, Any]) -> str: + kind = str(attrs.get("openinference.span.kind") or "").upper() + if kind in _OPENINFERENCE_KIND: + return _OPENINFERENCE_KIND[kind] + operation = str(attrs.get("gen_ai.operation.name") or "").lower() + if operation in _GEN_AI_OPERATION_TYPE: + return _GEN_AI_OPERATION_TYPE[operation] + if attrs.get("gen_ai.tool.name"): + return "tool" + return "generic" + + +def _number(value: Any) -> float | None: + return float(value) if isinstance(value, (int, float)) and not isinstance(value, bool) else None + + +def _status(span: dict[str, Any]) -> str: + code = (span.get("status") or {}).get("code", 0) + if code in (2, "STATUS_CODE_ERROR"): + return "error" + if code in (1, "STATUS_CODE_OK"): + return "ok" + return "unset" + + +def _seconds(nanos: Any) -> float | None: + try: + value = int(nanos) + except (TypeError, ValueError): + return None + return value / 1e9 if value > 0 else None + + +@dataclass +class Translated: + """Envelopes ready for ingestion plus the spans that could not be translated.""" + + envelopes: list[dict[str, Any]] + rejected: int + errors: list[str] + + +def translate( + request: dict[str, Any], + *, + project_id: str | None = None, + mode: str | None = None, +) -> Translated: + """OTLP ``ExportTraceServiceRequest`` (JSON-shaped dict) -> telemetry envelopes.""" + mode = mode or capture_mode() + per_trace: dict[str, dict[str, Any]] = {} + rejected = 0 + errors: list[str] = [] + for resource_spans in request.get("resourceSpans") or []: + resource = _attrs((resource_spans.get("resource") or {}).get("attributes")) + service = str(resource.get("service.name") or "otlp")[:128] or "otlp" + environment = str( + resource.get("deployment.environment.name") + or resource.get("deployment.environment") + or "default" + )[:64] + instance = str(resource.get("service.instance.id") or service)[:128] + scopes = resource_spans.get("scopeSpans") or resource_spans.get( + "instrumentationLibrarySpans" + ) + for scope_spans in scopes or []: + for span in scope_spans.get("spans") or []: + tid = _hex_id(span.get("traceId")) + sid = _hex_id(span.get("spanId")) + start = _seconds(span.get("startTimeUnixNano")) + end = _seconds(span.get("endTimeUnixNano")) + if ( + not re.fullmatch(r"[0-9a-f]{1,32}", tid) + or not re.fullmatch(r"[0-9a-f]{1,16}", sid) + or start is None + ): + rejected += 1 + errors.append("span with an invalid trace id, span id or start time") + continue + if end is not None and end < start: + end = start + raw = _attrs(span.get("attributes")) + attrs = _safe_attributes(raw, mode) + model = raw.get("gen_ai.response.model") or raw.get("gen_ai.request.model") + if model: + attrs["model"] = str(model)[:160] + span_type = _span_type(raw) + name = str(span.get("name") or "span") + if span_type == "tool" and raw.get("gen_ai.tool.name"): + name = str(raw["gen_ai.tool.name"]) + tokens_in = _number( + raw.get("gen_ai.usage.input_tokens", raw.get("gen_ai.usage.prompt_tokens")) + ) + tokens_out = _number( + raw.get("gen_ai.usage.output_tokens", raw.get("gen_ai.usage.completion_tokens")) + ) + parent = _hex_id(span.get("parentSpanId")) + item = { + "id": span_uuid(tid, sid), + "_parent": span_uuid(tid, parent) if parent.strip("0") else None, + "name": name[:255] or "span", + "span_type": span_type, + "start_time": start, + "end_time": end, + "status": _status(span), + "attributes": attrs or None, + "tokens_input": int(tokens_in) if tokens_in is not None else None, + "tokens_output": int(tokens_out) if tokens_out is not None else None, + "_root": not parent.strip("0"), + } + bucket = per_trace.setdefault( + tid, + { + "project": (project_id or service)[:128], + "environment": environment, + "instance": instance, + "session": str( + raw.get("session.id") or raw.get("gen_ai.conversation.id") or tid + )[:255], + "spans": {}, + }, + ) + bucket["spans"][item["id"]] = item + envelopes: list[dict[str, Any]] = [] + for tid, bucket in per_trace.items(): + items = list(bucket["spans"].values()) + ids = {i["id"] for i in items} + roots = [i for i in items if i["_root"]] + head = min(roots or items, key=lambda i: i["start_time"]) + starts = [i["start_time"] for i in items] + ends = [i["end_time"] for i in items if i["end_time"] is not None] + trace_status = ( + "error" + if any(i["status"] == "error" for i in items) + else ("ok" if all(i["status"] == "ok" for i in items) else "unset") + ) + for chunk_start in range(0, len(items), _SPANS_PER_ENVELOPE): + chunk = items[chunk_start : chunk_start + _SPANS_PER_ENVELOPE] + spans = [] + for i in chunk: + clean = {k: v for k, v in i.items() if not k.startswith("_") and v is not None} + parent = i["_parent"] + clean["parent_id"] = ( + parent if parent in ids else None + ) # may arrive in a later batch + clean["_external_parent"] = parent if parent not in ids else None + spans.append(clean) + body = { + "schema_version": "manitos.telemetry.v1", + "project_id": bucket["project"], + "environment": bucket["environment"], + "service_instance_id": bucket["instance"], + "session_id": bucket["session"], + "turn_id": tid[:128], + "actor_id_hash": None, + "trace": { + "id": trace_uuid(tid), + "name": head["name"], + "start_time": min(starts), + "end_time": max(ends) if ends else None, + "status": trace_status, + }, + "spans": spans, + } + digest = hashlib.sha256( + json.dumps(body, sort_keys=True, default=str).encode() + ).hexdigest() + body["idempotency_key"] = f"otlp:{digest[:48]}" + envelopes.append(body) + return Translated(envelopes=envelopes, rejected=rejected, errors=errors[:5]) + + +def request_from_protobuf(data: bytes) -> dict[str, Any]: + """Decode a protobuf ``ExportTraceServiceRequest`` into the OTLP/JSON shape.""" + try: + from google.protobuf.json_format import MessageToDict + from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import ( + ExportTraceServiceRequest, + ) + except ImportError as exc: # declared dependency; this is a broken install + raise OtlpError("protobuf support is not installed", 415) from exc + message = ExportTraceServiceRequest() + try: + message.ParseFromString(data) + except Exception as exc: # protobuf raises DecodeError and others + raise OtlpError("invalid protobuf body") from exc + data = MessageToDict(message, preserving_proto_field_name=False) + # protobuf's JSON mapping writes bytes as base64; OTLP/JSON (and this module) use hex ids. + for resource_spans in data.get("resourceSpans", []): + for scope_spans in resource_spans.get("scopeSpans", []): + for span in scope_spans.get("spans", []): + for key in ("traceId", "spanId", "parentSpanId"): + if span.get(key): + span[key] = base64.b64decode(span[key]).hex() + return data + + +def request_from_json(data: bytes) -> dict[str, Any]: + try: + value = json.loads(data) + except ValueError as exc: + raise OtlpError("invalid JSON body") from exc + if not isinstance(value, dict): + raise OtlpError("body must be a JSON object") + return value + + +def response_body(rejected: int, message: str, *, protobuf: bool) -> tuple[bytes, str]: + """``ExportTraceServiceResponse`` in the format the exporter used.""" + if protobuf: + from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import ( + ExportTraceServiceResponse, + ) + + response = ExportTraceServiceResponse() + if rejected: + response.partial_success.rejected_spans = rejected + response.partial_success.error_message = message[:500] + return response.SerializeToString(), "application/x-protobuf" + payload: dict[str, Any] = {} + if rejected: + payload["partialSuccess"] = {"rejectedSpans": str(rejected), "errorMessage": message[:500]} + return json.dumps(payload).encode(), "application/json" diff --git a/backend/pyproject.toml b/backend/pyproject.toml index cafde26..44782f6 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -21,6 +21,7 @@ dependencies = [ "slowapi>=0.1.9", "pyyaml>=6.0", "httpx>=0.27.0", + "opentelemetry-proto>=1.27,<2", ] [project.urls] diff --git a/backend/tests/test_otlp.py b/backend/tests/test_otlp.py new file mode 100644 index 0000000..e7ac407 --- /dev/null +++ b/backend/tests/test_otlp.py @@ -0,0 +1,233 @@ +import gzip +import json + +import pytest +from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import ( + ExportTraceServiceRequest, + ExportTraceServiceResponse, +) +from sqlalchemy import select + +from app.models import Span, Trace +from app.otlp import OtlpError, decompress, span_uuid, trace_uuid, translate + +TID = "5b8efff798038103d269b633813fc60c" +ROOT = "eee19b7ec3c1b174" +CHILD = "eee19b7ec3c1b175" +T0 = 1_760_000_000_000_000_000 + + +def _attr(key, value): + if isinstance(value, bool): + return {"key": key, "value": {"boolValue": value}} + if isinstance(value, int): + return {"key": key, "value": {"intValue": str(value)}} + return {"key": key, "value": {"stringValue": value}} + + +def _span(sid, name, *, parent="", start=0, ms=1000, status=1, attrs=None, tid=TID): + return { + "traceId": tid, + "spanId": sid, + "parentSpanId": parent, + "name": name, + "startTimeUnixNano": str(T0 + start), + "endTimeUnixNano": str(T0 + start + ms * 1_000_000), + "status": {"code": status}, + "attributes": [_attr(k, v) for k, v in (attrs or {}).items()], + } + + +def _request(spans, service="my-agent", env="prod", extra_resource=None): + resource = [_attr("service.name", service), _attr("deployment.environment.name", env)] + resource += [_attr(k, v) for k, v in (extra_resource or {}).items()] + return { + "resourceSpans": [{"resource": {"attributes": resource}, "scopeSpans": [{"spans": spans}]}] + } + + +def _demo(): + return _request( + [ + _span(ROOT, "agent.run", attrs={"gen_ai.operation.name": "invoke_agent"}), + _span( + CHILD, + "chat gpt", + parent=ROOT, + start=10_000_000, + attrs={ + "gen_ai.operation.name": "chat", + "gen_ai.request.model": "gpt-x", + "gen_ai.usage.input_tokens": 120, + "gen_ai.usage.output_tokens": 30, + "gen_ai.prompt": "PRIVATE PROMPT", + "gen_ai.completion": "PRIVATE REPLY", + "http.url": "https://x.test/?token=SECRET", + }, + ), + _span( + "eee19b7ec3c1b176", + "tool call", + parent=ROOT, + start=20_000_000, + status=2, + attrs={"gen_ai.operation.name": "execute_tool", "gen_ai.tool.name": "web_search"}, + ), + ] + ) + + +def test_translation_maps_types_status_ids_and_drops_content(): + out = translate(_demo(), mode="metadata") + assert out.rejected == 0 and len(out.envelopes) == 1 + env = out.envelopes[0] + assert env["project_id"] == "my-agent" and env["environment"] == "prod" + assert env["trace"]["id"] == trace_uuid(TID) and env["trace"]["name"] == "agent.run" + assert env["trace"]["status"] == "error" # one span failed + by_name = {s["name"]: s for s in env["spans"]} + assert by_name["agent.run"]["span_type"] == "agent" + llm = by_name["chat gpt"] + assert llm["span_type"] == "llm" and llm["tokens_input"] == 120 and llm["tokens_output"] == 30 + assert llm["attributes"]["model"] == "gpt-x" + assert llm["parent_id"] == span_uuid(TID, ROOT) + tool = by_name["web_search"] # a tool span is named after the tool, like Manitos' own + assert tool["span_type"] == "tool" and tool["status"] == "error" + blob = json.dumps(env) + for secret in ("PRIVATE", "SECRET"): + assert secret not in blob + assert "gen_ai.usage.input_tokens" in llm["attributes"] # counts are not content + + +def test_full_capture_keeps_content_attributes(): + env = translate(_demo(), mode="full").envelopes[0] + assert "PRIVATE PROMPT" in json.dumps(env) + + +def test_invalid_spans_are_rejected_not_fatal(): + bad = _request([_span("zz", "x"), _span(ROOT, "ok")]) + out = translate(bad) + assert out.rejected == 1 and len(out.envelopes[0]["spans"]) == 1 + + +def test_decompress_gzip_and_limits(): + data = b'{"a":1}' + assert decompress(gzip.compress(data), "gzip") == data + with pytest.raises(OtlpError, match="unsupported"): + decompress(data, "br") + with pytest.raises(OtlpError, match="invalid compressed"): + decompress(b"not gzip", "gzip") + import app.otlp as otlp + + old = otlp.MAX_BODY_BYTES + otlp.MAX_BODY_BYTES = 100 + try: + with pytest.raises(OtlpError) as exc: + decompress(gzip.compress(b"0" * 10_000), "gzip") + assert exc.value.status == 413 + finally: + otlp.MAX_BODY_BYTES = old + + +def _post(client, body, *, ctype="application/json", headers=None): + return client.post( + "/v1/otlp/v1/traces", + content=body, + headers={"content-type": ctype, **(headers or {})}, + ) + + +@pytest.mark.anyio +async def test_json_ingestion_stores_trace_and_spans_and_is_idempotent(client, db_session): + body = json.dumps(_demo()) + first = await _post(client, body) + assert first.status_code == 200 and first.json() == {} + again = await _post(client, body) # an exporter retry + assert again.status_code == 200 + trace = await db_session.get(Trace, trace_uuid(TID)) + assert trace.project_id == "my-agent" and trace.environment == "prod" + spans = ( + (await db_session.execute(select(Span).where(Span.trace_id == trace_uuid(TID)))) + .scalars() + .all() + ) + assert len(spans) == 3 # not duplicated by the retry + tool = next(s for s in spans if s.span_type == "tool") + assert tool.name == "web_search" and tool.status == "error" + child = next(s for s in spans if s.name == "chat gpt") + assert child.parent_span_id == span_uuid(TID, ROOT) + + +@pytest.mark.anyio +async def test_protobuf_gzip_ingestion_and_protobuf_response(client, db_session): + message = ExportTraceServiceRequest() + rs = message.resource_spans.add() + kv = rs.resource.attributes.add() + kv.key, kv.value.string_value = "service.name", "pb-agent" + span = rs.scope_spans.add().spans.add() + span.trace_id = bytes.fromhex(TID) + span.span_id = bytes.fromhex(ROOT) + span.name = "pb.root" + span.start_time_unix_nano = T0 + span.end_time_unix_nano = T0 + 5_000_000 + span.status.code = 1 + attr = span.attributes.add() + attr.key, attr.value.string_value = "gen_ai.operation.name", "chat" + res = await _post( + client, + gzip.compress(message.SerializeToString()), + ctype="application/x-protobuf", + headers={"content-encoding": "gzip"}, + ) + assert res.status_code == 200 and res.headers["content-type"].startswith("application/x-p") + assert ExportTraceServiceResponse.FromString(res.content).partial_success.rejected_spans == 0 + stored = await db_session.get(Trace, trace_uuid(TID)) + assert stored.project_id == "pb-agent" and stored.name == "pb.root" + + +@pytest.mark.anyio +async def test_spans_arriving_in_later_batches_link_and_keep_the_full_range(client, db_session): + late_child = _request([_span(CHILD, "child", parent=ROOT, start=3_000_000_000, ms=500)]) + root = _request([_span(ROOT, "root", start=0, ms=100)]) + assert (await _post(client, json.dumps(late_child))).status_code == 200 # parent not seen yet + assert (await _post(client, json.dumps(root))).status_code == 200 + trace = await db_session.get(Trace, trace_uuid(TID)) + assert trace.name # a name, from whichever batch came first + spans = ( + (await db_session.execute(select(Span).where(Span.trace_id == trace_uuid(TID)))) + .scalars() + .all() + ) + assert len(spans) == 2 + # Both batches belong to one trace spanning from the root's start to the child's end. + span = (trace.end_time - trace.start_time).total_seconds() + assert span == pytest.approx(3.5, abs=0.01) + third = _request([_span("eee19b7ec3c1b177", "grandchild", parent=CHILD, start=3_100_000_000)]) + await _post(client, json.dumps(third)) + gc = (await db_session.execute(select(Span).where(Span.name == "grandchild"))).scalar_one() + assert gc.parent_span_id == span_uuid(TID, CHILD) # linked to a parent from an earlier batch + + +@pytest.mark.anyio +async def test_project_header_and_http_errors(client, db_session): + body = json.dumps(_demo()) + ok = await _post(client, body, headers={"x-observer-project": "team-a"}) + assert ok.status_code == 200 + assert (await db_session.get(Trace, trace_uuid(TID))).project_id == "team-a" + assert (await _post(client, body, ctype="text/plain")).status_code == 415 + assert (await _post(client, "{nope")).status_code == 400 + assert (await _post(client, "[1]")).status_code == 400 + assert (await _post(client, b"\xff\xff", ctype="application/x-protobuf")).status_code == 400 + partial = await _post(client, json.dumps(_request([_span("zz", "bad")]))) + assert partial.status_code == 200 + assert partial.json()["partialSuccess"]["rejectedSpans"] == "1" + + +@pytest.mark.anyio +async def test_a_trace_owned_by_another_project_is_rejected_without_a_receipt(client, db_session): + body = json.dumps(_demo()) + assert (await _post(client, body, headers={"x-observer-project": "a"})).status_code == 200 + res = await _post( + client, body.replace("agent.run", "agent.run2"), headers={"x-observer-project": "b"} + ) + assert res.status_code == 200 + assert res.json()["partialSuccess"]["rejectedSpans"] == "3" diff --git a/docs/otlp.md b/docs/otlp.md new file mode 100644 index 0000000..1cdfa12 --- /dev/null +++ b/docs/otlp.md @@ -0,0 +1,66 @@ +# Standard OpenTelemetry (OTLP) ingestion + +Observer accepts traces from any stock OpenTelemetry exporter, so an app can be instrumented +without the Observer SDK. + +```bash +export OTEL_EXPORTER_OTLP_ENDPOINT=http://127.0.0.1:8100/v1/otlp # the exporter appends /v1/traces +export OTEL_EXPORTER_OTLP_HEADERS="Authorization=Bearer " # only when OBSERVATORY_API_KEY is set +``` + +The endpoint is `POST /v1/otlp/v1/traces`. It takes OTLP/HTTP with `Content-Type: +application/x-protobuf` (the usual default) or `application/json`, optionally gzip- or +deflate-compressed, and answers with an `ExportTraceServiceResponse` in the same format +(`partialSuccess.rejectedSpans` when some spans were refused). gRPC is not supported. + +## Where the data goes + +| OTLP | Observer | +|---|---| +| `resource` `service.name` | project (`project_id`); override with the `X-Observer-Project` header | +| `deployment.environment.name` / `deployment.environment` | environment | +| `service.instance.id` | service instance | +| span attribute `session.id` / `gen_ai.conversation.id` | session (default: the trace id) | +| trace id (16 bytes) | trace id, as a UUID | +| span id (8 bytes) | a stable UUID derived from the trace and span id (a re-sent span updates, never duplicates) | +| `parentSpanId` | parent span (linked also when the parent arrived in an earlier batch) | +| span `name` | span name; for tool spans the tool's name (`gen_ai.tool.name`) | +| `startTimeUnixNano`, `endTimeUnixNano` | start and end time | +| status `ERROR` / `OK` / unset | `error` / `ok` / `unset`; a trace is `error` if any span is | +| root span (no parent) | names the trace; the trace spans the earliest start to the latest end | +| `gen_ai.usage.input_tokens` (or `prompt_tokens`) / `output_tokens` (or `completion_tokens`) | `tokens_input` / `tokens_output` | +| `gen_ai.response.model` / `gen_ai.request.model` | attribute `model` | + +Span type comes from `openinference.span.kind` (`LLM`, `TOOL`, `AGENT`, `CHAIN`, `RETRIEVER`, +`EMBEDDING`, `RERANKER`) or from `gen_ai.operation.name`: `chat`, `text_completion`, +`generate_content` → `llm`; `execute_tool` → `tool`; `invoke_agent`, `create_agent` → `agent`; +`embeddings` → `embedding`. Anything else is `generic`. Tool spans therefore feed the same +test-case checks and `observer-regression` gate as Manitos' own tool spans. + +Events and links are ignored. + +## Privacy: metadata only by default + +Observer is metadata-only. `OBSERVATORY_OTLP_CAPTURE` chooses what is stored: + +* `metadata` (default): span attributes whose key looks like content are dropped before storage + (anything containing `prompt`, `completion`, `content`, `message`, `input`, `output`, `query`, + `statement`, `body`, `text`, `payload`, `argument`, `url`, `path`, `file`, `email`, `token`, + `key`, `secret`, `authorization`, `cookie`, and similar). Standard GenAI attributes that are not + content (`gen_ai.usage.*`, model, operation, tool and agent names, finish reasons) are kept. Strings + are cut at 200 characters. +* `full`: attributes are stored as sent (strings up to 4000 characters). Use it only for data you + are willing to keep in the Observer database. + +The filter looks at attribute **names**, not values, so a name that does not look like content can +still carry content. Treat `metadata` as a safety net, not a guarantee, and keep prompts out of +span attributes you do not want stored. Per-project capture modes are planned (roadmap step 7). + +## Limits and behaviour + +* Body limit 16 MB after decompression; `RATE_LIMIT_OTLP` (default `600/minute`) rate-limits it. +* Up to 500 spans per trace per request are processed in chunks; a trace owned by another project + is rejected and counted in `rejectedSpans`; a rejected trace leaves no partial data behind. +* A child that arrives before its parent is stored unlinked; it is linked only if its parent had + already been stored when the child arrived (an earlier parent is found, a later one is not + re-linked). diff --git a/docs/roadmap.md b/docs/roadmap.md index 83f482c..4f57f60 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -26,8 +26,8 @@ and write down what was learned or changed. Keep entries short. | 2 | Datasets from traces | done | this PR | | 3 | Compare runs by version | done | this PR | | 4 | Regression command for CI | done | this PR | -| 5 | Standard OTLP ingestion | planned | | -| 6 | Read-only MCP server | done | this PR | +| 5 | Standard OTLP ingestion | done | #56 | +| 6 | Read-only MCP server | done | #57 | | 7 | Configurable capture privacy | later | | | 8 | Ingestion reliability (persistent queue, backoff, metrics) | later | | @@ -94,7 +94,11 @@ threshold or a per-case expectation regresses. Today `OTLPExporter` posts Observer's own JSON to `/v1/traces/batch`; there is no OTLP receiver. -- Add an OTLP/HTTP receiver and a documented attribute mapping. +- **Built:** `POST /v1/otlp/v1/traces` (`app/api/otlp.py`, translation in `app/otlp.py`), protobuf + and JSON, gzip, partial success. It translates OTLP spans into the existing + `manitos.telemetry.v1` envelope and reuses its ingestion (idempotency, upsert), so nothing else + changes. Mapping, privacy modes and limits: `docs/otlp.md`. Metadata-only by default + (`OBSERVATORY_OTLP_CAPTURE`); gRPC is not supported. New dependency: `opentelemetry-proto`. - Done when: an app instrumented with a stock OpenTelemetry exporter shows up without the SDK. ## Step 6: read-only MCP server @@ -143,6 +147,10 @@ investigate without screenshots. Read-only first. export tool spans and reply text before the gate can verify them on real traces. - 2026-10-16: Manitos PR #36 merged: tool spans exported. Tool expectations are now verifiable on new traces; text phrases remain `unknown`. Observer services migrated and restarted. +- 2026-10-16: step 5 done. OTLP/HTTP receiver, verified with a stock `opentelemetry-sdk` + + `opentelemetry-exporter-otlp-proto-http` app (protobuf default): trace, spans, token counts and a + tool span arrive, prompt attributes are dropped. Learned: protobuf's JSON mapping writes ids as + base64, OTLP/JSON as hex, so ids are normalised to hex before translation. - 2026-10-16: step 6 done. `observer-mcp`: 11 read-only tools, no content by default. Verified over real stdio with an MCP client against a synthetic Observer (an OTLP-ingested trace with its tool span was listed and read). Steps 1-6 complete; what remains is under "Later".