Instrumenting MRV Pipelines with OpenTelemetry and structlog #

Instrumentation for a carbon pipeline has an unusual requirement: the same values must serve an operator staring at a dashboard tonight and a verifier reconstructing a figure in 2036. Those two consumers want different stores, different retention, and different query patterns — but if they are fed by different code paths they will eventually disagree, and an operator dashboard that contradicts the audit record is worse than either alone. This guide wires both from a single emission point, within MRV pipeline observability and failure modes in the pipeline orchestration and compliance reference stack.

The design has three commitments. Events carry a stable schema with typed fields, so they can be aggregated across runs and years. Every event is emitted once, into a structured log and an OpenTelemetry span simultaneously, from the same dictionary. And the fields are partitioned by retention horizon at emission time, so the operational chatter can expire in thirty days while the provenance record survives the crediting period.

Root Cause Analysis #

Instrumentation projects fail in three characteristic ways, and each is avoidable by a decision made early.

Prose logs that cannot be aggregated. A line reading Finished tile N00E010 in 42s is unusable at scale: it cannot be grouped, compared to last month, or joined to a lineage record. The fix is a typed event schema — the same keys, with the same types, in every emission — treated with the same care as a database schema, including a version field so a consumer can handle a change. Schema drift in logs is invisible until the day someone tries to compare across it.

Spans that fragment under fan-out. A carbon run fans out over thousands of tiles across a cluster, and the trace context does not propagate automatically across process boundaries in most distributed frameworks. Without explicit context propagation, each worker starts a fresh trace, and the run becomes ten thousand unrelated traces rather than one. The result is that per-tile latency is queryable but “which tiles did this run process” is not — precisely the question that matters. The fix is to serialise the trace context into the task payload and re-attach it in the worker.

Two code paths for the same value. The most damaging failure is subtle: a dashboard computes feature counts from one source and the audit record from another, and they drift. This usually starts innocently, when the evidence record is added later by a different person against the same underlying data. The cure is architectural — one function computes the signals, one emitter distributes them, and neither the dashboard nor the evidence store may compute anything itself.

Diagnostic Pipeline / Pre-Flight Validation #

Before instrumenting, validate that the emission contract holds: the schema is complete and typed, the trace context propagates across the fan-out boundary, and the retention routing sends each field where it belongs.

from dataclasses import dataclass, fields as dataclass_fields

import structlog

log = structlog.get_logger()

SCHEMA_VERSION = "mrv-signals/2"

# Which retention horizon each field belongs to. Routing is a property of the
# SCHEMA, not of the call site — otherwise a new field silently lands in the
# 30-day store and is gone when a verifier asks for it.
HORIZONS = {
    "ops": {"status", "duration_s", "retries", "queue_depth", "worker"},
    "trend": {"feature_count", "null_rate", "value_p50", "value_p99", "bbox"},
    "evidence": {"run_id", "stage", "crs", "total_area_ha", "input_digest",
                 "code_version", "factor_set_version", "parameters",
                 "invalid_geometry_count", "schema_version"},
}


def validate_schema(signals: dict) -> list[str]:
    """Every field must be routed and typed. An unrouted field is a field that
    will be discovered missing years later, which is the expensive way."""
    routed = set().union(*HORIZONS.values())
    problems = []

    unrouted = set(signals) - routed
    if unrouted:
        problems.append(f"unrouted fields: {sorted(unrouted)}")

    missing = HORIZONS["evidence"] - set(signals)
    if missing:
        problems.append(f"missing evidence fields: {sorted(missing)}")

    for key, value in signals.items():
        if not isinstance(value, (str, int, float, bool, tuple, list, dict, type(None))):
            problems.append(f"field {key} has non-serialisable type {type(value).__name__}")

    if problems:
        log.error("obs.schema.invalid", problems=problems, schema=SCHEMA_VERSION)
    return problems


def check_context_propagation(worker_traces: list[str], expected_trace: str) -> dict:
    """A run that fanned out to N workers must produce ONE trace, not N.

    Run this once against a small fan-out in staging; it is the cheapest way to
    discover that context is not crossing the process boundary.
    """
    distinct = set(worker_traces)
    propagated = distinct == {expected_trace}

    if not propagated:
        log.error("obs.trace.fragmented", expected=expected_trace,
                  distinct_traces=len(distinct), workers=len(worker_traces),
                  hint="serialise the trace context into the task payload and re-attach it")
    return {"propagated": propagated, "distinct_traces": len(distinct),
            "workers": len(worker_traces)}

Deterministic Transformation Logic #

The emitter below is the single point through which every stage reports. It stamps the schema version, splits fields by horizon, writes one structured log line, sets span attributes, and returns the evidence subset for the caller to persist alongside the artefact.

import json
from contextlib import contextmanager

import structlog
from opentelemetry import trace, context as otel_context
from opentelemetry.propagate import inject, extract

log = structlog.get_logger()
tracer = trace.get_tracer("mrv.pipeline")


def configure_logging() -> None:
    """Structured, typed, machine-readable. The renderer is JSON in production
    and a console renderer locally, but the EVENT is identical in both."""
    structlog.configure(
        processors=[
            structlog.contextvars.merge_contextvars,   # run_id set once, on every event
            structlog.processors.add_log_level,
            structlog.processors.TimeStamper(fmt="iso", utc=True),
            structlog.processors.EventRenamer("event"),
            structlog.processors.JSONRenderer(sort_keys=True),
        ],
        wrapper_class=structlog.make_filtering_bound_logger(20),
        cache_logger_on_first_use=True,
    )


@contextmanager
def run_context(run_id: str, period: str):
    """Bind run-scoped keys once so every event inside carries them without the
    call sites having to remember."""
    structlog.contextvars.bind_contextvars(run_id=run_id, period=period,
                                           schema_version=SCHEMA_VERSION)
    try:
        with tracer.start_as_current_span("mrv.run") as span:
            span.set_attribute("mrv.run_id", run_id)
            span.set_attribute("mrv.period", period)
            yield span
    finally:
        structlog.contextvars.clear_contextvars()


def emit(event: str, signals: dict) -> dict:
    """Emit ONCE. Log line and span attributes come from the same dictionary, so
    the dashboard and the audit record cannot disagree — they are the same values.

    Returns the evidence subset for the caller to persist WITH the artefact,
    because log retention is not an audit strategy.
    """
    problems = validate_schema(signals)
    if problems:
        raise ValueError(f"signal schema invalid: {problems}")

    payload = {**signals, "schema_version": SCHEMA_VERSION}

    span = trace.get_current_span()
    for key, value in payload.items():
        span.set_attribute(
            f"mrv.{key}",
            json.dumps(value) if isinstance(value, (dict, list, tuple)) else value,
        )

    log.info(event, **payload)
    return {k: v for k, v in payload.items() if k in HORIZONS["evidence"]}


def task_payload(tile_id: str, period: str) -> dict:
    """Serialise the trace context INTO the task so the worker can rejoin the
    run's trace instead of starting its own."""
    carrier: dict[str, str] = {}
    inject(carrier)
    return {"tile_id": tile_id, "period": period, "otel_context": carrier}


def run_worker_task(payload: dict) -> dict:
    """Re-attach the context, then work. Without the attach, this run's ten
    thousand tiles become ten thousand unrelated traces."""
    token = otel_context.attach(extract(payload.get("otel_context", {})))
    try:
        with tracer.start_as_current_span("mrv.tile") as span:
            span.set_attribute("mrv.tile_id", payload["tile_id"])
            result = process_tile(payload["tile_id"], payload["period"])
            return emit("mrv.tile.complete", result)
    finally:
        otel_context.detach(token)

Note what emit returns. The evidence subset goes back to the caller so it can be written beside the artefact — in the Parquet footer, a sidecar JSON, or the schema’s provenance columns. Relying on log retention for audit evidence is the most common instrumentation mistake in this domain, and it is discovered at exactly the wrong moment: a verifier asks for a 2027 bounding box in 2036, and the observability vendor’s thirty-day tier expired it nine years ago.

Compliance Gating & Audit Trail Generation #

The evidence stream is the compliance-bearing output, and three properties make it usable years later. It must be append-only, so a record cannot be quietly revised; self-describing, carrying its schema version so a future reader can interpret a field that has since changed meaning; and co-located with the artefact, so retrieving the data retrieves its provenance rather than requiring a join against a system that may not exist.

The shared keys deserve deliberate design. run_id, input_digest, and code_version appear in all three streams and are the only way an operator investigating a latency spike and a verifier reconstructing a tonnage can talk about the same event. Divergent identifiers between the ops stack and the evidence store are a common and painful defect, usually discovered mid-audit when nobody can map a dashboard incident to a data record.

Retention is a budgeting decision that is far cheaper made at design time. Operational telemetry at full granularity over thousands of tiles is expensive and worthless after a month; provenance records are tiny and must outlive the crediting period. Splitting them at emission — rather than retaining everything at the longest horizon, or worse, at the shortest — is what makes both affordable. Route the evidence stream into the same store that backs MRV data lineage and provenance tracking and conform its fields to the MRV data schema reference.

Production Integration #

1. Define the schema first, with a version field and an explicit horizon for every key; treat adding a field as a schema change requiring review.
2. Configure structlog once at process start, with context variables bound for run-scoped keys so call sites cannot forget them.
3. Propagate trace context into every task payload and re-attach it in the worker; verify it in staging with the fan-out check above.
4. Emit through one function, never logging or setting span attributes directly from a stage.
5. Persist the evidence subset with the artefact, not only to the log pipeline.
6. Route by horizon at the collector, and test the audit-horizon store by retrieving a year-old record deliberately, on a schedule.

For orchestrator integration, bind the run identifier from the orchestrator’s own run context so traces join to the scheduler’s view — the mapping differs between Airflow, Prefect and Dagster and is worth getting right once, as discussed in Prefect vs Airflow vs Dagster for MRV pipelines.

The retention split is the decision most teams get wrong in a way that only becomes visible years later, so it is worth making explicit at design time rather than inheriting from whatever the observability vendor’s default tier happens to be.

Frequently Asked Questions #

Should the evidence record live in the log pipeline or beside the data? #

Beside the data, always, with the log pipeline as a convenience copy. Logs are operational infrastructure with operational retention and operational access controls, and none of those match an audit horizon measured in decades. Writing the provenance subset into the Parquet footer, a sidecar JSON, or dedicated provenance columns means the evidence travels with the artefact through every copy, migration, and handover — including the ones that happen after your observability vendor has been replaced.

How much cardinality is safe in span attributes? #

Less than you want. Per-tile identifiers on spans are usually fine because spans are sampled and stored differently from metrics, but the same identifiers as metric labels will produce a cardinality explosion that either costs a great deal or gets silently dropped. The workable split is: high-cardinality identifiers on spans and in the evidence record, low-cardinality dimensions — stage, status, sensor — on metrics. Aggregate per-tile counts into per-stage metrics rather than emitting a metric per tile.

Do I need OpenTelemetry at all if I already have structured logs? #

You need it for one thing structured logs cannot do: relate events across a distributed fan-out into a single causal structure. If your pipeline runs in one process, well-designed structured logs with a run identifier carry most of the value and tracing is optional. Once work fans out across workers and stages, the trace is what turns ten thousand independent events into one run you can ask questions about — including the completeness question that catches partial-output failures.

What belongs in a span event versus a log line? #

With a single emitter the question mostly dissolves, which is the point. Practically: everything goes to the log line, and the span carries the same values as attributes so a trace-based query can filter on them. Where they diverge is volume — you may sample spans at a fraction of runs while keeping every log line, or the reverse for very high-frequency stages. Make that a sampling decision at the collector rather than a decision about which values to emit, so the two views never describe different things.

How do I instrument a pipeline that already exists without rewriting it? #

Work backwards from the evidence stream, which is both the most valuable and the least invasive. Start by wrapping the existing stage boundaries with a function that computes the signals from the output artefact — this requires no change to the stage’s internals at all, because the signals are derived rather than reported. That single step gives you the provenance record, the spatial invariants, and a baseline for run comparison, and it can be added to a legacy pipeline in a day. Only then add the trace context propagation, which does require touching the task dispatch path, and finally migrate the existing prose logs to the structured schema stage by stage. Teams that attempt the migration in the opposite order — logs first — spend weeks on the least valuable part and often stall before reaching the evidence stream.

Does the instrumentation itself need testing? #

Yes, and the test that matters is a retrieval test rather than an emission test. It is easy to verify that a stage emits an event; it is the retrieval a year later that actually fails, because a retention policy changed, a field was renamed, an index was dropped, or the store was migrated. Schedule a recurring job that fetches a deliberately old evidence record, asserts every expected field is present and parseable under the current schema version, and fails loudly when it is not. That job is the difference between believing you have an audit trail and knowing it.

How do I keep the schema stable while the pipeline evolves? #

Version it, add rather than repurpose, and never change a field’s meaning under a stable name. Adding null_rate_after_mask alongside null_rate is cheap; redefining null_rate to mean something else silently invalidates every historical comparison and every trend a verifier might reconstruct. Where a field genuinely must change semantics, bump the schema version and keep readers that understand both. The discipline is the same one you would apply to a database schema, and for the same reason: consumers you cannot see depend on it.

What sampling rate is appropriate for traces in a large fan-out? #

Sample by run rather than by span, and keep every span within a sampled run. Sampling individual spans in a ten-thousand-tile fan-out produces partial traces that cannot answer completeness questions, which is most of the reason to have traces at all. Sampling whole runs at, say, one in ten during steady operation and one hundred per cent for any run that will produce a reported figure keeps cost bounded without fragmenting the runs that matter.

How should sensitive values be kept out of the signal streams? #

By allow-listing the fields that may be emitted rather than deny-listing the ones that may not. A schema that enumerates permitted keys cannot leak a credential added to a parameters dictionary six months later, whereas a redaction filter will miss exactly that case. Where a value must be recorded but not exposed — a supplier identifier under a confidentiality agreement, say — record a stable hash in the observability stream and keep the mapping in the restricted evidence store.

Related guides #

• MRV Pipeline Observability & Failure Modes — the parent topic and the four signal classes.
• Failure Mode Catalog for Spatial MRV Pipelines — the invariants this instrumentation carries.
• Tracking Data Lineage with OpenLineage for ESG Audits — where the evidence stream becomes a lineage graph.
• Prefect vs Airflow vs Dagster for MRV Pipelines — how each orchestrator exposes the run context to bind against.