Trace fleet refresh, grouping, and wake routing

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<KnownDevice>,
     agent_devices: Vec<AgentDeviceWithChildren>,

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<AppState>,
     Query(query): Query<ListFleetDevicesQuery>,
@@ -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<AppState>,
     Json(req): Json<RefreshFleetDevicesRequest>,
@@ -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::<Vec<_>>()
@@ -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::<usize>();
+    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<AppState>,
     Json(req): Json<WakeFleetDeviceRequest>,
 ) -> Result<impl IntoResponse, ApiError> {
+    // 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)
 }