Skip to main content

saluki_components/config/
metrics_endpoint_routing.rs

1//! Builds endpoint routing groups from exact-name and literal-prefix metric allow lists.
2
3use std::collections::{BTreeMap, HashMap};
4
5use agent_data_plane_config::shared;
6use saluki_error::{generic_error, GenericError};
7
8use crate::common::datadog::routing::{RoutingTargetCatalog, RoutingTargetSet};
9
10/// Sorts literal prefixes and removes duplicates and prefixes covered by another prefix.
11pub fn compact_metric_prefixes(prefixes: &mut Vec<String>) {
12    prefixes.sort_unstable();
13    if !prefixes.is_empty() {
14        let mut retained = 0;
15        for next in 1..prefixes.len() {
16            if !prefixes[next].starts_with(&prefixes[retained]) {
17                retained += 1;
18                prefixes.swap(retained, next);
19            }
20        }
21        prefixes.truncate(retained + 1);
22    }
23}
24
25/// A group of configured endpoints that share exact-name and prefix metric allowlists.
26#[derive(Clone, Debug, Eq, PartialEq)]
27pub struct EndpointAllowlistGroup {
28    /// Configured endpoint identities targeted by this policy.
29    pub endpoints: Vec<String>,
30
31    /// Exact metric names permitted to reach these endpoints.
32    pub metric_allowlist: Vec<String>,
33    /// Literal metric-name prefixes permitted to reach these endpoints.
34    pub metric_prefix_allowlist: Vec<String>,
35}
36
37impl EndpointAllowlistGroup {
38    /// Returns the routing targets that receive this group's filtered metric stream.
39    ///
40    /// Every API key configured for one of the group's endpoints is a target.
41    pub fn targets(&self, catalog: &RoutingTargetCatalog) -> RoutingTargetSet {
42        catalog.select(|target| {
43            self.endpoints
44                .iter()
45                .any(|endpoint| endpoint == target.configured_endpoint())
46        })
47    }
48}
49
50/// Endpoint-aware selective metric-routing configuration.
51#[derive(Debug, Eq, PartialEq)]
52pub struct MetricsEndpointRoutingConfiguration {
53    selected_endpoints: Vec<String>,
54    policy_groups: Vec<EndpointAllowlistGroup>,
55}
56
57impl MetricsEndpointRoutingConfiguration {
58    /// Resolves endpoint policies against the configured primary and additional endpoints.
59    ///
60    /// An empty policy map preserves ordinary endpoint routing. Every policy key must exactly match a key in
61    /// `additional_endpoints` or the effective primary endpoint; this prevents a typo from silently leaving an intended
62    /// capacity-saving destination on the unfiltered path. Equivalent allow lists are grouped so they can share one
63    /// filtered encoder branch.
64    ///
65    /// # Errors
66    ///
67    /// Returns an error if a policy names neither the primary endpoint nor an endpoint configured in
68    /// `additional_endpoints`.
69    pub fn from_configuration(
70        metric_allowlists: &HashMap<String, Vec<String>>, metric_prefix_allowlists: &HashMap<String, Vec<String>>,
71        endpoints: &shared::Endpoints,
72    ) -> Result<Self, GenericError> {
73        let mut selected_endpoints = metric_allowlists
74            .keys()
75            .chain(metric_prefix_allowlists.keys())
76            .cloned()
77            .collect::<Vec<_>>();
78        selected_endpoints.sort_unstable();
79        selected_endpoints.dedup();
80        let mut grouped_endpoints = BTreeMap::<(Vec<String>, Vec<String>), Vec<String>>::new();
81        let primary_endpoint = endpoints.primary_endpoint();
82
83        for endpoint in &selected_endpoints {
84            if endpoint != &primary_endpoint && !endpoints.additional_endpoints.contains_key(endpoint) {
85                return Err(generic_error!(
86                    "Experimental metrics endpoint-routing policy endpoint '{}' does not match the configured primary \
87                     endpoint and is not present in `additional_endpoints`; correct the endpoint, add it and its API \
88                     key to `additional_endpoints`, or remove it from \
89                     `experimental.metrics_endpoint_routing.metric_allowlist` and \
90                     `experimental.metrics_endpoint_routing.metric_prefix_allowlist`.",
91                    endpoint
92                ));
93            }
94
95            let mut canonical_allowlist = metric_allowlists.get(endpoint).cloned().unwrap_or_default();
96            canonical_allowlist.sort_unstable();
97            canonical_allowlist.dedup();
98            let mut canonical_prefixes = metric_prefix_allowlists.get(endpoint).cloned().unwrap_or_default();
99            compact_metric_prefixes(&mut canonical_prefixes);
100            // Exact names covered by a prefix do not change the policy's matching behavior.
101            canonical_allowlist.retain(|name| {
102                let index = canonical_prefixes.partition_point(|prefix| prefix <= name);
103                index == 0 || !name.starts_with(&canonical_prefixes[index - 1])
104            });
105            if !canonical_allowlist.is_empty() || !canonical_prefixes.is_empty() {
106                grouped_endpoints
107                    .entry((canonical_allowlist, canonical_prefixes))
108                    .or_default()
109                    .push(endpoint.clone());
110            }
111        }
112
113        let policy_groups = grouped_endpoints
114            .into_iter()
115            .map(|((metric_allowlist, metric_prefix_allowlist), mut endpoints)| {
116                endpoints.sort_unstable();
117                EndpointAllowlistGroup {
118                    endpoints,
119                    metric_allowlist,
120                    metric_prefix_allowlist,
121                }
122            })
123            .collect();
124
125        Ok(Self {
126            selected_endpoints,
127            policy_groups,
128        })
129    }
130
131    /// Returns every configured endpoint removed from the ordinary unfiltered metric path.
132    pub fn selected_endpoints(&self) -> &[String] {
133        &self.selected_endpoints
134    }
135
136    /// Returns the routing targets that receive the unfiltered metric stream.
137    ///
138    /// These are the targets whose configured endpoint has no policy.
139    pub fn unfiltered_targets(&self, catalog: &RoutingTargetCatalog) -> RoutingTargetSet {
140        catalog.select(|target| {
141            self.selected_endpoints
142                .binary_search_by(|endpoint| endpoint.as_str().cmp(target.configured_endpoint()))
143                .is_err()
144        })
145    }
146
147    /// Returns the filtered routing groups that require an encoder branch.
148    ///
149    /// Selected endpoints with an empty allowlist are absent because they receive no metric payloads.
150    pub fn policy_groups(&self) -> &[EndpointAllowlistGroup] {
151        &self.policy_groups
152    }
153}
154
155#[cfg(test)]
156mod tests {
157    use std::collections::HashMap;
158
159    use agent_data_plane_config::ConfigValue;
160
161    use super::*;
162
163    const PRIMARY: &str = "https://primary.example.com";
164
165    fn endpoints() -> shared::Endpoints {
166        shared::Endpoints {
167            dd_url: ConfigValue::explicit(PRIMARY.to_string()),
168            additional_endpoints: HashMap::from([
169                ("https://secondary-a.example.com".to_string(), vec!["key-a".to_string()]),
170                ("https://secondary-b.example.com".to_string(), vec!["key-b".to_string()]),
171                ("https://secondary-c.example.com".to_string(), vec!["key-c".to_string()]),
172            ]),
173            ..Default::default()
174        }
175    }
176
177    #[test]
178    fn empty_policy_map_preserves_the_ordinary_endpoint_path() {
179        let allowlists = HashMap::new();
180
181        let config =
182            MetricsEndpointRoutingConfiguration::from_configuration(&allowlists, &HashMap::new(), &endpoints())
183                .expect("empty routing should be valid");
184        assert!(config.selected_endpoints().is_empty());
185        assert!(config.policy_groups().is_empty());
186    }
187
188    #[test]
189    fn groups_equivalent_primary_and_additional_allowlists_and_keeps_empty_policies_selected() {
190        let allowlists = HashMap::from([
191            (
192                PRIMARY.to_string(),
193                vec!["metric.b".to_string(), "metric.a".to_string(), "metric.a".to_string()],
194            ),
195            (
196                "https://secondary-b.example.com".to_string(),
197                vec!["metric.a".to_string(), "metric.b".to_string()],
198            ),
199            ("https://secondary-c.example.com".to_string(), Vec::new()),
200        ]);
201
202        let config =
203            MetricsEndpointRoutingConfiguration::from_configuration(&allowlists, &HashMap::new(), &endpoints())
204                .expect("configured endpoints should resolve");
205        assert_eq!(
206            config.selected_endpoints(),
207            [
208                "https://primary.example.com",
209                "https://secondary-b.example.com",
210                "https://secondary-c.example.com"
211            ]
212        );
213        assert_eq!(
214            config.policy_groups(),
215            [EndpointAllowlistGroup {
216                endpoints: vec![
217                    "https://primary.example.com".to_string(),
218                    "https://secondary-b.example.com".to_string()
219                ],
220                metric_allowlist: vec!["metric.a".to_string(), "metric.b".to_string()],
221                metric_prefix_allowlist: vec![],
222            }]
223        );
224    }
225
226    #[test]
227    fn accepts_a_site_derived_primary_endpoint() {
228        let endpoints = shared::Endpoints {
229            site: ConfigValue::explicit("us5.datadoghq.com".to_string()),
230            dd_url: ConfigValue::defaulted("https://app.datadoghq.com".to_string()),
231            ..Default::default()
232        };
233        let allowlists = HashMap::from([(
234            "https://app.us5.datadoghq.com".to_string(),
235            vec!["metric.a".to_string()],
236        )]);
237
238        let config = MetricsEndpointRoutingConfiguration::from_configuration(&allowlists, &HashMap::new(), &endpoints)
239            .expect("site-derived primary endpoint should resolve");
240        assert_eq!(config.selected_endpoints(), ["https://app.us5.datadoghq.com"]);
241    }
242
243    #[test]
244    fn rejects_a_policy_that_does_not_name_a_configured_endpoint() {
245        let allowlists = HashMap::from([("https://typo.example.com".to_string(), vec!["metric.a".to_string()])]);
246
247        let error = MetricsEndpointRoutingConfiguration::from_configuration(&allowlists, &HashMap::new(), &endpoints())
248            .expect_err("unknown endpoint should be rejected");
249        let message = error.to_string();
250        assert!(message.contains("https://typo.example.com"));
251        assert!(message.contains("primary endpoint"));
252        assert!(message.contains("additional_endpoints"));
253    }
254
255    #[test]
256    fn combines_policy_maps_and_groups_equivalent_name_and_prefix_lists() {
257        let names = HashMap::from([
258            (
259                PRIMARY.to_string(),
260                vec!["exact".to_string(), "a.covered".to_string(), "b.covered".to_string()],
261            ),
262            ("https://secondary-a.example.com".to_string(), vec!["exact".to_string()]),
263        ]);
264        let prefixes = HashMap::from([
265            (
266                PRIMARY.to_string(),
267                vec![
268                    "b.".to_string(),
269                    "a.".to_string(),
270                    "a.".to_string(),
271                    "a.nested.".to_string(),
272                ],
273            ),
274            (
275                "https://secondary-a.example.com".to_string(),
276                vec!["a.".to_string(), "b.".to_string()],
277            ),
278            ("https://secondary-b.example.com".to_string(), vec!["a.".to_string()]),
279            ("https://secondary-c.example.com".to_string(), vec![]),
280        ]);
281        let config = MetricsEndpointRoutingConfiguration::from_configuration(&names, &prefixes, &endpoints()).unwrap();
282        assert_eq!(config.selected_endpoints().len(), 4);
283        assert_eq!(config.policy_groups().len(), 2);
284        let combined = config
285            .policy_groups()
286            .iter()
287            .find(|p| !p.metric_allowlist.is_empty())
288            .unwrap();
289        assert_eq!(combined.endpoints, [PRIMARY, "https://secondary-a.example.com"]);
290        assert_eq!(combined.metric_allowlist, ["exact"]);
291        assert_eq!(combined.metric_prefix_allowlist, ["a.", "b."]);
292        let prefix_only = config
293            .policy_groups()
294            .iter()
295            .find(|p| p.metric_allowlist.is_empty())
296            .unwrap();
297        assert_eq!(prefix_only.endpoints, ["https://secondary-b.example.com"]);
298        assert_eq!(prefix_only.metric_prefix_allowlist, ["a."]);
299    }
300
301    #[test]
302    fn different_prefixes_do_not_share_an_encoder_group() {
303        let names = HashMap::from([
304            (PRIMARY.to_string(), vec!["exact".to_string()]),
305            ("https://secondary-a.example.com".to_string(), vec!["exact".to_string()]),
306        ]);
307        let prefixes = HashMap::from([(PRIMARY.to_string(), vec!["a.".to_string()])]);
308        let config = MetricsEndpointRoutingConfiguration::from_configuration(&names, &prefixes, &endpoints()).unwrap();
309        assert_eq!(config.policy_groups().len(), 2);
310    }
311
312    #[test]
313    fn policies_resolve_to_complementary_routing_targets() {
314        let endpoints = endpoints();
315        let catalog =
316            RoutingTargetCatalog::new(&endpoints.primary_endpoint(), &endpoints.additional_endpoints).unwrap();
317        let allowlists = HashMap::from([
318            (PRIMARY.to_string(), vec!["metric.a".to_string()]),
319            (
320                "https://secondary-b.example.com".to_string(),
321                vec!["metric.a".to_string()],
322            ),
323            ("https://secondary-c.example.com".to_string(), vec![]),
324        ]);
325        let config =
326            MetricsEndpointRoutingConfiguration::from_configuration(&allowlists, &HashMap::new(), &endpoints).unwrap();
327
328        let configured = |targets: &RoutingTargetSet| {
329            targets
330                .iter()
331                .map(|id| catalog.get(id).unwrap().configured_endpoint().to_string())
332                .collect::<Vec<_>>()
333        };
334        // Only the endpoint without a policy keeps the unfiltered stream.
335        assert_eq!(
336            configured(&config.unfiltered_targets(&catalog)),
337            ["https://secondary-a.example.com"]
338        );
339        let [group] = config.policy_groups() else {
340            panic!("expected one policy group");
341        };
342        assert_eq!(
343            configured(&group.targets(&catalog)),
344            [PRIMARY, "https://secondary-b.example.com"]
345        );
346    }
347
348    #[test]
349    fn rejects_unknown_endpoints_in_prefix_map_including_empty_policies() {
350        for list in [vec![], vec!["metric.".to_string()]] {
351            let prefixes = HashMap::from([("https://typo.example.com".to_string(), list)]);
352            let error =
353                MetricsEndpointRoutingConfiguration::from_configuration(&HashMap::new(), &prefixes, &endpoints())
354                    .unwrap_err();
355            assert!(error.to_string().contains("metric_prefix_allowlist"));
356        }
357    }
358}