# netring monitor — primitives by use case
A one-page tour of the `Monitor` toolkit, grouped by what you're trying
to do. Everything here is in `netring::prelude::*` unless noted. Mirrors
flowscope's `docs/discoverability.md` on the flow-tracking side.
## Build a monitor
```rust
use netring::prelude::*;
Monitor::builder().interface("eth0") /* … */ .build()?.run_until_signal().await?;
```
- `.interface(s)` / `.interfaces([…])` — capture source(s).
- `.all_l4()` — register Tcp + Udp + Icmp in one call (no "forgot Icmp"
foot-gun). `.all_l7()` — Http + Dns + Tls + TlsHandshake (feature-gated).
- `.protocol::<P>()` — register one protocol explicitly.
- `.name(s)` — label surfaced on `ctx.monitor_name`.
- `.label_table(t)` — custom well-known port → app-label table.
- Run: `.run_for(d)` / `.run_until(deadline)` / `.run_until_signal()` /
`.run_until_idle(window)` / `.replay()` (pcap).
## React to protocol roles (0.22 R1)
The type system enforces which events a protocol can produce:
- **`FlowProtocol`** (`Tcp`, `Udp`, `Icmp`) — emit lifecycle events
`FlowStarted` / `FlowEstablished` / `FlowEnded` / `FlowTick<P>` and the
flat `FlowPacket`. `on::<FlowStarted<Tcp>>(…)`.
- **`MessageProtocol`** (`Http`, `Dns`, `Tls`, `TlsHandshake`, `Icmp`) —
deliver parsed messages. `on::<Http>(|msg, ctx| …)`.
- `on::<Tcp>` and `FlowStarted<Http>` are **compile errors** — the roles
make invalid combinations unrepresentable.
## Handle events
- `.on::<E>(|payload| …)` — payload only.
- `.on_ctx::<E>(|payload, ctx| …)` — payload + `&mut Ctx`.
- `.on_async::<E>(handler)` — async handler.
- `.tick(period, |tick, ctx| …)` — periodic.
- `.detect(…)` / `detector!` / `pattern_detector!` — scored detectors.
## Per-packet & per-flow data (0.22 R2)
- `FlowPacket { proto, key, side, len, tcp, ts }` — one flat event for
every L4 packet; branch on `evt.proto`.
- `FlowTick<P>` — periodic per-flow `FlowStats` snapshot.
- `FlowStats` throughput helpers: `throughput_bps[_for(side)]`,
`throughput_pps[_for]`, `bytes_for`, `pkts_for`, `direction_skew`
(safe-divide; zero-duration → `0.0`).
## Bandwidth by application (0.22 §2.3)
```rust
Ok(())
})
```
- `bandwidth_by_app()` / `bandwidth_windowed(window, bucket)` — register
the primitive; read via `ctx.bandwidth() -> Option<BandwidthReport>`.
- `on_bandwidth(period, f)` — fused: auto-register + periodic report.
- `BandwidthReport`: `top(n)`, `rate(app)`, `total()`, `app_count()` —
no `Timestamp`/`RollingRate`/`Option` at the call site.
## ICMP triage (0.22 §2.4/2.5)
```rust
ctx.emit("Icmp", Severity::Warning).with("kind", err.kind.as_str()).emit();
Ok(())
})
```
- `IcmpError { family, kind, correlated_flow, stats, ts }` — unified
v4/v6, pre-classified, with the originating flow joined.
- `IcmpErrorKind`: `DestUnreachable(DestUnreachableKind)` / `TimeExceeded`
/ `ParameterProblem` / `MtuSignal(MtuSignalKind)`; `.as_str()`.
- `ctx.lookup_icmp_flow(inner)` — manual inner-tuple → flow + stats join.
## TCP resets (0.22 §2.6)
- `.on_tcp_reset(|rst, ctx| …)` — `TcpRst { key, stats, ts, zero_payload }`,
synthesised from `FlowEnded<Tcp>` with `reason == Rst`. `zero_payload`
separates "connection refused" from a mid-transfer abort.
## Sliding-window state (`netring::correlate`)
| `RollingRate<K, V>` | per-key per-second rate | bandwidth, request rates |
| `TimeBucketedCounter<K>` | per-key sliding count | bursts, storms |
| `TimeBucketedSet` | per-key distinct set | cardinality (port scans) |
| `TopK` | top-N by score | talkers, domains |
| `Ewma` | smoothed average | latency, jitter |
| `BurstDetector` | burst onset | beacons, floods |
| `KeyIndexed<K, V>` | TTL map (`iter_fresh` / `drain_expired`) | "expected B after A" |
| `FlowStateMap` | per-flow state, auto-evict | per-conversation accumulators |
## Emit anomalies
- `ctx.emit(kind, severity).with(k, v).with_metric(k, f).with_key(&key).emit()`.
- Sinks: `StdoutSink`, `StdoutJsonSink` (serde), `TracingSink`,
`ChannelSink`, `EveSink` (eve-sink), `MetricsSink` (metrics).
- Layers: `MinSeverity` (`info()`/`warning()`/`error()`, all `const`),
`DedupeAnomalies`, `RateLimitAnomalies`, `Sample`, `Tee` — stack via
`.layer(…)` (first = outermost).
## Periodic reports (0.22 §3)
A third output stream beside per-event anomalies and broadcast streams:
periodic typed snapshots of derived state (the Suricata `stats.log` /
Zeek `conn.log` shape).
- `.report(period, |snap| …)` — ad-hoc closure; `snap` exposes
`bandwidth()` / `state::<T>()` / `counter::<K>()` / `emit(…)` / `now()`.
- `.report_to(period, |snap| R, sink)` — ship a typed `Report` to a
`ReportSink`. Shipped sinks: `StdoutReportSink`, `JsonReportSink`
(serde, newline-JSON). `BandwidthSnapshot` is the reference `Report`
(`bw.to_snapshot(n)`).
- Example: `examples/monitor/report_stream.rs`.
## Observability surface (0.29)
Purpose-built periodic reports, each armed with one builder call plus an
`on_*(period, |report| …)` callback (and a serializable snapshot):
| Traffic aggregation (#121) | `.on_aggregate(p, …)` | top talkers / host-pair matrix / top domains / top SNI | `traffic_aggregation` |
| RED metrics (#122) | `.on_red(p, …)` | rate / error rate / duration quantiles per `RedProto` | `red_metrics` |
| Owner bandwidth (#130) | `.with_flow_attribution(hook)` + `.on_owner_bandwidth(p, …)` | throughput by tenant/container + unattributed bucket | `owner_bandwidth` |
**Bandwidth, by grouping** — pick the key that matches the question:
- **by app protocol** (`http`/`https`/`dns`…): `bandwidth_by_app()` / `on_bandwidth` — port-derived label.
- **by session (5-tuple)**: no dedicated builder — `FlowEnded.stats` gives cumulative bytes + duration per connection for free; for a rolling top-N of *active* sessions feed a `BandwidthByKey<FlowKey>` on `FlowPacket`. Example: `session_bandwidth`.
- **by owner / tenant / container**: `owner_bandwidth` above.
- **by OS process (pid)**: not a core API — passive capture sees no process identity. The `process_bandwidth` example resolves it host-side from `/proc/net` + `/proc/<pid>/fd` and feeds `with_flow_attribution` (local-host, best-effort — see the example's caveats).
## DNS visibility & names (0.29)
- `.on_dns(|view, ctx| …)` — structured DNS exchanges (client/server split,
qname, rcode, community id). `.name_map()` + `.on_name(…)` learn a passive
IP→name map; `ctx.names(ip)` resolves inside any handler.
- `.on_encrypted_dns(…)` — flag DoH/DoT/DoQ sessions + resolver.
- Examples: `dns_monitor`, `encrypted_dns`.
## Fingerprints & anti-evasion (0.29)
- `.on_ssh_fingerprint(…)` (HASSH), `.on_quic_fingerprint(…)`,
`TlsFingerprint.{app_protocol, pq_key_share}` — see FINGERPRINTS.md.
- `.reassemble_ip_fragments()` — reassemble IPv4 fragments before parsing
(defeats fragmentation evasion; example `ip_defrag`).
## Capture placement & recording (0.29)
- `NetNs::from_name(..)` + `CaptureBuilder::netns(&ns)` — capture inside a
container's network namespace (example `netns_capture`).
- `RotatingPcapWriter` + `TriggeredPcapWriter` (feature `pcap`) — rotating
capture with a pre-trigger ring (example `triggered_pcap`).
## Stream consumers
- `.with_broadcast::<P>()` + `monitor.subscribe::<P>()` →
`EventStream<P::Message>` (`futures_core::Stream`). Both bounded to
`MessageProtocol`.
## Scale out
- `ShardedRunner::new(iface, mode, group_id, n, build_shard)` — per-CPU
AF_PACKET fanout, one Monitor per shard.
- **Cross-shard aggregation (0.22 §5.1):** `.merge_state(period, fold)` /
`.state_auto_merge::<T: AddAssign>(period)` fold each shard's `T` slot
into a global total; observe it with `.on_merge(|total| …)`. Per-shard
state stays local (no hot-path locking); a merge-worker thread probes.
- **Per-shard layers:** `.layer(spec)` where `spec: LayerSpec` —
cloneable config layers pass directly; non-`Clone` layers (`Tee`) via
`LayerFactory(|| …)`. Example: `examples/monitor/sharded_runner.rs`.