saluki_components/transforms/autoscaling_failover_gateway/
mod.rs

1//! Autoscaling failover metrics gateway transform.
2
3use std::collections::HashSet;
4
5use async_trait::async_trait;
6use saluki_core::accounting::{MemoryBounds, MemoryBoundsBuilder};
7use saluki_core::{
8    components::{
9        transforms::{Transform, TransformBuilder, TransformContext},
10        BuildContext,
11    },
12    data_model::event::{metric::MetricValues, Event, EventType},
13    topology::{EventsBuffer, OutputDefinition},
14};
15use saluki_error::GenericError;
16use tokio::select;
17use tracing::{debug, error};
18
19use crate::config::AutoscalingFailoverConfiguration;
20
21/// Autoscaling failover metrics gateway transform configuration.
22///
23/// This transform sits between the shared metrics enrichment stage and the Cluster Agent encoder/forwarder branch. It
24/// forwards only configured series metrics to the Cluster Agent branch and drops everything else.
25pub struct AutoscalingFailoverGatewayConfiguration {
26    failover_config: AutoscalingFailoverConfiguration,
27}
28
29impl AutoscalingFailoverGatewayConfiguration {
30    /// Creates a new `AutoscalingFailoverGatewayConfiguration` from the given failover configuration.
31    pub fn new(failover_config: AutoscalingFailoverConfiguration) -> Self {
32        Self { failover_config }
33    }
34}
35
36/// Routing and filtering state for the autoscaling failover metrics gateway.
37#[derive(Debug)]
38enum GatewayMode {
39    /// Autoscaling failover is disabled or has no allowed metrics; drop all events.
40    Inactive,
41    /// Autoscaling failover is active; forward only matching series metrics.
42    FilteredForward { allowlist: HashSet<String> },
43}
44
45/// Autoscaling failover metrics gateway transform.
46pub struct AutoscalingFailoverGateway {
47    mode: GatewayMode,
48}
49
50impl AutoscalingFailoverGateway {
51    fn new(failover_config: AutoscalingFailoverConfiguration) -> Self {
52        let mode = Self::mode_for_config(&failover_config);
53
54        Self { mode }
55    }
56
57    fn mode_for_config(failover_config: &AutoscalingFailoverConfiguration) -> GatewayMode {
58        if !failover_config.is_branch_requested() {
59            GatewayMode::Inactive
60        } else {
61            GatewayMode::FilteredForward {
62                allowlist: failover_config.metrics().iter().cloned().collect(),
63            }
64        }
65    }
66
67    fn should_forward(&self, event: &Event) -> bool {
68        let GatewayMode::FilteredForward { allowlist } = &self.mode else {
69            return false;
70        };
71        let Event::Metric(metric) = event else {
72            return false;
73        };
74        if !matches!(
75            metric.values(),
76            MetricValues::Counter(..) | MetricValues::Rate(..) | MetricValues::Gauge(..) | MetricValues::Set(..)
77        ) {
78            return false;
79        }
80
81        allowlist.contains(metric.context().name().as_ref())
82    }
83
84    async fn process_event_batch(
85        &self, mut events: EventsBuffer, context: &mut TransformContext,
86    ) -> Result<(), GenericError> {
87        let input_events = events.len();
88        events.remove_if(|event| !self.should_forward(event));
89
90        let sent_events = context.dispatcher().buffered()?.send_all(events).await?;
91        let dropped_events = input_events - sent_events;
92        debug!(
93            sent_events,
94            dropped_events, "Autoscaling failover metrics gateway processed event batch."
95        );
96
97        Ok(())
98    }
99}
100
101#[async_trait]
102impl TransformBuilder for AutoscalingFailoverGatewayConfiguration {
103    async fn build(&self, _context: BuildContext) -> Result<Box<dyn Transform + Send>, GenericError> {
104        Ok(Box::new(AutoscalingFailoverGateway::new(self.failover_config.clone())))
105    }
106
107    fn input_event_type(&self) -> EventType {
108        EventType::Metric
109    }
110
111    fn outputs(&self) -> &[OutputDefinition<EventType>] {
112        static OUTPUTS: &[OutputDefinition<EventType>] = &[OutputDefinition::default_output(EventType::Metric)];
113        OUTPUTS
114    }
115}
116
117impl MemoryBounds for AutoscalingFailoverGatewayConfiguration {
118    fn specify_bounds(&self, builder: &mut MemoryBoundsBuilder) {
119        let metrics = self.failover_config.metrics();
120        builder
121            .minimum()
122            .with_single_value::<AutoscalingFailoverGateway>("component struct")
123            .with_fixed_amount("hashset overhead", std::mem::size_of::<HashSet<String>>())
124            .with_fixed_amount(
125                "allowlist strings",
126                metrics
127                    .iter()
128                    .map(|name| name.len() + std::mem::size_of::<String>())
129                    .sum::<usize>(),
130            )
131            .with_fixed_amount(
132                "hashset buckets",
133                metrics.len() * std::mem::size_of::<Option<String>>() * 2,
134            );
135    }
136}
137
138#[async_trait]
139impl Transform for AutoscalingFailoverGateway {
140    async fn run(self: Box<Self>, mut context: TransformContext) -> Result<(), GenericError> {
141        let mut health = context.take_health_handle();
142
143        health.mark_ready();
144        debug!(mode = ?self.mode, "Autoscaling failover metrics gateway transform started.");
145
146        loop {
147            select! {
148                _ = health.live() => continue,
149                maybe_events = context.events().next() => match maybe_events {
150                    Some(events) => {
151                        if let Err(e) = self.process_event_batch(events, &mut context).await {
152                            error!(error = %e, "Autoscaling failover metrics gateway failed to process event batch.");
153                        }
154                    }
155                    None => {
156                        debug!("Event stream terminated, shutting down autoscaling failover metrics gateway transform.");
157                        break;
158                    }
159                },
160            }
161        }
162
163        debug!("Autoscaling failover metrics gateway transform stopped.");
164        Ok(())
165    }
166}
167
168#[cfg(test)]
169mod tests {
170    use std::time::Duration;
171
172    use saluki_core::data_model::event::{metric::Metric, Event};
173
174    use super::*;
175
176    fn gateway_from_config(enabled: bool, metrics: &[&str]) -> AutoscalingFailoverGateway {
177        let metrics = metrics.iter().map(|name| (*name).to_string()).collect();
178        AutoscalingFailoverGateway::new(AutoscalingFailoverConfiguration::new(enabled, metrics))
179    }
180
181    #[test]
182    fn inactive_gateway_drops_everything() {
183        let gw = gateway_from_config(false, &["allowed.metric"]);
184
185        assert!(!gw.should_forward(&Event::Metric(Metric::counter("allowed.metric", 1.0))));
186    }
187
188    #[test]
189    fn empty_metric_allowlist_drops_everything() {
190        let gw = gateway_from_config(true, &[]);
191
192        assert!(!gw.should_forward(&Event::Metric(Metric::counter("allowed.metric", 1.0))));
193    }
194
195    #[test]
196    fn active_gateway_forwards_only_allowed_series_metrics() {
197        let gw = gateway_from_config(
198            true,
199            &["allowed.counter", "allowed.gauge", "allowed.rate", "allowed.set"],
200        );
201
202        assert!(gw.should_forward(&Event::Metric(Metric::counter("allowed.counter", 1.0))));
203        assert!(gw.should_forward(&Event::Metric(Metric::gauge("allowed.gauge", 1.0))));
204        assert!(gw.should_forward(&Event::Metric(Metric::rate(
205            "allowed.rate",
206            1.0,
207            Duration::from_secs(10)
208        ))));
209        assert!(gw.should_forward(&Event::Metric(Metric::set("allowed.set", "value"))));
210        assert!(!gw.should_forward(&Event::Metric(Metric::counter("blocked.counter", 1.0))));
211    }
212
213    #[test]
214    fn active_gateway_drops_sketch_metrics_even_when_allowed() {
215        let gw = gateway_from_config(true, &["allowed.histogram", "allowed.distribution"]);
216
217        assert!(!gw.should_forward(&Event::Metric(Metric::histogram("allowed.histogram", [1.0, 2.0, 3.0]))));
218        assert!(!gw.should_forward(&Event::Metric(Metric::distribution(
219            "allowed.distribution",
220            [1.0, 2.0, 3.0]
221        ))));
222    }
223}
saluki_components/transforms/autoscaling_failover_gateway/mod.rs

saluki_components/transforms/autoscaling_failover_gateway/
mod.rs
