diff --git a/docs/telemetry.md b/docs/telemetry.md index acb2fed..54272e2 100644 --- a/docs/telemetry.md +++ b/docs/telemetry.md @@ -90,6 +90,10 @@ existing debug events too; inspect them before enabling it broadly. - Inventory source spans separate neighbor, DHCP lease, and hook-store reads from merging. `db_*` spans cover snapshot replacement, fleet reads, and manual identity changes, including transaction waits and result assembly where applicable. +- Fleet spans cover list/load/grouping, refresh fan-out with per-agent outcomes, + and wake dispatch. Counts distinguish grouped, filtered, and returned rows; + wake events identify the chosen agent and whether the route was explicit, + without recording device identifiers or addresses. Instrument operation boundaries rather than individual rows or terminal chunks. Use `skip_all`, explicitly allow count fields, and instrument async futures rather diff --git a/wakey-control-plane/src/api/control/fleet/build.rs b/wakey-control-plane/src/api/control/fleet/build.rs index e8ff2a2..ed30ab5 100644 --- a/wakey-control-plane/src/api/control/fleet/build.rs +++ b/wakey-control-plane/src/api/control/fleet/build.rs @@ -61,6 +61,7 @@ impl Default for FleetAccumulator { } } +#[tracing::instrument(name = "fleet_group_devices", skip_all, fields(known = known_devices.len(), observed = agent_devices.len()))] pub(crate) fn build_fleet_devices( known_devices: Vec, agent_devices: Vec, diff --git a/wakey-control-plane/src/api/control/fleet/mod.rs b/wakey-control-plane/src/api/control/fleet/mod.rs index fecfc8c..99056d4 100644 --- a/wakey-control-plane/src/api/control/fleet/mod.rs +++ b/wakey-control-plane/src/api/control/fleet/mod.rs @@ -5,7 +5,7 @@ use axum::extract::{Query, State}; use axum::http::StatusCode; use axum::response::IntoResponse; use futures_util::StreamExt; -use tracing::warn; +use tracing::{Instrument, info, warn}; use wakey_agent::protocol::{AgentCommand, InventoryRequest, WakeRequest}; use crate::api::ApiError; @@ -29,6 +29,7 @@ use types::{ const REFRESH_CONCURRENCY: usize = 8; +#[tracing::instrument(name = "fleet_list", skip_all)] pub async fn list_fleet_devices( State(state): State, Query(query): Query, @@ -46,6 +47,7 @@ pub async fn list_fleet_devices( } } +#[tracing::instrument(name = "fleet_refresh", skip_all, fields(agents = tracing::field::Empty, succeeded = tracing::field::Empty, failed = tracing::field::Empty, accepted = tracing::field::Empty))] pub async fn refresh_fleet_devices( State(state): State, Json(req): Json, @@ -57,9 +59,12 @@ pub async fn refresh_fleet_devices( req.agent_ids.clone() }; let agent_ids = normalize_refresh_agent_ids(agent_ids); + tracing::Span::current().record("agents", agent_ids.len()); let results = futures_util::stream::iter(agent_ids.into_iter().map(|agent_id| { let state = state.clone(); + // Each concurrent future carries its own span through relay and storage. + let span = tracing::info_span!("fleet_refresh_agent", agent_id = %agent_id, outcome = tracing::field::Empty, accepted = tracing::field::Empty); async move { let response = relay_agent_command( &state, @@ -75,8 +80,16 @@ pub async fn refresh_fleet_devices( req.timeout_ms, ) .await; - refresh_fleet_agent_result(&state, &agent_id, response).await + let result = refresh_fleet_agent_result(&state, &agent_id, response).await; + tracing::Span::current().record("outcome", result.status.as_str()); + tracing::Span::current().record("accepted", result.accepted); + if result.status != "ok" { + // The response retains diagnostic detail; do not export payloads. + warn!("fleet agent refresh failed"); + } + result } + .instrument(span) })) .buffer_unordered(REFRESH_CONCURRENCY) .collect::>() @@ -85,6 +98,15 @@ pub async fn refresh_fleet_devices( let mut results = results; results.sort_by(|a, b| a.agent_id.cmp(&b.agent_id)); let total_accepted = results.iter().map(|result| result.accepted).sum::(); + let succeeded = results + .iter() + .filter(|result| result.status == "ok") + .count(); + let span = tracing::Span::current(); + span.record("succeeded", succeeded); + span.record("failed", results.len() - succeeded); + span.record("accepted", total_accepted); + info!("fleet refresh completed"); Ok(( StatusCode::OK, @@ -165,10 +187,13 @@ async fn refresh_fleet_agent_result( } } +#[tracing::instrument(name = "fleet_wake", skip_all, fields(explicit_route = req.route_id.is_some(), outcome = tracing::field::Empty))] pub async fn wake_fleet_device( State(state): State, Json(req): Json, ) -> Result { + // Early database/relay errors retain an outcome even without a command result. + tracing::Span::current().record("outcome", "error"); let query = ListFleetDevicesQuery { query: None, presence: None, @@ -190,6 +215,7 @@ pub async fn wake_fleet_device( .into_iter() .find(|device| device.device_key == req.device_key) .ok_or_else(|| { + tracing::Span::current().record("outcome", "device_not_found"); ApiError::new( StatusCode::NOT_FOUND, "fleet_device_not_found", @@ -205,6 +231,7 @@ pub async fn wake_fleet_device( None => device.recommended_route, } .ok_or_else(|| { + tracing::Span::current().record("outcome", "route_unavailable"); ApiError::new( StatusCode::BAD_REQUEST, "wake_route_unavailable", @@ -213,6 +240,7 @@ pub async fn wake_fleet_device( })?; let Some(mac) = route.mac else { + tracing::Span::current().record("outcome", "route_without_mac"); return Err(ApiError::new( StatusCode::BAD_REQUEST, "wake_route_unavailable", @@ -220,6 +248,7 @@ pub async fn wake_fleet_device( )); }; if !route.connected { + tracing::Span::current().record("outcome", "route_disconnected"); return Err(ApiError::new( StatusCode::BAD_REQUEST, "wake_route_unavailable", @@ -227,6 +256,7 @@ pub async fn wake_fleet_device( )); } if !route.wakeable { + tracing::Span::current().record("outcome", "route_not_wakeable"); return Err(ApiError::new( StatusCode::BAD_REQUEST, "wake_route_unavailable", @@ -234,6 +264,7 @@ pub async fn wake_fleet_device( )); } + info!(agent_id = %route.agent_id, has_ip = route.ip.is_some(), explicit_route = req.route_id.is_some(), "fleet wake route selected"); let command = relay_agent_command( &state, &route.agent_id, @@ -245,6 +276,7 @@ pub async fn wake_fleet_device( req.timeout_ms, ) .await?; + tracing::Span::current().record("outcome", command.status.as_str()); Ok(( StatusCode::OK, @@ -252,6 +284,7 @@ pub async fn wake_fleet_device( )) } +#[tracing::instrument(name = "fleet_load", skip_all, fields(grouped = tracing::field::Empty, filtered = tracing::field::Empty, returned = tracing::field::Empty))] async fn load_fleet_devices( state: &AppState, query: &ListFleetDevicesQuery, @@ -290,9 +323,12 @@ async fn load_fleet_devices( identifier_map, }; let mut devices = build_fleet_devices(known_devices, agent_devices, &context); + tracing::Span::current().record("grouped", devices.len()); filter_fleet_devices(&mut devices, query); + tracing::Span::current().record("filtered", devices.len()); let limit = query.limit.unwrap_or(500).clamp(1, 1000); // this is going to be 10 million times harder if i decide to do this lazily devices.truncate(limit); + tracing::Span::current().record("returned", devices.len()); Ok(devices) }