kumo_server_common/
log_backlog.rs

1//! Expose, as metrics, how far each out-of-process log consumer trails the
2//! newest data in the directories to which this node writes logs.
3//!
4//! A directory holding more than one log stream, such as several writers with
5//! distinct suffixes sharing one directory, is unsupported: the scan counts
6//! every segment matching the pattern, regardless of which writer produced it,
7//! so a consumer reading only one of the streams is reported against the
8//! combined backlog of all of them.
9
10pub use kumo_jsonl::WriterLocation;
11use kumo_prometheus::declare_metric;
12use std::collections::{HashMap, HashSet};
13use std::sync::atomic::{AtomicU64, Ordering};
14use std::sync::{LazyLock, Mutex, Once};
15use std::time::Duration;
16
17/// How often the monitor rescans every registered location, in milliseconds.
18static SCAN_INTERVAL_MS: AtomicU64 = AtomicU64::new(30_000);
19
20/// Set how often the monitor rescans registered locations, effective on its
21/// next cycle.
22pub fn set_scan_interval(interval: Duration) {
23    SCAN_INTERVAL_MS.store(interval.as_millis() as u64, Ordering::Relaxed);
24}
25
26fn scan_interval() -> Duration {
27    Duration::from_millis(SCAN_INTERVAL_MS.load(Ordering::Relaxed))
28}
29
30static LOCATIONS: LazyLock<Mutex<HashSet<WriterLocation>>> = LazyLock::new(Default::default);
31static MONITOR: Once = Once::new();
32
33declare_metric! {
34/// The number of whole log segments not yet processed by a consumer.
35///
36/// Labelled by:
37/// * `log_dir` - the log directory being scanned
38/// * `checkpoint_name` - the name set by the tailer of the consumer for its
39///   checkpoint.
40///
41/// A sample value is
42/// `log_consumer_segments_behind{log_dir="/var/log/kumomta",checkpoint_name="cp"} 3`.
43static SEGMENTS_BEHIND: IntGaugeVec(
44        "log_consumer_segments_behind",
45        &["log_dir", "checkpoint_name"]
46    );
47}
48
49declare_metric! {
50/// The compressed size, in bytes, of whole log segments awaiting processing by
51/// a consumer.
52///
53/// This counts whole unread segments only, not the remainder of the segment
54/// that the consumer is partway through processing.
55///
56/// Labelled by:
57/// * `log_dir` - the log directory being scanned
58/// * `checkpoint_name` - the name set by the tailer of the consumer for its
59///   checkpoint.
60static BYTES_BEHIND: IntGaugeVec(
61        "log_consumer_bytes_behind",
62        &["log_dir", "checkpoint_name"]
63    );
64}
65
66declare_metric! {
67/// Age in seconds of the oldest log segment not yet started by a consumer,
68/// counted from the modification time of that segment.
69///
70/// Such a segment does not exist once the consumer has reached the newest log
71/// segment. The gauge reports 0 in that case.
72///
73/// Labelled by:
74///
75/// * `log_dir` - the log directory being scanned
76/// * `checkpoint_name` - the name set by the tailer of the consumer for its
77///   checkpoint.
78static LAG_SECONDS: IntGaugeVec(
79        "log_consumer_lag_seconds",
80        &["log_dir", "checkpoint_name"]
81    );
82}
83
84/// Add `location` to the monitored set and start the monitor if it is not
85/// already running. Registering the same location more than once, including
86/// when a configuration reload re-registers the same directory, has the same
87/// effect as registering it once. A location is monitored for the lifetime of
88/// the process. A location cannot be deregistered.
89pub fn register_location(location: WriterLocation) {
90    LOCATIONS
91        .lock()
92        .expect("log backlog locations mutex poisoned")
93        .insert(location);
94
95    start_monitor();
96}
97
98/// Start the background monitor thread if it is not already running.
99pub fn start_monitor() {
100    MONITOR.call_once(|| {
101        // Allow tests to override the scan interval. Invalid values are
102        // silently ignored.
103        if let Some(interval) = std::env::var("KUMOD_LOG_BACKLOG_SCAN_INTERVAL")
104            .ok()
105            .and_then(|v| v.parse::<f64>().ok())
106            .and_then(|v| Duration::try_from_secs_f64(v).ok())
107        {
108            set_scan_interval(interval);
109        }
110        std::thread::Builder::new()
111            .name("log-backlog-monitor".to_string())
112            .spawn(monitor_thread)
113            .expect("failed to spawn log-backlog-monitor thread");
114    });
115}
116
117/// Returns the current set of locations that require backlog monitoring.
118fn monitored_locations() -> HashSet<WriterLocation> {
119    let mut locations: HashSet<WriterLocation> = LOCATIONS
120        .lock()
121        .expect("log backlog locations mutex poisoned")
122        .clone();
123    locations.extend(kumo_jsonl::active_writer_locations());
124    locations
125}
126
127/// Publish the backlog gauges for `location` and return the set of checkpoint
128/// names reported by it. Returns `None` on a scan error.
129fn scan_location(location: &WriterLocation) -> Option<HashSet<String>> {
130    let reports = match kumo_jsonl::scan_backlog(&location.directory, &location.pattern) {
131        Ok(reports) => reports,
132        Err(err) => {
133            tracing::error!("scanning log backlog in {}: {err:#}", location.directory);
134            return None;
135        }
136    };
137
138    let dir = location.directory.as_str();
139    let mut checkpoint_names = HashSet::new();
140    for report in reports {
141        let labels = [dir, report.checkpoint_name.as_str()];
142        if let Ok(gauge) = SEGMENTS_BEHIND.get_metric_with_label_values(&labels) {
143            gauge.set(report.segments_behind as i64);
144        }
145        if let Ok(gauge) = BYTES_BEHIND.get_metric_with_label_values(&labels) {
146            gauge.set(report.bytes_behind as i64);
147        }
148        if let Ok(gauge) = LAG_SECONDS.get_metric_with_label_values(&labels) {
149            gauge.set(report.lag.map(|d| d.as_secs()).unwrap_or(0) as i64);
150        }
151        checkpoint_names.insert(report.checkpoint_name);
152    }
153    Some(checkpoint_names)
154}
155
156/// Remove the gauges for a consumer whose checkpoint has disappeared since the
157/// previous scan to avoid leaving a stale reading behind.
158fn forget_checkpoint(dir: &str, checkpoint_name: &str) {
159    let labels = [dir, checkpoint_name];
160    let _ = SEGMENTS_BEHIND.remove_label_values(&labels);
161    let _ = BYTES_BEHIND.remove_label_values(&labels);
162    let _ = LAG_SECONDS.remove_label_values(&labels);
163}
164
165fn monitor_thread() {
166    let mut previous: HashMap<WriterLocation, HashSet<String>> = HashMap::new();
167    loop {
168        let locations = monitored_locations();
169
170        previous.retain(|location, names| {
171            if locations.contains(location) {
172                return true;
173            }
174            // Clear gauges for inactive locations because Prometheus retains
175            // their last values otherwise.
176            for name in names.iter() {
177                forget_checkpoint(location.directory.as_str(), name);
178            }
179            false
180        });
181
182        for location in &locations {
183            let Some(current) = scan_location(location) else {
184                continue;
185            };
186            if let Some(prior) = previous.get(location) {
187                for gone in prior.difference(&current) {
188                    forget_checkpoint(location.directory.as_str(), gone);
189                }
190            }
191            previous.insert(location.clone(), current);
192        }
193        std::thread::sleep(scan_interval());
194    }
195}
kumo_server_common/log_backlog.rs

kumo_server_common/
log_backlog.rs
