netring 0.26.0

High-performance zero-copy packet I/O for Linux (AF_PACKET TPACKET_V3 + AF_XDP)
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77

//! 0.25 W1f (C5): structured **JSON logging** of a Monitor's anomalies and
//! capture telemetry via `tracing-subscriber`'s JSON formatter — the
//! newline-delimited JSON shape Vector / Loki / Elasticsearch ingest directly,
//! with no custom netring sink.
//!
//! Two sources of structured events:
//! - **Anomalies** flow through [`TracingSink`], which re-emits each one as a
//!   `tracing` event at the matching level — so the JSON subscriber renders it.
//! - **Per-flow** + **capture-telemetry** events are emitted by hand with
//!   `tracing::info!` and structured fields, so they share the same JSON
//!   pipeline as the anomalies.
//!
//! Run (Ctrl-C to stop; `RUST_LOG` tunes the level):
//!
//! ```sh
//! cargo run --example monitor_tracing_json --features monitor -- eth0
//! # each line is a JSON object: {"timestamp":..,"level":"INFO","fields":{..},..}
//! # pipe into a collector:  ... | vector --config vector.toml
//! ```

use std::time::Duration;

use netring::monitor::Monitor;
use netring::prelude::TracingSink;
use netring::protocol::builtin::Tcp;
use netring::protocol::event_typed::FlowEnded;

#[tokio::main]
async fn main() -> Result<(), netring::Error> {
    // One JSON object per log line on stdout. `RUST_LOG` (env-filter) controls
    // the level; default to `info` so the demo prints without configuration.
    let filter = tracing_subscriber::EnvFilter::try_from_default_env()
        .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info"));
    tracing_subscriber::fmt()
        .json()
        .with_env_filter(filter)
        .with_current_span(false)
        .init();

    let iface = std::env::args().nth(1).unwrap_or_else(|| "lo".to_string());

    let monitor = Monitor::builder()
        .interface(&iface)
        .name("tracing-json-demo")
        .protocol::<Tcp>()
        // Anomalies → `tracing` events at their severity level (JSON-rendered).
        .sink(TracingSink::default())
        // One structured JSON event per completed TCP flow.
        .on::<FlowEnded<Tcp>>(|f: &FlowEnded<Tcp>| {
            tracing::info!(
                target: "netring.flow",
                src = %f.key.a,
                dst = %f.key.b,
                packets = f.stats.packets_initiator + f.stats.packets_responder,
                bytes = f.stats.bytes_initiator + f.stats.bytes_responder,
                reason = ?f.reason,
                "tcp flow ended"
            );
            Ok(())
        })
        // Capture telemetry every 2s → one structured JSON event per source.
        .on_capture_stats(Duration::from_secs(2), |t, _ctx| {
            tracing::info!(
                target: "netring.telemetry",
                source = t.source.0,
                packets = t.packets,
                drops = t.drops,
                freezes = t.freezes,
                "capture telemetry"
            );
            Ok(())
        })
        .build()?;

    println!("# logging TCP flows + telemetry as JSON on {iface} (Ctrl-C to stop)");
    monitor.run_until_signal().await
}