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/// Advanced autodiscovery identifier
69#[derive(Debug, Clone, PartialEq, Eq)]
70pub struct AdvancedADIdentifier {
71    /// Kubernetes service
72    pub kube_service: Option<KubeNamespacedName>,
73    /// Kubernetes endpoints
74    pub kube_endpoints: Option<KubeNamespacedName>,
75}
76
77impl From<ProtoAdvancedAdIdentifier> for AdvancedADIdentifier {
78    fn from(value: ProtoAdvancedAdIdentifier) -> Self {
79        Self {
80            kube_service: value.kube_service.map(Into::into),
81            kube_endpoints: value
82                .kube_endpoints
83                .and_then(|endpoint| endpoint.kube_namespaced_name)
84                .map(Into::into),
85        }
86    }
87}
88
89/// Raw data trait for configuration data
90pub trait RawData {
91    /// Get the value of the data
92    fn get_value(&self) -> &BTreeMap<MetaString, serde_yaml::Value>;
93
94    /// Get a value from the data
95    fn get(&self, key: &str) -> Option<&serde_yaml::Value> {
96        self.get_value().get(key)
97    }
98
99    /// Convert the data to bytes
100    fn to_bytes(&self) -> Result<Vec<u8>, GenericError> {
101        let mut buffer = Vec::new();
102        serde_yaml::to_writer(&mut buffer, &self.get_value())?;
103        Ok(buffer)
104    }
105}
106
107/// Generic map of key-value pairs
108#[derive(Debug, Default, Clone, PartialEq, Eq)]
109pub struct Data {
110    value: BTreeMap<MetaString, serde_yaml::Value>,
111}
112
113impl RawData for Data {
114    fn get_value(&self) -> &BTreeMap<MetaString, serde_yaml::Value> {
115        &self.value
116    }
117}
118
119/// Configuration for a check instance
120#[derive(Debug, Default, Clone, PartialEq, Eq)]
121pub struct Instance {
122    /// Instance ID
123    id: String,
124    /// Instance value
125    value: BTreeMap<MetaString, serde_yaml::Value>,
126}
127
128impl Instance {
129    /// Get the instance ID
130    pub fn id(&self) -> &String {
131        &self.id
132    }
133}
134
135impl RawData for Instance {
136    fn get_value(&self) -> &BTreeMap<MetaString, serde_yaml::Value> {
137        &self.value
138    }
139}
140
141/// Configuration for a check
142#[derive(Debug, Clone)]
143pub struct Config {
144    /// Configuration name/identifier
145    pub name: MetaString,
146    /// Raw configuration data
147    pub init_config: Data,
148    /// Instance configurations
149    pub instances: Vec<Data>,
150    /// Metric configuration
151    pub metric_config: Data,
152    /// Logs configuration
153    pub logs_config: Data,
154    /// Auto-discovery identifiers
155    pub ad_identifiers: Vec<MetaString>,
156    /// Advanced auto-discovery identifiers
157    pub advanced_ad_identifiers: Vec<AdvancedADIdentifier>,
158    /// Provider that discovered this config
159    pub provider: MetaString,
160    /// Service ID
161    pub service_id: MetaString,
162    /// Tagger entity
163    pub tagger_entity: MetaString,
164    /// Whether this is a cluster check
165    pub cluster_check: bool,
166    /// Node name
167    pub node_name: MetaString,
168    /// Source of the configuration
169    pub source: MetaString,
170    /// Whether to ignore autodiscovery tags
171    pub ignore_autodiscovery_tags: bool,
172    /// Whether metrics are excluded
173    pub metrics_excluded: bool,
174    /// Whether logs are excluded
175    pub logs_excluded: bool,
176}
177
178/// Check configuration
179#[derive(Debug, Clone, PartialEq, Eq)]
180pub struct CheckConfig {
181    /// Check name
182    pub name: MetaString,
183    /// Init configuration data
184    pub init_config: Data,
185    /// Instance configurations
186    pub instances: Vec<Instance>,
187    /// Source of the configuration
188    pub source: MetaString,
189}
190
191impl Config {
192    /// Get the digest for the config
193    pub fn digest(&self) -> u64 {
194        let mut h = XxHash64::with_seed(0);
195
196        h.write(self.name.as_bytes());
197        for i in &self.instances {
198            h.write(&i.to_bytes().unwrap_or_default());
199        }
200        h.write(&self.init_config.to_bytes().unwrap_or_default());
201        for i in &self.ad_identifiers {
202            h.write(i.as_bytes());
203        }
204        h.write(self.node_name.as_bytes());
205        h.write(&self.logs_config.to_bytes().unwrap_or_default());
206        h.write(self.service_id.as_bytes());
207        h.write(if self.ignore_autodiscovery_tags {
208            b"true"
209        } else {
210            b"false"
211        });
212
213        h.finish()
214    }
215}
216
217impl From<ProtoConfig> for Config {
218    fn from(proto: ProtoConfig) -> Self {
219        // Convert advanced AD identifiers from proto
220        let advanced_ad_identifiers = proto.advanced_ad_identifiers.into_iter().map(Into::into).collect();
221
222        let init_config = bytes_to_data(proto.init_config).unwrap_or_default();
223        let instances = proto
224            .instances
225            .into_iter()
226            .map(|instance| bytes_to_data(instance).unwrap_or_default())
227            .collect();
228
229        Self {
230            name: proto.name.into(),
231            init_config,
232            instances,
233            metric_config: bytes_to_data(proto.metric_config).unwrap_or_default(),
234            logs_config: bytes_to_data(proto.logs_config).unwrap_or_default(),
235            ad_identifiers: proto.ad_identifiers.into_iter().map(MetaString::from).collect(),
236            advanced_ad_identifiers,
237            provider: proto.provider.into(),
238            service_id: proto.service_id.into(),
239            tagger_entity: proto.tagger_entity.into(),
240            cluster_check: proto.cluster_check,
241            node_name: proto.node_name.into(),
242            source: proto.source.into(),
243            ignore_autodiscovery_tags: proto.ignore_autodiscovery_tags,
244            metrics_excluded: proto.metrics_excluded,
245            logs_excluded: proto.logs_excluded,
246        }
247    }
248}
249
250fn bytes_to_data(bytes: Vec<u8>) -> Result<Data, GenericError> {
251    let parse_bytes = String::from_utf8(bytes)?;
252
253    let map: BTreeMap<String, serde_yaml::Value> = serde_yaml::from_str(&parse_bytes)?;
254
255    let mut result = BTreeMap::<MetaString, serde_yaml::Value>::new();
256
257    for (key, value) in map {
258        result.insert(key.into(), value);
259    }
260
261    Ok(Data { value: result })
262}
263
264fn instance_id(name: &str, instance: &Data, digest: u64, init_config: &Data) -> String {
265    let mut h2 = FnvHasher::default();
266    h2.write_u64(digest);
267    h2.write(&instance.to_bytes().unwrap_or_default());
268    h2.write(&init_config.to_bytes().unwrap_or_default());
269
270    let instance_name = instance_name(instance);
271    let hash2 = h2.finish();
272
273    if !instance_name.is_empty() {
274        format!("{}:{}:{:X}", name, instance_name, hash2)
275    } else {
276        format!("{}:{:X}", name, hash2)
277    }
278}
279
280fn instance_name(instance: &Data) -> String {
281    if let Some(name) = instance.get("name") {
282        if let Some(value) = name.as_str() {
283            return value.to_string();
284        }
285    }
286    if let Some(namespace) = instance.get("namespace") {
287        if let Some(value) = namespace.as_str() {
288            return value.to_string();
289        }
290    }
291    "".to_string()
292}
293
294impl From<Config> for CheckConfig {
295    fn from(config: Config) -> Self {
296        let digest = config.digest();
297
298        CheckConfig {
299            name: config.name.clone(),
300            init_config: config.init_config.clone(),
301            instances: config
302                .instances
303                .into_iter()
304                .map(|instance_data| Instance {
305                    id: instance_id(&config.name, &instance_data, digest, &config.init_config),
306                    value: instance_data.value,
307                })
308                .collect(),
309            source: config.source,
310        }
311    }
312}
313
314impl From<ProtoConfig> for AutodiscoveryEvent {
315    fn from(proto: ProtoConfig) -> AutodiscoveryEvent {
316        let event_type = EventType::from(proto.event_type);
317
318        let config = Config::from(proto);
319
320        if !config.instances.is_empty() && !config.cluster_check {
321            let check_config = CheckConfig::from(config);
322
323            if event_type == EventType::Schedule {
324                return AutodiscoveryEvent::CheckSchedule { config: check_config };
325            } else {
326                return AutodiscoveryEvent::CheckUnscheduled { config: check_config };
327            }
328        }
329
330        if event_type == EventType::Schedule {
331            AutodiscoveryEvent::Schedule { config }
332        } else {
333            AutodiscoveryEvent::Unscheduled { config }
334        }
335    }
336}
337
338/// An autodiscovery event
339#[derive(Debug, Clone)]
340#[allow(clippy::large_enum_variant)]
341pub enum AutodiscoveryEvent {
342    /// Schedule a check configuration
343    CheckSchedule {
344        /// Configuration
345        config: CheckConfig,
346    },
347    /// Unschedule a check configuration
348    CheckUnscheduled {
349        /// Configuration
350        config: CheckConfig,
351    },
352    /// Schedule a generic configuration
353    Schedule {
354        /// Configuration
355        config: Config,
356    },
357    /// Unscheduled a generic configuration
358    Unscheduled {
359        /// Configuration
360        config: Config,
361    },
362}
363
364/// Provides autodiscovery functionality.
365///
366/// This trait is used to discover and monitor configuration files for checks.
367#[async_trait]
368pub trait AutodiscoveryProvider {
369    /// Subscribe to autodiscovery events.
370    ///
371    /// Returns `None` when no provider is configured.
372    async fn subscribe(&self) -> Option<Receiver<AutodiscoveryEvent>>;
373}
374
375#[async_trait]
376impl<T> AutodiscoveryProvider for Option<T>
377where
378    T: AutodiscoveryProvider + Sync,
379{
380    async fn subscribe(&self) -> Option<Receiver<AutodiscoveryEvent>> {
381        match self.as_ref() {
382            Some(provider) => provider.subscribe().await,
383            None => None,
384        }
385    }
386}
387
388#[cfg(test)]
389mod tests {
390    use datadog_protos::agent::{AdvancedAdIdentifier, KubeNamespacedName};
391
392    use super::*;
393
394    // Builds a `ProtoConfig` with representative defaults for every field, so tests can override only the fields
395    // they actually exercise (via struct-update syntax) instead of restating the full 17-field literal.
396    fn base_proto_config() -> ProtoConfig {
397        ProtoConfig {
398            name: "test-config".to_string(),
399            event_type: ConfigEventType::Schedule as i32,
400            init_config: b"key: value".to_vec(),
401            instances: vec![],
402            provider: "test-provider".to_string(),
403            ad_identifiers: vec!["id1".to_string(), "id2".to_string()],
404            cluster_check: false,
405            metric_config: b"metric_key: metric_value".to_vec(),
406            logs_config: b"log_key: log_value".to_vec(),
407            advanced_ad_identifiers: vec![],
408            service_id: "service-id".to_string(),
409            tagger_entity: "tagger-entity".to_string(),
410            node_name: "node-name".to_string(),
411            source: "source".to_string(),
412            ignore_autodiscovery_tags: false,
413            metrics_excluded: false,
414            logs_excluded: false,
415        }
416    }
417
418    #[test]
419    fn test_event_type_from_config_event_type() {
420        assert_eq!(EventType::from(ConfigEventType::Schedule), EventType::Schedule);
421        assert_eq!(EventType::from(ConfigEventType::Unschedule), EventType::Unschedule);
422    }
423
424    #[test]
425    fn test_event_type_from_i32() {
426        // Known values
427        assert_eq!(EventType::from(0), EventType::Schedule); // Schedule is 0
428        assert_eq!(EventType::from(1), EventType::Unschedule); // Unschedule is 1
429
430        // Unknown values should default to Schedule
431        assert_eq!(EventType::from(2), EventType::Schedule);
432        assert_eq!(EventType::from(-1), EventType::Schedule);
433    }
434
435    #[test]
436    fn test_check_config_instance_id() {
437        let proto_config = ProtoConfig {
438            name: "test-check".to_string(),
439            instances: vec![b"name: test".to_vec(), b"another_key: another_value".to_vec()],
440            ..base_proto_config()
441        };
442
443        let config = Config::from(proto_config);
444
445        let check_config = CheckConfig::from(config);
446
447        let id1 = &check_config.instances[0].id;
448        let id2 = &check_config.instances[1].id;
449
450        assert_ne!(id1, id2);
451
452        assert_eq!(id1, "test-check:test:369F074E36651C8");
453        assert_eq!(id2, "test-check:8C83712B7A572843");
454    }
455
456    #[test]
457    fn test_config_from_proto_config() {
458        // Create a ProtoConfig with test values
459        let mut proto_config = ProtoConfig {
460            instances: vec![
461                b"instance_key: instance_value".to_vec(),
462                b"another_key: another_value".to_vec(),
463            ],
464            cluster_check: true,
465            ..base_proto_config()
466        };
467
468        let kube_svc = KubeNamespacedName {
469            name: "nginx".to_string(),
470            namespace: "default".to_string(),
471        };
472
473        let adv_id = AdvancedAdIdentifier {
474            kube_service: Some(kube_svc),
475            kube_endpoints: None,
476        };
477
478        proto_config.advanced_ad_identifiers = vec![adv_id];
479
480        let config = Config::from(proto_config);
481
482        assert_eq!(config.name, "test-config");
483        assert_eq!(config.provider, "test-provider");
484        assert_eq!(
485            config.ad_identifiers,
486            vec![MetaString::from_static("id1"), MetaString::from_static("id2")]
487        );
488        assert!(config.cluster_check);
489        assert_eq!(config.service_id, "service-id");
490        assert_eq!(config.tagger_entity, "tagger-entity");
491        assert_eq!(config.node_name, "node-name");
492        assert_eq!(config.source, "source");
493        assert!(!config.ignore_autodiscovery_tags);
494        assert!(!config.metrics_excluded);
495        assert!(!config.logs_excluded);
496        assert_eq!(
497            config.init_config.get("key"),
498            Some(&serde_yaml::Value::String("value".to_string()))
499        );
500        assert_eq!(config.instances.len(), 2);
501        assert_eq!(
502            config.instances[0].get("instance_key"),
503            Some(&serde_yaml::Value::String("instance_value".to_string()))
504        );
505        assert_eq!(
506            config.instances[1].get("another_key"),
507            Some(&serde_yaml::Value::String("another_value".to_string()))
508        );
509        assert_eq!(
510            config.metric_config.get("metric_key"),
511            Some(&serde_yaml::Value::String("metric_value".to_string()))
512        );
513        assert_eq!(
514            config.logs_config.get("log_key"),
515            Some(&serde_yaml::Value::String("log_value".to_string()))
516        );
517
518        assert_eq!(config.advanced_ad_identifiers.len(), 1);
519        let adv_id = &config.advanced_ad_identifiers[0];
520        assert!(adv_id.kube_endpoints.is_none());
521        assert!(adv_id.kube_service.is_some());
522        let svc = adv_id.kube_service.as_ref().unwrap();
523        assert_eq!(svc.name, "nginx");
524        assert_eq!(svc.namespace, "default");
525    }
526
527    #[test]
528    fn test_autodiscovery_event_from_proto_config() {
529        // Create a ProtoConfig with test values
530        let mut proto_config = ProtoConfig {
531            init_config: b"init-data".to_vec(),
532            cluster_check: true,
533            metric_config: vec![],
534            logs_config: vec![],
535            ..base_proto_config()
536        };
537
538        let kube_svc = KubeNamespacedName {
539            name: "nginx".to_string(),
540            namespace: "default".to_string(),
541        };
542
543        let adv_id = AdvancedAdIdentifier {
544            kube_service: Some(kube_svc),
545            kube_endpoints: None,
546        };
547
548        proto_config.advanced_ad_identifiers = vec![adv_id];
549
550        let event = AutodiscoveryEvent::from(proto_config.clone());
551
552        match event {
553            AutodiscoveryEvent::Schedule { config: _config } => {}
554            _ => panic!("Expected an Schedule event"),
555        }
556
557        proto_config.event_type = ConfigEventType::Unschedule as i32;
558
559        let event = AutodiscoveryEvent::from(proto_config.clone());
560
561        match event {
562            AutodiscoveryEvent::Unscheduled { config } => {
563                assert_eq!(config.name, "test-config");
564            }
565            _ => panic!("Expected an Unscheduled event"),
566        }
567
568        proto_config.instances = vec![b"instance1".to_vec(), b"instance2".to_vec()];
569        proto_config.cluster_check = false;
570        proto_config.event_type = ConfigEventType::Schedule as i32;
571
572        let event = AutodiscoveryEvent::from(proto_config.clone());
573
574        match event {
575            AutodiscoveryEvent::CheckSchedule { config: _config } => {}
576            _ => panic!("Expected an CheckSchedule event"),
577        }
578
579        proto_config.event_type = ConfigEventType::Unschedule as i32;
580
581        let event = AutodiscoveryEvent::from(proto_config);
582
583        match event {
584            AutodiscoveryEvent::CheckUnscheduled { config: _config } => {}
585            _ => panic!("Expected an CheckUnscheduled event"),
586        }
587    }
588}
saluki_env/autodiscovery/mod.rs

saluki_env/autodiscovery/
mod.rs
