netring 0.25.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

//! Flow-record export — the NetFlow/IPFIX/`conn.log` output shape.
//!
//! Emits one [`FlowRecord`](netring::export::FlowRecord) per completed flow to
//! an exporter, plus **interim records for long-lived flows** on an active
//! timeout (0.25 W1c) — the standard NetFlow behaviour where a flow alive
//! longer than the timeout gets periodic snapshots, not just one record at the
//! end.
//!
//! ```sh
//! cargo run --example monitor_flow_export --features monitor -- eth0
//! # NDJSON flow records on stdout; pipe into Vector / Loki / jq:
//! #   cargo run ... | jq 'select(.proto=="Tcp")'
//! ```
//!
//! For IPFIX/NetFlow v10 wire output to a collector, swap `JsonFlowExporter`
//! for `IpfixExporter::new(socket, observation_domain_id)` (feature `ipfix`).

use std::time::Duration;

use netring::export::JsonFlowExporter;
use netring::monitor::Monitor;
use netring::protocol::builtin::Tcp;

#[tokio::main]
async fn main() -> Result<(), netring::Error> {
    let iface = std::env::args().nth(1).unwrap_or_else(|| "lo".to_string());

    let monitor = Monitor::builder()
        .interface(&iface)
        .protocol::<Tcp>()
        // One newline-delimited JSON record per completed flow.
        .export_flows(JsonFlowExporter::stdout())
        // 0.25 W1c: also emit an interim record (reason == null / is_ongoing)
        // every 30s for flows still alive — so a long download/stream shows up
        // before it finishes.
        .export_active_timeout(Duration::from_secs(30))
        // A second exporter: a closure is a FlowExporter too. Tag the ongoing
        // ones so the difference is visible on stdout.
        .export_flows(|rec: &netring::export::FlowRecord| {
            if rec.is_ongoing() {
                eprintln!(
                    "# ongoing  {:?} {} ↔ {} : {} bytes so far",
                    rec.proto,
                    rec.a,
                    rec.b,
                    rec.total_bytes()
                );
            }
        })
        .build()?;

    println!("# exporting TCP flow records on {iface} (Ctrl-C to stop)");
    monitor.run_until_signal().await
}