Observable Gauges and Asynchronous Instruments

OpenTelemetry has two kinds of instruments. Synchronous ones — counters, histograms, up-down counters — are called from application code when something happens. Asynchronous ones, called observable instruments, are the reverse: the SDK calls a function registered by the application whenever it collects metrics, and that function reports the current value. They are the right tool for state that exists independently of any single event — queue lengths, pool sizes, cache occupancy, values read from the operating system. This article covers when to use them, the three types, and how to write callbacks that neither block collection nor report stale values. It belongs to the OpenTelemetry metrics SDK in the Python metrics and instrumentation section.

Prerequisites

pip install "opentelemetry-sdk>=1.27.0,<2.0.0"

A meter provider with a metric reader, as in recording counters and histograms with OpenTelemetry.

Implementation steps

Step 1 — Decide: event or state? A request completing, a message being processed and a cache miss are events; they happen at a moment in code, and a synchronous counter or histogram records them there. The number of open connections, the length of a queue and the bytes held by a cache are states; they exist continuously and can be read at any time. Updating a synchronous up-down counter every time a connection opens or closes works, and reading the pool's size once per collection is simpler, cheaper and cannot drift if an update is missed.

Step 2 — Choose the observable type.

The distinction between the gauge and the up-down counter is subtle. Both report a current value. The up-down counter's values are additive across attributes — pool sizes for three pools can be summed into a total — while a gauge's may not be — averaging CPU temperatures is meaningful, summing them is not. Choose by whether a total across attributes means something.

Step 3 — Write the callback. A callback receives CallbackOptions and returns an iterable of Observation, each with a value and optional attributes.

from opentelemetry import metrics
from opentelemetry.metrics import CallbackOptions, Observation

meter = metrics.get_meter("orders-api")

def observe_pool(options: CallbackOptions):
    for name, pool in POOLS.items():                  # a small, known set
        yield Observation(pool.size(), {"pool": name, "state": "total"})
        yield Observation(pool.checkedout(), {"pool": name, "state": "in_use"})

meter.create_observable_up_down_counter(
    "db.client.connection.count", callbacks=[observe_pool],
    unit="{connection}", description="Database connections by pool and state")

Step 4 — Report every combination, every time. An attribute set that appears in one collection and not the next is treated as having no value in the second. For a Prometheus exporter that means the series disappears; for cumulative OTLP it means a gap. Yielding every known combination on every call — including zeros — keeps series continuous.

Step 5 — Keep callbacks fast and safe. The callback runs on the reader's collection thread, while all other instruments wait. It should read values that are already in memory and return. A callback that makes a network call or takes a lock contended by request handlers delays every metric's export and can stall the application.

def observe_queue(options: CallbackOptions):
    try:
        yield Observation(len(WORK_QUEUE), {"queue": "ingest"})    # O(1), no I/O
    except Exception:                                                 # never let it raise
        return

State owned by an event loop

Asynchronous services keep much of their state inside the event loop — the number of tasks, the backlog of an asyncio.Queue, the lag measured by a monitoring coroutine. Callbacks cannot await, and they run on the reader's thread, not the loop's. The pattern is to have the loop publish plain values that the callback reads.

import asyncio, time

class LoopStats:
    lag_seconds: float = 0.0
    tasks: int = 0

STATS = LoopStats()

async def monitor_loop(interval: float = 0.5):
    while True:
        t0 = time.perf_counter()
        await asyncio.sleep(interval)
        STATS.lag_seconds = max(0.0, time.perf_counter() - t0 - interval)
        STATS.tasks = len(asyncio.all_tasks())

meter.create_observable_gauge(
    "asyncio.event_loop.lag", unit="s",
    callbacks=[lambda o: [Observation(STATS.lag_seconds)]])

Assigning a float or an int to an attribute is atomic in CPython, so the callback reads a consistent value without a lock. Measuring asyncio event loop lag develops the measurement itself.

Reading from the operating system

Process-level state — resident memory, open file descriptors, CPU time — is the classic use for observable instruments, because the operating system maintains the value and the callback only reads it.

import os, resource, psutil

PROC = psutil.Process(os.getpid())

def observe_process(options):
    mem = PROC.memory_info()
    yield Observation(mem.rss, {"type": "rss"})
    yield Observation(PROC.num_fds(), {"type": "open_fds"})

meter.create_observable_gauge("process.memory.usage", unit="By", callbacks=[observe_process])
meter.create_observable_counter(
    "process.cpu.time", unit="s",
    callbacks=[lambda o: [Observation(sum(PROC.cpu_times()[:2]))]])

The CPU time is an observable counter because it is a monotonic total; the backend computes a rate. The OpenTelemetry system-metrics instrumentation package provides a ready-made set of these, and writing them by hand is useful when only a few are needed or when process-per-worker servers require the values to be tagged per worker.

What a sampled value can and cannot show

An observable gauge reports one value per collection. Whatever happened between collections is invisible, and that shapes which questions the metric can answer.

For slowly changing state — pool configuration, cache size, memory — a sample per interval is plenty. For bursty state — queue lengths, concurrent requests, lag — a sample can miss exactly the moments that matter. The high-water-mark pattern keeps the callback's simplicity: the application updates a maximum as values change, a cheap comparison, and the callback reports the maximum and resets it. The reported value then answers "how bad did it get in this interval", which is usually the question during an incident.

Resetting on read has one subtlety. With more than one reader — a Prometheus reader and a periodic OTLP reader on the same provider — each collection resets the maximum, so each reader sees only the peak since the other one read. A single reader per provider, or a separate maximum per reader, avoids it.

Collection timing also varies. A periodic reader collects on its interval; a Prometheus reader collects whenever it is scraped, and two Prometheus servers scraping the same target collect twice as often. Callbacks should not assume a fixed interval — rates computed inside a callback from the time since the last call are fragile. Report totals or current values and let the backend compute rates.

Observable instruments in multi-process servers

Under Gunicorn or a Celery prefork pool, each worker has its own meter provider and its own callbacks, and each reports its own view of state. For per-process state — that worker's memory, its event-loop lag — that is correct, and a worker identity in the resource attributes keeps the series apart. For shared state — the length of a Redis queue every worker can see — every worker reports the same value, and a query that sums across workers multiplies it by the worker count. Shared state is better observed once, from one process or a separate exporter, as in collecting metrics from Celery workers.

Configuration options

Verification

Collect with an in-memory reader in a test:

from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import InMemoryMetricReader

reader = InMemoryMetricReader()
provider = MeterProvider(metric_readers=[reader])
meter = provider.get_meter("test")
meter.create_observable_gauge("queue.length", callbacks=[lambda o: [Observation(7, {"queue": "ingest"})]])

data = reader.get_metrics_data()
point = data.resource_metrics[0].scope_metrics[0].metrics[0].data.data_points[0]
assert point.value == 7 and point.attributes == {"queue": "ingest"}

Expected Output: the test passes, and changing the callback to yield nothing makes the metric disappear from get_metrics_data, which demonstrates why every combination must be reported every time.

Common mistakes

Synchronous gauges updated on every change. Error signature: a pool-size gauge that drifts from the real size after an exception path skips a decrement. Root cause: event-driven tracking of state. Remediation: an observable instrument that reads the state.

I/O in the callback. Error signature: metric exports arriving late or timing out. Root cause: the reader thread waiting on the network. Remediation: read cached values updated elsewhere.

Omitting zero values. Error signature: series that appear and vanish, and dashboards with gaps. Root cause: combinations not reported every collection. Remediation: yield every known combination.

Observable counter for a gauge value. Error signature: negative or nonsensical rates. Root cause: a non-monotonic value reported as a counter. Remediation: an observable gauge.

Unbounded attributes from the callback. Error signature: series count growing with users or tenants. Root cause: iterating over request-driven collections. Remediation: aggregate before observing.

Shared state observed by every worker. Error signature: a queue length four times the real value on the dashboard. Root cause: each worker reporting the same shared value, then summed. Remediation: observe shared state from one place.

Frequently Asked Questions