Propagating Trace Context Across Prefect Tasks

To make a whole spatial flow appear as one OpenTelemetry trace, serialise the active context into a carrier dict with inject() in the parent task and rebuild it with extract() inside each Prefect task, ProcessPoolExecutor worker, or GDAL subprocess. Prefect’s task runners hand work to separate threads and processes whose thread-local OTel context is empty, so without an explicit carrier every task opens a detached root trace and the causal link between fetch, reproject, and load is lost. Passing the traceparent by hand restores a single parent-child span tree spanning the entire run.

When to Use This Pattern#

Reach for manual context propagation whenever a boundary sits between the code that opened a span and the code that should nest under it. Concretely:

• A Prefect flow submits tasks to a ConcurrentTaskRunner, DaskTaskRunner, or process-based runner and each task currently shows up as its own trace.
• Heavy GIS work is offloaded to a ProcessPoolExecutor or concurrent.futures pool where thread-local context does not survive the process fork.
• A task shells out to gdalwarp, gdal_translate, or ogr2ogr via subprocess and you want the CLI stage nested under the flow’s trace.
• You are extending the instrumentation from the parent guide on OpenTelemetry tracing for spatial tasks to a real multi-task Prefect deployment.

Why the Context Is Lost by Default#

The OpenTelemetry Python SDK stores the currently active span in a contextvars-backed thread-local Context. That storage is invisible across two boundaries that spatial pipelines cross constantly. The first is the process boundary: when Prefect’s task runner or a ProcessPoolExecutor forks or spawns a worker, the child process starts with a fresh, empty context and often no TracerProvider at all, so any span it opens becomes a new root. The second is the serialization boundary: Prefect may run tasks on entirely separate machines, and nothing about a Python object graph automatically carries the active traceparent with it. The W3C Trace Context standard exists precisely to bridge these gaps — it defines a compact traceparent header string that encodes the trace id, the parent span id, and the sampling flag, small enough to pass as an ordinary function argument, an environment variable, or a message header. inject() writes that string into a carrier; extract() reads it back. Everything in this page is an application of those two functions to the specific boundaries a Prefect spatial flow presents.

Complete Working Example#

The flow below fetches features in one task and reprojects them in a pooled worker. The parent injects the context into a carrier dict; the child extracts it and opens its span with that context, so both spans share one trace_id.

from concurrent.futures import ProcessPoolExecutor
from typing import Any

import geopandas as gpd
from pyproj import CRS

from prefect import flow, task, get_run_logger

from opentelemetry import trace
from opentelemetry.propagate import inject, extract
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter


def _init_tracing(service_name: str = "prefect-spatial") -> None:
    """Install a TracerProvider. Must run in EVERY process, including pool workers."""
    if isinstance(trace.get_tracer_provider(), TracerProvider):
        return  # already configured in this process
    provider = TracerProvider(resource=Resource.create({"service.name": service_name}))
    provider.add_span_processor(
        BatchSpanProcessor(OTLPSpanExporter(endpoint="otel-collector:4317", insecure=True))
    )
    trace.set_tracer_provider(provider)


_init_tracing()
tracer = trace.get_tracer("geo.prefect")


def _reproject_worker(carrier: dict[str, str], wkt: str, dst_epsg: int) -> int:
    """Runs in a separate process: rebuild context, then open a child span."""
    _init_tracing()  # the forked process has no provider yet
    ctx = extract(carrier)  # rebuild parent context from the traceparent header
    with tracer.start_as_current_span("reproject_worker", context=ctx) as span:
        span.set_attribute("geo.epsg", dst_epsg)
        gdf = gpd.GeoSeries.from_wkt([wkt], crs="EPSG:4326").to_crs(epsg=dst_epsg)
        span.set_attribute("geo.feature_count", len(gdf))
        return len(gdf)


@task
def fetch(wfs_url: str) -> gpd.GeoDataFrame:
    with tracer.start_as_current_span("fetch") as span:
        gdf: gpd.GeoDataFrame = gpd.read_file(wfs_url)  # EPSG:4326
        span.set_attribute("geo.epsg", gdf.crs.to_epsg())
        span.set_attribute("geo.feature_count", len(gdf))
        return gdf


@task
def reproject(gdf: gpd.GeoDataFrame, dst_epsg: int = 3857) -> int:
    with tracer.start_as_current_span("reproject") as span:
        carrier: dict[str, Any] = {}
        inject(carrier)  # serialise THIS span's context into the carrier
        wkt = gdf.geometry.iloc[0].wkt
        with ProcessPoolExecutor(max_workers=1) as pool:
            future = pool.submit(_reproject_worker, carrier, wkt, dst_epsg)
            count = future.result()
        span.set_attribute("geo.feature_count", count)
        span.set_attribute("geo.target_crs", CRS.from_epsg(dst_epsg).to_string())
        return count


@flow(name="spatial-trace-demo")
def spatial_flow(wfs_url: str) -> None:
    with tracer.start_as_current_span("spatial_flow") as root:
        root.set_attribute("geo.epsg", 4326)
        gdf = fetch(wfs_url)
        reproject(gdf, dst_epsg=3857)
        get_run_logger().info("flow complete under trace_id=%s",
                              format(root.get_span_context().trace_id, "032x"))

The key moves: inject(carrier) writes a traceparent string into the dict inside the parent span; the dict is passed as a plain argument through the process boundary; extract(carrier) rebuilds a Context; and start_as_current_span(..., context=ctx) opens the child under that parent instead of starting a new root. Because a GeoDataFrame does not pickle cleanly into pool workers in every environment, the example passes a WKT string and the target EPSG rather than the frame itself.

Two design choices deserve emphasis. First, _init_tracing() is idempotent: it checks whether a real TracerProvider is already installed before building one, so calling it in the main process and again in every pool worker is safe and avoids the duplicate-provider bug where a second provider silently drops spans. Second, the CRS is passed explicitly as dst_epsg and re-recorded on the worker span as geo.epsg rather than being inferred inside the worker. Making the target projection an explicit argument, and stamping it on the span, means the trace shows exactly which reprojection ran — invaluable when a fleet processes tiles into different projections and one misconfigured worker starts emitting EPSG:3857 output where EPSG:4326 was expected.

The same carrier technique extends to subprocess GDAL calls. Rather than passing the carrier as a Python argument, serialise it into the child process environment (for example OTEL_CARRIER_TRACEPARENT), read it back inside a thin GDAL wrapper, and call extract() on the reconstructed dict. The mechanics are identical; only the transport changes from a function argument to an environment variable. Message-queue-driven flows follow the same shape, injecting the carrier into message headers so a consumer on another host rebuilds the parent context from the payload it receives.

Parameter & Option Reference#

Verification & Testing#

Confirm parent and child share a trace_id and that the child’s parent is the reproject span. An in-memory exporter makes this assertable without a collector.

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter


def test_context_propagates_across_pool() -> None:
    exporter = InMemorySpanExporter()
    provider = TracerProvider()
    provider.add_span_processor(SimpleSpanProcessor(exporter))
    trace.set_tracer_provider(provider)

    spatial_flow("tests/fixtures/one_feature.geojson")

    spans = {s.name: s for s in exporter.get_finished_spans()}
    root = spans["spatial_flow"]
    reproj = spans["reproject"]
    # All spans belong to one trace.
    assert reproj.context.trace_id == root.context.trace_id
    # The reproject span's parent is the flow root.
    assert reproj.parent.span_id == root.context.span_id
    # The spatial attribute survived the round trip.
    assert reproj.attributes["geo.target_crs"] == "EPSG:3857"

The three assertions map to the three things that actually go wrong. The trace_id equality check catches a detached root — the classic symptom of a carrier that never reached the worker. The parent.span_id check catches the subtler failure where the child does share the trace but attaches to the wrong ancestor, which happens when inject() runs outside the intended span. The attribute check confirms the payload — here the target CRS — survived the process hop intact, guarding against a worker that ran but recorded nothing useful. Treat these as a permanent regression test: context propagation is exactly the kind of wiring that silently breaks when someone swaps a task runner or refactors the worker signature months later.

For a live run, open the trace in your backend (Jaeger, Tempo) and check that fetch, reproject, and reproject_worker render under a single spatial_flow waterfall with no orphaned roots. A correctly linked trace shows the reproject_worker bar nested strictly inside the reproject bar, which is nested inside spatial_flow; if any of them appears as a separate trace in the backend’s list view, propagation failed at that boundary. On the collector host you can also confirm ingestion with a quick check that spans carry matching trace IDs:

# Tail the collector's debug exporter and confirm one trace_id spans all stages
docker logs otel-collector 2>&1 | grep -A2 "Trace ID" | sort -u

Common Pitfalls#

• Forgetting to initialise the provider in pool workers. A forked ProcessPoolExecutor process starts with the default no-op tracer, so extract() succeeds but the child span is never recorded. Always call your _init_tracing() at the top of the worker function.
• Injecting before the parent span is current. If inject(carrier) runs outside the with tracer.start_as_current_span(...) block, the carrier captures an empty context and the child becomes a detached root. Inject inside the active span.
• Passing the carrier as global or module state. Under concurrent tile processing, a shared global carrier races and cross-links unrelated traces. Keep the carrier a local variable threaded through arguments — the same discipline used for state management in geospatial flows where per-run data must never leak between runs.
• Mismatched propagators. If one process uses the W3C TraceContext propagator and another uses B3, extract() silently finds nothing. Standardise the propagator across every worker image, exactly as you would pin GDAL and PROJ versions for DAG design principles for spatial ETL.
• Re-injecting a stale carrier for fan-out. When one parent task fans out to many tile workers, building the carrier once and reusing it is correct — every child should share the parent context. But if you instead capture the carrier before the parent span exists, or mutate it between submissions, children link inconsistently. Build the carrier inside the parent span and treat it as immutable once populated.
• Assuming Prefect result persistence carries context. Prefect can persist and rehydrate task results across runs, but the trace context is deliberately not part of that persisted state; it is regenerated per execution. Do not try to reconstruct a previous run’s trace from a cached result — start a fresh trace and, if you need to relate the two, link them with a span link rather than a shared parent.

Related: This page extends OpenTelemetry tracing for spatial tasks, pairs with tracing PostGIS query spans in spatial ETL for the load stage, correlates with structured logging for geospatial flows by trace_id, and builds on DAG design principles for spatial ETL. The W3C Trace Context specification documents the traceparent header format.

← Back to OpenTelemetry Tracing for Spatial Tasks