Streaming Collector Ingestion: gNMI and gRPC Dial-Out Telemetry at Line Rate #

Streaming collector ingestion is the always-on data plane that lives inside the Ingestion & Parsing Workflows pipeline, sitting where the legacy poll-and-scrape model gives way to persistent, push-based telemetry. Instead of walking SNMP tables on a timer, this stage holds open long-lived gNMI and gRPC dial-out subscriptions to routers, optical transport nodes, and virtualized network functions, and consumes a continuous flow of protobuf-encoded notifications the moment a counter crosses a threshold or an operational-state leaf changes. Its single job is to turn that firehose of vendor-shaped protobuf into normalized fault events that conform to the canonical contract defined in Event Schema Design, without ever letting a fast-talking device outrun the correlation tier behind it.

At carrier scale the difference matters. A modern platform can push interface counters, optical power levels, and BGP session state at sub-second cadence across tens of thousands of subscription paths. Polling that same surface would add seconds of blind time to every fault, and every second of blind time is a direct line item in mean-time-to-acknowledge. By subscribing instead of polling, this stage compresses the detection window from the poll interval down to the transport latency of a single notification — typically single-digit milliseconds — while keeping the ingestion host from drowning when a fiber cut makes ten thousand paths change state at once.

Operational Intent and Boundary #

What enters this stage is a set of authenticated, long-lived transport streams: gNMI Subscribe RPCs that the collector dials into a device, and gRPC dial-out sessions where the device itself initiates the connection and streams unsolicited. What exits is a sequence of validated fault-event objects, each carrying the mandatory contract fields — device_id, path, severity, event_time in UTC, and a raw_payload_hash for deduplication and audit — published onto the internal event bus for correlation. Everything in between is stream lifecycle management, protobuf decode, path-to-field mapping, and type coercion. Nothing more.

What is deliberately excluded is as important as what is included. This stage performs no root-cause analysis, no cross-event topology grouping, and no severity arbitration between concurrent notifications; those judgments belong to the rule tier. It does not own duplicate suppression across time windows — that is the responsibility of Async Batch Processing downstream. And it does not shape inbound volume by policy; sustained-rate protection is owned by Rate Limiting Strategies at the edge. This stage owns exactly one transformation boundary: raw protobuf notification in, one canonical event out, one undecodable frame diverted to a dead-letter queue. Keeping that boundary sharp is what lets the collector fleet scale by adding replicas, each owning a disjoint slice of the subscription space, with no shared mutable state to coordinate.

The child procedure for the transport itself — opening the Subscribe RPC, choosing SAMPLE versus ON_CHANGE mode, and draining notifications without blocking — is documented in Subscribing to gNMI Telemetry Streams in Python. This page covers how that transport slots into the wider async data plane and how its output reaches the bus.

Pipeline Architecture and Data Flow #

The collector is structured as a classic producer-consumer split joined by a bounded queue. On one side, a pool of subscription tasks each own a single device stream; they decode protobuf notifications and hand raw update tuples to the queue. On the other side, a pool of normalizer workers drain the queue, map each update path onto the canonical schema, validate it, and publish. The bounded queue between them is the backpressure boundary: when normalizers fall behind, the queue fills, and a full queue applies backpressure directly to the subscription tasks — which in gRPC terms means the collector simply stops reading from the socket, and HTTP/2 flow control pushes the pause back to the device. No frame is dropped in memory; the pressure propagates to the source.

Diagram: subscription tasks decode protobuf and feed a bounded queue that normalizer workers drain into the canonical schema, with HTTP/2 flow control carrying backpressure back to the devices and undecodable frames branching to a dead-letter queue.

This split is what keeps the event loop responsive under a storm. Protobuf decode is cheap and stays on the loop, but the moment a normalizer needs to do heavier work — a MIB-style path lookup, a schema construction, a hash — it is running on its own worker coroutine, decoupled from the socket read. A slow normalizer therefore slows the queue, not the wire, and the wire slows the device, and the whole chain degrades in an orderly, memory-bounded way rather than by exhausting the heap.

Production-Ready Async Collector #

The implementation below is the load-bearing shape of this stage: an async subscription task that drains a gNMI stream and hands raw updates to a bounded queue, and a normalizer worker that maps each update onto the canonical schema and publishes. The protobuf message types (SubscribeResponse, Notification, Update) come from the standard gNMI proto definitions; here they are consumed through a minimal typed view so the example stays runnable without vendor stubs. Validation uses Pydantic V2 in strict mode so a malformed update becomes a clean dead-letter entry rather than a corrupt event.

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

from pydantic import BaseModel, ConfigDict, Field, field_validator


class Severity(str, Enum):
    CRITICAL = "CRITICAL"
    MAJOR = "MAJOR"
    MINOR = "MINOR"
    WARNING = "WARNING"
    CLEARED = "CLEARED"


class FaultEvent(BaseModel):
    """Canonical fault-event contract emitted onto the bus."""
    # Strict mode rejects silent coercion; a bad update is dead-lettered,
    # never forwarded half-formed into correlation.
    model_config = ConfigDict(strict=True, extra="forbid", frozen=True)

    device_id: str
    path: str
    severity: Severity
    event_time: datetime
    raw_payload_hash: str = Field(min_length=64, max_length=64)

    @field_validator("event_time")
    @classmethod
    def require_utc(cls, value: datetime) -> datetime:
        # gNMI carries nanosecond epoch timestamps; the normalizer converts
        # them to timezone-aware UTC before construction. Naive values mean a
        # decode bug and must fail loudly rather than skew correlation windows.
        if value.tzinfo is None:
            raise ValueError("event_time must be timezone-aware UTC")
        return value.astimezone(timezone.utc)


# A path-to-severity map warmed from the fault taxonomy at load time. Keeping
# it in memory keeps the normalizer hot path to an O(1) dict lookup.
PATH_SEVERITY: dict[str, Severity] = {
    "/interfaces/interface/state/oper-status": Severity.MAJOR,
    "/components/component/optical-channel/state/input-power": Severity.MINOR,
    "/network-instances/network-instance/protocols/protocol/bgp/neighbors": Severity.CRITICAL,
}


class RawUpdate(BaseModel):
    """Decoded, still-untyped view of one gNMI Update leaf."""
    model_config = ConfigDict(frozen=True)

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


async def subscribe_task(
    stream: AsyncIterator[Any],
    device_id: str,
    queue: asyncio.Queue[RawUpdate],
) -> None:
    """Own one device stream: decode protobuf notifications, enqueue updates.

    The stream yields gNMI SubscribeResponse messages. We read the `update`
    Notification, flatten each Update leaf, and hand a RawUpdate to the queue.
    `queue.put` blocks when the queue is full, which stops us reading the
    socket and lets HTTP/2 flow control pause the device — backpressure, not
    frame loss.
    """
    async for response in stream:
        notification = response.update          # gnmi.Notification
        if notification is None:                # sync_response / heartbeat frame
            continue
        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=notification.timestamp,
            )
            await queue.put(raw)                # backpressure boundary


async def normalizer_worker(
    queue: asyncio.Queue[RawUpdate],
    publish: Callable[[FaultEvent], Any],
    dlq: asyncio.Queue[dict[str, Any]],
) -> None:
    """Drain raw updates, map onto the canonical schema, publish or dead-letter."""
    while True:
        raw = await queue.get()
        try:
            severity = PATH_SEVERITY.get(_strip_keys(raw.path))
            if severity is None:
                # Unknown path: emit nothing, but record it so the taxonomy
                # table gains coverage instead of silently swallowing signal.
                raise KeyError(f"unmapped path {raw.path}")
            event = FaultEvent(
                device_id=raw.device_id,
                path=raw.path,
                severity=severity,
                event_time=_ns_to_utc(raw.timestamp_ns),
                raw_payload_hash=hashlib.sha256(
                    f"{raw.device_id}{raw.path}{raw.value}".encode()
                ).hexdigest(),
            )
            await publish(event)               # single non-blocking bus write
        except Exception as exc:               # decode/schema/mapping failure
            await dlq.put({
                "device_id": raw.device_id,
                "path": raw.path,
                "reason": type(exc).__name__,
                "detail": str(exc),
            })
        finally:
            queue.task_done()


def _ns_to_utc(timestamp_ns: int) -> datetime:
    return datetime.fromtimestamp(timestamp_ns / 1_000_000_000, tz=timezone.utc)


def _strip_keys(path: str) -> str:
    # Collapse list keys like [name=Ethernet1] so the map keys on the schema path.
    out, depth = [], 0
    for ch in path:
        if ch == "[":
            depth += 1
        elif ch == "]":
            depth -= 1
        elif depth == 0:
            out.append(ch)
    return "".join(out)


def _join_path(path: Any) -> str:
    # Render a gnmi.Path (repeated PathElem) as a slash-joined string.
    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:
    # gnmi.TypedValue is a oneof; return whichever scalar field is set.
    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 def run(streams: dict[str, AsyncIterator[Any]], publish, *, maxsize: int = 20_000) -> None:
    """Wire subscription tasks and normalizer workers around one bounded queue."""
    queue: asyncio.Queue[RawUpdate] = asyncio.Queue(maxsize=maxsize)
    dlq: asyncio.Queue[dict[str, Any]] = asyncio.Queue(maxsize=maxsize)
    tasks = [
        asyncio.create_task(subscribe_task(stream, device_id, queue))
        for device_id, stream in streams.items()
    ]
    tasks += [
        asyncio.create_task(normalizer_worker(queue, publish, dlq))
        for _ in range(4)                      # scale normalizers to CPU budget
    ]
    await asyncio.gather(*tasks)

The raw_payload_hash is load-bearing exactly as it is elsewhere in the pipeline: it is the deduplication key that lets Async Batch Processing collapse a storm of identical oper-status flaps into one representative event with an occurrence count, and it is the audit anchor tying a normalized event back to the exact leaf update that produced it. Note that the normalizer never asserts topology or correlates across paths — an unmapped path is recorded for taxonomy coverage, not guessed, which keeps this stage stateless and horizontally scalable.

Topology and Schema Validation #

Because this is the first stage to produce a typed event from a stream, it enforces two narrow boundary checks and no more, so it never duplicates the deep validation owned downstream.

• Schema guard. The Pydantic model rejects any update that cannot populate the mandatory fields. A leaf that yields a naive timestamp, a malformed path, or a value that will not coerce raises a ValidationError; the worker catches it, tags the raw frame with a reason code, and diverts it to the dead-letter queue rather than emitting a half-formed event. This keeps the parser guarantees identical to those enforced by Logparser Integration for line-oriented feeds, so correlation sees one uniform contract regardless of whether an event began as a syslog line or a gNMI leaf.
• Path sanity. The normalizer carries a warmed map of known subscription paths keyed to the canonical schema. An update against an unknown path is not silently dropped — losing a genuine new leaf is worse than recording noise — but it is dead-lettered as unmapped_path so the taxonomy table can be extended. The stage never decides whether the path represents a real fault; that meaning is assigned by the correlation tier.

This split keeps ownership unambiguous: this stage guarantees shape and provenance, batching guarantees grouping and deduplication, and correlation guarantees meaning. No layer second-guesses another, and the same Event Schema Design contract governs the boundary in every case.

Configuration and Tuning Parameters #

The defaults below are starting points; tune per element class and re-measure against SLA targets.

Held to these defaults, a single collector replica sustains line-rate ingestion of high-frequency counters while holding P99 normalize-and-publish latency in the low tens of milliseconds and a post-decode schema-rejection rate below 0.1%.

Debugging Workflow and Observability #

Streaming failures must be isolated without tearing down healthy subscriptions. Work through this checklist when notification rate, latency, or drain depth regress.

1. Stream liveness — confirm each subscription is receiving heartbeats. A gauge of stream_last_message_age_seconds per device that exceeds the heartbeat interval means a session is silently dead and needs a reconnect, not a restart of the whole collector.
2. Queue depth — watch collector_queue_depth and queue_full_events_total. Sustained depth near maxsize means normalizers are under-provisioned or a downstream bus write is stalling; add normalizer workers or investigate the publish path before the backpressure reaches the devices.
3. Decode failures — every protobuf decode error is a counter, not just a log line. A burst of decode_error_total sharing one device_id usually points to a firmware change shifting a TypedValue encoding, traceable to the exact device.
4. Unmapped paths — gauge the share of updates dead-lettered as unmapped_path. A spike is the early signal that a device started streaming a subscription path the taxonomy table does not cover yet — extend the map before those events are lost to correlation.
5. Publish latency — histogram normalize_publish_latency_seconds. A rising P99 with flat queue depth points at the bus client, not the collector; a rising P99 with rising queue depth points at CPU saturation in the normalizer pool.

Expose these as standard metric types — counters for decode and dead-letter events, a histogram for publish latency, gauges for queue depth and stream age — and keep instrumentation lock-free on the hot path.

Failure Modes and Mitigation #

Streaming collectors fail in a small number of well-understood ways, each with a concrete containment strategy.

• Session death without notice. A device or controller can drop the TCP connection without a clean gRPC teardown, and an ON_CHANGE subscription on a quiet path will look identical to a healthy idle stream. The heartbeat interval defuses this: a missed keepalive triggers a reconnect with exponential backoff and jitter, so the fleet never reconnects in lockstep.
• Notification storm. A fiber cut can make every downstream interface change oper-status in the same instant, and dial-out devices will happily stream all of it. The bounded queue caps memory: it fills, backpressure pauses the streams via HTTP/2 flow control, and Async Batch Processing collapses the identical flaps by raw_payload_hash downstream. Sustained-rate abuse is shed by Rate Limiting Strategies before it reaches the queue at all.
• Protobuf schema drift. A firmware upgrade shifts a field encoding and decode fails on a subset of leaves. The dead-letter queue converts this into clean decode_error diversions carrying the raw bytes, so correlation never sees corrupt values and the offending device is identifiable in seconds.
• Graceful degradation. When the bus client stalls, a circuit breaker around the publish path opens and events spill to the dead-letter queue for replay rather than accumulating in memory and blocking the loop. CLEARED and informational leaves are shed before any MAJOR or CRITICAL under saturation, preserving the signal that matters.

Held to these mitigations, this stage delivers one deterministic, schema-compliant event per meaningful state change, cutting detection latency from the poll interval to single-digit milliseconds while keeping the ingestion host memory-bounded during the exact storms that matter most.

Frequently Asked Questions #

Why subscribe with gNMI instead of polling SNMP on a timer? Polling adds the full poll interval of blind time to every fault, because a change that happens just after a poll is not seen until the next one. A streaming subscription pushes the change the moment it happens, compressing detection latency from seconds to the single-digit-millisecond transport time of one notification. That reduction lands directly in mean-time-to-acknowledge, and it removes the load spike of walking large tables on every cycle.

How does backpressure reach the device without dropping frames? The collector puts decoded updates onto a bounded queue. When normalizers fall behind, the queue fills and the subscription task blocks on put, which stops it reading the socket. HTTP/2 flow control then stops advertising receive window to the device, so the device pauses sending. No frame is dropped in memory; the pressure propagates all the way back to the source and the whole chain degrades in a memory-bounded way.

What happens to a notification the collector cannot decode? It is diverted to the dead-letter queue carrying its raw bytes, a reason code, and the originating path, rather than being dropped or forwarded half-formed. A burst of decode failures sharing one device usually means a firmware upgrade shifted a field encoding, which the dead-letter queue makes visible and traceable to the exact device within seconds.

Does this stage decide severity or correlate events? No. It assigns only a baseline severity from a single leaf update in isolation, using a warmed path-to-severity map. Cross-event severity arbitration, topology grouping, and root-cause analysis are owned by the correlation tier, and duplicate suppression is owned by async batching. Keeping this stage stateless is what lets it scale by adding replicas that each own a disjoint slice of the subscription space.

Related #

• Up to the parent reference: Ingestion & Parsing Workflows
• The transport procedure it depends on: Subscribing to gNMI Telemetry Streams in Python
• Downstream grouping and deduplication: Async Batch Processing
• Edge volume protection: Rate Limiting Strategies
• Line-oriented normalization alongside this stage: Logparser Integration
• The event contract this stage emits: Event Schema Design