Adding OpenTelemetry Spans to Async Geometry Handlers

To trace an async geometry pipeline, configure one SDK TracerProvider, wrap each handler stage in tracer.start_as_current_span(...) so ingress, validate, reproject, and publish nest under a single root span, attach only bounded geo attributes (h3 cell, source EPSG, vertex count), and manually re-attach the captured context inside any asyncio.create_task coroutine so the child span stays in the same trace. Done correctly, one webhook delivery produces one connected trace you can read top to bottom.

This page belongs to Structured Logging & Tracing for Spatial Events, part of the Monitoring & Observability for Spatial Pipelines — the reference for making event-driven spatial systems debuggable in production.

When to use this pattern

Reach for distributed tracing over plain structured logs when:

  • A single webhook event fans out across several async coroutines or tasks — validate, reproject, enrich, publish — and you need to see the causal chain and per-stage latency for one delivery, not aggregate counters.
  • Slow or failing events are hard to pin down because logs from concurrent deliveries interleave, and you cannot tell which log line belongs to which payload.
  • You already ship spans from an upstream service (an API gateway, a Kafka producer) and want the geometry worker to continue the same trace rather than start a disconnected one.

It is not the right tool when you only need rates, ratios, or histograms — geometry validation failure rate, queue depth, reprojection duration percentiles. Those are aggregate signals and belong in metrics; see Geo-Specific Metrics & Instrumentation for that layer. Tracing answers “what happened to this event”; metrics answer “how is the fleet doing”.

Why async breaks a naive trace

OpenTelemetry tracks the “current span” in a context variable. In synchronous code, opening a span with start_as_current_span sets that context for the duration of the with block, and any span opened inside becomes its child automatically. Async pipelines break this in two places.

First, await boundaries are fine — the context travels with the coroutine — but asyncio.create_task schedules a coroutine that may not inherit the active OpenTelemetry context. A span opened inside that task can silently start a brand-new root trace, so your publish stage appears as an orphan disconnected from ingress. Second, geometry payloads tempt you to attach the whole coordinate array as an attribute for “easier debugging”, which bloats every span and can leak location data.

The diagram shows the shape you want: one root span per delivery, three nested child spans, and a fourth span that crosses a create_task boundary yet stays inside the same trace because the context was re-attached by hand.

A single trace across four async geometry stages A root span labelled process_geometry_event spans the full width. Beneath it, three nested child spans — validate_geometry, reproject, and publish — sit in sequence. The publish span sits below a dashed create_task boundary, with an arrow showing the OpenTelemetry context being carried across it so publish remains a child of the root rather than a new trace. process_geometry_event (root span · trace_id shared by all children) validate_geometry reproject span attributes: geo.h3_cell · geo.source_epsg · geo.vertex_count (never the raw coordinates) asyncio.create_task boundary — context attached by hand carry ctx publish still a child of the root — same trace_id, parent = process_geometry_event

Complete runnable implementation

The module below is self-contained. It configures a tracer, instruments four async stages, attaches bounded geo attributes, and shows the manual context handoff across asyncio.create_task. It uses only opentelemetry-api and opentelemetry-sdk (pip install opentelemetry-api opentelemetry-sdk); the h3 call is illustrative and stubbed so the file runs standalone.

python 
import asyncio
from opentelemetry import trace, context
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor, ConsoleSpanExporter
from opentelemetry.sdk.trace.sampling import ParentBased, TraceIdRatioBased
from opentelemetry.trace import SpanKind, Status, StatusCode

# --- One-time tracer setup (do this at process startup) -------------------
# ParentBased(TraceIdRatioBased(rate)) makes the sampling decision once at the
# root and every child span inherits it, so heavy geometry traces are kept or
# dropped as a whole rather than half-recorded.
provider = TracerProvider(sampler=ParentBased(TraceIdRatioBased(0.10)))
provider.add_span_processor(SimpleSpanProcessor(ConsoleSpanExporter()))
trace.set_tracer_provider(provider)

tracer = trace.get_tracer("geospatial.webhook.geometry_handler")


def _bounded_geo_attrs(feature: dict) -> dict:
    """Derive low-cardinality, size-safe span attributes from a GeoJSON feature.

    We deliberately never attach the coordinate array itself: it can be
    kilobytes per span, inflates export payloads, and may leak location data.
    Instead we record an h3 cell (spatial bucket), the source CRS, the vertex
    count, and the geometry type — all small and bounded.
    """
    geom = feature.get("geometry") or {}
    coords = geom.get("coordinates") or []

    def _count_vertices(c) -> int:
        if not isinstance(c, list):
            return 0
        if c and isinstance(c[0], (int, float)):
            return 1  # a single [lon, lat] position
        return sum(_count_vertices(x) for x in c)

    lon, lat = 0.0, 0.0
    if geom.get("type") == "Point" and len(coords) >= 2:
        lon, lat = coords[0], coords[1]

    return {
        # h3.latlng_to_cell(lat, lon, 8) in real code; stubbed here to stay standalone.
        "geo.h3_cell": f"stub-h3-r8-{round(lat, 2)}-{round(lon, 2)}",
        "geo.source_epsg": int(feature.get("source_epsg", 4326)),  # EPSG:4326 (WGS84)
        "geo.geometry_type": geom.get("type", "unknown"),
        "geo.vertex_count": _count_vertices(coords),
    }


# --- Stage handlers, each wrapped in its own span -------------------------
async def validate_geometry(feature: dict) -> dict:
    with tracer.start_as_current_span("validate_geometry") as span:
        span.set_attributes(_bounded_geo_attrs(feature))
        await asyncio.sleep(0)  # stand-in for real async validation I/O
        if not feature.get("geometry"):
            span.set_status(Status(StatusCode.ERROR, "null geometry"))
            raise ValueError("feature has no geometry")
        return feature


async def reproject(feature: dict, target_epsg: int = 4326) -> dict:
    with tracer.start_as_current_span("reproject") as span:
        span.set_attribute("geo.source_epsg", int(feature.get("source_epsg", 4326)))
        span.set_attribute("geo.target_epsg", target_epsg)  # EPSG:4326 (WGS84)
        await asyncio.sleep(0)  # stand-in for pyproj transform
        feature["source_epsg"] = target_epsg
        return feature


async def publish(feature: dict) -> None:
    # Runs inside a task scheduled by the ingress handler. The parent context
    # is attached by the caller before we open this span (see below).
    with tracer.start_as_current_span("publish", kind=SpanKind.PRODUCER) as span:
        span.set_attribute("geo.h3_cell", _bounded_geo_attrs(feature)["geo.h3_cell"])
        await asyncio.sleep(0)  # stand-in for Kafka/Redis produce


# --- Ingress: root span + manual context propagation ---------------------
async def process_geometry_event(feature: dict) -> None:
    with tracer.start_as_current_span("process_geometry_event", kind=SpanKind.CONSUMER):
        feature = await validate_geometry(feature)
        feature = await reproject(feature)

        # Capture the ACTIVE context now, while the root span is current.
        parent_ctx = context.get_current()

        async def _publish_task() -> None:
            # Re-attach the captured context as the first action in the task,
            # otherwise the publish span may start a brand-new root trace.
            token = context.attach(parent_ctx)
            try:
                await publish(feature)
            finally:
                context.detach(token)

        task = asyncio.create_task(_publish_task())
        await task  # await so the span exports before the demo exits


if __name__ == "__main__":
    demo = {
        "type": "Feature",
        "source_epsg": 4326,  # EPSG:4326 (WGS84)
        "geometry": {"type": "Point", "coordinates": [-73.965355, 40.782865]},
        "properties": {"sensor_id": "SN-42"},
    }
    asyncio.run(process_geometry_event(demo))

Run it and the console exporter prints four spans that share one trace_id, with validate_geometry, reproject, and publish all pointing at the process_geometry_event span as their parent.

Parameter reference

Span / attribute Type Spatial constraint Default
process_geometry_event span (CONSUMER) Root span per delivery; owns the sampling decision —
validate_geometry span Set ERROR status on invalid/null geometry —
reproject span Carry both geo.source_epsg and geo.target_epsg —
publish span (PRODUCER) Must inherit context across create_task —
geo.h3_cell str One h3 index at a fixed resolution (7–9); low cardinality per span —
geo.source_epsg int Numeric EPSG code, e.g. 4326 (WGS84) or 3857 (Web Mercator) 4326
geo.vertex_count int Derived count, never the coordinate array itself 0
geo.geometry_type str Bounded enum: Point, LineString, Polygon, … per RFC 7946 "unknown"
sampler ratio float 0.0–1.0; keep low for high-volume geometry traffic 0.10

Gotchas and spatial edge cases

  1. Context lost across create_task. This is the most common failure. asyncio.create_task does not reliably carry the OpenTelemetry context, so a span opened in the task becomes a new root. Always context.get_current() in the scheduling coroutine and context.attach() it as the first line of the task, detaching in a finally.

  2. Raw coordinates as attributes. A large Polygon or MultiPolygon coordinate array is kilobytes per span. Attaching it multiplies export volume, can breach exporter size limits, and leaks precise location data. Record geo.vertex_count, geo.h3_cell, and a bounding-box area instead.

  3. Attribute cardinality creep. Span attributes tolerate high cardinality far better than metric labels, but backends still index them. Avoid per-request unique strings like a full delivery UUID as a searchable dimension unless you need it; prefer the h3 cell at a coarse resolution for spatial grouping.

  4. Sampling that splits a trace. A head sampler that decides independently per span can keep validate_geometry but drop publish, giving you half a trace. Use ParentBased(TraceIdRatioBased(rate)) so one decision at the root propagates to all children.

  5. CRS mismatch hidden by tracing. Recording only geo.target_epsg hides reprojection bugs. Always record geo.source_epsg and geo.target_epsg so a trace shows, for example, an EPSG:3857 (Web Mercator) payload that was never normalized to EPSG:4326 (WGS84).

  6. Spans not exported in short-lived scripts. With SimpleSpanProcessor/BatchSpanProcessor, spans flush asynchronously. In tests or one-shot workers call provider.force_flush() (or await the task as above) before the process exits, or the last spans vanish.

  7. Instrumenting CPU-bound offload. Heavy reprojection or validation often runs in a thread or process pool. run_in_executor does not carry context automatically either — capture and re-attach it inside the executor callable, the same way you do for create_task. See Async Processing for Heavy Geometries for the offload patterns this applies to.

Verification

This test drives the pipeline against an InMemorySpanExporter and asserts the span tree by trace_id and parent linkage — proving the publish span really is a child and not an orphan. Run with pytest.

python 
import asyncio
import pytest
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

import geometry_handler as gh  # the module shown above


@pytest.fixture
def spans():
    """Swap in a provider backed by an in-memory exporter for assertions."""
    exporter = InMemorySpanExporter()
    provider = TracerProvider()
    provider.add_span_processor(SimpleSpanProcessor(exporter))
    trace.set_tracer_provider(provider)
    # Rebind the module's tracer to the test provider.
    gh.tracer = trace.get_tracer("test.geometry_handler")
    yield exporter
    exporter.clear()


def test_single_connected_trace(spans):
    feature = {
        "type": "Feature",
        "source_epsg": 3857,  # EPSG:3857 (Web Mercator) — should be recorded
        "geometry": {"type": "Point", "coordinates": [-73.965355, 40.782865]},
    }
    asyncio.run(gh.process_geometry_event(feature))

    finished = {s.name: s for s in spans.get_finished_spans()}
    assert set(finished) == {
        "process_geometry_event", "validate_geometry", "reproject", "publish",
    }

    root = finished["process_geometry_event"]
    # All four spans share ONE trace_id.
    trace_ids = {s.context.trace_id for s in finished.values()}
    assert len(trace_ids) == 1

    # publish crossed a create_task boundary but is still a child of the root.
    publish = finished["publish"]
    assert publish.parent is not None
    assert publish.parent.span_id == root.context.span_id
    assert root.parent is None  # the root has no parent


def test_geo_attributes_are_bounded(spans):
    feature = {
        "type": "Feature",
        "source_epsg": 4326,  # EPSG:4326 (WGS84)
        "geometry": {"type": "Point", "coordinates": [-73.965355, 40.782865]},
    }
    asyncio.run(gh.process_geometry_event(feature))

    validate = next(
        s for s in spans.get_finished_spans() if s.name == "validate_geometry"
    )
    attrs = dict(validate.attributes)
    assert attrs["geo.vertex_count"] == 1
    assert attrs["geo.source_epsg"] == 4326
    # No raw coordinate array leaked onto any attribute value.
    assert all("coordinates" not in str(v) for v in attrs.values())

FAQ

Why does my publish span show up as a separate trace instead of a child?

asyncio.create_task does not carry the OpenTelemetry context into the new task on all runtimes and instrumentation setups, so a span opened inside the task starts a fresh root. Capture context.get_current() in the scheduling coroutine and context.attach() it as the first line of the background coroutine, detaching the token in a finally block.

Can I put the geometry coordinates in a span attribute for debugging?

No. A raw coordinate array can be kilobytes per span, it inflates trace storage and export payloads, and it can leak sensitive location data into your observability backend. Record derived, bounded values instead — an h3 cell index, the vertex count, the bounding-box area, and the source EPSG code.

How do I keep high-volume geometry traces from overwhelming the backend?

Use a parent-based, ratio sampler (ParentBased(TraceIdRatioBased(rate))) so the sampling decision is made once at ingress and inherited by every child span in the pipeline. Sample a small fraction of normal traffic and pair it with tail-based sampling in a collector to always keep error and high-latency traces.

Should h3 cell index be a span attribute or a metric label?

As a span attribute it is safe because spans are individual records, not aggregated time series. As a Prometheus metric label it would explode cardinality — millions of h3 cells become millions of series. Keep the cell on the span for per-request debugging and aggregate to a coarse resolution before using it as a metric label.