Implement pipeline observability with OTel, Prometheus, Grafana

Introduction

When your embedding pipeline latency doubles overnight, you cannot tell whether the cause is larger input documents, a degraded GPU node, Kafka consumer lag, or a DVC checkout bottleneck unless every stage emits traces and metrics. Teams that ship GenAI pipelines without observability burn hours of on-call time correlating logs by hand during every incident, and miss cost spikes until the monthly bill lands. By the end of this lesson you'll be able to instrument Dagster assets, Argo containers, and Kafka consumers with OpenTelemetry spans and metrics, expose them to Prometheus, and visualise pipeline health, per-stage latency, and run history in Grafana.

Key Terminology

• OpenTelemetry (OTel) span: a timed unit of work in a pipeline stage that carries attributes (stage name, record count, run id) and propagates trace context across Dagster, Argo, and Kafka boundaries.
• TracerProvider: the root OTel factory configured once per process at startup; it owns the span processors and exporters, so every span emitted by every stage in the run flows through the same export pipeline.
• BatchSpanProcessor: an OTel span processor that buffers completed spans in memory and ships them to the collector in batches, keeping the hot path fast and avoiding one network call per span when a pipeline emits thousands of stages per minute.
• OTel Collector: a Kubernetes-resident agent that receives OTLP traces and metrics from pipeline components, batches and enriches them with pod metadata, and fans them out to Prometheus and a trace backend.
• Prometheus histogram: the metric type used for per-stage duration (pipeline.stage.duration_seconds) — bucketed so Grafana can render p50/p95/p99 latency and Alertmanager can fire on SLA bucket breaches.

Concepts

Pipeline observability rests on three signals — traces, metrics, and a shared correlation id — emitted from every stage of the embedding pipeline. Traces give per-stage timing and parent-child causality across Dagster assets, Argo containers, and KEDA-scaled Kafka consumers, so a slowdown can be attributed to the specific stage and node that caused it. Metrics — a duration histogram, a records-processed counter, and an error counter, all labelled by stage — feed Prometheus alert rules for SLA breaches and Grafana panels for throughput and error-rate trends. The pipeline_run_id propagated as an OTel trace context attribute, an Argo workflow parameter, a Kafka header, and a DVC commit prefix is what stitches a single run together across all four systems, replacing manual log correlation with one-click drill-down from a failed run to the exact span that failed.

Code Walkthrough

Building on the three signals introduced in Concepts — spans, metrics, and the shared pipeline_run_id — the module below provides a reusable instrumentation wrapper that creates OTel spans for pipeline stages and records custom metrics. The PipelineInstrumentor class wraps any processing function with tracing and metrics collection, automatically recording duration, data volume, and error status for each execution, so a single decorator gives Dagster assets, Argo containers, and Kafka consumers consistent traces and metrics. The observability architecture spans three layers: at the application layer each pipeline component emits OTel traces and metrics; at the collection layer an OTel Collector batches, enriches, and exports them to Prometheus and a trace backend; at the visualization layer Grafana queries both to render dashboards combining time-series metrics with drill-down trace views.

1from opentelemetry import trace, metrics
2from opentelemetry.sdk.trace import TracerProvider
3from opentelemetry.sdk.metrics import MeterProvider
4from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
5from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter
6from opentelemetry.sdk.trace.export import BatchSpanProcessor
7from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
8import time
9import functools
10
11def init_telemetry(service_name: str, otlp_endpoint: str) -> None:
12    """Initialize OTel tracing and metrics with OTLP export."""
13    tracer_provider = TracerProvider()
14    tracer_provider.add_span_processor(
15        BatchSpanProcessor(OTLPSpanExporter(endpoint=otlp_endpoint))
16    )
17    trace.set_tracer_provider(tracer_provider)
18
19    reader = PeriodicExportingMetricReader(
20        OTLPMetricExporter(endpoint=otlp_endpoint),
21        export_interval_millis=15000,
22    )
23    meter_provider = MeterProvider(metric_readers=[reader])
24    metrics.set_meter_provider(meter_provider)
25
26class PipelineInstrumentor:
27    """Instrument pipeline stages with OTel traces and metrics."""
28
29    def __init__(self, service_name: str = "genai-pipeline"):
30        self.tracer = trace.get_tracer(service_name)
31        meter = metrics.get_meter(service_name)
32        self.duration_hist = meter.create_histogram(
33            "pipeline.stage.duration_seconds",
34            description="Processing duration per pipeline stage",
35        )
36        self.records_counter = meter.create_counter(
37            "pipeline.stage.records_processed",
38            description="Number of records processed per stage",
39        )
40        self.error_counter = meter.create_counter(
41            "pipeline.stage.errors_total",
42            description="Total errors per pipeline stage",
43        )
44
45    def instrument(self, stage_name: str):
46        """Decorator that wraps a function with OTel tracing and metrics."""
47        def decorator(func):
48            @functools.wraps(func)
49            def wrapper(*args, **kwargs):
50                with self.tracer.start_as_current_span(stage_name) as span:
51                    span.set_attribute("pipeline.stage", stage_name)
52                    start = time.monotonic()
53                    try:
54                        result = func(*args, **kwargs)
55                        record_count = len(result) if hasattr(result, "__len__") else 1
56                        span.set_attribute("pipeline.record_count", record_count)
57                        self.records_counter.add(record_count, {"stage": stage_name})
58                        return result
59                    except Exception as exc:
60                        span.set_status(trace.StatusCode.ERROR, str(exc))
61                        self.error_counter.add(1, {"stage": stage_name})
62                        raise
63                    finally:
64                        elapsed = time.monotonic() - start
65                        self.duration_hist.record(elapsed, {"stage": stage_name})
66            return wrapper
67        return decorator

With this instrumentation in place, every Dagster asset and Argo container stage emits traces and metrics that flow through the OTel Collector to Prometheus. The Prometheus metrics enable three critical alert rules for production pipeline operations: stage duration exceeding the SLA threshold (indicating performance degradation), error rate exceeding the acceptable percentage (indicating data quality or infrastructure problems), and records-processed rate dropping below the expected throughput (indicating upstream starvation or consumer scaling issues).

Grafana dashboards built on these metrics provide operational visibility at three levels. The overview dashboard shows pipeline run history with success/failure rates, total processing time, and cost per run. The stage-level dashboard breaks down each pipeline execution into its constituent stages, showing duration distributions, record counts, and error rates per stage. The infrastructure dashboard correlates pipeline metrics with Kubernetes resource utilization (CPU, memory, GPU from Kueue), Kafka consumer lag (from KEDA metrics), and DVC operation latency, enabling operators to identify infrastructure bottlenecks that affect pipeline performance.

The correlation ID is the thread that ties all of these signals together. Every pipeline run generates a unique pipeline_run_id that is propagated as an OTel trace context attribute through Dagster, injected as an Argo Workflow parameter, passed as a Kafka message header, and recorded as a DVC commit message prefix. When an operator sees a failed pipeline run in the Grafana overview, they click through to the trace view, which shows the complete execution timeline across all systems with a single pipeline_run_id. This end-to-end traceability eliminates the manual log correlation that traditionally consumes hours of on-call engineering time during incident investigation.

Do's and Don'ts

Building on the instrumentation pattern and dashboard wiring above, the following guidelines steer you away from the mistakes that most often degrade pipeline observability in production.

Do's

1. ✓Emit duration as a Prometheus histogram (not a gauge or summary) so percentile-based SLA alerts and Grafana p95/p99 panels work without re-aggregation.
2. ✓Propagate a single pipeline_run_id as an OTel trace context attribute, an Argo workflow parameter, a Kafka message header, and a DVC commit prefix so one query stitches a run together across all four systems.
3. ✓Run the OTel Collector as a DaemonSet and ship to it over OTLP/gRPC — that lets the collector enrich spans with Kubernetes pod metadata before exporting to Prometheus and Tempo/Jaeger.

Don'ts

1. ✗Don't push metrics directly from each pod to Prometheus — Prometheus scrapes, and a push-based shortcut bypasses the OTel Collector's batching, enrichment, and resource attribution.
2. ✗Don't record stage duration with time.time() — wall-clock jumps from NTP corrections produce negative or inflated samples that poison the histogram; use time.monotonic() as the snippet above does.
3. ✗Don't omit the stage label on the duration histogram and error counter — without it Grafana cannot break latency down per stage and you lose the ability to tell whether the regression is in embed, chunk, or index.