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

saluki_components/destinations/dsd_debug_log/
mod.rs
