sgnmon

Monitoring and metrics for SGN streaming pipelines

Observe SGN pipelines from the outside: collect Prometheus metrics (gap/non-gap content, latencies, throughput, end-of-stream state) from running pipelines, serve them over HTTP, and evaluate pipeline health checks suitable for container orchestration and monitoring plugins.

Resources

• Documentation
• Source Code
• Issue Tracker

Installation

pip install sgnmon

Features

• Non-invasive taps -- attach probes to any existing pipeline's pads without modifying the graph, at three observation points per element (frames produced, frames received, and post-audioadapter consumption) so latency can be attributed per element
• Monitoring elements -- a pass-through MonitorTransform and terminal MonitorSink for explicit in-graph instrumentation
• Prometheus metrics -- frame counts, gap/data seconds, sample throughput, latency histograms, per-pad execution time, freshness timestamps, EOS state, and opt-in process CPU/memory metrics
• Custom metrics -- declare application metrics on your own elements (MetricSpec/MetricsMixin) or create families directly on the monitor; an EventLatency element measures per-event latency for event streams
• Windowed statistics -- rates, gap fractions, and latencies aggregated natively over a 5s...6h trailing-window ladder, for sub-minute resolution and consumers without PromQL
• Health checks -- a configuration-free default set (data freshness, throughput, latency drift) that self-paces its startup, plus opt-in absolute-latency, gap-fraction, and end-of-stream checks, aggregated into a JSON report
• Built-in webserver -- /metrics, /health (HTTP 503 when unhealthy), /status, /graph, and live updates over server-sent events, served from a background thread
• Live dashboard -- a self-contained web page drawing the pipeline graph with live per-element rates, latencies, gap fractions, and health state; composed elements expand in place to show their internal elements
• CLI -- sgnmon run -- <app> ... attaches monitoring to any sgn application with zero code changes; sgnmon demo runs a monitored example pipeline; sgnmon check <url> queries a running pipeline's health with plugin-friendly exit codes

Quickstart

Monitor any sgn application, zero code changes

$ sgnmon run --port 9090 -- sgn-cal --config config.yaml --model model.npz
[sgnmon] dashboard: http://localhost:9090/

Tap an existing pipeline

from sgnmon import Monitor, MonitorServer

monitor = Monitor()
monitor.tap(pipeline)  # observe every pad, no graph changes

with MonitorServer(monitor, port=9090):
    pipeline.run()

Default health checks (data freshness, throughput, latency drift) are registered out of the box; tune or extend them with monitor.add_default_checks(...). Then point Prometheus at http://<host>:9090/metrics, or check health by hand:

$ sgnmon check http://localhost:9090

Instrument with elements

from sgnmon import Monitor, MonitorTransform

monitor = Monitor()
mon = MonitorTransform(name="mon", sink_pad_names=["H1"], monitor=monitor)
pipeline.insert(src, mon, snk)
pipeline.link({
    "mon:snk:H1": "src:src:H1",
    "snk:snk:H1": "mon:src:H1",
})

Try the demo

$ sgnmon demo --port 9090
dashboard: http://localhost:9090/
metrics:  http://localhost:9090/metrics
health:   http://localhost:9090/health
status:   http://localhost:9090/status

Open the dashboard in a browser to watch the pipeline graph update live.