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    use crate::config::{DatadogRemapper, KEY_ALIASES};
355
356    fn test_default_log_file_path() -> PathBuf {
357        PathBuf::from("/tmp/default-dogstatsd-stats.log")
358    }
359
360    async fn deser_config(raw_json: &str) -> DogStatsDDebugLogConfiguration {
361        let value = serde_json::from_str(raw_json).expect("test config should be valid JSON");
362        DogStatsDDebugLogConfiguration::from_configuration(&config_from(value).await, test_default_log_file_path())
363            .expect("DogStatsDDebugLogConfiguration should deserialize")
364    }
365
366    #[tokio::test]
367    async fn defaults_match_core_agent() {
368        let config = deser_config("{}").await;
369
370        assert!(config.enabled());
371        assert!(!config.metrics_stats_enabled);
372        assert!(config.logging_enabled);
373        assert_eq!(config.log_file(), test_default_log_file_path());
374        assert_eq!(config.log_file_max_size(), ByteSize::mb(10));
375        assert_eq!(config.log_file_max_rolls(), 3);
376    }
377
378    #[tokio::test]
379    async fn logging_enabled_controls_topology_wiring() {
380        let config = deser_config(r#"{ "dogstatsd_metrics_stats_enable": true }"#).await;
381        assert!(config.enabled());
382        assert!(config.metrics_stats_enabled);
383
384        let config = deser_config(
385            r#"{
386                "dogstatsd_metrics_stats_enable": true,
387                "dogstatsd_logging_enabled": true
388            }"#,
389        )
390        .await;
391        assert!(config.enabled());
392        assert!(config.metrics_stats_enabled);
393
394        let config = deser_config(
395            r#"{
396                "dogstatsd_metrics_stats_enable": true,
397                "dogstatsd_logging_enabled": false
398            }"#,
399        )
400        .await;
401        assert!(!config.enabled());
402        assert!(config.metrics_stats_enabled);
403
404        let config = deser_config(
405            r#"{
406                "dogstatsd_metrics_stats_enable": false,
407                "dogstatsd_logging_enabled": true
408            }"#,
409        )
410        .await;
411        assert!(config.enabled());
412        assert!(!config.metrics_stats_enabled);
413    }
414
415    #[tokio::test]
416    async fn explicit_log_file_path_is_preserved() {
417        let config = deser_config(r#"{ "dogstatsd_log_file": "/tmp/dsd-debug.log" }"#).await;
418
419        assert_eq!(config.log_file(), std::path::Path::new("/tmp/dsd-debug.log"));
420    }
421
422    #[tokio::test]
423    async fn negative_log_file_max_rolls_is_rejected() {
424        let value = json!({ "dogstatsd_log_file_max_rolls": -1 });
425        let config = config_from(value).await;
426
427        let result = DogStatsDDebugLogConfiguration::from_configuration(&config, test_default_log_file_path());
428        assert!(result.is_err());
429    }
430
431    #[tokio::test]
432    async fn smoke_test() {
433        run_config_smoke_tests(
434            structs::DOGSTATSD_DEBUG_LOG_CONFIGURATION,
435            &[],
436            json!({}),
437            |cfg| {
438                DogStatsDDebugLogConfiguration::from_configuration(&cfg, test_default_log_file_path())
439                    .expect("DogStatsDDebugLogConfiguration should deserialize")
440            },
441            KEY_ALIASES,
442            DatadogRemapper::from_env_vars,
443        )
444        .await
445    }
446
447    async fn test_config(log_file: PathBuf, max_size: ByteSize, max_rolls: usize) -> DogStatsDDebugLogConfiguration {
448        let config = config_from(json!({
449            "dogstatsd_metrics_stats_enable": true,
450            "dogstatsd_logging_enabled": true,
451            "dogstatsd_log_file": log_file.display().to_string(),
452            "dogstatsd_log_file_max_size": max_size.as_u64(),
453            "dogstatsd_log_file_max_rolls": max_rolls,
454        }))
455        .await;
456
457        DogStatsDDebugLogConfiguration::from_configuration(&config, test_default_log_file_path())
458            .expect("DogStatsDDebugLogConfiguration should deserialize")
459    }
460
461    fn read_log_files(log_file: &Path, max_rolls: usize) -> String {
462        let mut output = String::new();
463
464        for roll in (0..=max_rolls).rev() {
465            let path = rolled_path(log_file, roll);
466            if path.exists() {
467                output.push_str(&fs::read_to_string(&path).expect("debug log file should be readable"));
468            }
469        }
470
471        output
472    }
473
474    fn rolled_path(log_file: &Path, roll: usize) -> PathBuf {
475        if roll == 0 {
476            log_file.to_path_buf()
477        } else {
478            PathBuf::from(format!("{}.{}", log_file.display(), roll))
479        }
480    }
481
482    fn tagged_metric() -> Metric {
483        let context = Context::from_static_parts("custom.metric", &["env:prod", "service:web"]);
484        Metric::counter(context, 1.0)
485    }
486
487    #[tokio::test]
488    async fn writes_metric_debug_lines_and_updates_count() {
489        let tempdir = tempdir().expect("temporary directory should be created");
490        let log_file = tempdir.path().join("dogstatsd-stats.log");
491        let config = test_config(log_file.clone(), ByteSize::kb(64), 3).await;
492        let metric = tagged_metric();
493
494        let mut destination =
495            DogStatsDDebugLog::from_configuration(&config).expect("debug log destination should be built");
496        destination
497            .write_metric(&metric)
498            .expect("first metric should be written");
499        destination
500            .write_metric(&metric)
501            .expect("second metric should be written");
502        drop(destination);
503
504        let output = read_log_files(&log_file, config.log_file_max_rolls());
505        let lines = output.lines().collect::<Vec<_>>();
506
507        assert_eq!(lines.len(), 2);
508        assert!(lines[0].contains("Metric Name: custom.metric"));
509        assert!(lines[0].contains("Tags: {env:prod service:web}"));
510        assert!(lines[0].contains("Count: 1"));
511        assert!(lines[0].contains("Last Seen: "));
512        assert!(lines[1].contains("Count: 2"));
513    }
514
515    #[tokio::test]
516    async fn dynamically_drops_until_metrics_stats_are_enabled() {
517        use std::time::Duration;
518
519        use saluki_config::dynamic::ConfigUpdate;
520
521        let tempdir = tempdir().expect("temporary directory should be created");
522        let log_file = tempdir.path().join("dogstatsd-stats.log");
523        let (config, sender) = saluki_config::ConfigurationLoader::for_tests(
524            Some(json!({
525                "dogstatsd_log_file": log_file.display().to_string(),
526                "dogstatsd_log_file_max_size": "64kb",
527                "dogstatsd_log_file_max_rolls": 3,
528                "dogstatsd_logging_enabled": true
529            })),
530            None,
531            true,
532        )
533        .await;
534        let sender = sender.expect("dynamic sender should be present");
535        sender
536            .send(ConfigUpdate::Snapshot(json!({})))
537            .await
538            .expect("initial dynamic snapshot should be sent");
539        config.ready().await;
540
541        let dsd_config = DogStatsDDebugLogConfiguration::from_configuration(&config, test_default_log_file_path())
542            .expect("DogStatsDDebugLogConfiguration should deserialize");
543        assert!(dsd_config.enabled());
544        assert!(!dsd_config.metrics_stats_enabled);
545
546        let metric = tagged_metric();
547        let mut destination =
548            DogStatsDDebugLog::from_configuration(&dsd_config).expect("debug log destination should be built");
549        let mut watcher = destination
550            .configuration
551            .watch_for_updates(super::DOGSTATSD_METRICS_STATS_ENABLE_KEY);
552        destination
553            .process_metric(&metric)
554            .expect("disabled metric should be dropped cleanly");
555        assert!(!log_file.exists());
556
557        sender
558            .send(ConfigUpdate::Partial {
559                key: "dogstatsd_metrics_stats_enable".to_string(),
560                value: json!(true),
561            })
562            .await
563            .expect("dynamic update should be sent");
564
565        let (_, maybe_enabled) = tokio::time::timeout(Duration::from_secs(2), watcher.changed::<bool>())
566            .await
567            .expect("metrics stats flag should receive enabled update");
568        destination.metrics_stats_enabled = maybe_enabled.expect("metrics stats update should have a new value");
569
570        destination
571            .process_metric(&metric)
572            .expect("enabled metric should be written");
573
574        sender
575            .send(ConfigUpdate::Partial {
576                key: "dogstatsd_metrics_stats_enable".to_string(),
577                value: json!(false),
578            })
579            .await
580            .expect("dynamic update should be sent");
581
582        let (_, maybe_enabled) = tokio::time::timeout(Duration::from_secs(2), watcher.changed::<bool>())
583            .await
584            .expect("metrics stats flag should receive disabled update");
585        destination.metrics_stats_enabled = maybe_enabled.expect("metrics stats update should have a new value");
586
587        destination
588            .process_metric(&metric)
589            .expect("disabled metric should be dropped cleanly");
590        drop(destination);
591
592        let output = read_log_files(&log_file, dsd_config.log_file_max_rolls());
593        let lines = output.lines().collect::<Vec<_>>();
594
595        assert_eq!(lines.len(), 1);
596        assert!(lines[0].contains("Metric Name: custom.metric"));
597        assert!(lines[0].contains("Count: 1"));
598    }
599
600    #[tokio::test]
601    async fn rotates_log_file_at_configured_size() {
602        let tempdir = tempdir().expect("temporary directory should be created");
603        let log_file = tempdir.path().join("dogstatsd-stats.log");
604        let min_debug_line_len =
605            "Metric Name: custom.metric | Tags: {env:prod service:web} | Count: 1 | Last Seen: ".len();
606        let config = test_config(log_file.clone(), ByteSize::b(min_debug_line_len as u64), 2).await;
607        let metric = tagged_metric();
608
609        let mut destination =
610            DogStatsDDebugLog::from_configuration(&config).expect("debug log destination should be built");
611        for _ in 0..12 {
612            destination.write_metric(&metric).expect("metric should be written");
613        }
614        drop(destination);
615
616        assert!(log_file.exists());
617        assert!(rolled_path(&log_file, 1).exists());
618        assert!(rolled_path(&log_file, 2).exists());
619        assert!(!rolled_path(&log_file, 3).exists());
620
621        let output = read_log_files(&log_file, config.log_file_max_rolls());
622        assert!(output.contains("Metric Name: custom.metric"));
623    }
624
625    #[tokio::test]
626    async fn build_error_mentions_log_file_config_key_and_path() {
627        let tempdir = tempdir().expect("temporary directory should be created");
628        let blocked_parent = tempdir.path().join("not-a-directory");
629        fs::write(&blocked_parent, "not a directory").expect("blocking file should be written");
630        let log_file = blocked_parent.join("dogstatsd-stats.log");
631        let config = test_config(log_file.clone(), ByteSize::kb(64), 3).await;
632
633        let err = match DogStatsDDebugLog::from_configuration(&config) {
634            Ok(_) => panic!("build should fail"),
635            Err(err) => err,
636        };
637        let err = err.to_string();
638
639        assert!(err.contains("dogstatsd_log_file"));
640        assert!(err.contains(&log_file.display().to_string()));
641    }
642}
saluki_components/destinations/dsd_debug_log/mod.rs

saluki_components/destinations/dsd_debug_log/
mod.rs
