1use crate::{
8 CanisterProtocolError, call_canister_with_arg,
9 fleet_ensure::CurrentFleetResolution,
10 icp::IcpCli,
11 protocol_binding::{ProtocolBindingError, resolve_registry_protocol_binding},
12 query_canister_with_arg,
13 registry::RegistryEntry,
14};
15use candid::{CandidType, Deserialize, Principal, types::principal::PrincipalError};
16use canic_core::{
17 dto::{
18 canister::{CanisterInspectionRequest, CanisterStatusResponse},
19 observability::{
20 CanisterObservabilityRequest, CanisterObservabilityResponse,
21 FleetCanisterObservabilityRequest,
22 },
23 page::Page,
24 },
25 ids::CanisterRole,
26 protocol,
27};
28use std::{collections::BTreeSet, path::Path};
29use thiserror::Error as ThisError;
30
31#[derive(CandidType)]
32enum RootCommandFragment {
33 InspectCanister(CanisterInspectionRequest),
34 ObserveCanister(FleetCanisterObservabilityRequest),
35}
36
37#[derive(CandidType, Deserialize)]
38enum RootCommandResponseFragment {
39 InspectCanister(Box<CanisterStatusResponse>),
40 InspectionReserveRequired(canic_core::dto::canister::CanisterInspectionReserveResponse),
41 ObserveCanister(CanisterObservabilityResponse),
42}
43
44#[derive(CandidType)]
45enum RootStatusRequestFragment {
46 ChildFunding(Principal),
47 CycleBalance,
48 CycleHistory(canic_core::dto::page::PageRequest),
49 MemoryAllocations,
50 Metrics(canic_core::dto::role::MetricsStatusRequest),
51}
52
53#[derive(CandidType, Deserialize)]
54enum RootStatusResponseFragment {
55 ChildFunding(canic_core::dto::observability::ChildFundingUsage),
56 CycleBalance(canic_core::dto::role::CycleBalanceStatusResponse),
57 CycleHistory(Page<canic_core::dto::cycles::CycleTrackerEntry>),
58 MemoryAllocations(canic_core::dto::memory::MemoryAllocationsResponse),
59 Metrics(Page<canic_core::dto::metrics::MetricEntry>),
60}
61
62#[derive(CandidType)]
63enum StoreStatusRequestFragment {
64 CycleHistory(canic_core::dto::page::PageRequest),
65}
66
67#[derive(CandidType, Deserialize)]
68enum StoreStatusResponseFragment {
69 CycleHistory(Page<canic_core::dto::cycles::CycleTrackerEntry>),
70}
71
72#[derive(Debug, ThisError)]
74pub enum FleetObservabilityError {
75 #[error("Canister {canister} has no owning Fleet Subnet Root in current topology")]
76 MissingRoot { canister: String },
77
78 #[error("current Fleet observability topology contains a parent cycle at {canister}")]
79 ParentCycle { canister: String },
80
81 #[error("invalid {field} Principal {value}: {source}")]
82 Principal {
83 field: &'static str,
84 value: String,
85 #[source]
86 source: PrincipalError,
87 },
88
89 #[error(transparent)]
90 Protocol(#[from] CanisterProtocolError),
91
92 #[error(transparent)]
93 ProtocolBinding(#[from] ProtocolBindingError),
94
95 #[error("Fleet Subnet Root does not expose cycle top-up history")]
96 RootCycleTopupsUnsupported,
97
98 #[error("Wasm Store does not maintain child funding grants")]
99 StoreChildFundingUnsupported,
100
101 #[error("Wasm Store does not expose cycle top-up history")]
102 StoreCycleTopupsUnsupported,
103
104 #[error("Wasm Store does not expose current memory allocations")]
105 StoreMemoryAllocationsUnsupported,
106
107 #[error("Wasm Store does not expose runtime metrics")]
108 StoreMetricsUnsupported,
109
110 #[error("Fleet Subnet Root returned a different observability response variant")]
111 UnexpectedRootResponse,
112
113 #[error("Canister {canister} cycle balance exceeds u128")]
114 CycleBalanceOverflow { canister: String },
115}
116
117pub fn observe_fleet_canister(
119 icp: &IcpCli,
120 icp_root: &Path,
121 environment: &str,
122 fleet: &CurrentFleetResolution,
123 entry: &RegistryEntry,
124 request: CanisterObservabilityRequest,
125) -> Result<CanisterObservabilityResponse, FleetObservabilityError> {
126 if entry.role.as_deref() == Some(CanisterRole::ROOT.as_str()) {
127 return observe_root(icp, icp_root, environment, entry, request);
128 }
129
130 let root = fleet_subnet_root_entry(&fleet.registry.entries, entry)?;
131 let binding = resolve_registry_protocol_binding(icp_root, environment, root)?;
132 let root_canister = parse_principal("Fleet Subnet Root", &root.pid)?;
133 let target = parse_principal("observability target", &entry.pid)?;
134 if matches!(&request, CanisterObservabilityRequest::CycleBalance) {
135 return icp.measure_canister_inspection(root_canister, target, || {
136 crate::canister_protocol::inspection::preflight_inspection(
137 icp,
138 &binding.candid_path,
139 root_canister,
140 target,
141 )?;
142 let response: RootCommandResponseFragment = call_canister_with_arg(
143 icp,
144 &binding,
145 root_canister,
146 protocol::CANIC_ROOT_COMMAND,
147 &RootCommandFragment::InspectCanister(CanisterInspectionRequest {
148 canister_id: target,
149 }),
150 )?;
151 let response = match response {
152 RootCommandResponseFragment::InspectCanister(response) => response,
153 RootCommandResponseFragment::InspectionReserveRequired(evidence) => {
154 return Err(CanisterProtocolError::inspection_reserve(
155 root_canister,
156 target,
157 evidence,
158 )
159 .into());
160 }
161 RootCommandResponseFragment::ObserveCanister(_) => {
162 return Err(FleetObservabilityError::UnexpectedRootResponse);
163 }
164 };
165 let cycles = u128::try_from(response.cycles.0).map_err(|_| {
166 FleetObservabilityError::CycleBalanceOverflow {
167 canister: entry.pid.clone(),
168 }
169 })?;
170 Ok(CanisterObservabilityResponse::CycleBalance(
171 canic_core::dto::role::CycleBalanceStatusResponse { cycles },
172 ))
173 });
174 }
175 if entry.role.as_deref() == Some(CanisterRole::WASM_STORE.as_str()) {
176 return observe_store(icp, icp_root, environment, entry, request);
177 }
178 let expected_child = child_funding_target(&request);
179 let response: RootCommandResponseFragment = call_canister_with_arg(
180 icp,
181 &binding,
182 root_canister,
183 protocol::CANIC_ROOT_COMMAND,
184 &RootCommandFragment::ObserveCanister(FleetCanisterObservabilityRequest {
185 canister_id: target,
186 request,
187 }),
188 )?;
189 let RootCommandResponseFragment::ObserveCanister(response) = response else {
190 return Err(FleetObservabilityError::UnexpectedRootResponse);
191 };
192 verify_child_funding_response(response, target, expected_child)
193}
194
195fn observe_store(
196 icp: &IcpCli,
197 icp_root: &Path,
198 environment: &str,
199 store: &RegistryEntry,
200 request: CanisterObservabilityRequest,
201) -> Result<CanisterObservabilityResponse, FleetObservabilityError> {
202 let request = match request {
203 CanisterObservabilityRequest::ChildFunding(_) => {
204 return Err(FleetObservabilityError::StoreChildFundingUnsupported);
205 }
206 CanisterObservabilityRequest::CycleHistory(page) => {
207 StoreStatusRequestFragment::CycleHistory(page)
208 }
209 CanisterObservabilityRequest::CycleTopups(_) => {
210 return Err(FleetObservabilityError::StoreCycleTopupsUnsupported);
211 }
212 CanisterObservabilityRequest::Metrics(_) => {
213 return Err(FleetObservabilityError::StoreMetricsUnsupported);
214 }
215 CanisterObservabilityRequest::MemoryAllocations => {
216 return Err(FleetObservabilityError::StoreMemoryAllocationsUnsupported);
217 }
218 CanisterObservabilityRequest::CycleBalance => {
219 unreachable!("Store CycleBalance uses Root management inspection");
220 }
221 };
222 let binding = resolve_registry_protocol_binding(icp_root, environment, store)?;
223 let store_canister = parse_principal("Wasm Store", &store.pid)?;
224 let response: StoreStatusResponseFragment = query_canister_with_arg(
225 icp,
226 &binding,
227 store_canister,
228 protocol::CANIC_OBSERVABILITY,
229 &request,
230 )?;
231 let StoreStatusResponseFragment::CycleHistory(page) = response;
232 Ok(CanisterObservabilityResponse::CycleHistory(page))
233}
234
235fn observe_root(
236 icp: &IcpCli,
237 icp_root: &Path,
238 environment: &str,
239 root: &RegistryEntry,
240 request: CanisterObservabilityRequest,
241) -> Result<CanisterObservabilityResponse, FleetObservabilityError> {
242 let expected_child = child_funding_target(&request);
243 let request = match request {
244 CanisterObservabilityRequest::ChildFunding(child) => {
245 RootStatusRequestFragment::ChildFunding(child)
246 }
247 CanisterObservabilityRequest::CycleBalance => RootStatusRequestFragment::CycleBalance,
248 CanisterObservabilityRequest::CycleHistory(page) => {
249 RootStatusRequestFragment::CycleHistory(page)
250 }
251 CanisterObservabilityRequest::CycleTopups(_) => {
252 return Err(FleetObservabilityError::RootCycleTopupsUnsupported);
253 }
254 CanisterObservabilityRequest::Metrics(request) => {
255 RootStatusRequestFragment::Metrics(request)
256 }
257 CanisterObservabilityRequest::MemoryAllocations => {
258 RootStatusRequestFragment::MemoryAllocations
259 }
260 };
261 let binding = resolve_registry_protocol_binding(icp_root, environment, root)?;
262 let root_canister = parse_principal("Fleet Subnet Root", &root.pid)?;
263 let response: RootStatusResponseFragment = query_canister_with_arg(
264 icp,
265 &binding,
266 root_canister,
267 protocol::CANIC_OBSERVABILITY,
268 &request,
269 )?;
270 let response = match response {
271 RootStatusResponseFragment::ChildFunding(response) => {
272 CanisterObservabilityResponse::ChildFunding(response)
273 }
274 RootStatusResponseFragment::CycleBalance(response) => {
275 CanisterObservabilityResponse::CycleBalance(response)
276 }
277 RootStatusResponseFragment::CycleHistory(response) => {
278 CanisterObservabilityResponse::CycleHistory(response)
279 }
280 RootStatusResponseFragment::Metrics(response) => {
281 CanisterObservabilityResponse::Metrics(response)
282 }
283 RootStatusResponseFragment::MemoryAllocations(response) => {
284 CanisterObservabilityResponse::MemoryAllocations(response)
285 }
286 };
287 verify_child_funding_response(response, root_canister, expected_child)
288}
289
290const fn child_funding_target(request: &CanisterObservabilityRequest) -> Option<Principal> {
291 match request {
292 CanisterObservabilityRequest::ChildFunding(child) => Some(*child),
293 _ => None,
294 }
295}
296
297fn verify_child_funding_response(
298 response: CanisterObservabilityResponse,
299 parent: Principal,
300 child: Option<Principal>,
301) -> Result<CanisterObservabilityResponse, FleetObservabilityError> {
302 if let Some(child) = child {
303 let CanisterObservabilityResponse::ChildFunding(value) = &response else {
304 return Err(FleetObservabilityError::UnexpectedRootResponse);
305 };
306 if (value.parent, value.child) != (parent, child) {
307 return Err(FleetObservabilityError::UnexpectedRootResponse);
308 }
309 }
310 Ok(response)
311}
312
313fn fleet_subnet_root_entry<'a>(
314 registry: &'a [RegistryEntry],
315 entry: &RegistryEntry,
316) -> Result<&'a RegistryEntry, FleetObservabilityError> {
317 let mut current = entry;
318 let mut visited = BTreeSet::new();
319 loop {
320 if !visited.insert(current.pid.as_str()) {
321 return Err(FleetObservabilityError::ParentCycle {
322 canister: current.pid.clone(),
323 });
324 }
325 if current.role.as_deref() == Some(CanisterRole::ROOT.as_str()) {
326 return registry
327 .iter()
328 .find(|candidate| candidate.pid == current.pid)
329 .ok_or_else(|| FleetObservabilityError::MissingRoot {
330 canister: entry.pid.clone(),
331 });
332 }
333 let Some(parent) = current.parent_pid.as_deref() else {
334 return Err(FleetObservabilityError::MissingRoot {
335 canister: entry.pid.clone(),
336 });
337 };
338 current = registry
339 .iter()
340 .find(|candidate| candidate.pid == parent)
341 .ok_or_else(|| FleetObservabilityError::MissingRoot {
342 canister: entry.pid.clone(),
343 })?;
344 }
345}
346
347fn parse_principal(field: &'static str, value: &str) -> Result<Principal, FleetObservabilityError> {
348 Principal::from_text(value).map_err(|source| FleetObservabilityError::Principal {
349 field,
350 value: value.to_string(),
351 source,
352 })
353}
354
355#[cfg(test)]
356mod tests {
357 use super::*;
358
359 #[test]
360 fn child_funding_response_binds_parent_and_child_for_direct_and_relayed_reads() {
361 let parent = Principal::from_slice(&[1]);
362 let child = Principal::from_slice(&[2]);
363 let usage = canic_core::dto::observability::ChildFundingUsage {
364 parent,
365 child,
366 observed_at_ns: 10,
367 accounted_cycles: 0.into(),
368 last_accounted_at_secs: 0,
369 pending_operations: 0,
370 reserved_cycles: Some(0.into()),
371 };
372 assert!(
373 verify_child_funding_response(
374 CanisterObservabilityResponse::ChildFunding(usage.clone()),
375 parent,
376 Some(child)
377 )
378 .is_ok()
379 );
380 for (wrong_parent, wrong_child) in [(child, child), (parent, parent)] {
381 assert!(matches!(
382 verify_child_funding_response(
383 CanisterObservabilityResponse::ChildFunding(usage.clone()),
384 wrong_parent,
385 Some(wrong_child)
386 ),
387 Err(FleetObservabilityError::UnexpectedRootResponse)
388 ));
389 }
390 }
391
392 fn entry(pid: &str, role: &str, parent: Option<&str>) -> RegistryEntry {
393 RegistryEntry {
394 pid: pid.to_string(),
395 role: Some(role.to_string()),
396 parent_pid: parent.map(str::to_string),
397 module_hash: None,
398 protocol_binding: None,
399 }
400 }
401
402 #[test]
403 fn descendant_observability_resolves_the_exact_owning_root() {
404 let coordinator = entry("coordinator", "fleet_coordinator", None);
405 let root = entry("root", "root", Some("coordinator"));
406 let parent = entry("parent", "hub", Some("root"));
407 let child = entry("child", "leaf", Some("parent"));
408 let registry = vec![coordinator, root.clone(), parent, child.clone()];
409
410 assert_eq!(
411 fleet_subnet_root_entry(®istry, &child)
412 .expect("owning Root")
413 .pid,
414 root.pid
415 );
416 }
417}