Building EPCIS Ingestion Dashboards with Python

An ingestion pipeline that silently falls behind is worse than one that crashes loudly — a crash pages someone, but a consumer that quietly accumulates ten minutes of lag behind live packaging-line traffic can let a commission event sit unindexed while a verification request for that same serial is already in flight. This guide is a code-first companion within Real-Time Monitoring & Alerting — the area covering how the ingestion side of Serialization Data Ingestion & EPCIS Event Sync proves, in near real time, that it is keeping pace with the packaging line. It shows how to pick the handful of metrics that actually predict a compliance-relevant failure, expose them from an async consumer with the Prometheus Python client, and lay out the Grafana panels that turn those numbers into something an on-call engineer can act on in seconds.

Prerequisites #

• Python 3.10+ for the match statement and X | Y union types used below.
• prometheus_client ≥ 0.20 — the official Python instrumentation library, giving you Counter, Gauge, and Histogram primitives plus a text-format exposition helper.
• aiohttp ≥ 3.9 — to run the async consumer and a lightweight /metrics HTTP endpoint in the same event loop, so scraping never competes with a separate WSGI process for the same data.
• A running Prometheus server and Grafana instance — this guide assumes both are already deployed and reachable; it focuses on what the pipeline exposes and how the panels are structured, not on installing the observability stack itself.
• DSCSA data prerequisites — an ingestion consumer already producing typed records per the four EPCIS event types (ObjectEvent, AggregationEvent, TransactionEvent, TransformationEvent), each carrying eventTime, a business step, and a trading-partner identifier, ideally the same consumer described in Building async batch processors for serialization events.

Step-by-Step Solution #

Step 1 — Define the metrics that actually predict a compliance failure #

Resist the urge to instrument everything the consumer touches. Five signals map directly onto DSCSA-relevant failure modes: throughput by event type, ingestion lag against wall clock, dead-letter queue depth, the commission-to-verify gap, and per-partner error rate. Declare each as a module-level metric object so every part of the consumer shares one registry.

from prometheus_client import Counter, Gauge, Histogram

EVENTS_TOTAL = Counter(
    "epcis_events_total",
    "EPCIS events successfully ingested",
    ["event_type", "partner_id"],
)

INGESTION_LAG_SECONDS = Histogram(
    "epcis_ingestion_lag_seconds",
    "Seconds between EPCIS eventTime and the moment the record was committed",
    buckets=(0.5, 1, 2, 5, 10, 30, 60, 120, 300, 600, 1800),
)

DLQ_DEPTH = Gauge(
    "epcis_dlq_depth",
    "Current number of messages sitting in the dead-letter queue",
)

COMMISSION_VERIFY_GAP = Gauge(
    "epcis_commission_verify_gap",
    "Serials commissioned but not yet observed by a verification query",
)

PARTNER_ERRORS_TOTAL = Counter(
    "epcis_partner_errors_total",
    "Ingestion errors attributable to a specific trading partner",
    ["partner_id", "error_type"],
)

DSCSA/GS1 note: labeling EVENTS_TOTAL by event_type keeps ObjectEvent, AggregationEvent, TransactionEvent, and TransformationEvent throughput separately visible, because a drop in one event type alone (say, aggregations stalling while commissioning continues) is exactly the kind of partial failure that corrupts a packaging hierarchy without ever showing up in an aggregate events/sec number.

Step 2 — Instrument the async consumer’s hot path #

Every metric update happens where the event is already being handled — never in a side channel that could drift out of sync with what was actually committed. The lag histogram converts the EPCIS eventTime to UTC before subtracting, since a naive comparison against local wall clock silently fabricates skew.

import asyncio
from datetime import datetime, timezone

async def handle_event(record: dict, repo, partner_id: str) -> None:
    """Commit one parsed EPCIS event and record its observability signals."""
    event_time = record["event_time"]
    if event_time.tzinfo is None:
        event_time = event_time.replace(tzinfo=timezone.utc)

    await repo.commit_event(record)

    EVENTS_TOTAL.labels(event_type=record["event_type"], partner_id=partner_id).inc()
    lag = (datetime.now(timezone.utc) - event_time.astimezone(timezone.utc)).total_seconds()
    INGESTION_LAG_SECONDS.observe(max(lag, 0.0))


async def consume(stream, repo, dlq) -> None:
    async for raw in stream:
        try:
            await handle_event(raw.record, repo, raw.partner_id)
        except Exception as exc:
            error_type = type(exc).__name__
            PARTNER_ERRORS_TOTAL.labels(partner_id=raw.partner_id, error_type=error_type).inc()
            await dlq.put({"payload": raw.record, "error": error_type, "offset": raw.offset})

DSCSA/GS1 note: incrementing EVENTS_TOTAL only after repo.commit_event succeeds means the counter reflects records that actually earned a place in the six-year audit trail, not merely records the consumer attempted — a distinction an auditor will ask about the first time throughput and repository row counts disagree.

Step 3 — Track dead-letter depth and the commission-verify gap as background gauges #

Counter fits things that only go up; DLQ depth and the commission-verify gap are levels, not totals, so they belong on a Gauge refreshed by a periodic background task rather than updated inline with every event. This keeps the hot path free of extra queries.

async def refresh_level_metrics(dlq, repo, interval_seconds: int = 15) -> None:
    """Poll queue depth and the commission/verify gap on a fixed cadence."""
    while True:
        DLQ_DEPTH.set(await dlq.qsize())
        gap = await repo.count_commissioned_not_verified()
        COMMISSION_VERIFY_GAP.set(gap)
        await asyncio.sleep(interval_seconds)

DSCSA/GS1 note: count_commissioned_not_verified should query for ObjectEvent commissioning records with no matching verification query response within a rolling window — a persistently rising gap is an early signal of exactly the kind of dispenser-side confusion that Suspect Product Investigation Workflows exist to resolve, and it is the metric an alert rule in configuring SLA alerts for serialization pipelines should watch first. Sustained DLQ growth deserves the same scrutiny, and is the signal that drives draining dead-letter queues safely.

Step 4 — Expose a /metrics endpoint alongside the consumer #

Run the HTTP endpoint and the consumer loop in the same process with asyncio.gather, so a single deployable artifact both ingests events and answers scrapes. generate_latest renders the current registry state in the Prometheus text exposition format.

from aiohttp import web
from prometheus_client import CONTENT_TYPE_LATEST, generate_latest

async def metrics_handler(_request: web.Request) -> web.Response:
    return web.Response(body=generate_latest(), content_type=CONTENT_TYPE_LATEST)


def build_metrics_app() -> web.Application:
    app = web.Application()
    app.router.add_get("/metrics", metrics_handler)
    return app


async def run_pipeline(stream, repo, dlq) -> None:
    runner = web.AppRunner(build_metrics_app())
    await runner.setup()
    site = web.TCPSite(runner, "0.0.0.0", 9100)
    await site.start()

    await asyncio.gather(
        consume(stream, repo, dlq),
        refresh_level_metrics(dlq, repo),
    )

DSCSA/GS1 note: binding /metrics on a dedicated port (9100 here, following the Prometheus exporter convention) keeps observability traffic separate from any partner-facing verification API, so a scrape storm can never compete with a time-sensitive DSCSA verification response.

Step 5 — Structure the Grafana panels around the metric families #

Each metric family maps to one panel type and one PromQL query. Building the dashboard from this table keeps every panel traceable back to the code that emits it, which matters when a compliance reviewer asks how a graph on a screen relates to a specific EPCIS field.

DSCSA/GS1 note: using rate() rather than the raw counter value means every panel survives a consumer restart gracefully — Counter values reset to zero on process restart, and rate() correctly handles that reset instead of rendering a misleading negative spike.

Step 6 — Guard label cardinality before it reaches the time-series database #

partner_id and error_type are useful dimensions, but an unbounded or attacker-influenced label value can multiply time series explosively. Validate labels against a known partner registry before they ever reach a .labels() call.

KNOWN_PARTNERS: set[str] = set()  # populated from the trading-partner registry


def safe_partner_label(partner_id: str) -> str:
    """Collapse any unregistered or malformed partner id to a bounded label."""
    return partner_id if partner_id in KNOWN_PARTNERS else "unregistered"

DSCSA/GS1 note: this bound matters as much operationally as it does for observability cost — a partner GLN that has not completed onboarding under Trading Partner Verification & Onboarding should never silently mint its own label series, since that would make a data-quality problem look like normal traffic growth on the dashboard.

Verification #

Confirm the instrumentation before trusting the dashboard in an incident. A unit test against the registry catches label typos and counter-vs-gauge mistakes cheaply:

import pytest
from prometheus_client import REGISTRY

def test_events_total_increments():
    before = REGISTRY.get_sample_value(
        "epcis_events_total", {"event_type": "ObjectEvent", "partner_id": "TESTGLN"}
    ) or 0.0
    EVENTS_TOTAL.labels(event_type="ObjectEvent", partner_id="TESTGLN").inc()
    after = REGISTRY.get_sample_value(
        "epcis_events_total", {"event_type": "ObjectEvent", "partner_id": "TESTGLN"}
    )
    assert after == before + 1.0

For an end-to-end check, start the pipeline against a staging stream, run curl http://localhost:9100/metrics, and confirm epcis_events_total, epcis_ingestion_lag_seconds_bucket, epcis_dlq_depth, epcis_commission_verify_gap, and epcis_partner_errors_total are all present with the expected label sets. Finally, feed a synthetic malformed event through the consumer and verify both that it lands in the dead-letter queue and that epcis_partner_errors_total incremented for the correct partner — a mismatch between the two means the exception handler and the DLQ write are not sharing the same code path.

Gotchas & Edge Cases #

• Counters reset on restart. A raw epcis_events_total value dropping to zero after a deploy is expected, not an outage — always graph with rate() or increase(), never the raw counter, or every restart will look like a traffic collapse.
• Unbounded label cardinality. partner_id and error_type labels multiply; an unvalidated partner identifier or an exception message used as a label value can create thousands of unique time series and degrade the whole Prometheus instance, not just this dashboard.
• UTC vs. local eventTime. Computing ingestion lag against a naive or local-zone timestamp fabricates drift; always convert eventTime to UTC before subtracting, the same discipline required when parsing EPCIS XML with lxml.
• Histogram buckets tuned for the wrong regime. Buckets sized for sub-second lag will saturate into the +Inf bucket the moment a partner replays a backlog after an outage, hiding exactly the spike you need to see; size buckets for both steady-state and recovery scenarios.
• Level metrics updated on the hot path. Setting DLQ_DEPTH or COMMISSION_VERIFY_GAP inline with every event adds a query per event; keep those on the periodic background task in Step 3, not inside handle_event.

Related #

• Up to the parent topic: Real-Time Monitoring & Alerting
• Configuring SLA alerts for serialization pipelines — turning these metrics into paging rules
• Draining dead-letter queues safely — what to do once epcis_dlq_depth is rising
• Serialization Data Ingestion & EPCIS Event Sync — the pipeline this dashboard observes