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

saluki_components/config/
metrics_endpoint_routing.rs
