feat(observability): add OpenTelemetry + Jaeger distributed tracing (Rec #8)

pmoves/config/grafana/provisioning/datasources/jaeger.yaml

@@ -0,0 +1,32 @@
+# =============================================================================
+# PMOVES.AI - Grafana Jaeger Datasource
+# =============================================================================
+# Purpose: Provision Jaeger as the default tracing datasource in Grafana
+# Recommendation: PMOVES Rec #8 - Distributed Tracing
+#
+# Enables "Explore > Traces" in Grafana with direct Jaeger integration.
+# The Jaeger query service must be reachable at http://jaeger:16686.
+# =============================================================================
+
+apiVersion: 1
+
+datasources:
+  - name: Jaeger
+    type: jaeger
+    access: proxy
+    url: http://jaeger:16686
+    isDefault: false
+    editable: true
+    jsonData:
+      tracesToMetrics:
+        datasourceUid: prometheus
+        tags:
+          - key: service.name
+            value: service
+      nodeGraph:
+        enabled: true
+      traceQuery:
+        timeShiftEnabled: true
+        spanBar:
+          type: TagName
+          tagName: http.method

pmoves/config/otel/otel-collector-config.yaml

@@ -0,0 +1,59 @@
+# =============================================================================
+# PMOVES.AI - OpenTelemetry Collector Configuration
+# =============================================================================
+# Purpose: Central trace collection and export to Jaeger
+# Recommendation: PMOVES Rec #8 - Distributed Tracing
+#
+# Receivers:
+#   - OTLP gRPC (:4317) and HTTP (:4318) for service instrumentation
+# Processors:
+#   - batch: groups spans into batches for efficient export
+#   - memory_limiter: prevents OOM under load
+# Exporters:
+#   - jaeger: sends traces to Jaeger all-in-one
+#   - prometheus: exposes trace metrics for Grafana
+#   - logging: debug exporter for development
+# =============================================================================
+
+receivers:
+  otlp:
+    protocols:
+      grpc:
+        endpoint: 0.0.0.0:4317
+      http:
+        endpoint: 0.0.0.0:4318
+
+processors:
+  batch:
+    send_batch_size: 1024
+    timeout: 5s
+    send_batch_max_size: 2048
+  memory_limiter:
+    check_interval: 5s
+    limit_mib: 400
+    spike_limit_mib: 100
+
+exporters:
+  jaeger:
+    endpoint: jaeger:14250
+    tls:
+      insecure: true
+  prometheus:
+    endpoint: 0.0.0.0:8889
+    namespace: pmoves_otel
+  logging:
+    loglevel: warn
+
+service:
+  pipelines:
+    traces:
+      receivers: [otlp]
+      processors: [memory_limiter, batch]
+      exporters: [jaeger, logging]
+    metrics:
+      receivers: [otlp]
+      processors: [memory_limiter, batch]
+      exporters: [prometheus, logging]
+  telemetry:
+    logs:
+      level: info

pmoves/docker-compose.tracing.yml

@@ -0,0 +1,97 @@
+# =============================================================================
+# PMOVES.AI - Distributed Tracing Docker Compose Overlay
+# =============================================================================
+# Purpose: Jaeger + OpenTelemetry Collector for distributed tracing
+# Recommendation: PMOVES Rec #8 - Distributed Tracing
+#
+# Usage:
+#   docker compose -f docker-compose.yml -f docker-compose.tracing.yml up -d
+#
+# Environment:
+#   JAEGER_UI_PORT       - Jaeger UI port (default: 16686)
+#   OTEL_GRPC_PORT       - OTLP gRPC receiver port (default: 4327)
+#   OTEL_HTTP_PORT       - OTLP HTTP receiver port (default: 4318)
+#   OTEL_METRICS_PORT    - OTEL Collector Prometheus metrics port (default: 8889)
+#
+# Services:
+#   jaeger         - Jaeger all-in-one (UI + collector + agent)
+#   otel-collector - OpenTelemetry Collector (receives, processes, exports)
+# =============================================================================
+
+services:
+  # ---------------------------------------------------------------------------
+  # Jaeger All-in-One - Distributed tracing UI and storage
+  # ---------------------------------------------------------------------------
+  jaeger:
+    image: jaegertracing/all-in-one:1.62
+    container_name: pmoves-jaeger
+    restart: unless-stopped
+    environment:
+      - COLLECTOR_OTLP_ENABLED=true
+      - LOG_LEVEL=info
+      - MEMORY_MAX_TRACES=100000
+    ports:
+      - "${JAEGER_BIND:-127.0.0.1}:${JAEGER_UI_PORT:-16686}:16686"
+      - "14268:14268"
+      - "14250:14250"
+      - "4317:4317"
+    networks:
+      - pmoves_monitoring
+    healthcheck:
+      test: ["CMD", "wget", "--spider", "-q", "http://localhost:16686"]
+      interval: 30s
+      timeout: 10s
+      retries: 3
+      start_period: 15s
+    deploy:
+      resources:
+        limits:
+          memory: 512M
+          cpus: "1.0"
+        reservations:
+          memory: 128M
+          cpus: "0.1"
+    security_opt:
+      - no-new-privileges:true
+    read_only: true
+    tmpfs:
+      - /tmp:size=100M,mode=1777
+
+  # ---------------------------------------------------------------------------
+  # OpenTelemetry Collector - Central trace receiver and processor
+  # ---------------------------------------------------------------------------
+  otel-collector:
+    image: otel/opentelemetry-collector-contrib:0.112.0
+    container_name: pmoves-otel-collector
+    restart: unless-stopped
+    command: ["--config=/etc/otelcol/config.yaml"]
+    volumes:
+      - ./config/otel/otel-collector-config.yaml:/etc/otelcol/config.yaml:ro
+    ports:
+      - "${OTEL_GRPC_PORT:-4327}:4317"
+      - "${OTEL_HTTP_PORT:-4318}:4318"
+      - "${OTEL_METRICS_PORT:-8889}:8889"
+    networks:
+      - pmoves_monitoring
+    depends_on:
+      jaeger:
+        condition: service_healthy
+    healthcheck:
+      test: ["CMD", "wget", "--spider", "-q", "http://localhost:13133/"]
+      interval: 30s
+      timeout: 10s
+      retries: 3
+      start_period: 10s
+    deploy:
+      resources:
+        limits:
+          memory: 512M
+          cpus: "1.0"
+        reservations:
+          memory: 128M
+          cpus: "0.1"
+    security_opt:
+      - no-new-privileges:true
+    read_only: true
+    tmpfs:
+      - /tmp:size=100M,mode=1777

pmoves/services/common/nats_client.py

@@ -5,6 +5,9 @@
 
 Consolidates the NATS boilerplate repeated across 30+ services.
 
+Includes optional OpenTelemetry trace context propagation for
+distributed tracing across NATS message boundaries.
+
 Usage::
 
     from services.common.nats_client import create_nats_connection
@@ -21,7 +24,7 @@
 
 from __future__ import annotations
 
-import asyncio
+import json
 import logging
 import os
 from contextlib import asynccontextmanager
@@ -143,6 +146,110 @@ async def nats_connection(
                 pass
 
 
+
+
+# -----------------------------------------------------------------------
+# Trace-propagating publish / subscribe helpers
+# -----------------------------------------------------------------------
+
+# Graceful import — tracing is opt-in
+try:
+    from opentelemetry.trace import SpanKind
+    from services.common.tracing import (
+        inject_trace_headers,
+        extract_trace_headers,
+        trace_nats_message,
+        get_tracer,
+    )
+    _trace_available = True
+except (ImportError, Exception):
+    _trace_available = False
+
+
+async def traced_publish(
+    nc: Any,
+    subject: str,
+    payload: bytes,
+    headers: dict[str, str] | None = None,
+) -> None:
+    """Publish a NATS message with trace context propagation.
+
+    Injects the current W3C ``traceparent`` header into the NATS message
+    headers so downstream consumers can link their spans to the producer.
+
+    Falls back to a plain ``nc.publish()`` when tracing is not available.
+
+    Args:
+        nc: A connected NATS client.
+        subject: NATS subject to publish to.
+        payload: Message body as bytes.
+        headers: Optional existing headers (trace context will be merged in).
+    """
+    hdrs = dict(headers) if headers else {}
+
+    if _trace_available:
+        try:
+            hdrs = inject_trace_headers(hdrs)
+        except Exception as exc:
+            logger.debug("Failed to inject trace headers: %s", exc)
+
+    await nc.publish(subject, payload, headers=hdrs or None)
+
+
+def make_traced_callback(
+    service_name: str,
+    subject: str,
+    handler: Callable,
+) -> Callable:
+    """Wrap a NATS subscription callback with trace context extraction.
+
+    When a message arrives, the wrapper extracts ``traceparent`` from the
+    message headers, creates a child span, and calls *handler* inside that
+    span context.  The handler receives the original NATS message.
+
+    Falls back to calling *handler* directly when tracing is not available.
+
+    Usage::
+
+        async def on_message(msg):
+            ...
+
+        await nc.subscribe(
+            "compute.nodes.heartbeat",
+            cb=make_traced_callback("my-service", "compute.nodes.heartbeat", on_message),
+        )
+    """
+    if not _trace_available:
+        return handler
+
+    async def _wrapped(msg: Any) -> None:
+        # NATS headers are a dict-like; coerce to plain dict for propagator.
+        msg_headers: dict[str, str] = {}
+        if hasattr(msg, "headers") and msg.headers:
+            msg_headers = dict(msg.headers)
+
+        ctx = extract_trace_headers(msg_headers)
+        tracer = get_tracer(service_name)
+
+        with tracer.start_as_current_span(
+            f"nats.receive {subject}",
+            context=ctx,
+            kind=SpanKind.CONSUMER if _trace_available else None,
+            attributes={
+                "messaging.system": "nats",
+                "messaging.destination": subject,
+                "messaging.operation": "receive",
+            },
+        ) as span:
+            if span is not None:
+                if ctx is not None:
+                    span.set_attribute("messaging.trace_propagated", True)
+                if hasattr(msg, "data") and msg.data is not None:
+                    span.set_attribute("message.size", len(msg.data))
+            await handler(msg)
+
+    return _wrapped
+
 # -----------------------------------------------------------------------
 # Callback helpers
 # -----------------------------------------------------------------------
@@ -184,5 +291,7 @@ async def _reconnected_cb() -> None:
     "make_logging_error_cb",
     "make_logging_disconnected_cb",
     "make_logging_reconnected_cb",
+    "traced_publish",
+    "make_traced_callback",
     "DEFAULT_NATS_URL",
 ]