saddle_observability/
pressure.rs

1//! Fixed, bounded records of process-scoped execution pressure.
2use crate::{DiagnosticSubmission, EmergencyDiagnosticHandle, Observer};
3use saddle_admission::pressure::{PressureState, PressureWindow};
4use serde::Serialize;
5
6pub struct ExecutionPressureFacts {
7    pub os_error: Option<i32>,
8    pub window: PressureWindow,
9    pub decision_at: std::time::Duration,
10    pub source: &'static str,
11    pub state: PressureState,
12    pub workers: usize,
13    pub global_queue_depth: usize,
14    pub invalid_reason: Option<&'static str>,
15}
16
17impl Observer {
18    pub fn record_execution_pressure(
19        &self,
20        output: Option<&EmergencyDiagnosticHandle>,
21        facts: ExecutionPressureFacts,
22    ) -> DiagnosticSubmission {
23        #[derive(Serialize)]
24        struct Record {
25            schema_version: u8,
26            event: &'static str,
27            context: Option<()>,
28            source: &'static str,
29            scope: &'static str,
30            observed_monotonic_ms: u128,
31            decision_monotonic_ms: u128,
32            window_age_ms: u128,
33            runtime_identity: u64,
34            process_cpu_ratio: f64,
35            worker_busy_ratio: f64,
36            scheduling_delay_ms: u128,
37            window_valid: bool,
38            valid: bool,
39            state: &'static str,
40            workers: usize,
41            global_queue_depth: usize,
42            original_os_error: Option<i32>,
43            original_error: Option<&'static str>,
44        }
45        let Some(output) = output else {
46            return DiagnosticSubmission::OutputUnavailable;
47        };
48        output.submit_fixed_record(&Record {
49            schema_version: 1,
50            event: "framework.execution_pressure",
51            context: None,
52            source: facts.source,
53            scope: "process_and_runtime",
54            observed_monotonic_ms: facts.window.observed_at.as_millis(),
55            decision_monotonic_ms: facts.decision_at.as_millis(),
56            window_age_ms: facts.decision_at.saturating_sub(facts.window.observed_at).as_millis(),
57            runtime_identity: facts.window.runtime_identity,
58            process_cpu_ratio: facts.window.process_cpu_ratio,
59            worker_busy_ratio: facts.window.worker_busy_ratio,
60            scheduling_delay_ms: facts.window.scheduling_delay.as_millis(),
61            window_valid: facts.window.valid,
62            valid: facts.window.valid && facts.state != PressureState::Invalid,
63            state: match facts.state {
64                PressureState::Cold => "cold",
65                PressureState::Open => "open",
66                PressureState::Pressured => "pressured",
67                PressureState::Invalid => "invalid",
68            },
69            workers: facts.workers,
70            global_queue_depth: facts.global_queue_depth,
71            original_error: facts.invalid_reason,
72            original_os_error: facts.os_error,
73        })
74    }
75}
saddle_observability/pressure.rs

saddle_observability/
pressure.rs
