Distributed Tracing¶

# From backend/core/telemetry.py:135-180
def setup_telemetry(
    service_name: str = "nemotron-backend",
    *,
    enable_auto_instrumentation: bool = True,
    otel_endpoint: str | None = None,
    sample_rate: float = 1.0,
) -> None:
    """Initialize OpenTelemetry with tracing and optional metrics."""
    if not settings.otel_enabled:
        logger.info("OpenTelemetry disabled by configuration")
        return

    # Create resource with service info
    resource = Resource.create({
        ResourceAttributes.SERVICE_NAME: service_name,
        ResourceAttributes.SERVICE_VERSION: get_app_version(),
        ResourceAttributes.DEPLOYMENT_ENVIRONMENT: settings.environment,
    })

    # Configure trace provider
    provider = TracerProvider(resource=resource)
    trace.set_tracer_provider(provider)

    # Configure exporter
    endpoint = otel_endpoint or settings.otel_exporter_otlp_endpoint
    exporter = OTLPSpanExporter(endpoint=endpoint)

    # Configure batch processor
    processor = BatchSpanProcessor(
        exporter,
        max_queue_size=2048,
        max_export_batch_size=512,
        schedule_delay_millis=5000,
    )
    provider.add_span_processor(processor)

./scripts/init-elasticsearch.sh

# From backend/core/telemetry.py:380-420
@contextmanager
def create_span(
    name: str,
    *,
    kind: SpanKind = SpanKind.INTERNAL,
    attributes: dict[str, Any] | None = None,
    record_exception: bool = True,
) -> Generator[Span, None, None]:
    """Create a traced span for custom operations."""
    tracer = trace.get_tracer(__name__)
    with tracer.start_as_current_span(
        name,
        kind=kind,
        attributes=attributes or {},
    ) as span:
        try:
            yield span
        except Exception as e:
            if record_exception:
                span.record_exception(e)
                span.set_status(Status(StatusCode.ERROR, str(e)))
            raise

from backend.core.telemetry import create_span
from opentelemetry.trace import SpanKind

async def analyze_image(image_path: str) -> dict:
    with create_span(
        "analyze_image",
        kind=SpanKind.CLIENT,
        attributes={"image.path": image_path},
    ) as span:
        result = await run_analysis(image_path)
        span.set_attribute("detection.count", len(result.detections))
        return result

# From backend/core/telemetry.py:453-490
@asynccontextmanager
async def create_async_span(
    name: str,
    *,
    kind: SpanKind = SpanKind.INTERNAL,
    attributes: dict[str, Any] | None = None,
) -> AsyncGenerator[Span, None]:
    """Create a traced span for async operations."""
    tracer = trace.get_tracer(__name__)
    with tracer.start_as_current_span(name, kind=kind, attributes=attributes or {}) as span:
        try:
            yield span
        except Exception as e:
            span.record_exception(e)
            span.set_status(Status(StatusCode.ERROR, str(e)))
            raise

traceparent: 00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01
tracestate: vendor1=value1,vendor2=value2

baggage: camera.id=front_door,event.priority=high,request.source=api

from backend.api.middleware.baggage import set_pipeline_baggage

# At the start of the detection pipeline
set_pipeline_baggage(
    camera_id="front_door",
    event_priority="high",
    request_source="api"
)

from backend.api.middleware.baggage import (
    get_camera_id_from_baggage,
    get_event_priority_from_baggage,
    get_request_source_from_baggage,
)

# In downstream services
camera_id = get_camera_id_from_baggage()
if get_event_priority_from_baggage() == "high":
    # Fast-track processing
    pass

# Context is automatically injected by httpx instrumentation
async def call_yolo26(image_data: bytes) -> dict:
    async with httpx.AsyncClient() as client:
        # traceparent header is automatically added
        response = await client.post(
            f"{yolo26_url}/detect",
            files={"image": image_data},
        )
        return response.json()

# From backend/core/telemetry.py:505-530
def inject_context_to_dict(carrier: dict[str, str]) -> None:
    """Inject current trace context into a dictionary."""
    propagator = get_global_textmap()
    propagator.inject(carrier)

def extract_context_from_dict(carrier: dict[str, str]) -> Context:
    """Extract trace context from a dictionary."""
    propagator = get_global_textmap()
    return propagator.extract(carrier)

# Producer
message = {"data": payload}
inject_context_to_dict(message)  # Adds traceparent, tracestate
await queue.push(message)

# Consumer
message = await queue.pop()
ctx = extract_context_from_dict(message)
with trace.use_span(trace.get_current_span(), end_on_exit=False):
    context.attach(ctx)
    # Process message with trace context restored

tracesToMetrics:
  datasourceUid: PBFA97CFB590B2093
  spanStartTimeShift: '-5m'
  spanEndTimeShift: '5m'
  tags:
    - key: 'service.name'
      value: 'service'
    - key: 'http.method'
      value: 'method'
    - key: 'http.status_code'
      value: 'status_code'
  queries:
    - name: 'Pipeline Errors/min'
      query: 'rate(hsi_pipeline_errors_total[1m]) * 60'
    - name: 'Detection Queue Depth'
      query: 'hsi_detection_queue_depth'
    - name: 'YOLO26 Latency (p95)'
      query: 'histogram_quantile(0.95, rate(yolo26_inference_latency_seconds_bucket[5m]))'

derivedFields:
  - name: TraceID
    matcherRegex: 'trace_id=([a-f0-9]{32})'
    url: '${__value.raw}'
    datasourceUid: PC9A941E8F2E49454
    urlDisplayLabel: 'View Trace'

2024-01-15 10:30:45 | INFO | backend.services | trace_id=0af7651916cd43dd8448eb211c80319c span_id=b7ad6b7169203331 | Processing detection

{
  "targets": [
    {
      "queryType": "search",
      "service": "nemotron-backend",
      "operation": "analysis_processing",
      "limit": 15
    }
  ]
}

{
  "targets": [
    {
      "queryType": "search",
      "service": "nemotron-backend",
      "tags": "error=true",
      "limit": 20
    }
  ]
}

# From backend/core/telemetry.py:185-200
if sample_rate < 1.0:
    sampler = TraceIdRatioBased(sample_rate)
else:
    sampler = AlwaysOnSampler()

provider = TracerProvider(
    resource=resource,
    sampler=sampler,
)

uv run pytest backend/tests/unit/core/test_telemetry.py -v