hydracache/grid/
capacity.rs

1use std::fmt;
2
3use serde::{Deserialize, Serialize};
4
5use crate::cluster::{ClusterEpoch, ClusterNodeId, PartitionId};
6use crate::grid::elasticity::{
7    validate_move_preserves_zone_quorum, MovePhase, NodeTopology, PartitionMove, RegionId,
8    ReshardPlan, ReshardPlanError, UpgradeGuard, UpgradeGuardError, UpgradeStep,
9    ZoneAwareReplicaSet,
10};
11use crate::grid::ReplicationConfig;
12
13/// Recommendation emitted for an external autoscaler or operator.
14#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
15#[serde(rename_all = "snake_case")]
16pub enum ScaleRecommendation {
17    /// No scaling action should be taken.
18    Hold,
19    /// Add nodes and backfill before quorum admission.
20    ScaleOut {
21        /// Suggested number of nodes to add.
22        suggested: usize,
23    },
24    /// Drain listed nodes before removal.
25    ScaleIn {
26        /// Nodes that can be drained by an external autoscaler.
27        drain: Vec<ClusterNodeId>,
28    },
29    /// Rebalance hot partitions without changing membership.
30    Rebalance,
31}
32
33/// One capacity sample consumed by the recommendation engine.
34#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
35pub struct CapacitySample {
36    /// Region that owns this sample.
37    pub region: RegionId,
38    /// Memory pressure, normalized to `0.0..=1.0+`.
39    pub memory_pressure: f32,
40    /// Cross-region replication lag.
41    pub replication_lag: u64,
42    /// Hot-partition skew ratio.
43    pub hot_partition_skew: f32,
44    /// Repair debt that can block safe scaling.
45    pub repair_debt: u64,
46    /// Seconds since the last accepted scale action.
47    pub seconds_since_last_scale: u64,
48    /// Nodes eligible for scale-in drain.
49    pub scale_in_candidates: Vec<ClusterNodeId>,
50}
51
52impl CapacitySample {
53    /// Create a capacity sample with no scale-in candidates.
54    pub fn new(region: impl Into<RegionId>) -> Self {
55        Self {
56            region: region.into(),
57            memory_pressure: 0.0,
58            replication_lag: 0,
59            hot_partition_skew: 0.0,
60            repair_debt: 0,
61            seconds_since_last_scale: u64::MAX,
62            scale_in_candidates: Vec::new(),
63        }
64    }
65}
66
67/// Tunables for capacity recommendations.
68#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)]
69pub struct CapacityThresholds {
70    /// Memory pressure at which scale-out is recommended.
71    pub scale_out_memory_pressure: f32,
72    /// Memory pressure below which scale-in can be recommended.
73    pub scale_in_memory_pressure: f32,
74    /// Replication lag at which scale-out is recommended.
75    pub replication_lag_limit: u64,
76    /// Hot-partition skew at which rebalance is recommended.
77    pub hot_partition_skew_limit: f32,
78    /// Repair debt at which rebalance is recommended.
79    pub repair_debt_limit: u64,
80    /// Minimum seconds between accepted scale actions.
81    pub minimum_dwell_secs: u64,
82    /// Suggested node count for scale-out recommendations.
83    pub scale_out_suggested: usize,
84}
85
86impl Default for CapacityThresholds {
87    fn default() -> Self {
88        Self {
89            scale_out_memory_pressure: 0.85,
90            scale_in_memory_pressure: 0.25,
91            replication_lag_limit: 1_000,
92            hot_partition_skew_limit: 2.0,
93            repair_debt_limit: 100,
94            minimum_dwell_secs: 300,
95            scale_out_suggested: 1,
96        }
97    }
98}
99
100/// Structured capacity signal exported by status/metrics.
101#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
102pub struct CapacitySignal {
103    /// Region this signal describes.
104    pub region: RegionId,
105    /// Memory pressure, normalized to `0.0..=1.0+`.
106    pub memory_pressure: f32,
107    /// Cross-region replication lag.
108    pub replication_lag: u64,
109    /// Hot-partition skew ratio.
110    pub hot_partition_skew: f32,
111    /// Repair debt that can block safe scaling.
112    pub repair_debt: u64,
113    /// Autoscaler recommendation.
114    pub recommendation: ScaleRecommendation,
115}
116
117/// Compute one capacity recommendation with dwell-time hysteresis.
118pub fn evaluate_capacity(sample: CapacitySample, thresholds: CapacityThresholds) -> CapacitySignal {
119    let in_dwell_window = sample.seconds_since_last_scale < thresholds.minimum_dwell_secs;
120    let recommendation = if in_dwell_window {
121        ScaleRecommendation::Hold
122    } else if sample.memory_pressure >= thresholds.scale_out_memory_pressure
123        || sample.replication_lag > thresholds.replication_lag_limit
124    {
125        ScaleRecommendation::ScaleOut {
126            suggested: thresholds.scale_out_suggested.max(1),
127        }
128    } else if sample.hot_partition_skew >= thresholds.hot_partition_skew_limit
129        || sample.repair_debt > thresholds.repair_debt_limit
130    {
131        ScaleRecommendation::Rebalance
132    } else if sample.memory_pressure <= thresholds.scale_in_memory_pressure
133        && !sample.scale_in_candidates.is_empty()
134    {
135        ScaleRecommendation::ScaleIn {
136            drain: sample.scale_in_candidates.clone(),
137        }
138    } else {
139        ScaleRecommendation::Hold
140    };
141
142    CapacitySignal {
143        region: sample.region,
144        memory_pressure: sample.memory_pressure,
145        replication_lag: sample.replication_lag,
146        hot_partition_skew: sample.hot_partition_skew,
147        repair_debt: sample.repair_debt,
148        recommendation,
149    }
150}
151
152/// Autoscaler membership intent accepted through the guarded admission surface.
153#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
154#[serde(rename_all = "snake_case")]
155pub enum AutoscalerIntent {
156    /// Add a node, backfill partitions, then count it toward quorum.
157    ScaleOut {
158        /// Joining node.
159        node: ClusterNodeId,
160        /// Authoritative topology for the joining node.
161        topology: NodeTopology,
162        /// Candidate replica set after the join.
163        candidate: ZoneAwareReplicaSet,
164        /// Backfill work from existing owners into the joining node.
165        backfill_sources: Vec<(PartitionId, ClusterNodeId, u64)>,
166        /// Compatibility step advertised by the joining binary.
167        compat: UpgradeStep,
168    },
169    /// Drain a node before removing it.
170    ScaleIn {
171        /// Node being drained.
172        drain: ClusterNodeId,
173        /// Remaining voters after removal.
174        remaining_voters: usize,
175        /// Drain movements to surviving owners.
176        drain_targets: Vec<(PartitionId, ClusterNodeId, u64)>,
177        /// Compatibility step advertised by the operator/controller.
178        compat: UpgradeStep,
179    },
180    /// Rebalance partitions without a membership change.
181    Rebalance {
182        /// Candidate reshard plan.
183        plan: ReshardPlan,
184        /// Compatibility step advertised by the operator/controller.
185        compat: UpgradeStep,
186    },
187}
188
189/// Guardrails used to admit an autoscaler intent.
190#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
191pub struct AutoscalerAdmissionPolicy {
192    /// Current control-plane epoch.
193    pub epoch: ClusterEpoch,
194    /// Replication/quorum configuration.
195    pub replication: ReplicationConfig,
196    /// Compatibility guard.
197    pub upgrade_guard: UpgradeGuard,
198    /// Maximum concurrent reshard moves.
199    pub max_concurrent_moves: usize,
200}
201
202impl AutoscalerAdmissionPolicy {
203    /// Create a policy with normalized concurrency.
204    pub fn new(
205        epoch: ClusterEpoch,
206        replication: ReplicationConfig,
207        upgrade_guard: UpgradeGuard,
208        max_concurrent_moves: usize,
209    ) -> Self {
210        Self {
211            epoch,
212            replication,
213            upgrade_guard,
214            max_concurrent_moves: max_concurrent_moves.max(1),
215        }
216    }
217}
218
219/// Accepted scale action.
220#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
221#[serde(rename_all = "snake_case")]
222pub enum ScaleAction {
223    /// Add a node.
224    ScaleOut,
225    /// Drain/remove a node.
226    ScaleIn,
227    /// Rebalance partitions.
228    Rebalance,
229}
230
231/// Accepted autoscaler admission result.
232#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
233pub struct AutoscalerAdmission {
234    /// Action accepted by the guard.
235    pub action: ScaleAction,
236    /// Reshard/drain/backfill work to execute before final membership effect.
237    pub plan: ReshardPlan,
238    /// Whether a joining node can count toward quorum immediately.
239    pub quorum_eligible: bool,
240    /// Whether a draining node may be removed immediately.
241    pub removal_allowed: bool,
242}
243
244/// Validate an autoscaler intent against zone, quorum, and COMPAT invariants.
245pub fn admit_autoscaler_intent(
246    intent: AutoscalerIntent,
247    policy: AutoscalerAdmissionPolicy,
248) -> Result<AutoscalerAdmission, AutoscalerIntentError> {
249    policy
250        .replication
251        .validate()
252        .map_err(|error| AutoscalerIntentError::new(error.to_string()))?;
253
254    match intent {
255        AutoscalerIntent::ScaleOut {
256            node,
257            topology: _,
258            candidate,
259            backfill_sources,
260            compat,
261        } => {
262            policy
263                .upgrade_guard
264                .check(compat)
265                .map_err(AutoscalerIntentError::from)?;
266            validate_move_preserves_zone_quorum(&candidate, policy.replication.write_quorum)
267                .map_err(AutoscalerIntentError::from)?;
268            let moves = backfill_sources
269                .into_iter()
270                .map(|(partition, from, bytes)| {
271                    PartitionMove::new(partition, from, node.clone(), bytes)
272                })
273                .collect();
274            Ok(AutoscalerAdmission {
275                action: ScaleAction::ScaleOut,
276                plan: ReshardPlan::new(policy.epoch, moves, policy.max_concurrent_moves),
277                quorum_eligible: false,
278                removal_allowed: false,
279            })
280        }
281        AutoscalerIntent::ScaleIn {
282            drain,
283            remaining_voters,
284            drain_targets,
285            compat,
286        } => {
287            policy
288                .upgrade_guard
289                .check(compat)
290                .map_err(AutoscalerIntentError::from)?;
291            if remaining_voters < policy.replication.write_quorum {
292                return Err(AutoscalerIntentError::new(
293                    "autoscaler intent would break write quorum",
294                ));
295            }
296            Ok(AutoscalerAdmission {
297                action: ScaleAction::ScaleIn,
298                plan: ReshardPlan::drain_node(
299                    policy.epoch,
300                    drain,
301                    drain_targets,
302                    policy.max_concurrent_moves,
303                ),
304                quorum_eligible: true,
305                removal_allowed: false,
306            })
307        }
308        AutoscalerIntent::Rebalance { plan, compat } => {
309            policy
310                .upgrade_guard
311                .check(compat)
312                .map_err(AutoscalerIntentError::from)?;
313            Ok(AutoscalerAdmission {
314                action: ScaleAction::Rebalance,
315                plan,
316                quorum_eligible: true,
317                removal_allowed: true,
318            })
319        }
320    }
321}
322
323/// Return whether every move has reached the committed owner.
324pub fn scale_out_counts_toward_quorum(plan: &ReshardPlan) -> bool {
325    !plan.moves.is_empty()
326        && plan
327            .moves
328            .iter()
329            .all(|movement| movement.phase >= MovePhase::Commit)
330}
331
332/// Return whether a scale-in drain completed and the node can be removed.
333pub fn scale_in_removal_allowed(plan: &ReshardPlan) -> bool {
334    !plan.moves.is_empty()
335        && plan
336            .moves
337            .iter()
338            .all(|movement| movement.phase == MovePhase::Cleanup)
339}
340
341/// Error returned by autoscaler intent admission.
342#[derive(Debug, Clone, PartialEq, Eq)]
343pub struct AutoscalerIntentError {
344    message: String,
345}
346
347impl AutoscalerIntentError {
348    fn new(message: impl Into<String>) -> Self {
349        Self {
350            message: message.into(),
351        }
352    }
353}
354
355impl From<UpgradeGuardError> for AutoscalerIntentError {
356    fn from(error: UpgradeGuardError) -> Self {
357        Self::new(error.to_string())
358    }
359}
360
361impl From<ReshardPlanError> for AutoscalerIntentError {
362    fn from(error: ReshardPlanError) -> Self {
363        Self::new(error.to_string())
364    }
365}
366
367impl fmt::Display for AutoscalerIntentError {
368    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
369        formatter.write_str(&self.message)
370    }
371}
372
373impl std::error::Error for AutoscalerIntentError {}
374
375/// Bounded metric snapshot for capacity/autoscaler surfaces.
376#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
377pub struct CapacityAutoscalerMetrics {
378    /// Last bounded recommendation label.
379    pub capacity_recommendation: ScaleRecommendation,
380    /// Total accepted scale actions.
381    pub scale_actions_total: u64,
382}
hydracache/grid/capacity.rs

hydracache/grid/
capacity.rs
