Instrumenting asyncpg Queries with OpenTelemetry

← Back to Observability for Spatial Endpoints

This page shows how to wrap PostGIS calls in spans that carry enough spatial context to diagnose a slow request from the trace alone, without re-running anything.

# Context & When to Use

A request span that says “912 ms” tells you a request was slow. A database span that says “912 ms, statement features_bbox, 12 rows, envelope 0.004 deg²” tells you it was slow for no good reason — a tiny area returning almost nothing should never take that long, so the index or the cache is the suspect. The same span reading “912 ms, 41 000 rows, envelope 380 deg²” tells you the client asked for a continent and got one.

That difference — between a number and an explanation — comes down to three attributes: what the statement was, how much ground it covered, and how much came back. None of them are available to generic database instrumentation, because none of them are visible in the SQL text alone.

The wrapper below is deliberately thin. It does not replace the automatic instrumentation’s job of timing the call; it adds the spatial context and it controls what leaves the process, which matters because a naive span attribute containing substituted SQL exports coordinates into a tracing backend — the exposure discussed in Redacting Coordinate Precision in Application Logs.

# Runnable Implementation

import time
from contextlib import asynccontextmanager
from typing import Any, Sequence

import asyncpg
from opentelemetry import trace
from opentelemetry.trace import SpanKind, Status, StatusCode

tracer = trace.get_tracer("geospatial-api.db")


def magnitude_bucket(area_deg2: float) -> str:
    if area_deg2 < 0.01:
        return "xs"
    if area_deg2 < 1:
        return "s"
    if area_deg2 < 25:
        return "m"
    if area_deg2 < 500:
        return "l"
    return "xl"


@asynccontextmanager
async def traced_pool_acquire(pool: asyncpg.Pool):
    """Separate 'waiting for a connection' from 'running the query'."""
    started = time.perf_counter()
    async with pool.acquire() as conn:
        wait_ms = (time.perf_counter() - started) * 1000
        span = trace.get_current_span()
        span.set_attribute("db.pool.wait_ms", round(wait_ms, 2))
        span.set_attribute("db.pool.size", pool.get_size())
        span.set_attribute("db.pool.idle", pool.get_idle_size())
        yield conn


async def traced_fetch(
    conn: asyncpg.Connection,
    name: str,
    sql: str,
    *args: Any,
    operation: str,
    envelope_deg2: float | None = None,
) -> Sequence[asyncpg.Record]:
    """Run one statement inside a span carrying its spatial context."""
    with tracer.start_as_current_span(f"db.{name}", kind=SpanKind.CLIENT) as span:
        span.set_attribute("db.system", "postgresql")
        # The NAME, never the substituted text — parameters carry coordinates
        span.set_attribute("db.operation", name)
        span.set_attribute("geo.operation", operation)
        if envelope_deg2 is not None:
            span.set_attribute("geo.envelope_deg2", round(envelope_deg2, 4))
            span.set_attribute("geo.magnitude", magnitude_bucket(envelope_deg2))
        try:
            rows = await conn.fetch(sql, *args)
        except asyncpg.PostgresError as exc:
            span.set_status(Status(StatusCode.ERROR, exc.__class__.__name__))
            span.set_attribute("db.postgres.sqlstate", getattr(exc, "sqlstate", ""))
            raise
        span.set_attribute("db.rows", len(rows))
        # The ratio is the diagnostic: rows per unit of ground asked for
        if envelope_deg2:
            span.set_attribute("geo.rows_per_deg2", round(len(rows) / envelope_deg2, 1))
        return rows

# Key Parameters & Options

Never add the substituted SQL or the parameter values. The statement name plus the source file is a complete reference for what ran, without exporting coordinates.

# What the attributes let you ask afterwards

# Gotchas & Failure Modes

• Span per row. Creating a span inside a result loop turns a 900-row query into 900 spans and dominates the request. One span per statement, attributes for the aggregate.
• Head-based sampling at 1 %. The slow requests are by definition rare, so a uniform sample almost never keeps one. Use tail-based sampling, or force sampled=true when the request exceeds its budget.
• Coordinates in attributes. db.statement with substituted parameters, or a bbox attribute copied verbatim, exports precise locations to the tracing backend. Record the area, not the box.
• Pool wait invisible. If the acquire happens outside the span, connection starvation looks exactly like a slow query and the wrong thing gets optimised — the interaction described in Connection Pooling & PgBouncer Setup.
• Exceptions swallowing the span status. Catching PostgresError without calling set_status leaves a failed query recorded as a success, and the error rate derived from traces silently under-reports.
• Cardinality leaking from attributes into metrics. Span attributes tolerate high cardinality; metric labels do not. Keep geo.magnitude for metrics and geo.envelope_deg2 for spans only.

# Wiring it into the route without repeating yourself

Threading operation and envelope_deg2 through every call site by hand goes stale within a sprint. Derive both once, in the dependency that already parses the bounding box, and stash them on the request so the data layer can read them without another argument.

from dataclasses import dataclass
from typing import Annotated

from fastapi import Depends, HTTPException, Query, Request


@dataclass(frozen=True)
class SpatialRequestContext:
    operation: str
    envelope_deg2: float | None

    @property
    def magnitude(self) -> str:
        return magnitude_bucket(self.envelope_deg2) if self.envelope_deg2 else "na"


def spatial_context(
    request: Request,
    bbox: Annotated[str | None, Query()] = None,
) -> SpatialRequestContext:
    area = None
    if bbox:
        try:
            minx, miny, maxx, maxy = (float(v) for v in bbox.split(","))
        except ValueError:
            raise HTTPException(422, detail={"error": "bbox_must_be_four_numbers"})
        area = abs(maxx - minx) * abs(maxy - miny)

    ctx = SpatialRequestContext(operation=classify(request.url.path), envelope_deg2=area)
    request.state.spatial = ctx        # available to middleware and metrics too
    return ctx

Every route then passes one object, and adding a future attribute — the tenant’s scope area, say, or the requested output projection — means changing one dataclass rather than every query call. The middleware described in Observability for Spatial Endpoints reads the same object for its metric labels, so the span and the histogram can never disagree about what a request was.

# Verification Snippet

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


async def test_span_carries_spatial_context(pool):
    exporter = InMemorySpanExporter()
    trace.get_tracer_provider().add_span_processor(SimpleSpanProcessor(exporter))

    async with traced_pool_acquire(pool) as conn:
        await traced_fetch(conn, "features_bbox", FEATURES_SQL,
                           -0.2, 51.4, 0.0, 51.6, 200,
                           operation="bbox", envelope_deg2=0.04)

    span = next(s for s in exporter.get_finished_spans() if s.name == "db.features_bbox")
    assert span.attributes["geo.magnitude"] == "s"
    assert span.attributes["db.rows"] > 0
    # No coordinates anywhere in the exported attributes
    assert not any("51.5" in str(v) for v in span.attributes.values())

# Confirm attributes arrive in the collector
otel-cli span --service test --name db.features_bbox --verbose 2>&1 | grep geo.

# Related

• Observability for Spatial Endpoints — the wider signal design these spans feed
• Connection Pooling & PgBouncer Setup — what a large pool.wait_ms is telling you
• Redacting Coordinate Precision in Application Logs — the same exposure question for traces

← Back to Observability for Spatial Endpoints