tests/observability/test_signoz_smoke.py

"""OTLP/HTTP smoke test against a fake local receiver (Task 6, Step 2).

This exercises the real `full`-mode export path -- `app.observability.bootstrap`'s
actual `OTLPSpanExporter` / `OTLPMetricExporter` / `OTLPLogExporter`, wired the
same way `_build_tracer_provider` / `_build_meter_provider` / `_build_logger_provider`
already do for a live SigNoz endpoint -- pointed at a throwaway local HTTP
server instead of a live SigNoz instance. It does not hand-roll OTLP wire
handling on the *sending* side: every span/metric/log is produced through the
same `observe_operation` boundary product code uses (see `test_rag_spans.py`
for the equivalent local-mode span-tree assertions this mirrors). The fake
receiver only needs to decode what arrives, using the real
`opentelemetry-proto` generated message classes (the same wire format SigNoz's
own OTLP collector parses), so this test is a genuine protocol-level check,
not a mock of `full` mode's own behavior.

Root RAG span -> nested retrieval span -> nested vector-search span that is
marked as an error (mirroring the real `chroma.vector_search` failure path
covered by `test_rag_spans.py::test_missing_collection_retrieval_span_tree_...`).
The vector-search span's error status makes `observe_operation` emit one
correlated error log automatically (the same mechanism `_ERROR_STATUSES`
triggers in production). One duration metric (`operation_duration_ms`) is
recorded for every operation, so three `rag.*`/`chroma.*` spans also produce
three histogram records under one shared `service.name` resource.

Assertions: all three `/v1/*` paths receive at least one export; the traces
export contains the full span tree under one shared `trace_id`; the error
span's status is `STATUS_CODE_ERROR`; the metrics export carries the
canonical `operation_duration_ms` histogram under the same resource
(`service.name=research-buddy-backend`); the logs export contains a record
whose `trace_id`/`span_id` match the erroring span exactly -- proving the
"navigate from an error span to its correlated log" query (see
`ops/signoz/queries/rag-investigation.md`) has real data to find.
"""

from __future__ import annotations

import gzip
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

import pytest
from opentelemetry.proto.collector.logs.v1 import logs_service_pb2
from opentelemetry.proto.collector.metrics.v1 import metrics_service_pb2
from opentelemetry.proto.collector.trace.v1 import trace_service_pb2

from app.observability import bootstrap
from app.observability.config import ObservabilityConfig
from app.observability.operation import observe_operation


class _Capture:
    """Thread-safe store of raw request bodies received per OTLP path."""

    def __init__(self) -> None:
        self._lock = threading.Lock()
        self.bodies: dict[str, list[bytes]] = {"/v1/traces": [], "/v1/metrics": [], "/v1/logs": []}
        self.content_types: dict[str, list[str]] = {"/v1/traces": [], "/v1/metrics": [], "/v1/logs": []}

    def add(self, path: str, body: bytes, content_type: str) -> None:
        with self._lock:
            self.bodies.setdefault(path, []).append(body)
            self.content_types.setdefault(path, []).append(content_type)

    def latest(self, path: str) -> bytes:
        with self._lock:
            return self.bodies[path][-1]


def _make_handler(capture: _Capture) -> type[BaseHTTPRequestHandler]:
    class _Handler(BaseHTTPRequestHandler):
        def do_POST(self) -> None:  # noqa: N802 - stdlib method name
            length = int(self.headers.get("Content-Length", 0))
            body = self.rfile.read(length)
            if self.headers.get("Content-Encoding") == "gzip":
                body = gzip.decompress(body)
            capture.add(self.path, body, self.headers.get("Content-Type", ""))
            self.send_response(200)
            self.send_header("Content-Type", "application/x-protobuf")
            self.send_header("Content-Length", "0")
            self.end_headers()

        def log_message(self, format: str, *args: object) -> None:  # noqa: A002 - silence stdlib access log
            pass

    return _Handler


@pytest.fixture
def fake_otlp_receiver():
    """A throwaway local HTTP server standing in for SigNoz's OTLP collector."""
    capture = _Capture()
    server = ThreadingHTTPServer(("127.0.0.1", 0), _make_handler(capture))
    port = server.server_address[1]
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    try:
        yield capture, f"http://127.0.0.1:{port}"
    finally:
        server.shutdown()
        server.server_close()
        thread.join(timeout=2.0)


@pytest.fixture(autouse=True)
def _shutdown_observability_after():
    yield
    bootstrap.shutdown_observability(timeout_seconds=5.0)


def _full_config(endpoint: str, artifact_root) -> ObservabilityConfig:
    return ObservabilityConfig(
        enabled=True,
        mode="full",
        otlp_endpoint=endpoint,
        signoz_ui_url="http://127.0.0.1:8080",
        artifact_root=artifact_root,
    )


def _spans_from(body: bytes) -> tuple[list, str | None]:
    req = trace_service_pb2.ExportTraceServiceRequest()
    req.ParseFromString(body)
    spans: list = []
    service_name: str | None = None
    for resource_spans in req.resource_spans:
        for attr in resource_spans.resource.attributes:
            if attr.key == "service.name":
                service_name = attr.value.string_value
        for scope_spans in resource_spans.scope_spans:
            spans.extend(scope_spans.spans)
    return spans, service_name


def _metric_names_and_service(body: bytes) -> tuple[set[str], str | None]:
    req = metrics_service_pb2.ExportMetricsServiceRequest()
    req.ParseFromString(body)
    names: set[str] = set()
    service_name: str | None = None
    for resource_metrics in req.resource_metrics:
        for attr in resource_metrics.resource.attributes:
            if attr.key == "service.name":
                service_name = attr.value.string_value
        for scope_metrics in resource_metrics.scope_metrics:
            for metric in scope_metrics.metrics:
                names.add(metric.name)
    return names, service_name


def _log_records(body: bytes) -> list:
    req = logs_service_pb2.ExportLogsServiceRequest()
    req.ParseFromString(body)
    records: list = []
    for resource_logs in req.resource_logs:
        for scope_logs in resource_logs.scope_logs:
            records.extend(scope_logs.log_records)
    return records


# --- Brief Step 2 example -----------------------------------------------


def test_signoz_smoke_root_span_metric_and_correlated_error_log(fake_otlp_receiver, tmp_path):
    capture, endpoint = fake_otlp_receiver
    bootstrap.initialize_observability(config=_full_config(endpoint, tmp_path))

    with observe_operation("rag.request", subsystem="rag", consumer="chat") as root:
        root.set("pipeline.version", "rag-naive-v1")
        with observe_operation("rag.retrieval", subsystem="retrieval") as retrieval:
            retrieval.add_stage("vector_search_ms", 12.5)
            with observe_operation("chroma.vector_search", subsystem="retrieval") as vector_search:
                vector_search.mark_error("vector_search_failed")

    # Force every processor/reader to export immediately instead of waiting on
    # the batch schedule -- mirrors what a real process shutdown does.
    bootstrap.shutdown_observability(timeout_seconds=10.0)

    assert capture.bodies["/v1/traces"], "fake receiver never got a POST to /v1/traces"
    assert capture.bodies["/v1/metrics"], "fake receiver never got a POST to /v1/metrics"
    assert capture.bodies["/v1/logs"], "fake receiver never got a POST to /v1/logs"

    for path in ("/v1/traces", "/v1/metrics", "/v1/logs"):
        assert "application/x-protobuf" in capture.content_types[path][-1]

    # -- traces: full nested tree under one shared trace_id --
    spans, trace_service_name = _spans_from(capture.latest("/v1/traces"))
    names = {s.name for s in spans}
    assert {"rag.request", "rag.retrieval", "chroma.vector_search"} <= names
    assert trace_service_name == bootstrap.SERVICE_NAME

    root_span = next(s for s in spans if s.name == "rag.request")
    retrieval_span = next(s for s in spans if s.name == "rag.retrieval")
    vector_search_span = next(s for s in spans if s.name == "chroma.vector_search")
    assert retrieval_span.trace_id == root_span.trace_id
    assert vector_search_span.trace_id == root_span.trace_id
    assert vector_search_span.parent_span_id == retrieval_span.span_id
    assert retrieval_span.parent_span_id == root_span.span_id
    STATUS_CODE_ERROR = 2
    assert vector_search_span.status.code == STATUS_CODE_ERROR

    # -- metrics: the one canonical duration histogram, same resource identity --
    metric_names, metrics_service_name = _metric_names_and_service(capture.latest("/v1/metrics"))
    assert bootstrap.DURATION_HISTOGRAM_NAME in metric_names
    assert metrics_service_name == bootstrap.SERVICE_NAME

    # -- logs: one correlated error log tied to the exact erroring span --
    log_records = _log_records(capture.latest("/v1/logs"))
    assert log_records
    correlated = [r for r in log_records if r.trace_id == vector_search_span.trace_id]
    assert correlated, "no log record shares the vector-search span's trace_id"
    assert any(r.span_id == vector_search_span.span_id for r in correlated), (
        "no log record shares the vector-search span's exact span_id -- "
        "'navigate from an error span to its correlated log' would have nothing to find"
    )