feat(observability): dump HTTP query, request body, and response body to trace spans (#2052)

Add an opt-in middleware that attaches the request and response bodies
(truncated, content-type filtered) onto the active OpenTelemetry root span,
and surface the URL query string as `url.query`. Off by default — bodies
may contain secrets and high-cardinality content; enable via
`server.observability.dump_body.enabled` and bound payload size with
`max_bytes`.

The dump middleware is registered before the HTTP observability middleware
so it nests inside the trace span (Starlette executes later-registered
middleware first). Streaming, multipart, and binary content types are
skipped, and any capture failure is swallowed so the request path is never
affected.

Co-authored-by: chenpengfei <chenpengfei@bytedance.com>

Author: KCHENPENGFEI

openviking/observability/http_observability_middleware.py

@@ -826,6 +826,7 @@ async def middleware(request: Request, call_next: Callable) -> Response:
 
         # Extract request information
         raw_path = str(request.url.path)
+        raw_query = request.url.query or None
         route_template = _get_route_template(request)
         request_id = request.headers.get("x-request-id") or str(uuid.uuid4())
 
@@ -835,6 +836,7 @@ async def middleware(request: Request, call_next: Callable) -> Response:
             http_route=route_template,
             request_id=request_id,
             url_path=raw_path,
+            url_query=raw_query,
             url_scheme=request.url.scheme,
             http_host=request.url.netloc,
             source_type=request.headers.get("x-source-type"),

openviking/server/app.py

@@ -320,7 +320,30 @@ async def _oauth_gc_loop(store) -> None:  # noqa: ANN001
         allow_headers=["*"],
     )
 
-    # Add HTTP observability middleware first (metrics, tracing)
+    # Body dump middleware must be registered BEFORE observability so it ends up
+    # nested inside the trace span (in Starlette, middleware added later wraps
+    # earlier-added ones — so earlier registration = inner layer).
+    if config.observability.dump_body.enabled:
+        from openviking.server.body_dump_middleware import (
+            create_dump_http_body_middleware,
+        )
+
+        _dump_body_fn = create_dump_http_body_middleware(
+            max_bytes=config.observability.dump_body.max_bytes,
+        )
+
+        @app.middleware("http")
+        async def dump_http_body(request: Request, call_next: Callable):
+            return await _dump_body_fn(request, call_next)
+
+        logger.info(
+            "HTTP body dump middleware enabled (max_bytes=%d) — bodies will be "
+            "attached to trace spans. Disable in production via "
+            "server.observability.dump_body.enabled=false.",
+            config.observability.dump_body.max_bytes,
+        )
+
+    # Add HTTP observability middleware (metrics, tracing).
     # Note: In FastAPI/Starlette, middleware added later executes first (outer layer).
     # We want timing to be the outermost layer to measure the full request duration.
     from openviking.observability.http_observability_middleware import (

openviking/server/body_dump_middleware.py

@@ -0,0 +1,128 @@
+# Copyright (c) 2026 Beijing Volcano Engine Technology Co., Ltd.
+# SPDX-License-Identifier: AGPL-3.0
+"""HTTP request/response body dump middleware for trace debugging.
+
+Attaches the request and response bodies as attributes on the active OpenTelemetry
+root span so they can be inspected in trace UIs (Jaeger, Tempo, etc.). The middleware
+must run inside the trace span context — register it before the http_observability
+middleware in ``create_app``.
+"""
+
+from __future__ import annotations
+
+from typing import Awaitable, Callable
+
+from fastapi import Request
+from starlette.responses import Response
+
+try:
+    from opentelemetry import trace as otel_trace
+except ImportError:  # pragma: no cover - OTel optional
+    otel_trace = None
+
+# Skip body capture for content types that are binary, streamed, or otherwise
+# pointless to materialize as a span attribute.
+_SKIP_CONTENT_TYPE_PREFIXES = (
+    "multipart/form-data",
+    "application/octet-stream",
+    "text/event-stream",
+    "audio/",
+    "video/",
+    "image/",
+)
+
+
+def _should_skip(content_type: str | None) -> bool:
+    if not content_type:
+        return False
+    ct = content_type.lower()
+    return any(ct.startswith(p) for p in _SKIP_CONTENT_TYPE_PREFIXES)
+
+
+def _truncate(data: bytes, max_bytes: int) -> str:
+    total = len(data)
+    head = data[:max_bytes]
+    text = head.decode("utf-8", errors="replace")
+    if total > max_bytes:
+        return f"{text}…[+{total - max_bytes}B truncated, total {total}B]"
+    return text
+
+
+def _set_span_attr(key: str, value: object) -> None:
+    if otel_trace is None:
+        return
+    try:
+        span = otel_trace.get_current_span()
+        if span is None or not span.is_recording():
+            return
+        span.set_attribute(key, value)
+    except Exception:
+        # Body dump must never break the request path.
+        pass
+
+
+def create_dump_http_body_middleware(
+    max_bytes: int = 4096,
+) -> Callable[[Request, Callable], Awaitable[Response]]:
+    """Build a body-dump middleware bound to ``max_bytes``.
+
+    The middleware skips streaming/binary content types and truncates payloads to
+    keep span attributes bounded.
+    """
+
+    async def middleware(
+        request: Request,
+        call_next: Callable[[Request], Awaitable[Response]],
+    ) -> Response:
+        req_ct = request.headers.get("content-type", "")
+        if not _should_skip(req_ct):
+            try:
+                body = await request.body()
+                if body:
+                    _set_span_attr("http.request.body", _truncate(body, max_bytes))
+                    _set_span_attr("http.request.body.size", len(body))
+                if req_ct:
+                    _set_span_attr("http.request.content_type", req_ct)
+            except Exception:
+                pass
+
+        response = await call_next(request)
+
+        resp_ct = response.headers.get("content-type", "")
+        if _should_skip(resp_ct):
+            return response
+
+        # Once we start iterating ``response.body_iterator`` we own the bytes;
+        # capture failures must not silently truncate the response sent to the
+        # client, so we always rebuild a Response from whatever we've collected.
+        chunks: list[bytes] = []
+        try:
+            async for chunk in response.body_iterator:
+                chunks.append(chunk)
+            body_bytes = b"".join(chunks)
+            if body_bytes:
+                _set_span_attr("http.response.body", _truncate(body_bytes, max_bytes))
+                _set_span_attr("http.response.body.size", len(body_bytes))
+            if resp_ct:
+                _set_span_attr("http.response.content_type", resp_ct)
+        except Exception:
+            body_bytes = b"".join(chunks)
+            _set_span_attr("http.response.body.capture_error", True)
+
+        try:
+            new_headers = {
+                k: v for k, v in response.headers.items() if k.lower() != "content-length"
+            }
+            return Response(
+                content=body_bytes,
+                status_code=response.status_code,
+                headers=new_headers,
+                media_type=response.media_type,
+            )
+        except Exception:
+            # Fall back to the original response object as a last resort. Its
+            # body_iterator is exhausted at this point, so this only fires if
+            # the rebuild path itself is broken.
+            return response
+
+    return middleware

openviking/server/config.py

@@ -104,13 +104,28 @@ class UsageAuditConfig(BaseModel):
     model_config = {"extra": "forbid"}
 
 
+class TraceDumpBodyConfig(BaseModel):
+    """HTTP body dump configuration.
+
+    Attaches request/response bodies as attributes on the active trace span so
+    they can be inspected in trace UIs. Off by default — bodies may contain
+    secrets and high-cardinality content.
+    """
+
+    enabled: bool = False
+    max_bytes: int = 4096
+
+    model_config = {"extra": "forbid"}
+
+
 class ObservabilityConfig(BaseModel):
     """Server-side observability configuration."""
 
     metrics: MetricsConfig = Field(default_factory=MetricsConfig)
     usage_audit: UsageAuditConfig = Field(default_factory=UsageAuditConfig)
     traces: OTelExporterConfig = Field(default_factory=OTelExporterConfig)
     logs: OTelExporterConfig = Field(default_factory=OTelExporterConfig)
+    dump_body: TraceDumpBodyConfig = Field(default_factory=TraceDumpBodyConfig)
 
     model_config = {"extra": "forbid"}
 

openviking/telemetry/span_models.py

@@ -44,6 +44,9 @@ class RootSpanAttributes:
     url_path: Optional[str] = None
     """Raw request path for debugging. This may be high-cardinality."""
 
+    url_query: Optional[str] = None
+    """Raw query string (without leading '?'). May contain secrets — handle accordingly."""
+
     url_scheme: Optional[str] = None
     """URL scheme, such as `http` or `https`."""
 
@@ -78,6 +81,8 @@ def to_otel_attributes(self) -> Dict[str, Any]:
             attrs["http.status_code"] = self.http_status_code
         if self.url_path is not None:
             attrs["url.path"] = self.url_path
+        if self.url_query is not None and self.url_query != "":
+            attrs["url.query"] = self.url_query
         if self.url_scheme is not None:
             attrs["url.scheme"] = self.url_scheme
         if self.http_host is not None:
@@ -290,6 +295,7 @@ def create_root_span_attributes(
     http_route: str,
     request_id: str,
     url_path: Optional[str] = None,
+    url_query: Optional[str] = None,
     url_scheme: Optional[str] = None,
     http_host: Optional[str] = None,
     source_type: Optional[str] = None,
@@ -301,6 +307,7 @@ def create_root_span_attributes(
         http_route=http_route,
         request_id=request_id,
         url_path=url_path,
+        url_query=url_query,
         url_scheme=url_scheme,
         http_host=http_host,
         source_type=source_type,