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
//! Basic Monitor example — single-protocol flow lifecycle.
//!
//! Counts TCP flows started + ended on the given interface. Emits
//! one anomaly per flow start to stdout via `StdoutSink`.
//!
//! Runs on the default multi-thread tokio runtime — `Monitor` is
//! `Send` as of 0.21 (flowscope 0.13's `Driver<E>: Send + Sync`):
//!
//! ```sh
//! cargo run --example monitor_basic --features "tokio,flow" -- eth0
//! ```
//!
//! `lo` works for smoke-testing — start `ping -c 1 127.0.0.1`
//! while the example is running and you should see one
//! `FlowStarted<Tcp>` per outbound TCP attempt (none for ICMP echo).
//! Use `monitor_async_handler` if you also want ICMP lifecycle.
use std::time::Duration;
use netring::prelude::*;
#[derive(Default)]
struct FlowCounters {
started: u64,
ended: u64,
}
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
let iface = std::env::args().nth(1).unwrap_or_else(|| "lo".into());
let dur_secs: u64 = std::env::args()
.nth(2)
.and_then(|s| s.parse().ok())
.unwrap_or(30);
eprintln!("monitor_basic: capturing on {iface} for {dur_secs}s");
Monitor::builder()
.interface(&iface)
.protocol::<Tcp>() // FlowStarted/Ended<Tcp> lifecycle events
.state::<FlowCounters>()
.on_ctx::<FlowStarted<Tcp>>(|_evt: &FlowStarted<Tcp>, ctx: &mut Ctx<'_>| {
// 0.21 A.3: `split_state_sink::<T>()` projects two
// disjoint borrows in one step — no manual borrow-
// shortening scope needed. 0.21 A.2: `ctx.emit(kind,
// sev)` is the one-line shortcut for the begin chain;
// it captures `ctx.ts` automatically.
let ts = ctx.ts;
let (counters, sink) = ctx.split_state_sink::<FlowCounters>();
counters.started += 1;
sink.begin("FlowStartedTcp", Severity::Info, ts)
.with_metric("started_total", counters.started as f64)
.emit();
Ok(())
})
.on_ctx::<FlowEnded<Tcp>>(|_evt: &FlowEnded<Tcp>, ctx: &mut Ctx<'_>| {
ctx.state_mut::<FlowCounters>().ended += 1;
Ok(())
})
.sink(StdoutSink::default())
.build()?
.run_for(Duration::from_secs(dur_secs))
.await?;
eprintln!("monitor_basic: done");
Ok(())
}