saluki_components/destinations/dsd_debug_log/
mod.rs

1use std::{
2    io::Write,
3    path::{Path, PathBuf},
4};
5
6use async_trait::async_trait;
7use bytesize::ByteSize;
8use chrono::{DateTime, Utc};
9use saluki_common::collections::FastHashMap;
10use saluki_config::GenericConfiguration;
11use saluki_context::tags::TagSet;
12use saluki_core::accounting::{MemoryBounds, MemoryBoundsBuilder};
13use saluki_core::{
14    components::{
15        destinations::{Destination, DestinationBuilder, DestinationContext},
16        ComponentContext,
17    },
18    data_model::event::{metric::Metric, Event, EventType},
19};
20use saluki_error::{generic_error, GenericError};
21use serde::Deserialize;
22use stringtheory::MetaString;
23use tokio::select;
24use tracing::{debug, warn};
25use tracing_appender::non_blocking::{NonBlocking, NonBlockingBuilder, WorkerGuard};
26use tracing_rolling_file::{RollingConditionBase, RollingFileAppenderBase};
27
28const DEFAULT_DOGSTATSD_LOG_FILE_MAX_SIZE: ByteSize = ByteSize::mb(10);
29const DEFAULT_DOGSTATSD_LOG_FILE_MAX_ROLLS: usize = 3;
30const DEBUG_LOG_WRITER_BUFFER_LINES: usize = 4096;
31const DOGSTATSD_METRICS_STATS_ENABLE_KEY: &str = "dogstatsd_metrics_stats_enable";
32
33const fn default_true() -> bool {
34    true
35}
36
37const fn default_log_file_max_size() -> ByteSize {
38    DEFAULT_DOGSTATSD_LOG_FILE_MAX_SIZE
39}
40
41const fn default_log_file_max_rolls() -> usize {
42    DEFAULT_DOGSTATSD_LOG_FILE_MAX_ROLLS
43}
44
45/// Configuration for the DogStatsD debug log destination.
46#[derive(Clone, Debug, Deserialize)]
47#[cfg_attr(test, derive(serde::Serialize))]
48pub struct DogStatsDDebugLogConfiguration {
49    /// Whether DogStatsD metric-level statistics are enabled.
50    ///
51    /// The debug log destination is always present when `dogstatsd_logging_enabled` is `true`,
52    /// but it drops metrics until this runtime flag becomes `true`, and only during that period.
53    ///
54    /// Defaults to `false`.
55    #[serde(rename = "dogstatsd_metrics_stats_enable", default)]
56    metrics_stats_enabled: bool,
57
58    #[serde(skip)]
59    configuration: Option<GenericConfiguration>,
60
61    /// Whether DogStatsD metric-level statistics should also be written to a log file.
62    ///
63    /// This controls whether the destination is added to the topology.
64    /// Defaults to `true`.
65    #[serde(rename = "dogstatsd_logging_enabled", default = "default_true")]
66    logging_enabled: bool,
67
68    /// Path to the DogStatsD debug log file.
69    ///
70    /// This defaults to the platform-specific core Agent DogStatsD stats log path when the configured value is empty.
71    #[serde(rename = "dogstatsd_log_file", default)]
72    log_file: PathBuf,
73
74    /// Maximum size of the active debug log file before rotation.
75    ///
76    /// Defaults to `10Mb`.
77    #[serde(rename = "dogstatsd_log_file_max_size", default = "default_log_file_max_size")]
78    log_file_max_size: ByteSize,
79
80    /// Number of rotated debug log files to keep.
81    ///
82    /// Defaults to `3`.
83    #[serde(rename = "dogstatsd_log_file_max_rolls", default = "default_log_file_max_rolls")]
84    log_file_max_rolls: usize,
85}
86
87#[cfg(test)]
88impl PartialEq for DogStatsDDebugLogConfiguration {
89    fn eq(&self, other: &Self) -> bool {
90        self.metrics_stats_enabled == other.metrics_stats_enabled
91            && self.logging_enabled == other.logging_enabled
92            && self.log_file == other.log_file
93            && self.log_file_max_size == other.log_file_max_size
94            && self.log_file_max_rolls == other.log_file_max_rolls
95    }
96}
97
98/// DogStatsD destination that writes metric debug lines to a rotating file.
99struct DogStatsDDebugLog {
100    log_file: PathBuf,
101    log_file_max_size: ByteSize,
102    log_file_max_rolls: usize,
103    writer: Option<DebugLogWriter>,
104    metrics_stats_enabled: bool,
105    configuration: GenericConfiguration,
106    stats: FastHashMap<ContextNoOrigin, MetricSample>,
107}
108
109struct DebugLogWriter {
110    writer: NonBlocking,
111    _guard: WorkerGuard,
112}
113
114#[derive(Debug, Default)]
115struct MetricSample {
116    count: u64,
117    last_seen: u64,
118}
119
120#[derive(Eq, Hash, PartialEq)]
121struct ContextNoOrigin {
122    name: MetaString,
123    tags: TagSet,
124}
125
126impl DogStatsDDebugLogConfiguration {
127    /// Creates a new `DogStatsDDebugLogConfiguration` from the given configuration.
128    ///
129    /// If `dogstatsd_log_file` is empty, `default_log_file_path` is used.
130    pub fn from_configuration(
131        config: &GenericConfiguration, default_log_file_path: PathBuf,
132    ) -> Result<Self, GenericError> {
133        let mut cfg: Self = config.as_typed()?;
134
135        if cfg.log_file.as_os_str().is_empty() {
136            cfg.log_file = default_log_file_path;
137        }
138
139        if cfg.log_file.to_str().is_none() {
140            return Err(generic_error!(
141                "dogstatsd_log_file must be valid UTF-8, got '{}'",
142                cfg.log_file.display()
143            ));
144        }
145
146        cfg.configuration = Some(config.clone());
147
148        Ok(cfg)
149    }
150
151    /// Returns `true` if the debug log destination should be added to the topology.
152    pub const fn enabled(&self) -> bool {
153        self.logging_enabled
154    }
155
156    /// Returns the DogStatsD debug log file path.
157    pub fn log_file(&self) -> &Path {
158        &self.log_file
159    }
160
161    /// Returns the maximum size of the active debug log file before rotation.
162    pub const fn log_file_max_size(&self) -> ByteSize {
163        self.log_file_max_size
164    }
165
166    /// Returns the number of rotated debug log files to keep.
167    pub const fn log_file_max_rolls(&self) -> usize {
168        self.log_file_max_rolls
169    }
170}
171
172impl DogStatsDDebugLog {
173    fn from_configuration(config: &DogStatsDDebugLogConfiguration) -> Result<Self, GenericError> {
174        let mut destination = Self {
175            log_file: config.log_file.clone(),
176            log_file_max_size: config.log_file_max_size,
177            log_file_max_rolls: config.log_file_max_rolls,
178            writer: None,
179            metrics_stats_enabled: config.metrics_stats_enabled,
180            configuration: config
181                .configuration
182                .clone()
183                .expect("configuration must be set via from_configuration"),
184            stats: FastHashMap::default(),
185        };
186
187        if destination.metrics_stats_enabled {
188            destination.ensure_writer()?;
189        }
190
191        Ok(destination)
192    }
193
194    fn process_metric(&mut self, metric: &Metric) -> Result<(), GenericError> {
195        if !self.metrics_stats_enabled {
196            return Ok(());
197        }
198
199        self.write_metric(metric)
200    }
201
202    fn write_metric(&mut self, metric: &Metric) -> Result<(), GenericError> {
203        self.ensure_writer()?;
204
205        let context = metric.context();
206        let metric_context = ContextNoOrigin {
207            name: context.name().clone(),
208            tags: context.tags().clone(),
209        };
210
211        let timestamp = saluki_common::time::get_coarse_unix_timestamp();
212        let sample = self.stats.entry(metric_context).or_default();
213        sample.count += 1;
214        sample.last_seen = timestamp;
215
216        let writer = self.writer.as_mut().expect("writer should be initialized");
217        writeln!(
218            writer.writer,
219            "Metric Name: {} | Tags: {{{}}} | Count: {} | Last Seen: {}",
220            context.name(),
221            format_tags(context.tags()),
222            sample.count,
223            format_timestamp(sample.last_seen)
224        )
225        .map_err(|e| {
226            generic_error!(
227                "Failed to write to DogStatsD debug log file '{}': {}",
228                self.log_file.display(),
229                e
230            )
231        })
232    }
233
234    fn ensure_writer(&mut self) -> Result<(), GenericError> {
235        if self.writer.is_some() {
236            return Ok(());
237        }
238
239        let appender = RollingFileAppenderBase::new(
240            &self.log_file,
241            RollingConditionBase::new().max_size(self.log_file_max_size.as_u64()),
242            self.log_file_max_rolls,
243        )
244        .map_err(|e| generic_error!("Failed to open dogstatsd_log_file '{}': {}", self.log_file.display(), e))?;
245
246        let (writer, guard) = NonBlockingBuilder::default()
247            .thread_name("dsd-dbg-writer")
248            .buffered_lines_limit(DEBUG_LOG_WRITER_BUFFER_LINES)
249            // Drop debug log lines rather than slow DogStatsD metric ingestion.
250            .lossy(true)
251            .finish(appender);
252
253        self.writer = Some(DebugLogWriter { writer, _guard: guard });
254
255        Ok(())
256    }
257}
258
259#[async_trait]
260impl Destination for DogStatsDDebugLog {
261    async fn run(mut self: Box<Self>, mut context: DestinationContext) -> Result<(), GenericError> {
262        let mut health = context.take_health_handle();
263        health.mark_ready();
264
265        let mut metrics_stats_enabled_watcher =
266            self.configuration.watch_for_updates(DOGSTATSD_METRICS_STATS_ENABLE_KEY);
267
268        loop {
269            select! {
270                _ = health.live() => continue,
271                maybe_events = context.events().next() => match maybe_events {
272                    Some(events) => {
273                        for event in events {
274                            if let Event::Metric(metric) = event {
275                                if let Err(error) = self.process_metric(&metric) {
276                                    warn!(error = %error, "Failed to write DogStatsD debug log line; continuing.");
277                                }
278                            }
279                        }
280                    },
281                    None => break,
282                },
283                (_, maybe_metrics_stats_enabled) = metrics_stats_enabled_watcher.changed::<bool>() => {
284                    if let Some(metrics_stats_enabled) = maybe_metrics_stats_enabled {
285                        self.metrics_stats_enabled = metrics_stats_enabled;
286                        debug!(metrics_stats_enabled, "Updated DogStatsD metrics stats debug logging gate.");
287                    }
288                },
289            }
290        }
291
292        Ok(())
293    }
294}
295
296#[async_trait]
297impl DestinationBuilder for DogStatsDDebugLogConfiguration {
298    fn input_event_type(&self) -> EventType {
299        EventType::Metric
300    }
301
302    async fn build(&self, _context: ComponentContext) -> Result<Box<dyn Destination + Send>, GenericError> {
303        DogStatsDDebugLog::from_configuration(self)
304            .map(|destination| Box::new(destination) as Box<dyn Destination + Send>)
305    }
306}
307
308impl MemoryBounds for DogStatsDDebugLogConfiguration {
309    fn specify_bounds(&self, builder: &mut MemoryBoundsBuilder) {
310        builder
311            .minimum()
312            .with_single_value::<DogStatsDDebugLog>("component struct");
313    }
314}
315
316fn format_tags(tags: &TagSet) -> String {
317    let mut formatted = String::new();
318
319    for tag in tags {
320        if !formatted.is_empty() {
321            formatted.push(' ');
322        }
323        formatted.push_str(tag.as_str());
324    }
325
326    formatted
327}
328
329fn format_timestamp(timestamp: u64) -> String {
330    i64::try_from(timestamp)
331        .ok()
332        .and_then(|ts| DateTime::<Utc>::from_timestamp(ts, 0))
333        .map(|dt| dt.format("%Y-%m-%d %H:%M:%S +0000 UTC").to_string())
334        .unwrap_or_else(|| timestamp.to_string())
335}
336
337#[cfg(test)]
338mod tests {
339    use std::{
340        fs,
341        path::{Path, PathBuf},
342    };
343
344    use bytesize::ByteSize;
345    use datadog_agent_config_testing::config_registry::structs;
346    use datadog_agent_config_testing::run_config_smoke_tests;
347    use saluki_config::config_from;
348    use saluki_context::Context;
349    use saluki_core::data_model::event::metric::Metric;
350    use serde_json::json;
351    use tempfile::tempdir;
352
353    use super::{DogStatsDDebugLog, DogStatsDDebugLogConfiguration};
354
355    fn test_default_log_file_path() -> PathBuf {
356        PathBuf::from("/tmp/default-dogstatsd-stats.log")
357    }
358
359    async fn deser_config(raw_json: &str) -> DogStatsDDebugLogConfiguration {
360        let value = serde_json::from_str(raw_json).expect("test config should be valid JSON");
361        DogStatsDDebugLogConfiguration::from_configuration(&config_from(value).await, test_default_log_file_path())
362            .expect("DogStatsDDebugLogConfiguration should deserialize")
363    }
364
365    #[tokio::test]
366    async fn defaults_match_core_agent() {
367        let config = deser_config("{}").await;
368
369        assert!(config.enabled());
370        assert!(!config.metrics_stats_enabled);
371        assert!(config.logging_enabled);
372        assert_eq!(config.log_file(), test_default_log_file_path());
373        assert_eq!(config.log_file_max_size(), ByteSize::mb(10));
374        assert_eq!(config.log_file_max_rolls(), 3);
375    }
376
377    #[tokio::test]
378    async fn logging_enabled_controls_topology_wiring() {
379        let config = deser_config(r#"{ "dogstatsd_metrics_stats_enable": true }"#).await;
380        assert!(config.enabled());
381        assert!(config.metrics_stats_enabled);
382
383        let config = deser_config(
384            r#"{
385                "dogstatsd_metrics_stats_enable": true,
386                "dogstatsd_logging_enabled": true
387            }"#,
388        )
389        .await;
390        assert!(config.enabled());
391        assert!(config.metrics_stats_enabled);
392
393        let config = deser_config(
394            r#"{
395                "dogstatsd_metrics_stats_enable": true,
396                "dogstatsd_logging_enabled": false
397            }"#,
398        )
399        .await;
400        assert!(!config.enabled());
401        assert!(config.metrics_stats_enabled);
402
403        let config = deser_config(
404            r#"{
405                "dogstatsd_metrics_stats_enable": false,
406                "dogstatsd_logging_enabled": true
407            }"#,
408        )
409        .await;
410        assert!(config.enabled());
411        assert!(!config.metrics_stats_enabled);
412    }
413
414    #[tokio::test]
415    async fn explicit_log_file_path_is_preserved() {
416        let config = deser_config(r#"{ "dogstatsd_log_file": "/tmp/dsd-debug.log" }"#).await;
417
418        assert_eq!(config.log_file(), std::path::Path::new("/tmp/dsd-debug.log"));
419    }
420
421    #[tokio::test]
422    async fn negative_log_file_max_rolls_is_rejected() {
423        let value = json!({ "dogstatsd_log_file_max_rolls": -1 });
424        let config = config_from(value).await;
425
426        let result = DogStatsDDebugLogConfiguration::from_configuration(&config, test_default_log_file_path());
427        assert!(result.is_err());
428    }
429
430    #[tokio::test]
431    async fn smoke_test() {
432        run_config_smoke_tests(structs::DOGSTATSD_DEBUG_LOG_CONFIGURATION, &[], json!({}), |cfg| {
433            DogStatsDDebugLogConfiguration::from_configuration(&cfg, test_default_log_file_path())
434                .expect("DogStatsDDebugLogConfiguration should deserialize")
435        })
436        .await
437    }
438
439    async fn test_config(log_file: PathBuf, max_size: ByteSize, max_rolls: usize) -> DogStatsDDebugLogConfiguration {
440        let config = config_from(json!({
441            "dogstatsd_metrics_stats_enable": true,
442            "dogstatsd_logging_enabled": true,
443            "dogstatsd_log_file": log_file.display().to_string(),
444            "dogstatsd_log_file_max_size": max_size.as_u64(),
445            "dogstatsd_log_file_max_rolls": max_rolls,
446        }))
447        .await;
448
449        DogStatsDDebugLogConfiguration::from_configuration(&config, test_default_log_file_path())
450            .expect("DogStatsDDebugLogConfiguration should deserialize")
451    }
452
453    fn read_log_files(log_file: &Path, max_rolls: usize) -> String {
454        let mut output = String::new();
455
456        for roll in (0..=max_rolls).rev() {
457            let path = rolled_path(log_file, roll);
458            if path.exists() {
459                output.push_str(&fs::read_to_string(&path).expect("debug log file should be readable"));
460            }
461        }
462
463        output
464    }
465
466    fn rolled_path(log_file: &Path, roll: usize) -> PathBuf {
467        if roll == 0 {
468            log_file.to_path_buf()
469        } else {
470            PathBuf::from(format!("{}.{}", log_file.display(), roll))
471        }
472    }
473
474    fn tagged_metric() -> Metric {
475        let context = Context::from_static_parts("custom.metric", &["env:prod", "service:web"]);
476        Metric::counter(context, 1.0)
477    }
478
479    #[tokio::test]
480    async fn writes_metric_debug_lines_and_updates_count() {
481        let tempdir = tempdir().expect("temporary directory should be created");
482        let log_file = tempdir.path().join("dogstatsd-stats.log");
483        let config = test_config(log_file.clone(), ByteSize::kb(64), 3).await;
484        let metric = tagged_metric();
485
486        let mut destination =
487            DogStatsDDebugLog::from_configuration(&config).expect("debug log destination should be built");
488        destination
489            .write_metric(&metric)
490            .expect("first metric should be written");
491        destination
492            .write_metric(&metric)
493            .expect("second metric should be written");
494        drop(destination);
495
496        let output = read_log_files(&log_file, config.log_file_max_rolls());
497        let lines = output.lines().collect::<Vec<_>>();
498
499        assert_eq!(lines.len(), 2);
500        assert!(lines[0].contains("Metric Name: custom.metric"));
501        assert!(lines[0].contains("Tags: {env:prod service:web}"));
502        assert!(lines[0].contains("Count: 1"));
503        assert!(lines[0].contains("Last Seen: "));
504        assert!(lines[1].contains("Count: 2"));
505    }
506
507    #[tokio::test]
508    async fn dynamically_drops_until_metrics_stats_are_enabled() {
509        use std::time::Duration;
510
511        use saluki_config::dynamic::{ConfigSetting, ConfigUpdate};
512
513        let tempdir = tempdir().expect("temporary directory should be created");
514        let log_file = tempdir.path().join("dogstatsd-stats.log");
515        let (config, sender) = saluki_config::ConfigurationLoader::for_tests(
516            Some(json!({
517                "dogstatsd_log_file": log_file.display().to_string(),
518                "dogstatsd_log_file_max_size": "64kb",
519                "dogstatsd_log_file_max_rolls": 3,
520                "dogstatsd_logging_enabled": true
521            })),
522            None,
523            true,
524        )
525        .await;
526        let sender = sender.expect("dynamic sender should be present");
527        sender
528            .send(ConfigUpdate::snapshot([]))
529            .await
530            .expect("initial dynamic snapshot should be sent");
531        config.ready().await;
532
533        let dsd_config = DogStatsDDebugLogConfiguration::from_configuration(&config, test_default_log_file_path())
534            .expect("DogStatsDDebugLogConfiguration should deserialize");
535        assert!(dsd_config.enabled());
536        assert!(!dsd_config.metrics_stats_enabled);
537
538        let metric = tagged_metric();
539        let mut destination =
540            DogStatsDDebugLog::from_configuration(&dsd_config).expect("debug log destination should be built");
541        let mut watcher = destination
542            .configuration
543            .watch_for_updates(super::DOGSTATSD_METRICS_STATS_ENABLE_KEY);
544        destination
545            .process_metric(&metric)
546            .expect("disabled metric should be dropped cleanly");
547        assert!(!log_file.exists());
548
549        sender
550            .send(ConfigUpdate::Partial(ConfigSetting::explicit(
551                "dogstatsd_metrics_stats_enable",
552                json!(true),
553            )))
554            .await
555            .expect("dynamic update should be sent");
556
557        let (_, maybe_enabled) = tokio::time::timeout(Duration::from_secs(2), watcher.changed::<bool>())
558            .await
559            .expect("metrics stats flag should receive enabled update");
560        destination.metrics_stats_enabled = maybe_enabled.expect("metrics stats update should have a new value");
561
562        destination
563            .process_metric(&metric)
564            .expect("enabled metric should be written");
565
566        sender
567            .send(ConfigUpdate::Partial(ConfigSetting::explicit(
568                "dogstatsd_metrics_stats_enable",
569                json!(false),
570            )))
571            .await
572            .expect("dynamic update should be sent");
573
574        let (_, maybe_enabled) = tokio::time::timeout(Duration::from_secs(2), watcher.changed::<bool>())
575            .await
576            .expect("metrics stats flag should receive disabled update");
577        destination.metrics_stats_enabled = maybe_enabled.expect("metrics stats update should have a new value");
578
579        destination
580            .process_metric(&metric)
581            .expect("disabled metric should be dropped cleanly");
582        drop(destination);
583
584        let output = read_log_files(&log_file, dsd_config.log_file_max_rolls());
585        let lines = output.lines().collect::<Vec<_>>();
586
587        assert_eq!(lines.len(), 1);
588        assert!(lines[0].contains("Metric Name: custom.metric"));
589        assert!(lines[0].contains("Count: 1"));
590    }
591
592    #[tokio::test]
593    async fn rotates_log_file_at_configured_size() {
594        let tempdir = tempdir().expect("temporary directory should be created");
595        let log_file = tempdir.path().join("dogstatsd-stats.log");
596        let min_debug_line_len =
597            "Metric Name: custom.metric | Tags: {env:prod service:web} | Count: 1 | Last Seen: ".len();
598        let config = test_config(log_file.clone(), ByteSize::b(min_debug_line_len as u64), 2).await;
599        let metric = tagged_metric();
600
601        let mut destination =
602            DogStatsDDebugLog::from_configuration(&config).expect("debug log destination should be built");
603        for _ in 0..12 {
604            destination.write_metric(&metric).expect("metric should be written");
605        }
606        drop(destination);
607
608        assert!(log_file.exists());
609        assert!(rolled_path(&log_file, 1).exists());
610        assert!(rolled_path(&log_file, 2).exists());
611        assert!(!rolled_path(&log_file, 3).exists());
612
613        let output = read_log_files(&log_file, config.log_file_max_rolls());
614        assert!(output.contains("Metric Name: custom.metric"));
615    }
616
617    #[tokio::test]
618    async fn build_error_mentions_log_file_config_key_and_path() {
619        let tempdir = tempdir().expect("temporary directory should be created");
620        let blocked_parent = tempdir.path().join("not-a-directory");
621        fs::write(&blocked_parent, "not a directory").expect("blocking file should be written");
622        let log_file = blocked_parent.join("dogstatsd-stats.log");
623        let config = test_config(log_file.clone(), ByteSize::kb(64), 3).await;
624
625        let err = match DogStatsDDebugLog::from_configuration(&config) {
626            Ok(_) => panic!("build should fail"),
627            Err(err) => err,
628        };
629        let err = err.to_string();
630
631        assert!(err.contains("dogstatsd_log_file"));
632        assert!(err.contains(&log_file.display().to_string()));
633    }
634}
saluki_components/destinations/dsd_debug_log/mod.rs

saluki_components/destinations/dsd_debug_log/
mod.rs
