saluki_env/autodiscovery/
mod.rs

1//! Autodiscovery provider.
2//!
3//! This module provides the `Autodiscovery` trait, which deals with providing information about autodiscovery.
4
5use std::collections::BTreeMap;
6use std::hash::Hasher;
7
8use async_trait::async_trait;
9use datadog_protos::agent::{
10    AdvancedAdIdentifier as ProtoAdvancedAdIdentifier, Config as ProtoConfig, ConfigEventType,
11    KubeNamespacedName as ProtoKubeNamespacedName,
12};
13use fnv::FnvHasher;
14use saluki_error::GenericError;
15use stringtheory::MetaString;
16use tokio::sync::mpsc::Receiver;
17use twox_hash::XxHash64;
18
19pub mod providers;
20
21/// Configuration event type
22#[derive(Debug, Clone, PartialEq, Eq)]
23pub enum EventType {
24    /// Schedule a configuration
25    Schedule,
26    /// Unschedule a configuration
27    Unschedule,
28}
29
30impl From<ConfigEventType> for EventType {
31    fn from(event_type: ConfigEventType) -> Self {
32        match event_type {
33            ConfigEventType::Schedule => EventType::Schedule,
34            ConfigEventType::Unschedule => EventType::Unschedule,
35        }
36    }
37}
38
39impl From<i32> for EventType {
40    fn from(value: i32) -> Self {
41        if value == ConfigEventType::Unschedule as i32 {
42            EventType::Unschedule
43        } else {
44            // Default to Schedule for unknown values
45            EventType::Schedule
46        }
47    }
48}
49
50/// Kubernetes namespaced name
51#[derive(Debug, Clone, PartialEq, Eq)]
52pub struct KubeNamespacedName {
53    /// Kubernetes resource name
54    pub name: MetaString,
55    /// Kubernetes namespace
56    pub namespace: MetaString,
57}
58
59impl From<ProtoKubeNamespacedName> for KubeNamespacedName {
60    fn from(value: ProtoKubeNamespacedName) -> Self {
61        Self {
62            name: value.name.into(),
63            namespace: value.namespace.into(),
64        }
65    }
66}
67
68/// Kubernetes endpoints identifier
69#[derive(Debug, Clone, PartialEq, Eq)]
70pub struct KubeEndpointsIdentifier {
71    /// Kubernetes endpoints resource
72    pub kube_namespaced_name: KubeNamespacedName,
73    /// Endpoints resolution mode
74    pub resolve: MetaString,
75}
76
77/// Advanced autodiscovery identifier
78#[derive(Debug, Clone, PartialEq, Eq)]
79pub struct AdvancedADIdentifier {
80    /// Kubernetes service
81    pub kube_service: Option<KubeNamespacedName>,
82    /// Kubernetes endpoints
83    pub kube_endpoints: Option<KubeEndpointsIdentifier>,
84}
85
86impl From<ProtoAdvancedAdIdentifier> for AdvancedADIdentifier {
87    fn from(value: ProtoAdvancedAdIdentifier) -> Self {
88        Self {
89            kube_service: value.kube_service.map(Into::into),
90            kube_endpoints: value.kube_endpoints.and_then(|endpoints| {
91                endpoints
92                    .kube_namespaced_name
93                    .map(|namespaced_name| KubeEndpointsIdentifier {
94                        kube_namespaced_name: namespaced_name.into(),
95                        resolve: endpoints.resolve.into(),
96                    })
97            }),
98        }
99    }
100}
101
102/// Raw data trait for configuration data
103pub trait RawData {
104    /// Get the value of the data
105    fn get_value(&self) -> &BTreeMap<MetaString, serde_yaml::Value>;
106
107    /// Get a value from the data
108    fn get(&self, key: &str) -> Option<&serde_yaml::Value> {
109        self.get_value().get(key)
110    }
111
112    /// Convert the data to bytes
113    fn to_bytes(&self) -> Result<Vec<u8>, GenericError> {
114        let mut buffer = Vec::new();
115        serde_yaml::to_writer(&mut buffer, &self.get_value())?;
116        Ok(buffer)
117    }
118}
119
120/// Generic map of key-value pairs
121#[derive(Debug, Default, Clone, PartialEq, Eq)]
122pub struct Data {
123    value: BTreeMap<MetaString, serde_yaml::Value>,
124}
125
126impl RawData for Data {
127    fn get_value(&self) -> &BTreeMap<MetaString, serde_yaml::Value> {
128        &self.value
129    }
130}
131
132/// Configuration for a check instance
133#[derive(Debug, Default, Clone, PartialEq, Eq)]
134pub struct Instance {
135    /// Instance ID
136    id: String,
137    /// Instance value
138    value: BTreeMap<MetaString, serde_yaml::Value>,
139}
140
141impl Instance {
142    /// Get the instance ID
143    pub fn id(&self) -> &String {
144        &self.id
145    }
146}
147
148impl RawData for Instance {
149    fn get_value(&self) -> &BTreeMap<MetaString, serde_yaml::Value> {
150        &self.value
151    }
152}
153
154/// Configuration for a check
155#[derive(Debug, Clone)]
156pub struct Config {
157    /// Configuration name/identifier
158    pub name: MetaString,
159    /// Raw configuration data
160    pub init_config: Data,
161    /// Instance configurations
162    pub instances: Vec<Data>,
163    /// Metric configuration
164    pub metric_config: Data,
165    /// Logs configuration
166    pub logs_config: Data,
167    /// Auto-discovery identifiers
168    pub ad_identifiers: Vec<MetaString>,
169    /// Advanced auto-discovery identifiers
170    pub advanced_ad_identifiers: Vec<AdvancedADIdentifier>,
171    /// Provider that discovered this config
172    pub provider: MetaString,
173    /// Service ID
174    pub service_id: MetaString,
175    /// Tagger entity
176    pub tagger_entity: MetaString,
177    /// Whether this is a cluster check
178    pub cluster_check: bool,
179    /// Node name
180    pub node_name: MetaString,
181    /// Source of the configuration
182    pub source: MetaString,
183    /// Whether to ignore autodiscovery tags
184    pub ignore_autodiscovery_tags: bool,
185    /// Whether metrics are excluded
186    pub metrics_excluded: bool,
187    /// Whether logs are excluded
188    pub logs_excluded: bool,
189}
190
191/// Check configuration
192#[derive(Debug, Clone, PartialEq, Eq)]
193pub struct CheckConfig {
194    /// Check name
195    pub name: MetaString,
196    /// Init configuration data
197    pub init_config: Data,
198    /// Instance configurations
199    pub instances: Vec<Instance>,
200    /// Metric configuration
201    pub metric_config: Data,
202    /// Logs configuration
203    pub logs_config: Data,
204    /// Auto-discovery identifiers
205    pub ad_identifiers: Vec<MetaString>,
206    /// Advanced auto-discovery identifiers
207    pub advanced_ad_identifiers: Vec<AdvancedADIdentifier>,
208    /// Provider that discovered this config
209    pub provider: MetaString,
210    /// Service ID
211    pub service_id: MetaString,
212    /// Tagger entity
213    pub tagger_entity: MetaString,
214    /// Whether this is a cluster check
215    pub cluster_check: bool,
216    /// Node name
217    pub node_name: MetaString,
218    /// Source of the configuration
219    pub source: MetaString,
220    /// Whether to ignore autodiscovery tags
221    pub ignore_autodiscovery_tags: bool,
222    /// Whether metrics are excluded
223    pub metrics_excluded: bool,
224    /// Whether logs are excluded
225    pub logs_excluded: bool,
226}
227
228impl Config {
229    /// Get the digest for the config
230    pub fn digest(&self) -> u64 {
231        let mut h = XxHash64::with_seed(0);
232
233        h.write(self.name.as_bytes());
234        for i in &self.instances {
235            h.write(&i.to_bytes().unwrap_or_default());
236        }
237        h.write(&self.init_config.to_bytes().unwrap_or_default());
238        for i in &self.ad_identifiers {
239            h.write(i.as_bytes());
240        }
241        h.write(self.node_name.as_bytes());
242        h.write(&self.logs_config.to_bytes().unwrap_or_default());
243        h.write(self.service_id.as_bytes());
244        h.write(if self.ignore_autodiscovery_tags {
245            b"true"
246        } else {
247            b"false"
248        });
249
250        h.finish()
251    }
252}
253
254impl From<ProtoConfig> for Config {
255    fn from(proto: ProtoConfig) -> Self {
256        // Convert advanced AD identifiers from proto
257        let advanced_ad_identifiers = proto.advanced_ad_identifiers.into_iter().map(Into::into).collect();
258
259        let init_config = bytes_to_data(proto.init_config).unwrap_or_default();
260        let instances = proto
261            .instances
262            .into_iter()
263            .map(|instance| bytes_to_data(instance).unwrap_or_default())
264            .collect();
265
266        Self {
267            name: proto.name.into(),
268            init_config,
269            instances,
270            metric_config: bytes_to_data(proto.metric_config).unwrap_or_default(),
271            logs_config: bytes_to_data(proto.logs_config).unwrap_or_default(),
272            ad_identifiers: proto.ad_identifiers.into_iter().map(MetaString::from).collect(),
273            advanced_ad_identifiers,
274            provider: proto.provider.into(),
275            service_id: proto.service_id.into(),
276            tagger_entity: proto.tagger_entity.into(),
277            cluster_check: proto.cluster_check,
278            node_name: proto.node_name.into(),
279            source: proto.source.into(),
280            ignore_autodiscovery_tags: proto.ignore_autodiscovery_tags,
281            metrics_excluded: proto.metrics_excluded,
282            logs_excluded: proto.logs_excluded,
283        }
284    }
285}
286
287fn bytes_to_data(bytes: Vec<u8>) -> Result<Data, GenericError> {
288    let parse_bytes = String::from_utf8(bytes)?;
289
290    let map: BTreeMap<String, serde_yaml::Value> = serde_yaml::from_str(&parse_bytes)?;
291
292    let mut result = BTreeMap::<MetaString, serde_yaml::Value>::new();
293
294    for (key, value) in map {
295        result.insert(key.into(), value);
296    }
297
298    Ok(Data { value: result })
299}
300
301fn instance_id(name: &str, instance: &Data, digest: u64, init_config: &Data) -> String {
302    let mut h2 = FnvHasher::default();
303    h2.write_u64(digest);
304    h2.write(&instance.to_bytes().unwrap_or_default());
305    h2.write(&init_config.to_bytes().unwrap_or_default());
306
307    let instance_name = instance_name(instance);
308    let hash2 = h2.finish();
309
310    if !instance_name.is_empty() {
311        format!("{}:{}:{:X}", name, instance_name, hash2)
312    } else {
313        format!("{}:{:X}", name, hash2)
314    }
315}
316
317fn instance_name(instance: &Data) -> String {
318    if let Some(name) = instance.get("name") {
319        if let Some(value) = name.as_str() {
320            return value.to_string();
321        }
322    }
323    if let Some(namespace) = instance.get("namespace") {
324        if let Some(value) = namespace.as_str() {
325            return value.to_string();
326        }
327    }
328    "".to_string()
329}
330
331impl From<Config> for CheckConfig {
332    fn from(config: Config) -> Self {
333        let digest = config.digest();
334        let Config {
335            name,
336            init_config,
337            instances,
338            metric_config,
339            logs_config,
340            ad_identifiers,
341            advanced_ad_identifiers,
342            provider,
343            service_id,
344            tagger_entity,
345            cluster_check,
346            node_name,
347            source,
348            ignore_autodiscovery_tags,
349            metrics_excluded,
350            logs_excluded,
351        } = config;
352        let instances = instances
353            .into_iter()
354            .map(|instance_data| Instance {
355                id: instance_id(&name, &instance_data, digest, &init_config),
356                value: instance_data.value,
357            })
358            .collect();
359
360        Self {
361            name,
362            init_config,
363            instances,
364            metric_config,
365            logs_config,
366            ad_identifiers,
367            advanced_ad_identifiers,
368            provider,
369            service_id,
370            tagger_entity,
371            cluster_check,
372            node_name,
373            source,
374            ignore_autodiscovery_tags,
375            metrics_excluded,
376            logs_excluded,
377        }
378    }
379}
380
381impl From<ProtoConfig> for AutodiscoveryEvent {
382    fn from(proto: ProtoConfig) -> AutodiscoveryEvent {
383        let event_type = EventType::from(proto.event_type);
384
385        let config = Config::from(proto);
386
387        if !config.instances.is_empty() && !config.cluster_check {
388            let check_config = CheckConfig::from(config);
389
390            if event_type == EventType::Schedule {
391                return AutodiscoveryEvent::CheckSchedule { config: check_config };
392            } else {
393                return AutodiscoveryEvent::CheckUnscheduled { config: check_config };
394            }
395        }
396
397        if event_type == EventType::Schedule {
398            AutodiscoveryEvent::Schedule { config }
399        } else {
400            AutodiscoveryEvent::Unscheduled { config }
401        }
402    }
403}
404
405/// An autodiscovery event
406#[derive(Debug, Clone)]
407#[allow(clippy::large_enum_variant)]
408pub enum AutodiscoveryEvent {
409    /// Schedule a check configuration
410    CheckSchedule {
411        /// Configuration
412        config: CheckConfig,
413    },
414    /// Unschedule a check configuration
415    CheckUnscheduled {
416        /// Configuration
417        config: CheckConfig,
418    },
419    /// Schedule a generic configuration
420    Schedule {
421        /// Configuration
422        config: Config,
423    },
424    /// Unscheduled a generic configuration
425    Unscheduled {
426        /// Configuration
427        config: Config,
428    },
429}
430
431/// Provides autodiscovery functionality.
432///
433/// This trait is used to discover and monitor configuration files for checks.
434#[async_trait]
435pub trait AutodiscoveryProvider {
436    /// Subscribe to autodiscovery events.
437    ///
438    /// Returns `None` when no provider is configured.
439    async fn subscribe(&self) -> Option<Receiver<AutodiscoveryEvent>>;
440}
441
442#[async_trait]
443impl<T> AutodiscoveryProvider for Option<T>
444where
445    T: AutodiscoveryProvider + Sync,
446{
447    async fn subscribe(&self) -> Option<Receiver<AutodiscoveryEvent>> {
448        match self.as_ref() {
449            Some(provider) => provider.subscribe().await,
450            None => None,
451        }
452    }
453}
454
455#[cfg(test)]
456mod tests {
457    use datadog_protos::agent::{
458        AdvancedAdIdentifier, KubeEndpointsIdentifier as ProtoKubeEndpointsIdentifier, KubeNamespacedName,
459    };
460
461    use super::*;
462
463    // Builds a `ProtoConfig` with representative defaults for every field, so tests can override only the fields
464    // they actually exercise (via struct-update syntax) instead of restating the full 17-field literal.
465    fn base_proto_config() -> ProtoConfig {
466        ProtoConfig {
467            name: "test-config".to_string(),
468            event_type: ConfigEventType::Schedule as i32,
469            init_config: b"key: value".to_vec(),
470            instances: vec![],
471            provider: "test-provider".to_string(),
472            ad_identifiers: vec!["id1".to_string(), "id2".to_string()],
473            cluster_check: false,
474            metric_config: b"metric_key: metric_value".to_vec(),
475            logs_config: b"log_key: log_value".to_vec(),
476            advanced_ad_identifiers: vec![],
477            service_id: "service-id".to_string(),
478            tagger_entity: "tagger-entity".to_string(),
479            node_name: "node-name".to_string(),
480            source: "source".to_string(),
481            ignore_autodiscovery_tags: false,
482            metrics_excluded: false,
483            logs_excluded: false,
484        }
485    }
486
487    #[test]
488    fn test_event_type_from_config_event_type() {
489        assert_eq!(EventType::from(ConfigEventType::Schedule), EventType::Schedule);
490        assert_eq!(EventType::from(ConfigEventType::Unschedule), EventType::Unschedule);
491    }
492
493    #[test]
494    fn test_event_type_from_i32() {
495        // Known values
496        assert_eq!(EventType::from(0), EventType::Schedule); // Schedule is 0
497        assert_eq!(EventType::from(1), EventType::Unschedule); // Unschedule is 1
498
499        // Unknown values should default to Schedule
500        assert_eq!(EventType::from(2), EventType::Schedule);
501        assert_eq!(EventType::from(-1), EventType::Schedule);
502    }
503
504    #[test]
505    fn test_check_config_instance_id() {
506        let proto_config = ProtoConfig {
507            name: "test-check".to_string(),
508            instances: vec![b"name: test".to_vec(), b"another_key: another_value".to_vec()],
509            ..base_proto_config()
510        };
511
512        let config = Config::from(proto_config);
513
514        let check_config = CheckConfig::from(config);
515
516        let id1 = &check_config.instances[0].id;
517        let id2 = &check_config.instances[1].id;
518
519        assert_ne!(id1, id2);
520
521        assert_eq!(id1, "test-check:test:369F074E36651C8");
522        assert_eq!(id2, "test-check:8C83712B7A572843");
523    }
524
525    #[test]
526    fn test_config_from_proto_config() {
527        // Create a ProtoConfig with test values
528        let mut proto_config = ProtoConfig {
529            instances: vec![
530                b"instance_key: instance_value".to_vec(),
531                b"another_key: another_value".to_vec(),
532            ],
533            cluster_check: true,
534            ..base_proto_config()
535        };
536
537        let kube_svc = KubeNamespacedName {
538            name: "nginx".to_string(),
539            namespace: "default".to_string(),
540        };
541
542        let adv_id = AdvancedAdIdentifier {
543            kube_service: Some(kube_svc),
544            kube_endpoints: None,
545        };
546
547        proto_config.advanced_ad_identifiers = vec![adv_id];
548
549        let config = Config::from(proto_config);
550
551        assert_eq!(config.name, "test-config");
552        assert_eq!(config.provider, "test-provider");
553        assert_eq!(
554            config.ad_identifiers,
555            vec![MetaString::from_static("id1"), MetaString::from_static("id2")]
556        );
557        assert!(config.cluster_check);
558        assert_eq!(config.service_id, "service-id");
559        assert_eq!(config.tagger_entity, "tagger-entity");
560        assert_eq!(config.node_name, "node-name");
561        assert_eq!(config.source, "source");
562        assert!(!config.ignore_autodiscovery_tags);
563        assert!(!config.metrics_excluded);
564        assert!(!config.logs_excluded);
565        assert_eq!(
566            config.init_config.get("key"),
567            Some(&serde_yaml::Value::String("value".to_string()))
568        );
569        assert_eq!(config.instances.len(), 2);
570        assert_eq!(
571            config.instances[0].get("instance_key"),
572            Some(&serde_yaml::Value::String("instance_value".to_string()))
573        );
574        assert_eq!(
575            config.instances[1].get("another_key"),
576            Some(&serde_yaml::Value::String("another_value".to_string()))
577        );
578        assert_eq!(
579            config.metric_config.get("metric_key"),
580            Some(&serde_yaml::Value::String("metric_value".to_string()))
581        );
582        assert_eq!(
583            config.logs_config.get("log_key"),
584            Some(&serde_yaml::Value::String("log_value".to_string()))
585        );
586
587        assert_eq!(config.advanced_ad_identifiers.len(), 1);
588        let adv_id = &config.advanced_ad_identifiers[0];
589        assert!(adv_id.kube_endpoints.is_none());
590        assert!(adv_id.kube_service.is_some());
591        let svc = adv_id.kube_service.as_ref().unwrap();
592        assert_eq!(svc.name, "nginx");
593        assert_eq!(svc.namespace, "default");
594    }
595
596    #[test]
597    fn test_advanced_ad_identifier_from_proto_retains_endpoints_resolve() {
598        fn proto_id(resolve: &str) -> AdvancedAdIdentifier {
599            AdvancedAdIdentifier {
600                kube_service: None,
601                kube_endpoints: Some(ProtoKubeEndpointsIdentifier {
602                    kube_namespaced_name: Some(KubeNamespacedName {
603                        name: "nginx".to_string(),
604                        namespace: "default".to_string(),
605                    }),
606                    resolve: resolve.to_string(),
607                }),
608            }
609        }
610
611        let adv_id = AdvancedADIdentifier::from(proto_id("ip"));
612
613        assert!(adv_id.kube_service.is_none());
614        let endpoints = adv_id.kube_endpoints.as_ref().expect("endpoints should be present");
615        assert_eq!(endpoints.kube_namespaced_name.name, "nginx");
616        assert_eq!(endpoints.kube_namespaced_name.namespace, "default");
617        assert_eq!(endpoints.resolve, "ip");
618
619        // Identifiers that only differ by their resolution mode must remain distinguishable.
620        assert_ne!(adv_id, AdvancedADIdentifier::from(proto_id("auto")));
621    }
622
623    #[test]
624    fn test_advanced_ad_identifier_from_proto_drops_endpoints_without_namespaced_name() {
625        let proto_id = AdvancedAdIdentifier {
626            kube_service: None,
627            kube_endpoints: Some(ProtoKubeEndpointsIdentifier {
628                kube_namespaced_name: None,
629                resolve: "ip".to_string(),
630            }),
631        };
632
633        assert!(AdvancedADIdentifier::from(proto_id).kube_endpoints.is_none());
634    }
635
636    #[test]
637    fn test_autodiscovery_event_from_proto_config() {
638        // Create a ProtoConfig with test values
639        let mut proto_config = ProtoConfig {
640            init_config: b"init-data".to_vec(),
641            cluster_check: true,
642            metric_config: vec![],
643            logs_config: vec![],
644            ..base_proto_config()
645        };
646
647        let kube_svc = KubeNamespacedName {
648            name: "nginx".to_string(),
649            namespace: "default".to_string(),
650        };
651
652        let adv_id = AdvancedAdIdentifier {
653            kube_service: Some(kube_svc),
654            kube_endpoints: None,
655        };
656
657        proto_config.advanced_ad_identifiers = vec![adv_id];
658
659        let event = AutodiscoveryEvent::from(proto_config.clone());
660
661        match event {
662            AutodiscoveryEvent::Schedule { config: _config } => {}
663            _ => panic!("Expected an Schedule event"),
664        }
665
666        proto_config.event_type = ConfigEventType::Unschedule as i32;
667
668        let event = AutodiscoveryEvent::from(proto_config.clone());
669
670        match event {
671            AutodiscoveryEvent::Unscheduled { config } => {
672                assert_eq!(config.name, "test-config");
673            }
674            _ => panic!("Expected an Unscheduled event"),
675        }
676
677        proto_config.instances = vec![b"instance1".to_vec(), b"instance2".to_vec()];
678        proto_config.cluster_check = false;
679        proto_config.event_type = ConfigEventType::Schedule as i32;
680
681        let event = AutodiscoveryEvent::from(proto_config.clone());
682
683        match event {
684            AutodiscoveryEvent::CheckSchedule { config: _config } => {}
685            _ => panic!("Expected an CheckSchedule event"),
686        }
687
688        proto_config.event_type = ConfigEventType::Unschedule as i32;
689
690        let event = AutodiscoveryEvent::from(proto_config);
691
692        match event {
693            AutodiscoveryEvent::CheckUnscheduled { config: _config } => {}
694            _ => panic!("Expected an CheckUnscheduled event"),
695        }
696    }
697}
saluki_env/autodiscovery/mod.rs

saluki_env/autodiscovery/
mod.rs
