poolsim_core/
telemetry.rs

1//! Telemetry import models and recommendation-diff helpers.
2//!
3//! This module keeps telemetry-derived sizing workflows in the core crate so
4//! the CLI and web layers can share one recommendation implementation.
5
6use serde::{Deserialize, Serialize};
7
8use crate::{
9    error::PoolsimError,
10    evaluate, simulate,
11    types::{
12        EvaluationResult, PoolConfig, SaturationLevel, SimulationOptions, SimulationReport,
13        WorkloadConfig,
14    },
15};
16
17/// Imported production telemetry for a single pool sizing decision.
18#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
19pub struct TelemetrySnapshot {
20    /// Optional service, application, or pool name.
21    #[serde(default)]
22    pub service_name: Option<String>,
23    /// Optional telemetry window label, such as `15m`, `1h`, or `2026-05-15T10:00Z/2026-05-15T11:00Z`.
24    #[serde(default)]
25    pub window: Option<String>,
26    /// Optional timestamp or timestamp range indicating when the snapshot was observed.
27    #[serde(default)]
28    pub observed_at: Option<String>,
29    /// Current production pool size to compare against the recommendation.
30    pub current_pool_size: u32,
31    /// Workload inferred from telemetry.
32    pub workload: WorkloadConfig,
33    /// Pool bounds and backend connection limits to use for recommendation.
34    pub pool: PoolConfig,
35}
36
37impl TelemetrySnapshot {
38    /// Validates the imported telemetry snapshot before recommendation.
39    ///
40    /// # Errors
41    ///
42    /// Returns [`PoolsimError::InvalidInput`] when the telemetry payload, workload,
43    /// pool constraints, or current production pool size are invalid.
44    pub fn validate(&self) -> Result<(), PoolsimError> {
45        self.workload.validate()?;
46        self.pool.validate()?;
47
48        if self.current_pool_size == 0 {
49            return Err(PoolsimError::invalid_input(
50                "INVALID_CURRENT_POOL_SIZE",
51                "current_pool_size must be greater than 0",
52                None,
53            ));
54        }
55
56        if self.current_pool_size > self.pool.max_server_connections {
57            return Err(PoolsimError::invalid_input(
58                "INVALID_CURRENT_POOL_SIZE",
59                "current_pool_size cannot exceed max_server_connections",
60                Some(serde_json::json!({
61                    "current_pool_size": self.current_pool_size,
62                    "max_server_connections": self.pool.max_server_connections,
63                })),
64            ));
65        }
66
67        Ok(())
68    }
69}
70
71/// Direction of change between the current and recommended pool sizes.
72#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
73pub enum PoolSizeChange {
74    /// The recommendation needs more connections than production currently uses.
75    Increase,
76    /// The recommendation uses fewer connections than production currently uses.
77    Decrease,
78    /// The recommendation matches the current production pool size.
79    Keep,
80}
81
82impl PoolSizeChange {
83    fn from_delta(delta: i64) -> Self {
84        match delta.cmp(&0) {
85            std::cmp::Ordering::Greater => Self::Increase,
86            std::cmp::Ordering::Less => Self::Decrease,
87            std::cmp::Ordering::Equal => Self::Keep,
88        }
89    }
90}
91
92/// Recommendation diff between an imported production pool and a computed recommendation.
93#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
94pub struct PoolRecommendationDiff {
95    /// Current production pool size from telemetry.
96    pub current_pool_size: u32,
97    /// Recommended pool size from the sizing engine.
98    pub recommended_pool_size: u32,
99    /// Signed difference: `recommended_pool_size - current_pool_size`.
100    pub pool_size_delta: i64,
101    /// Direction of the sizing change.
102    pub change: PoolSizeChange,
103    /// Extra connections required when [`Self::change`] is [`PoolSizeChange::Increase`].
104    pub additional_connections_required: u32,
105    /// Connections that can be removed when [`Self::change`] is [`PoolSizeChange::Decrease`].
106    pub removable_connections: u32,
107    /// Signed percentage change relative to the current pool size.
108    pub connection_change_percent: f64,
109    /// Evaluation of the current production pool size.
110    pub current_evaluation: EvaluationResult,
111    /// Full simulation report for the recommended pool size.
112    pub recommended_report: SimulationReport,
113}
114
115impl PoolRecommendationDiff {
116    fn new(
117        current_pool_size: u32,
118        current_evaluation: EvaluationResult,
119        recommended_report: SimulationReport,
120    ) -> Self {
121        let recommended_pool_size = recommended_report.optimal_pool_size;
122        let pool_size_delta = i64::from(recommended_pool_size) - i64::from(current_pool_size);
123        let change = PoolSizeChange::from_delta(pool_size_delta);
124        let additional_connections_required = pool_size_delta.max(0) as u32;
125        let removable_connections = (-pool_size_delta).max(0) as u32;
126        let connection_change_percent = (pool_size_delta as f64 / current_pool_size as f64) * 100.0;
127
128        Self {
129            current_pool_size,
130            recommended_pool_size,
131            pool_size_delta,
132            change,
133            additional_connections_required,
134            removable_connections,
135            connection_change_percent,
136            current_evaluation,
137            recommended_report,
138        }
139    }
140
141    /// Returns the worst saturation level across the current and recommended evaluations.
142    pub fn worst_saturation(&self) -> SaturationLevel {
143        if self.current_evaluation.saturation == SaturationLevel::Critical
144            || self.recommended_report.saturation == SaturationLevel::Critical
145        {
146            SaturationLevel::Critical
147        } else if self.current_evaluation.saturation == SaturationLevel::Warning
148            || self.recommended_report.saturation == SaturationLevel::Warning
149        {
150            SaturationLevel::Warning
151        } else {
152            SaturationLevel::Ok
153        }
154    }
155}
156
157/// Recommendation output produced from imported telemetry.
158#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
159pub struct TelemetryRecommendation {
160    /// Optional service, application, or pool name copied from telemetry.
161    #[serde(default)]
162    pub service_name: Option<String>,
163    /// Optional telemetry window copied from telemetry.
164    #[serde(default)]
165    pub window: Option<String>,
166    /// Optional observation timestamp copied from telemetry.
167    #[serde(default)]
168    pub observed_at: Option<String>,
169    /// Detailed current-vs-recommended diff.
170    pub diff: PoolRecommendationDiff,
171}
172
173/// Computes a pool-size recommendation and diff from imported telemetry.
174///
175/// # Errors
176///
177/// Returns [`PoolsimError`] when telemetry validation, fixed-pool evaluation, or
178/// full simulation fails.
179pub fn recommend_from_telemetry(
180    snapshot: &TelemetrySnapshot,
181    opts: &SimulationOptions,
182) -> Result<TelemetryRecommendation, PoolsimError> {
183    snapshot.validate()?;
184    opts.validate()?;
185
186    let current_evaluation = evaluate(&snapshot.workload, snapshot.current_pool_size, opts)?;
187    let recommended_report = simulate(&snapshot.workload, &snapshot.pool, opts)?;
188    let diff = PoolRecommendationDiff::new(
189        snapshot.current_pool_size,
190        current_evaluation,
191        recommended_report,
192    );
193
194    Ok(TelemetryRecommendation {
195        service_name: snapshot.service_name.clone(),
196        window: snapshot.window.clone(),
197        observed_at: snapshot.observed_at.clone(),
198        diff,
199    })
200}
201
202#[cfg(test)]
203mod tests {
204    use super::*;
205    use crate::types::{DistributionModel, QueueModel};
206
207    fn snapshot(current_pool_size: u32) -> TelemetrySnapshot {
208        TelemetrySnapshot {
209            service_name: Some("checkout-api".to_string()),
210            window: Some("1h".to_string()),
211            observed_at: Some("2026-05-15T10:00:00Z".to_string()),
212            current_pool_size,
213            workload: WorkloadConfig {
214                requests_per_second: 180.0,
215                latency_p50_ms: 8.0,
216                latency_p95_ms: 28.0,
217                latency_p99_ms: 75.0,
218                raw_samples_ms: None,
219                step_load_profile: None,
220            },
221            pool: PoolConfig {
222                max_server_connections: 100,
223                connection_overhead_ms: 1.5,
224                idle_timeout_ms: Some(60_000),
225                min_pool_size: 2,
226                max_pool_size: 20,
227            },
228        }
229    }
230
231    fn options() -> SimulationOptions {
232        SimulationOptions {
233            iterations: 1_200,
234            seed: Some(42),
235            distribution: DistributionModel::LogNormal,
236            queue_model: QueueModel::MMC,
237            target_wait_p99_ms: 45.0,
238            max_acceptable_rho: 0.85,
239        }
240    }
241
242    #[test]
243    fn validates_current_pool_size_bounds() {
244        let zero = snapshot(0);
245        let err = zero
246            .validate()
247            .expect_err("zero current pool size should fail");
248        assert_eq!(err.code(), "INVALID_CURRENT_POOL_SIZE");
249
250        let too_large = snapshot(101);
251        let err = too_large
252            .validate()
253            .expect_err("current pool above backend limit should fail");
254        assert_eq!(err.code(), "INVALID_CURRENT_POOL_SIZE");
255    }
256
257    #[test]
258    fn recommends_from_telemetry_and_keeps_metadata() {
259        let recommendation = recommend_from_telemetry(&snapshot(6), &options())
260            .expect("telemetry recommendation should run");
261
262        assert_eq!(recommendation.service_name.as_deref(), Some("checkout-api"));
263        assert_eq!(recommendation.window.as_deref(), Some("1h"));
264        assert_eq!(recommendation.diff.current_pool_size, 6);
265        assert!(recommendation.diff.recommended_pool_size >= 2);
266        assert!(recommendation.diff.recommended_pool_size <= 20);
267        assert_eq!(
268            recommendation.diff.pool_size_delta,
269            i64::from(recommendation.diff.recommended_pool_size) - 6
270        );
271        assert!(matches!(
272            recommendation.diff.change,
273            PoolSizeChange::Increase | PoolSizeChange::Decrease | PoolSizeChange::Keep
274        ));
275    }
276
277    fn evaluation(pool_size: u32, saturation: SaturationLevel) -> EvaluationResult {
278        EvaluationResult {
279            pool_size,
280            utilisation_rho: 0.50,
281            mean_queue_wait_ms: 1.0,
282            p99_queue_wait_ms: 5.0,
283            saturation,
284            warnings: Vec::new(),
285        }
286    }
287
288    fn report(pool_size: u32, saturation: SaturationLevel) -> SimulationReport {
289        SimulationReport {
290            optimal_pool_size: pool_size,
291            confidence_interval: (5, 7),
292            cold_start_min_pool_size: 5,
293            utilisation_rho: 0.90,
294            mean_queue_wait_ms: 2.0,
295            p99_queue_wait_ms: 20.0,
296            saturation,
297            sensitivity: Vec::new(),
298            step_load_analysis: Vec::new(),
299            warnings: Vec::new(),
300        }
301    }
302
303    #[test]
304    fn computes_decrease_diff_and_warning_saturation() {
305        let diff = PoolRecommendationDiff::new(
306            8,
307            evaluation(8, SaturationLevel::Ok),
308            report(6, SaturationLevel::Warning),
309        );
310        assert_eq!(diff.change, PoolSizeChange::Decrease);
311        assert_eq!(diff.pool_size_delta, -2);
312        assert_eq!(diff.additional_connections_required, 0);
313        assert_eq!(diff.removable_connections, 2);
314        assert_eq!(diff.connection_change_percent, -25.0);
315        assert_eq!(diff.worst_saturation(), SaturationLevel::Warning);
316    }
317
318    #[test]
319    fn computes_increase_and_keep_diffs() {
320        let increase = PoolRecommendationDiff::new(
321            8,
322            evaluation(8, SaturationLevel::Ok),
323            report(10, SaturationLevel::Ok),
324        );
325        assert_eq!(increase.change, PoolSizeChange::Increase);
326        assert_eq!(increase.pool_size_delta, 2);
327        assert_eq!(increase.additional_connections_required, 2);
328        assert_eq!(increase.removable_connections, 0);
329        assert_eq!(increase.connection_change_percent, 25.0);
330        assert_eq!(increase.worst_saturation(), SaturationLevel::Ok);
331
332        let keep = PoolRecommendationDiff::new(
333            8,
334            evaluation(8, SaturationLevel::Critical),
335            report(8, SaturationLevel::Ok),
336        );
337        assert_eq!(keep.change, PoolSizeChange::Keep);
338        assert_eq!(keep.pool_size_delta, 0);
339        assert_eq!(keep.additional_connections_required, 0);
340        assert_eq!(keep.removable_connections, 0);
341        assert_eq!(keep.connection_change_percent, 0.0);
342        assert_eq!(keep.worst_saturation(), SaturationLevel::Critical);
343    }
344}
poolsim_core/telemetry.rs

poolsim_core/
telemetry.rs
