Subscribing to gNMI Telemetry Streams in Python #

The specific operational failure this page addresses is blind time. A NOC that polls interface counters every thirty seconds is, on average, fifteen seconds behind reality — and during a fiber cut those fifteen seconds are the difference between an alarm that fires while the on-call engineer is still at their desk and one that fires after the customer has already opened a ticket. A gNMI Subscribe RPC removes that gap by pushing each change the instant it happens, but only if the client drains the stream correctly. A subscription that blocks its own event loop on a slow downstream write will fall behind the device, and a device that outruns its collector will either buffer without bound or have its stream torn down — both of which reintroduce exactly the blind time the subscription was meant to remove. Getting the drain loop right is what keeps detection latency in single-digit milliseconds and holds this contribution to mean-time-to-acknowledge near zero.

Schema alignment and taxonomy anchor #

This page is the transport-level procedure beneath Streaming Collector Ingestion, the stage that owns the streaming data plane within the Ingestion & Parsing Workflows pipeline. That parent stage owns the producer-consumer architecture, the normalizer pool, and the publish path; this page owns the narrower job of opening a Subscribe RPC, choosing the right subscription mode, decoding each protobuf notification, and handing raw updates across the bounded-queue boundary without blocking. Every update that leaves this loop is destined for the canonical contract defined in Event Schema Design, so the decode step must preserve the device identity, the leaf path, and the nanosecond timestamp intact for the normalizer that follows.

SAMPLE versus ON_CHANGE #

gNMI offers two subscription modes that solve two different problems, and choosing wrongly is the most common source of either missed transitions or self-inflicted storms.

• ON_CHANGE streams a notification only when a leaf’s value actually changes. It is the correct mode for state — oper-status, BGP session state, admin-status — where every transition matters and the value is otherwise stable. It is bandwidth-frugal but silent on a healthy path, which is why a heartbeat is mandatory to distinguish a quiet path from a dead session.
• SAMPLE streams a leaf’s current value at a fixed interval regardless of change. It is the correct mode for counters and gauges — byte counts, optical input power, error tallies — where the trend over time is the signal and there is no discrete transition to catch. The interval is a direct cost knob: a one-second sample on ten thousand paths is ten thousand notifications per second whether or not anything moved.

Diagram: ON_CHANGE emits only on transitions and relies on heartbeats to prove liveness, while SAMPLE emits every interval regardless of change.

A production subscription usually mixes both: SAMPLE on the counter paths that need trending, ON_CHANGE on the state paths where a single transition is the whole event, and a heartbeat on the ON_CHANGE paths so a silent session is caught within one interval.

Building and draining the subscription #

The loop below builds a SubscribeRequest with a mix of SAMPLE and ON_CHANGE paths, opens the stream, and drains it into a bounded queue. The gNMI protobuf types are consumed through a thin typed view so the example runs without vendor stubs; in production the stub.Subscribe call comes from the generated gNMI gRPC service. The critical property is that the drain loop does only cheap work — decode and enqueue — and the queue.put is the single point where backpressure is allowed to pause the read.

import asyncio
from datetime import datetime, timezone
from enum import Enum
from typing import Any, AsyncIterator

from pydantic import BaseModel, ConfigDict, Field


class SubMode(str, Enum):
    ON_CHANGE = "ON_CHANGE"
    SAMPLE = "SAMPLE"


class RawUpdate(BaseModel):
    """Decoded view of one gNMI Update leaf handed to the normalizer."""
    model_config = ConfigDict(frozen=True)

    device_id: str
    path: str
    value: Any
    timestamp_ns: int = Field(gt=0)
    event_time: datetime


# Subscription plan: SAMPLE the counters, watch state ON_CHANGE.
SUBSCRIPTION_PLAN: dict[str, SubMode] = {
    "/interfaces/interface/state/counters/in-errors": SubMode.SAMPLE,
    "/components/component/optical-channel/state/input-power": SubMode.SAMPLE,
    "/interfaces/interface/state/oper-status": SubMode.ON_CHANGE,
}


def build_subscribe_request(plan: dict[str, SubMode], *, sample_ns: int, heartbeat_ns: int) -> Any:
    """Assemble a gnmi.SubscribeRequest with a STREAM subscription list.

    Each path gets its mode; SAMPLE paths carry a sample_interval, ON_CHANGE
    paths carry a heartbeat_interval so a quiet leaf still proves liveness.
    """
    subscriptions = []
    for path, mode in plan.items():
        sub = {"path": path, "mode": mode.value}
        if mode is SubMode.SAMPLE:
            sub["sample_interval"] = sample_ns
        else:
            sub["heartbeat_interval"] = heartbeat_ns
        subscriptions.append(sub)
    # STREAM mode keeps the RPC open and pushes updates as they occur.
    return {"subscribe": {"mode": "STREAM", "subscription": subscriptions}}


async def drain_subscription(
    stub: Any,
    device_id: str,
    queue: asyncio.Queue[RawUpdate],
    *,
    sample_s: float = 10.0,
    heartbeat_s: float = 30.0,
) -> None:
    """Open a gNMI SUBSCRIBE stream and drain it into the bounded queue."""
    request = build_subscribe_request(
        SUBSCRIPTION_PLAN,
        sample_ns=int(sample_s * 1e9),
        heartbeat_ns=int(heartbeat_s * 1e9),
    )
    stream: AsyncIterator[Any] = stub.Subscribe(_single_request(request))
    async for response in stream:
        notification = response.update             # gnmi.Notification or None
        if notification is None:                   # sync_response / heartbeat only
            continue
        base_ns = notification.timestamp
        prefix = _join_path(notification.prefix)
        for update in notification.update:         # repeated gnmi.Update
            raw = RawUpdate(
                device_id=device_id,
                path=prefix + _join_path(update.path),
                value=_scalar(update.val),         # unwrap gnmi.TypedValue
                timestamp_ns=base_ns,
                event_time=datetime.fromtimestamp(base_ns / 1e9, tz=timezone.utc),
            )
            # The ONLY place the loop is allowed to pause. A full queue blocks
            # here, stops the socket read, and lets HTTP/2 flow control pause
            # the device instead of buffering without bound.
            await queue.put(raw)


async def _single_request(request: dict[str, Any]) -> AsyncIterator[dict[str, Any]]:
    # gNMI Subscribe is client-streaming; a STREAM subscription sends exactly
    # one SubscribeRequest and then reads responses until teardown.
    yield request


def _join_path(path: Any) -> str:
    if path is None:
        return ""
    parts = []
    for elem in getattr(path, "elem", []):
        keys = "".join(f"[{k}={v}]" for k, v in getattr(elem, "key", {}).items())
        parts.append(f"{elem.name}{keys}")
    return "/" + "/".join(parts) if parts else ""


def _scalar(typed_value: Any) -> Any:
    for field in ("json_val", "string_val", "int_val", "uint_val", "bool_val"):
        val = getattr(typed_value, field, None)
        if val is not None:
            return val
    return None

Async ingestion hook #

This loop is a producer. It is started once per device and handed the shared bounded queue that the normalizer pool drains, exactly as wired in the parent Streaming Collector Ingestion implementation. Because the only awaited call inside the drain loop is queue.put, the coroutine yields cleanly to the event loop between updates, so a single-threaded runtime can hold thousands of concurrent device streams. The decode work — path flattening and TypedValue unwrapping — is pure CPU and cheap, so it stays inline; nothing here needs a worker thread. The moment heavier work appears (a taxonomy lookup, a hash, a schema construction) it belongs to the normalizer on the other side of the queue, never in this loop, so a slow map lookup can never stall the socket read.

async def run_device(stub: Any, device_id: str, queue: asyncio.Queue) -> None:
    """Supervise one device stream with reconnect backoff and jitter."""
    import random
    backoff = 1.0
    while True:
        try:
            await drain_subscription(stub, device_id, queue)
        except Exception:                          # stream teardown / transport error
            # Never let ten thousand collectors reconnect in lockstep after a
            # controller blip; jittered exponential backoff spreads the retry.
            await asyncio.sleep(backoff + random.uniform(0, backoff))
            backoff = min(backoff * 2, 30.0)
        else:
            backoff = 1.0

Mitigation and hardening #

When a subscription misbehaves, contain it without disturbing healthy streams. These are the concrete failure paths a production deployment should implement.

1. Heartbeat-gated liveness. An ON_CHANGE path on a healthy device is silent, and a dead TCP session is also silent — indistinguishable without a heartbeat. Set heartbeat_interval on every ON_CHANGE subscription and track stream_last_message_age_seconds; when it exceeds the interval, tear down and reconnect that one stream rather than restarting the collector.
2. Bounded-queue backpressure, not buffering. Never drain into an unbounded structure. The bounded queue.put is the safety valve: a storm fills it, the read pauses, HTTP/2 flow control pauses the device, and no frame is lost to an out-of-memory kill. An unbounded queue simply moves the storm from the wire into the heap.
3. Jittered reconnect. A controller restart drops every session at once. Exponential backoff with jitter, capped at 30 seconds, spreads the reconnect so the collectors do not become a synchronized thundering herd against a device that is still recovering.
4. Decode isolation. A single leaf with an unexpected TypedValue encoding after a firmware upgrade must not abort the stream. Wrap the per-update decode so a bad leaf is dead-lettered with its raw bytes while the rest of the notification proceeds, keeping the session alive.

Operational hardening notes #

Tuning this loop is about keeping the per-update cost flat and the drain ahead of the wire. Keep the drain loop free of any awaited I/O other than queue.put; a stray await on a lookup or a log flush turns the socket read into a bottleneck the moment that call slows. Prefer a small number of SAMPLE paths at aggressive intervals over a large number at fine intervals — sample cost is the product of path count and rate, and it is the easiest way to accidentally build a self-inflicted storm. Where the device supports it, enable ON_CHANGE dampening so a flapping interface coalesces its toggles at the source before they ever hit the queue. Sized this way, a single collector coroutine per device sustains line-rate counter subscriptions while holding decode-and-enqueue latency in the low hundreds of microseconds, and the subscription contributes single-digit-millisecond detection latency to the platform’s MTTA budget. The gNMI messages themselves are consumed through generated gRPC stubs whose async client patterns follow the non-blocking model in the official Python asyncio documentation.

Frequently Asked Questions #

When should I use SAMPLE versus ON_CHANGE? #

Use ON_CHANGE for state leaves like oper-status or BGP session state, where every transition is the event and the value is otherwise stable, and always pair it with a heartbeat so a quiet path is distinguishable from a dead session. Use SAMPLE for counters and gauges like byte counts or optical power, where the trend over the interval is the signal and there is no discrete transition to catch. Mixing both in one subscription is normal.

Why does a full bounded queue not lose telemetry? #

Because the drain loop blocks on the queue put instead of dropping the update. Blocking there stops the socket read, and HTTP 2 flow control then stops advertising receive window to the device, so the device pauses sending. The backpressure propagates all the way to the source, and no frame is buffered without bound or dropped in memory. An unbounded queue would instead move the storm into the heap and risk an out-of-memory kill.

How do I detect a gNMI session that has silently died? #

Set a heartbeat interval on every ON_CHANGE subscription and track the age of the last message received per device. A healthy but idle ON_CHANGE path is silent, so without a heartbeat a dead session looks identical to a quiet one. When the last-message age exceeds the heartbeat interval, tear down and reconnect that single stream with jittered backoff rather than restarting the whole collector.

Should path-to-fault mapping happen inside the subscribe loop? #

No. The drain loop should only decode the protobuf and enqueue a raw update, keeping its only awaited call the queue put. Taxonomy lookups, hashing, and schema construction belong to the normalizer workers on the other side of the bounded queue, so a slow lookup can never stall the socket read and reintroduce the blind time the subscription exists to remove.

Related #

• Up to the parent reference: Streaming Collector Ingestion
• The wider data plane it feeds: Ingestion & Parsing Workflows
• Downstream grouping and deduplication: Async Batch Processing
• The event contract each update is normalized into: Event Schema Design