dial9 0.5.0

Low-overhead async runtime telemetry: event recording, Tokio integration, CPU/memory profiling, and a trace viewer CLI
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
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124

//! On-trigger pipeline runs: time windows and concurrent dumps.
//!
//! Companion to `on_trigger_dump.rs`, which covers the minimal case
//! (`dump_current_data` plus a debounce gate). This example exercises the rest
//! of the `DumpTrigger` surface:
//!
//!   * `dump_time_range(lookback, lookforward)` with a pure look-back window
//!     (capture the history the ring still holds, no forward window), and with
//!     a look-forward window (keep the dump open and capture segments as they
//!     seal after the trigger; the await only resolves once the forward
//!     deadline elapses).
//!   * Two overlapping dumps running concurrently: each gets its own `DumpId`,
//!     and a segment whose `[creation, seal]` span falls inside both windows is
//!     captured by both. Off S3 (here) each dump just resolves its own
//!     receipt; against S3 the shared object key would land in both dumps'
//!     manifests.
//!
//! Disk + `gzip` + `write_back`, no AWS setup. Dumped segments land as
//! `*.bin.gz` in the trace dir.
//!
//!   cargo run -p dial9-tokio-telemetry --example on_trigger_dump_windows
//!
//! Inspect a dumped segment afterwards:
//!   gunzip /tmp/dial9-on-trigger-windows/trace.0.bin.gz

use std::time::Duration;

use dial9::Dial9Handle;
use dial9::DiskBuffer;

const TRACE_DIR: &str = "/tmp/dial9-on-trigger-windows";

#[dial9::main(config = || {
    use dial9::{Dial9HandleTokioExt, RecorderPipelineExt, TokioAttachOptions};
    let _ = std::fs::remove_dir_all(TRACE_DIR);
    let _ = std::fs::create_dir_all(TRACE_DIR);
    let writer = DiskBuffer::builder()
        .base_path(TRACE_DIR)
        // Fast-rotating writer so the demo seals a segment every ~half second.
        .max_file_size(4 * 1024)
        .max_total_size(10 * 1024 * 1024)
        .rotation_period(Duration::from_millis(500))
        .build()
        .expect("open trace writer");
    let recorder = dial9::recorder(writer)
        // No debounce here: we want the two concurrent dumps to each register,
        // not fold into one another.
        .with_custom_pipeline(|p| p.gzip().write_back())
        .with_dump_trigger(|_| {})
        .build();

    let mut builder = tokio::runtime::Builder::new_multi_thread();
    builder.enable_all().worker_threads(2);

    let runtime = recorder
        .handle()
        .attach_tokio_runtime(builder, TokioAttachOptions::default())?;
    Ok((recorder, runtime))
})]
async fn main() {
    // Reach the dump trigger through the ambient handle; the runtime stashed
    // it when `with_dump_trigger` was configured.
    let trigger = Dial9Handle::current()
        .dump_trigger()
        .expect("on-demand mode enabled");

    // Steady workload so the ring keeps sealing segments for the whole demo.
    // The pipeline stays parked: nothing is gzipped or written back until a
    // dump asks for data.
    for id in 0..8 {
        dial9::spawn(async move {
            for _ in 0..800 {
                tokio::time::sleep(Duration::from_millis(25)).await;
                std::hint::black_box(id);
            }
        });
    }

    // Let some history accumulate in the ring before the first dump.
    tokio::time::sleep(Duration::from_secs(3)).await;

    // 1) Pure look-back: everything the ring retained over the last hour, no
    //    forward window. Resolves as soon as the matching ring segments finish
    //    the pipeline.
    let receipt = trigger
        .dump_time_range(Duration::from_secs(3600), Duration::ZERO)
        .with_metadata("reason", "look-back")
        .await
        .expect("look-back dump");
    println!(
        "look-back dump {}: {} segment(s), span {:?}",
        receipt.dump_id, receipt.segments_processed, receipt.time_range,
    );

    // 2) Look-forward: a short look-back plus a 3s forward window. The dump
    //    stays open and captures segments as they seal; the await resolves
    //    only after the forward deadline elapses.
    let receipt = trigger
        .dump_time_range(Duration::from_secs(1), Duration::from_secs(3))
        .with_metadata("reason", "look-forward")
        .await
        .expect("look-forward dump");
    println!(
        "look-forward dump {}: {} segment(s), span {:?}",
        receipt.dump_id, receipt.segments_processed, receipt.time_range,
    );

    // 3) Two overlapping dumps at once. Both carry a forward window, so they
    //    stay open concurrently; a segment sealing inside both windows is
    //    captured by each. They get distinct ids and resolve independently.
    let (a, b) = tokio::join!(
        trigger.dump_time_range(Duration::from_secs(1), Duration::from_secs(3)),
        trigger.dump_time_range(Duration::from_secs(1), Duration::from_secs(3)),
    );
    let a = a.expect("concurrent dump a");
    let b = b.expect("concurrent dump b");
    assert_ne!(a.dump_id, b.dump_id, "concurrent dumps get distinct ids");
    println!(
        "concurrent dumps {} ({} seg) and {} ({} seg) ran independently",
        a.dump_id, a.segments_processed, b.dump_id, b.segments_processed,
    );

    println!("processed to disk: run `ls {TRACE_DIR}/*.bin.gz`");
}