saluki_components/destinations/dsd_debug_log/
mod.rs

1use std::{io::Write, path::PathBuf};
2
3use agent_data_plane_config::Live;
4use async_trait::async_trait;
5use chrono::{DateTime, Utc};
6use saluki_common::collections::FastHashMap;
7use saluki_core::{
8    accounting::{MemoryBounds, MemoryBoundsBuilder},
9    components::{
10        destinations::{Destination, DestinationBuilder, DestinationContext},
11        BuildContext,
12    },
13    data_model::{
14        event::{metric::Metric, Event, EventType},
15        tags::TagSet,
16    },
17};
18use saluki_error::{generic_error, GenericError};
19use stringtheory::MetaString;
20use tokio::select;
21use tracing::{debug, warn};
22use tracing_appender::non_blocking::{NonBlocking, NonBlockingBuilder, WorkerGuard};
23use tracing_rolling_file::{RollingConditionBase, RollingFileAppenderBase};
24
25const DEBUG_LOG_WRITER_BUFFER_LINES: usize = 4096;
26
27/// Configuration for the DogStatsD debug log destination.
28pub struct DogStatsDDebugLogConfiguration {
29    /// Whether DogStatsD metric-level statistics are enabled.
30    ///
31    /// The destination drops metrics while this runtime setting is `false`.
32    pub metrics_stats_enabled: Live<bool>,
33
34    /// Path to the DogStatsD debug log file.
35    pub log_file: PathBuf,
36
37    /// Maximum size of the active debug log file before rotation, in bytes.
38    pub log_file_max_size: u64,
39
40    /// Number of rotated debug log files to keep.
41    pub log_file_max_rolls: usize,
42}
43
44/// DogStatsD destination that writes metric debug lines to a rotating file.
45struct DogStatsDDebugLog {
46    log_file: PathBuf,
47    log_file_max_size: u64,
48    log_file_max_rolls: usize,
49    writer: Option<DebugLogWriter>,
50    metrics_stats_enabled: Live<bool>,
51    stats: FastHashMap<ContextNoOrigin, MetricSample>,
52}
53
54struct DebugLogWriter {
55    writer: NonBlocking,
56    _guard: WorkerGuard,
57}
58
59#[derive(Debug, Default)]
60struct MetricSample {
61    count: u64,
62    last_seen: u64,
63}
64
65#[derive(Eq, Hash, PartialEq)]
66struct ContextNoOrigin {
67    name: MetaString,
68    tags: TagSet,
69}
70
71impl DogStatsDDebugLog {
72    fn new(config: &DogStatsDDebugLogConfiguration) -> Result<Self, GenericError> {
73        let mut destination = Self {
74            log_file: config.log_file.clone(),
75            log_file_max_size: config.log_file_max_size,
76            log_file_max_rolls: config.log_file_max_rolls,
77            writer: None,
78            metrics_stats_enabled: config.metrics_stats_enabled.clone(),
79            stats: FastHashMap::default(),
80        };
81
82        if *destination.metrics_stats_enabled {
83            destination.ensure_writer()?;
84        }
85
86        Ok(destination)
87    }
88
89    fn process_metric(&mut self, metric: &Metric) -> Result<(), GenericError> {
90        if !*self.metrics_stats_enabled {
91            return Ok(());
92        }
93
94        self.write_metric(metric)
95    }
96
97    fn write_metric(&mut self, metric: &Metric) -> Result<(), GenericError> {
98        self.ensure_writer()?;
99
100        let context = metric.context();
101        let metric_context = ContextNoOrigin {
102            name: context.name().clone(),
103            tags: context.tags().clone(),
104        };
105
106        let timestamp = saluki_common::time::get_coarse_unix_timestamp();
107        let sample = self.stats.entry(metric_context).or_default();
108        sample.count += 1;
109        sample.last_seen = timestamp;
110
111        let writer = self.writer.as_mut().expect("writer should be initialized");
112        writeln!(
113            writer.writer,
114            "Metric Name: {} | Tags: {{{}}} | Count: {} | Last Seen: {}",
115            context.name(),
116            format_tags(context.tags()),
117            sample.count,
118            format_timestamp(sample.last_seen)
119        )
120        .map_err(|e| {
121            generic_error!(
122                "Failed to write to DogStatsD debug log file '{}': {}",
123                self.log_file.display(),
124                e
125            )
126        })
127    }
128
129    fn ensure_writer(&mut self) -> Result<(), GenericError> {
130        if self.writer.is_some() {
131            return Ok(());
132        }
133
134        let appender = RollingFileAppenderBase::new(
135            &self.log_file,
136            RollingConditionBase::new().max_size(self.log_file_max_size),
137            self.log_file_max_rolls,
138        )
139        .map_err(|e| generic_error!("Failed to open dogstatsd_log_file '{}': {}", self.log_file.display(), e))?;
140
141        let (writer, guard) = NonBlockingBuilder::default()
142            .thread_name("dsd-dbg-writer")
143            .buffered_lines_limit(DEBUG_LOG_WRITER_BUFFER_LINES)
144            // Drop debug log lines rather than slow DogStatsD metric ingestion.
145            .lossy(true)
146            .finish(appender);
147
148        self.writer = Some(DebugLogWriter { writer, _guard: guard });
149
150        Ok(())
151    }
152}
153
154#[async_trait]
155impl Destination for DogStatsDDebugLog {
156    async fn run(mut self: Box<Self>, mut context: DestinationContext) -> Result<(), GenericError> {
157        let mut health = context.take_health_handle();
158        health.mark_ready();
159
160        loop {
161            select! {
162                _ = health.live() => continue,
163                maybe_events = context.events().next() => match maybe_events {
164                    Some(events) => {
165                        for event in events {
166                            if let Event::Metric(metric) = event {
167                                if let Err(error) = self.process_metric(&metric) {
168                                    warn!(error = %error, "Failed to write DogStatsD debug log line; continuing.");
169                                }
170                            }
171                        }
172                    },
173                    None => break,
174                },
175                metrics_stats_enabled = self.metrics_stats_enabled.changed() => {
176                    debug!(metrics_stats_enabled, "Updated DogStatsD metrics stats debug logging gate.");
177                },
178            }
179        }
180
181        Ok(())
182    }
183}
184
185#[async_trait]
186impl DestinationBuilder for DogStatsDDebugLogConfiguration {
187    fn input_event_type(&self) -> EventType {
188        EventType::Metric
189    }
190
191    async fn build(&self, _context: BuildContext) -> Result<Box<dyn Destination + Send>, GenericError> {
192        DogStatsDDebugLog::new(self).map(|destination| Box::new(destination) as Box<dyn Destination + Send>)
193    }
194}
195
196impl MemoryBounds for DogStatsDDebugLogConfiguration {
197    fn specify_bounds(&self, builder: &mut MemoryBoundsBuilder) {
198        builder
199            .minimum()
200            .with_single_value::<DogStatsDDebugLog>("component struct");
201    }
202}
203
204fn format_tags(tags: &TagSet) -> String {
205    let mut formatted = String::new();
206
207    for tag in tags {
208        if !formatted.is_empty() {
209            formatted.push(' ');
210        }
211        formatted.push_str(tag.as_str());
212    }
213
214    formatted
215}
216
217fn format_timestamp(timestamp: u64) -> String {
218    i64::try_from(timestamp)
219        .ok()
220        .and_then(|ts| DateTime::<Utc>::from_timestamp(ts, 0))
221        .map(|dt| dt.format("%Y-%m-%d %H:%M:%S +0000 UTC").to_string())
222        .unwrap_or_else(|| timestamp.to_string())
223}
224
225#[cfg(test)]
226mod tests {
227    use std::{
228        fs,
229        path::{Path, PathBuf},
230    };
231    use std::{sync::Arc, time::Duration};
232
233    use agent_data_plane_config::{Live, SalukiConfiguration};
234    use saluki_core::{
235        components::test_util::TestComponentDriver,
236        data_model::event::{
237            metric::{context::Context, Metric},
238            Event,
239        },
240    };
241    use tempfile::tempdir;
242
243    use super::{DogStatsDDebugLog, DogStatsDDebugLogConfiguration};
244
245    fn test_config(log_file: PathBuf, max_size: u64, max_rolls: usize) -> DogStatsDDebugLogConfiguration {
246        DogStatsDDebugLogConfiguration {
247            metrics_stats_enabled: Live::new_fixed(true),
248            log_file,
249            log_file_max_size: max_size,
250            log_file_max_rolls: max_rolls,
251        }
252    }
253
254    fn read_log_files(log_file: &Path, max_rolls: usize) -> String {
255        let mut output = String::new();
256
257        for roll in (0..=max_rolls).rev() {
258            let path = rolled_path(log_file, roll);
259            if path.exists() {
260                output.push_str(&fs::read_to_string(&path).expect("debug log file should be readable"));
261            }
262        }
263
264        output
265    }
266
267    fn rolled_path(log_file: &Path, roll: usize) -> PathBuf {
268        if roll == 0 {
269            log_file.to_path_buf()
270        } else {
271            PathBuf::from(format!("{}.{}", log_file.display(), roll))
272        }
273    }
274
275    fn tagged_metric() -> Metric {
276        let context = Context::from_static_parts("custom.metric", &["env:prod", "service:web"]);
277        Metric::counter(context, 1.0)
278    }
279
280    #[tokio::test]
281    async fn writes_metric_debug_lines_and_updates_count() {
282        let tempdir = tempdir().expect("temporary directory should be created");
283        let log_file = tempdir.path().join("dogstatsd-stats.log");
284        let config = test_config(log_file.clone(), 64_000, 3);
285        let metric = tagged_metric();
286
287        let mut destination = DogStatsDDebugLog::new(&config).expect("debug log destination should be built");
288        destination
289            .write_metric(&metric)
290            .expect("first metric should be written");
291        destination
292            .write_metric(&metric)
293            .expect("second metric should be written");
294        drop(destination);
295
296        let output = read_log_files(&log_file, config.log_file_max_rolls);
297        let lines = output.lines().collect::<Vec<_>>();
298
299        assert_eq!(lines.len(), 2);
300        assert!(lines[0].contains("Metric Name: custom.metric"));
301        assert!(lines[0].contains("Tags: {env:prod service:web}"));
302        assert!(lines[0].contains("Count: 1"));
303        assert!(lines[0].contains("Last Seen: "));
304        assert!(lines[1].contains("Count: 2"));
305    }
306
307    #[tokio::test]
308    async fn run_starts_and_stops_logging_with_metrics_stats_setting() {
309        let tempdir = tempdir().expect("temporary directory should be created");
310        let log_file = tempdir.path().join("dogstatsd-stats.log");
311        let cell = Arc::new(arc_swap::ArcSwap::from_pointee(SalukiConfiguration::default()));
312        let (tick_tx, tick_rx) = tokio::sync::watch::channel(());
313        let mut config = test_config(log_file.clone(), 64_000, 3);
314        config.metrics_stats_enabled = Live::new_dynamic(Arc::clone(&cell), tick_rx, |config| {
315            &config.domains.dogstatsd.debug_log.metrics_stats_enable
316        });
317        let log_file_max_rolls = config.log_file_max_rolls;
318        let control = TestComponentDriver::destination(config)
319            .await
320            .expect("debug log destination should be built");
321
322        control.send_event(Event::Metric(tagged_metric())).await;
323        control.wait_until_input_drained().await;
324        assert!(!log_file.exists());
325
326        let mut updated = (*cell.load_full()).clone();
327        updated.domains.dogstatsd.debug_log.metrics_stats_enable = true;
328        cell.store(Arc::new(updated));
329        tick_tx.send_replace(());
330
331        tokio::time::timeout(Duration::from_secs(2), async {
332            loop {
333                control.send_event(Event::Metric(tagged_metric())).await;
334                tokio::time::sleep(Duration::from_millis(10)).await;
335                if fs::read_to_string(&log_file).is_ok_and(|output| output.contains("Metric Name: custom.metric")) {
336                    break;
337                }
338            }
339        })
340        .await
341        .expect("metrics should be logged after the runtime setting is enabled");
342
343        let mut updated = (*cell.load_full()).clone();
344        updated.domains.dogstatsd.debug_log.metrics_stats_enable = false;
345        cell.store(Arc::new(updated));
346        tick_tx.send_replace(());
347
348        let line_count_after_disable = tokio::time::timeout(Duration::from_secs(2), async {
349            let mut previous_line_count = read_log_files(&log_file, log_file_max_rolls).lines().count();
350            let mut unchanged_samples = 0;
351
352            loop {
353                control.send_event(Event::Metric(tagged_metric())).await;
354                control.wait_until_input_drained().await;
355                tokio::time::sleep(Duration::from_millis(20)).await;
356
357                let current_line_count = read_log_files(&log_file, log_file_max_rolls).lines().count();
358                if current_line_count == previous_line_count {
359                    unchanged_samples += 1;
360                    if unchanged_samples == 5 {
361                        break current_line_count;
362                    }
363                } else {
364                    previous_line_count = current_line_count;
365                    unchanged_samples = 0;
366                }
367            }
368        })
369        .await
370        .expect("metrics should stop being logged after the runtime setting is disabled");
371
372        for _ in 0..3 {
373            control.send_event(Event::Metric(tagged_metric())).await;
374        }
375        control.wait_until_input_drained().await;
376        tokio::time::sleep(Duration::from_millis(100)).await;
377
378        let output = read_log_files(&log_file, log_file_max_rolls);
379        assert_eq!(output.lines().count(), line_count_after_disable);
380
381        control.shutdown().await.expect("destination should stop cleanly");
382    }
383
384    #[tokio::test]
385    async fn rotates_log_file_at_configured_size() {
386        let tempdir = tempdir().expect("temporary directory should be created");
387        let log_file = tempdir.path().join("dogstatsd-stats.log");
388        let min_debug_line_len =
389            "Metric Name: custom.metric | Tags: {env:prod service:web} | Count: 1 | Last Seen: ".len();
390        let config = test_config(log_file.clone(), min_debug_line_len as u64, 2);
391        let metric = tagged_metric();
392
393        let mut destination = DogStatsDDebugLog::new(&config).expect("debug log destination should be built");
394        for _ in 0..12 {
395            destination.write_metric(&metric).expect("metric should be written");
396        }
397        drop(destination);
398
399        assert!(log_file.exists());
400        assert!(rolled_path(&log_file, 1).exists());
401        assert!(rolled_path(&log_file, 2).exists());
402        assert!(!rolled_path(&log_file, 3).exists());
403
404        let output = read_log_files(&log_file, config.log_file_max_rolls);
405        assert!(output.contains("Metric Name: custom.metric"));
406    }
407
408    #[tokio::test]
409    async fn build_error_mentions_log_file_config_key_and_path() {
410        let tempdir = tempdir().expect("temporary directory should be created");
411        let blocked_parent = tempdir.path().join("not-a-directory");
412        fs::write(&blocked_parent, "not a directory").expect("blocking file should be written");
413        let log_file = blocked_parent.join("dogstatsd-stats.log");
414        let config = test_config(log_file.clone(), 64_000, 3);
415
416        let err = match DogStatsDDebugLog::new(&config) {
417            Ok(_) => panic!("build should fail"),
418            Err(err) => err,
419        };
420        let err = err.to_string();
421
422        assert!(err.contains("dogstatsd_log_file"));
423        assert!(err.contains(&log_file.display().to_string()));
424    }
425}
saluki_components/destinations/dsd_debug_log/mod.rs

saluki_components/destinations/dsd_debug_log/
mod.rs
