tracer-cloud--opensre/platform/observability/otlp_parser.py

"""Shared parsing for OTLP/JSON trace payloads.

Used by any client that fetches a single trace in OpenTelemetry JSON form
(``{"batches": [{"resource": ..., "scopeSpans": [{"spans": [...]}]}]}``):
the standalone Tempo client and the Grafana Cloud Tempo mixin both consume it.
"""

from __future__ import annotations

from typing import Any

# OTLP/JSON scalar value kinds, in the order an attribute's single-key value dict
# may carry them (int64 arrives as a string per the OTLP spec — kept as-is).
_OTLP_SCALAR_KINDS = ("stringValue", "intValue", "boolValue", "doubleValue")

#: Nanoseconds → milliseconds for OTLP span duration.
_NANOSECONDS_PER_MILLISECOND = 1_000_000
#: Decimal places kept on parsed OTLP durations (ms).
_OTLP_DURATION_MS_DECIMAL_PLACES = 4
_EMPTY_DURATION_MS = 0.0
_UNKNOWN_SPAN_NAME = "unknown"


def extract_span_attributes(span: dict[str, Any]) -> dict[str, Any]:
    """Flatten an OTLP attribute list into a plain key -> value mapping.

    Handles the common OTLP/JSON scalar value kinds. Attributes without a key or
    with an unsupported value kind are skipped.
    """
    attributes: dict[str, Any] = {}
    for attr in span.get("attributes", []):
        key = attr.get("key", "")
        if not key:
            continue
        value = attr.get("value", {})
        for kind in _OTLP_SCALAR_KINDS:
            if kind in value:
                attributes[key] = value[kind]
                break
    return attributes


def _duration_ms(start_unix_nano: Any, end_unix_nano: Any) -> float:
    """Span duration in milliseconds from OTLP nanosecond timestamps."""
    try:
        start = int(start_unix_nano)
        end = int(end_unix_nano)
    except (TypeError, ValueError):
        return _EMPTY_DURATION_MS
    if end <= start:
        return _EMPTY_DURATION_MS
    return round(
        (end - start) / _NANOSECONDS_PER_MILLISECOND,
        _OTLP_DURATION_MS_DECIMAL_PLACES,
    )


def parse_otlp_trace(trace_data: dict[str, Any]) -> list[dict[str, Any]]:
    """Parse an OTLP/JSON trace into a flat list of span dicts.

    The ``service_name`` is lifted from each batch's resource attributes so
    callers can correlate spans to services without a nested lookup.
    """
    spans: list[dict[str, Any]] = []

    for batch in trace_data.get("batches", []):
        if not isinstance(batch, dict):
            continue
        resource_attributes = extract_span_attributes(batch.get("resource", {}))
        service_name = str(resource_attributes.get("service.name", ""))

        for scope in batch.get("scopeSpans", []):
            if not isinstance(scope, dict):
                continue
            for span in scope.get("spans", []):
                if not isinstance(span, dict):
                    continue
                status = span.get("status") or {}
                spans.append(
                    {
                        "name": span.get("name", _UNKNOWN_SPAN_NAME),
                        "span_id": span.get("spanId", ""),
                        "parent_span_id": span.get("parentSpanId", ""),
                        "trace_id": span.get("traceId", ""),
                        "kind": span.get("kind", ""),
                        "service_name": service_name,
                        "duration_ms": _duration_ms(
                            span.get("startTimeUnixNano"),
                            span.get("endTimeUnixNano"),
                        ),
                        "status_code": status.get("code", ""),
                        "status_message": status.get("message", ""),
                        "attributes": extract_span_attributes(span),
                    }
                )

    return spans