lenso_runner/replicated/
diagnostics.rs

1use std::{
2    collections::BTreeMap,
3    sync::{
4        Arc,
5        atomic::{AtomicU64, Ordering},
6    },
7    time::Duration,
8};
9
10use lenso_app_plan::{ExecutionLaneId, ResolvedAppPlan};
11use lenso_kernel::{NativeApp, RuntimeInvocationProbe};
12
13#[derive(Debug)]
14pub(super) struct LaneDiagnosticsState {
15    plan: Arc<ResolvedAppPlan>,
16    lane_cpu_nanos: BTreeMap<String, AtomicU64>,
17    instance_queue_depths: BTreeMap<String, AtomicU64>,
18    total_messages: AtomicU64,
19    cross_lane_messages: AtomicU64,
20}
21
22impl LaneDiagnosticsState {
23    pub(super) fn new(plan: Arc<ResolvedAppPlan>) -> Self {
24        let lane_cpu_nanos = plan
25            .execution_lanes()
26            .iter()
27            .map(|lane| (lane.id().to_string(), AtomicU64::new(0)))
28            .collect();
29        let instance_queue_depths = plan
30            .plugin_instances()
31            .iter()
32            .map(|instance| (instance.instance_key().to_owned(), AtomicU64::new(0)))
33            .collect();
34        Self {
35            plan,
36            lane_cpu_nanos,
37            instance_queue_depths,
38            total_messages: AtomicU64::new(0),
39            cross_lane_messages: AtomicU64::new(0),
40        }
41    }
42
43    pub(super) fn record_invocation(
44        &self,
45        observing_lane: &ExecutionLaneId,
46        caller: &str,
47        provider: &str,
48    ) {
49        let Some(caller_lane) = self
50            .plan
51            .plugin_instance(caller)
52            .map(lenso_app_plan::PluginInstancePlan::execution_lane)
53        else {
54            return;
55        };
56        if caller_lane != observing_lane {
57            return;
58        }
59        let Some(provider_lane) = self
60            .plan
61            .plugin_instance(provider)
62            .map(lenso_app_plan::PluginInstancePlan::execution_lane)
63        else {
64            return;
65        };
66        self.total_messages.fetch_add(1, Ordering::Relaxed);
67        if caller_lane != provider_lane {
68            self.cross_lane_messages.fetch_add(1, Ordering::Relaxed);
69        }
70    }
71
72    pub(super) fn publish_lane(&self, lane: &ExecutionLaneId, app: &NativeApp, cpu_time: Duration) {
73        if let Some(cpu_nanos) = self.lane_cpu_nanos.get(lane.as_str()) {
74            cpu_nanos.store(duration_nanos(cpu_time), Ordering::Relaxed);
75        }
76        for (instance, depth) in app.instance_queue_depths() {
77            if let Some(queue_depth) = self.instance_queue_depths.get(&instance) {
78                queue_depth.store(u64::try_from(depth).unwrap_or(u64::MAX), Ordering::Relaxed);
79            }
80        }
81    }
82
83    pub(super) fn snapshot(&self) -> LaneDiagnosticsSnapshot {
84        LaneDiagnosticsSnapshot {
85            lane_cpu_time: self
86                .lane_cpu_nanos
87                .iter()
88                .map(|(lane, nanos)| {
89                    (
90                        lane.clone(),
91                        Duration::from_nanos(nanos.load(Ordering::Relaxed)),
92                    )
93                })
94                .collect(),
95            instance_queue_depths: self
96                .instance_queue_depths
97                .iter()
98                .map(|(instance, depth)| {
99                    (
100                        instance.clone(),
101                        usize::try_from(depth.load(Ordering::Relaxed)).unwrap_or(usize::MAX),
102                    )
103                })
104                .collect(),
105            total_messages: self.total_messages.load(Ordering::Relaxed),
106            cross_lane_messages: self.cross_lane_messages.load(Ordering::Relaxed),
107        }
108    }
109}
110
111#[derive(Debug)]
112pub(super) struct LaneInvocationProbe {
113    state: Arc<LaneDiagnosticsState>,
114    lane: ExecutionLaneId,
115}
116
117impl LaneInvocationProbe {
118    pub(super) fn new(state: Arc<LaneDiagnosticsState>, lane: ExecutionLaneId) -> Self {
119        Self { state, lane }
120    }
121}
122
123impl RuntimeInvocationProbe for LaneInvocationProbe {
124    fn record(&self, caller_instance: &str, provider_instance: &str) {
125        self.state
126            .record_invocation(&self.lane, caller_instance, provider_instance);
127    }
128}
129
130fn duration_nanos(duration: Duration) -> u64 {
131    u64::try_from(duration.as_nanos()).unwrap_or(u64::MAX)
132}
133
134/// One immutable structural snapshot used to evaluate Plan placement.
135#[derive(Clone, Debug, Default, PartialEq)]
136pub struct LaneDiagnosticsSnapshot {
137    lane_cpu_time: BTreeMap<String, Duration>,
138    instance_queue_depths: BTreeMap<String, usize>,
139    total_messages: u64,
140    cross_lane_messages: u64,
141}
142
143impl LaneDiagnosticsSnapshot {
144    /// Returns CPU time consumed by each lane's owner thread.
145    pub fn lane_cpu_time(&self) -> &BTreeMap<String, Duration> {
146        &self.lane_cpu_time
147    }
148
149    /// Returns the latest bounded request queue depth for one Plugin Instance.
150    pub fn instance_queue_depth(&self, instance: &str) -> Option<usize> {
151        self.instance_queue_depths.get(instance).copied()
152    }
153
154    /// Returns the number of observed App request messages.
155    pub const fn total_messages(&self) -> u64 {
156        self.total_messages
157    }
158
159    /// Returns the number of observed messages whose binding crosses lanes.
160    pub const fn cross_lane_messages(&self) -> u64 {
161        self.cross_lane_messages
162    }
163
164    /// Returns the share of observed request messages whose binding crosses lanes.
165    #[allow(clippy::cast_precision_loss)]
166    pub fn cross_lane_message_share(&self) -> f64 {
167        if self.total_messages == 0 {
168            0.0
169        } else {
170            self.cross_lane_messages as f64 / self.total_messages as f64
171        }
172    }
173}
lenso_runner/replicated/diagnostics.rs

lenso_runner/replicated/
diagnostics.rs
