millipede-core 0.1.0

Core primitives for the Millipede web crawler: request model, storage traits, events, errors, configuration.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74

use super::Snapshotter;
use tokio::time::Instant;

/// The concurrency adjustment recommended by current system load.
#[non_exhaustive]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ScaleDecision {
    /// Increase desired concurrency.
    ScaleUp,
    /// Decrease desired concurrency.
    ScaleDown,
    /// Keep desired concurrency unchanged.
    Hold,
}

/// Options controlling load-history evaluation.
#[derive(Debug, Clone, Default)]
#[non_exhaustive]
#[must_use = "system status options do nothing unless passed to SystemStatus::new"]
pub struct SystemStatusOptions {
    /// Minimum observations a signal needs before it contributes to scale-up.
    pub min_samples: usize,
}

/// Evaluates load-signal histories into scaling decisions.
pub struct SystemStatus {
    options: SystemStatusOptions,
}

impl SystemStatus {
    /// Creates a status evaluator from the supplied options.
    pub fn new(options: SystemStatusOptions) -> Self {
        Self { options }
    }

    /// Evaluates all configured signals at the current Tokio-clock instant.
    pub fn evaluate(
        &self,
        snapshotter: &Snapshotter,
        desired_utilization_ratio: f32,
        _now: Instant,
    ) -> ScaleDecision {
        let min_samples = self.options.min_samples.max(1);
        let mut ratios = Vec::new();

        for signal in snapshotter.signals() {
            let samples = signal.sample(snapshotter.window());

            if samples
                .iter()
                .max_by_key(|sample| sample.at)
                .is_some_and(|sample| sample.overloaded)
            {
                return ScaleDecision::ScaleDown;
            }

            if samples.len() >= min_samples {
                let healthy = samples.iter().filter(|sample| !sample.overloaded).count();
                ratios.push(healthy as f32 / samples.len() as f32);
            }
        }

        if ratios.is_empty() {
            return ScaleDecision::Hold;
        }

        let mean = ratios.iter().sum::<f32>() / ratios.len() as f32;
        if mean >= desired_utilization_ratio {
            ScaleDecision::ScaleUp
        } else {
            ScaleDecision::Hold
        }
    }
}