Skip to main content

a3s_box_runtime/
operator.rs

1//! BoxAutoscaler controller — reconciliation logic for the K8s operator.
2//!
3//! Evaluates metrics against the BoxAutoscaler spec and computes the
4//! desired replica count. This is the core decision engine that can be
5//! used standalone or inside a kube-rs controller.
6
7use std::time::Instant;
8
9use a3s_box_core::operator::{
10    AutoscalerCondition, BoxAutoscalerSpec, BoxAutoscalerStatus, MetricType, MetricValue,
11};
12use chrono::Utc;
13use serde::{Deserialize, Serialize};
14
15/// Input metrics for the reconciler.
16#[derive(Debug, Clone, Default)]
17pub struct ObservedMetrics {
18    /// Average CPU utilization (0-100).
19    pub avg_cpu_percent: Option<f32>,
20    /// Average memory utilization (0-100).
21    pub avg_memory_percent: Option<f32>,
22    /// Total in-flight requests.
23    pub total_inflight: Option<u32>,
24    /// Requests per second.
25    pub rps: Option<u32>,
26    /// Custom metric values (name → value).
27    pub custom: std::collections::HashMap<String, u32>,
28}
29
30/// Result of a reconciliation cycle.
31#[derive(Debug, Clone, Serialize, Deserialize)]
32pub struct ReconcileResult {
33    /// Computed desired replicas.
34    pub desired_replicas: u32,
35    /// Whether a scale action is needed.
36    pub scale_needed: bool,
37    /// Direction: "up", "down", or "none".
38    pub direction: String,
39    /// Reason for the decision.
40    pub reason: String,
41    /// Updated metric values for status.
42    pub metric_values: Vec<MetricValue>,
43}
44
45/// Autoscaler controller that computes desired replicas from metrics.
46pub struct AutoscalerController {
47    /// Last time a scale-up was performed.
48    last_scale_up: Option<Instant>,
49    /// Last time a scale-down was performed.
50    last_scale_down: Option<Instant>,
51}
52
53impl Default for AutoscalerController {
54    fn default() -> Self {
55        Self::new()
56    }
57}
58
59impl AutoscalerController {
60    /// Create a new controller.
61    pub fn new() -> Self {
62        Self {
63            last_scale_up: None,
64            last_scale_down: None,
65        }
66    }
67
68    /// Reconcile the desired state against observed metrics.
69    ///
70    /// Returns the computed desired replica count and whether scaling is needed.
71    pub fn reconcile(
72        &mut self,
73        spec: &BoxAutoscalerSpec,
74        current_replicas: u32,
75        metrics: &ObservedMetrics,
76    ) -> ReconcileResult {
77        if spec.metrics.is_empty() {
78            return ReconcileResult {
79                desired_replicas: current_replicas,
80                scale_needed: false,
81                direction: "none".to_string(),
82                reason: "No metrics configured".to_string(),
83                metric_values: Vec::new(),
84            };
85        }
86
87        let mut computed_desired: Option<u32> = None;
88        let mut metric_values = Vec::new();
89        let mut reasons = Vec::new();
90
91        for metric_spec in &spec.metrics {
92            let current_value = match metric_spec.metric_type {
93                MetricType::Cpu => metrics.avg_cpu_percent.map(|v| v as u32),
94                MetricType::Memory => metrics.avg_memory_percent.map(|v| v as u32),
95                MetricType::Inflight => metrics.total_inflight,
96                MetricType::Rps => metrics.rps,
97                MetricType::Custom => None, // Custom metrics not evaluated here
98            };
99
100            let current = match current_value {
101                Some(v) => v,
102                None => continue,
103            };
104
105            metric_values.push(MetricValue {
106                metric_type: metric_spec.metric_type,
107                current,
108                target: metric_spec.target,
109            });
110
111            // Compute desired replicas for this metric
112            let desired = compute_desired_replicas(
113                current_replicas,
114                current,
115                metric_spec.target,
116                metric_spec.tolerance_percent,
117                spec.min_replicas,
118                spec.max_replicas,
119            );
120
121            // Take the maximum across all metrics (most aggressive)
122            computed_desired =
123                Some(computed_desired.map_or(desired, |prev: u32| prev.max(desired)));
124
125            if desired > current_replicas {
126                reasons.push(format!(
127                    "{} at {}% (target {}%)",
128                    metric_spec.metric_type, current, metric_spec.target
129                ));
130            }
131        }
132
133        // No metric data available — hold at current
134        let max_desired = match computed_desired {
135            Some(d) => d,
136            None => {
137                return ReconcileResult {
138                    desired_replicas: current_replicas,
139                    scale_needed: false,
140                    direction: "none".to_string(),
141                    reason: "No metric data available".to_string(),
142                    metric_values,
143                };
144            }
145        };
146
147        let desired = max_desired.clamp(spec.min_replicas, spec.max_replicas);
148
149        let (scale_needed, direction, reason) = if desired > current_replicas {
150            // Check scale-up stabilization
151            if let Some(last) = self.last_scale_up {
152                if last.elapsed().as_secs() < spec.behavior.scale_up.stabilization_window_secs {
153                    return ReconcileResult {
154                        desired_replicas: current_replicas,
155                        scale_needed: false,
156                        direction: "none".to_string(),
157                        reason: "Scale-up stabilization window active".to_string(),
158                        metric_values,
159                    };
160                }
161            }
162            self.last_scale_up = Some(Instant::now());
163            (
164                true,
165                "up".to_string(),
166                if reasons.is_empty() {
167                    "Metrics above target".to_string()
168                } else {
169                    reasons.join("; ")
170                },
171            )
172        } else if desired < current_replicas {
173            // Check scale-down stabilization
174            if let Some(last) = self.last_scale_down {
175                if last.elapsed().as_secs() < spec.behavior.scale_down.stabilization_window_secs {
176                    return ReconcileResult {
177                        desired_replicas: current_replicas,
178                        scale_needed: false,
179                        direction: "none".to_string(),
180                        reason: "Scale-down stabilization window active".to_string(),
181                        metric_values,
182                    };
183                }
184            }
185            self.last_scale_down = Some(Instant::now());
186            (true, "down".to_string(), "Metrics below target".to_string())
187        } else {
188            (false, "none".to_string(), "At target".to_string())
189        };
190
191        ReconcileResult {
192            desired_replicas: desired,
193            scale_needed,
194            direction,
195            reason,
196            metric_values,
197        }
198    }
199
200    /// Build an updated status from a reconcile result.
201    pub fn build_status(
202        &self,
203        current_replicas: u32,
204        result: &ReconcileResult,
205    ) -> BoxAutoscalerStatus {
206        let mut conditions = Vec::new();
207
208        conditions.push(AutoscalerCondition {
209            condition_type: "Ready".to_string(),
210            status: "True".to_string(),
211            last_transition_time: Some(Utc::now()),
212            reason: "Reconciled".to_string(),
213            message: result.reason.clone(),
214        });
215
216        if result.scale_needed {
217            conditions.push(AutoscalerCondition {
218                condition_type: "ScalingActive".to_string(),
219                status: "True".to_string(),
220                last_transition_time: Some(Utc::now()),
221                reason: format!(
222                    "Scale{}",
223                    if result.direction == "up" {
224                        "Up"
225                    } else {
226                        "Down"
227                    }
228                ),
229                message: format!(
230                    "{} → {} replicas",
231                    current_replicas, result.desired_replicas
232                ),
233            });
234        }
235
236        BoxAutoscalerStatus {
237            current_replicas,
238            desired_replicas: result.desired_replicas,
239            last_scale_time: if result.scale_needed {
240                Some(Utc::now())
241            } else {
242                None
243            },
244            current_metrics: result.metric_values.clone(),
245            conditions,
246        }
247    }
248}
249
250/// Compute desired replicas for a single metric using the ratio algorithm.
251///
252/// `desired = ceil(current_replicas * (current_value / target_value))`
253///
254/// Respects tolerance: no change if within ±tolerance% of target.
255fn compute_desired_replicas(
256    current_replicas: u32,
257    current_value: u32,
258    target_value: u32,
259    tolerance_percent: u32,
260    min_replicas: u32,
261    max_replicas: u32,
262) -> u32 {
263    if target_value == 0 || current_replicas == 0 {
264        return min_replicas;
265    }
266
267    let ratio = current_value as f64 / target_value as f64;
268    let tolerance = tolerance_percent as f64 / 100.0;
269
270    // Within tolerance band — no change
271    if (ratio - 1.0).abs() <= tolerance {
272        return current_replicas;
273    }
274
275    let desired = (current_replicas as f64 * ratio).ceil() as u32;
276    desired.clamp(min_replicas, max_replicas)
277}
278
279#[cfg(test)]
280mod tests {
281    use super::*;
282    use a3s_box_core::operator::{
283        BoxAutoscalerSpec, MetricSpec, MetricType, ScalingBehavior, ScalingRules, TargetRef,
284    };
285
286    fn test_spec(max: u32) -> BoxAutoscalerSpec {
287        BoxAutoscalerSpec {
288            target_ref: TargetRef {
289                kind: "BoxService".to_string(),
290                name: "test".to_string(),
291                namespace: "default".to_string(),
292            },
293            min_replicas: 1,
294            max_replicas: max,
295            metrics: vec![MetricSpec {
296                metric_type: MetricType::Cpu,
297                target: 70,
298                tolerance_percent: 10,
299            }],
300            behavior: ScalingBehavior {
301                scale_up: ScalingRules {
302                    stabilization_window_secs: 0,
303                    max_scale_per_minute: 10,
304                },
305                scale_down: ScalingRules {
306                    stabilization_window_secs: 0,
307                    max_scale_per_minute: 10,
308                },
309            },
310            cooldown_secs: 0,
311        }
312    }
313
314    #[test]
315    fn test_compute_desired_replicas_above_target() {
316        // 3 replicas, CPU at 90%, target 70% → ratio 1.28 → ceil(3 * 1.28) = 4
317        let desired = compute_desired_replicas(3, 90, 70, 10, 1, 10);
318        assert_eq!(desired, 4);
319    }
320
321    #[test]
322    fn test_compute_desired_replicas_below_target() {
323        // 5 replicas, CPU at 30%, target 70% → ratio 0.43 → ceil(5 * 0.43) = 3
324        let desired = compute_desired_replicas(5, 30, 70, 10, 1, 10);
325        assert_eq!(desired, 3);
326    }
327
328    #[test]
329    fn test_compute_desired_replicas_within_tolerance() {
330        // 3 replicas, CPU at 75%, target 70%, tolerance 10% → ratio 1.07 → within ±10%
331        let desired = compute_desired_replicas(3, 75, 70, 10, 1, 10);
332        assert_eq!(desired, 3); // No change
333    }
334
335    #[test]
336    fn test_compute_desired_replicas_clamped_max() {
337        // Would want 15 but max is 10
338        let desired = compute_desired_replicas(5, 200, 70, 10, 1, 10);
339        assert_eq!(desired, 10);
340    }
341
342    #[test]
343    fn test_compute_desired_replicas_clamped_min() {
344        // Would want 0 but min is 1
345        let desired = compute_desired_replicas(5, 1, 70, 10, 1, 10);
346        assert_eq!(desired, 1);
347    }
348
349    #[test]
350    fn test_compute_desired_replicas_zero_target() {
351        let desired = compute_desired_replicas(3, 50, 0, 10, 1, 10);
352        assert_eq!(desired, 1); // min_replicas
353    }
354
355    #[test]
356    fn test_reconcile_scale_up() {
357        let mut ctrl = AutoscalerController::new();
358        let spec = test_spec(10);
359        let metrics = ObservedMetrics {
360            avg_cpu_percent: Some(95.0),
361            ..Default::default()
362        };
363
364        let result = ctrl.reconcile(&spec, 3, &metrics);
365        assert!(result.scale_needed);
366        assert_eq!(result.direction, "up");
367        assert!(result.desired_replicas > 3);
368        assert!(!result.metric_values.is_empty());
369    }
370
371    #[test]
372    fn test_reconcile_scale_down() {
373        let mut ctrl = AutoscalerController::new();
374        let spec = test_spec(10);
375        let metrics = ObservedMetrics {
376            avg_cpu_percent: Some(20.0),
377            ..Default::default()
378        };
379
380        let result = ctrl.reconcile(&spec, 5, &metrics);
381        assert!(result.scale_needed);
382        assert_eq!(result.direction, "down");
383        assert!(result.desired_replicas < 5);
384    }
385
386    #[test]
387    fn test_reconcile_no_change() {
388        let mut ctrl = AutoscalerController::new();
389        let spec = test_spec(10);
390        let metrics = ObservedMetrics {
391            avg_cpu_percent: Some(72.0), // Within 10% tolerance of 70
392            ..Default::default()
393        };
394
395        let result = ctrl.reconcile(&spec, 3, &metrics);
396        assert!(!result.scale_needed);
397        assert_eq!(result.direction, "none");
398        assert_eq!(result.desired_replicas, 3);
399    }
400
401    #[test]
402    fn test_reconcile_no_metrics_configured() {
403        let mut ctrl = AutoscalerController::new();
404        let mut spec = test_spec(10);
405        spec.metrics.clear();
406
407        let metrics = ObservedMetrics::default();
408        let result = ctrl.reconcile(&spec, 3, &metrics);
409        assert!(!result.scale_needed);
410        assert_eq!(result.desired_replicas, 3);
411    }
412
413    #[test]
414    fn test_reconcile_no_metric_data() {
415        let mut ctrl = AutoscalerController::new();
416        let spec = test_spec(10);
417        let metrics = ObservedMetrics::default(); // No CPU data
418
419        let result = ctrl.reconcile(&spec, 3, &metrics);
420        assert!(!result.scale_needed);
421    }
422
423    #[test]
424    fn test_reconcile_stabilization_window() {
425        let mut ctrl = AutoscalerController::new();
426        let mut spec = test_spec(10);
427        spec.behavior.scale_up.stabilization_window_secs = 3600; // 1 hour
428
429        let metrics = ObservedMetrics {
430            avg_cpu_percent: Some(95.0),
431            ..Default::default()
432        };
433
434        // First reconcile: scale up
435        let r1 = ctrl.reconcile(&spec, 3, &metrics);
436        assert!(r1.scale_needed);
437
438        // Second reconcile immediately: blocked by stabilization
439        let r2 = ctrl.reconcile(&spec, 3, &metrics);
440        assert!(!r2.scale_needed);
441        assert!(r2.reason.contains("stabilization"));
442    }
443
444    #[test]
445    fn test_reconcile_multiple_metrics() {
446        let mut ctrl = AutoscalerController::new();
447        let mut spec = test_spec(10);
448        spec.metrics = vec![
449            MetricSpec {
450                metric_type: MetricType::Cpu,
451                target: 70,
452                tolerance_percent: 10,
453            },
454            MetricSpec {
455                metric_type: MetricType::Inflight,
456                target: 50,
457                tolerance_percent: 10,
458            },
459        ];
460
461        let metrics = ObservedMetrics {
462            avg_cpu_percent: Some(65.0), // Within tolerance
463            total_inflight: Some(200),   // Way above target
464            ..Default::default()
465        };
466
467        let result = ctrl.reconcile(&spec, 2, &metrics);
468        assert!(result.scale_needed);
469        assert_eq!(result.direction, "up");
470        // Should scale based on the highest demand metric (inflight)
471        assert!(result.desired_replicas > 2);
472    }
473
474    #[test]
475    fn test_reconcile_respects_max_replicas() {
476        let mut ctrl = AutoscalerController::new();
477        let spec = test_spec(5); // max 5
478
479        let metrics = ObservedMetrics {
480            avg_cpu_percent: Some(200.0), // Extreme load
481            ..Default::default()
482        };
483
484        let result = ctrl.reconcile(&spec, 4, &metrics);
485        assert!(result.desired_replicas <= 5);
486    }
487
488    #[test]
489    fn test_reconcile_respects_min_replicas() {
490        let mut ctrl = AutoscalerController::new();
491        let mut spec = test_spec(10);
492        spec.min_replicas = 2;
493
494        let metrics = ObservedMetrics {
495            avg_cpu_percent: Some(1.0), // Almost idle
496            ..Default::default()
497        };
498
499        let result = ctrl.reconcile(&spec, 5, &metrics);
500        assert!(result.desired_replicas >= 2);
501    }
502
503    #[test]
504    fn test_build_status_no_scale() {
505        let ctrl = AutoscalerController::new();
506        let result = ReconcileResult {
507            desired_replicas: 3,
508            scale_needed: false,
509            direction: "none".to_string(),
510            reason: "At target".to_string(),
511            metric_values: vec![],
512        };
513
514        let status = ctrl.build_status(3, &result);
515        assert_eq!(status.current_replicas, 3);
516        assert_eq!(status.desired_replicas, 3);
517        assert!(status.last_scale_time.is_none());
518        assert_eq!(status.conditions.len(), 1);
519        assert_eq!(status.conditions[0].condition_type, "Ready");
520    }
521
522    #[test]
523    fn test_build_status_with_scale() {
524        let ctrl = AutoscalerController::new();
525        let result = ReconcileResult {
526            desired_replicas: 5,
527            scale_needed: true,
528            direction: "up".to_string(),
529            reason: "cpu at 90% (target 70%)".to_string(),
530            metric_values: vec![MetricValue {
531                metric_type: MetricType::Cpu,
532                current: 90,
533                target: 70,
534            }],
535        };
536
537        let status = ctrl.build_status(3, &result);
538        assert_eq!(status.current_replicas, 3);
539        assert_eq!(status.desired_replicas, 5);
540        assert!(status.last_scale_time.is_some());
541        assert_eq!(status.conditions.len(), 2); // Ready + ScalingActive
542        assert_eq!(status.conditions[1].condition_type, "ScalingActive");
543    }
544
545    #[test]
546    fn test_controller_new() {
547        let ctrl = AutoscalerController::new();
548        assert!(ctrl.last_scale_up.is_none());
549        assert!(ctrl.last_scale_down.is_none());
550    }
551}