flyby 0.1.1

A high-performance Rust framework for composable data-ingestion pipelines.
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

//! Runtime telemetry keys and lifecycle event hooks.

use crate::api::{MetricKey, MetricsCollector};

/// Metric keys emitted by the runtime / pipeline driver.
#[derive(Debug, Clone, Copy)]
pub enum RuntimeMetricKey {
    /// Pipeline steps completed.
    Steps,
    /// Messages successfully written to sinks.
    MessagesOut,
    /// Messages dropped by back-pressure policy.
    MessagesDropped,
    /// Back-pressure events observed.
    BackpressureEvents,
    /// Decode / preprocess idle skips.
    IdleSkips,
    /// Wall-time nanoseconds for one step (histogram).
    StepDurationNs,
    /// Current runtime phase as a gauge (ordinal).
    Phase,
    /// Scheduler worker count.
    Workers,
}

impl MetricKey for RuntimeMetricKey {
    fn name(&self) -> &str {
        match self {
            Self::Steps => "runtime.steps",
            Self::MessagesOut => "runtime.messages_out",
            Self::MessagesDropped => "runtime.messages_dropped",
            Self::BackpressureEvents => "runtime.backpressure_events",
            Self::IdleSkips => "runtime.idle_skips",
            Self::StepDurationNs => "runtime.step_duration_ns",
            Self::Phase => "runtime.phase",
            Self::Workers => "runtime.workers",
        }
    }
}

/// Lifecycle / scheduling events for logs and future OpenTelemetry spans.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum RuntimeEvent {
    /// Runtime constructed.
    Built,
    /// Configuration validated.
    Validated,
    /// Pipeline and sinks initialized.
    Initialized,
    /// Sources started / run loop entered.
    Started,
    /// Draining remaining pending work after source exhaustion.
    Draining,
    /// Shutdown requested or completed.
    Shutdown,
    /// Resources cleaned up.
    Cleanup,
    /// Scheduler tick / step completed.
    Step {
        /// Step counter.
        n: u64,
    },
}

/// Emit a lifecycle counter when metrics are enabled.
pub fn emit_lifecycle(metrics: &dyn MetricsCollector, enabled: bool, event: &RuntimeEvent) {
    if !enabled {
        return;
    }
    let key = match event {
        RuntimeEvent::Built => "runtime.event.built",
        RuntimeEvent::Validated => "runtime.event.validated",
        RuntimeEvent::Initialized => "runtime.event.initialized",
        RuntimeEvent::Started => "runtime.event.started",
        RuntimeEvent::Draining => "runtime.event.draining",
        RuntimeEvent::Shutdown => "runtime.event.shutdown",
        RuntimeEvent::Cleanup => "runtime.event.cleanup",
        RuntimeEvent::Step { .. } => "runtime.event.step",
    };
    metrics.record_counter(&StrKey(key), 1);
}

#[derive(Debug)]
struct StrKey(&'static str);
impl MetricKey for StrKey {
    fn name(&self) -> &str {
        self.0
    }
}