Trace inventory sources and device storage operations

docs/telemetry-recipes.md

@@ -0,0 +1,131 @@
+# Wakey Metrics Recipes
+
+Use Grafana Explore with your Prometheus-compatible data source and the Code
+query editor. These recipes assume standard OTLP-to-Prometheus name conversion.
+Discover your actual series first:
+
+```promql
+count by (__name__, job) ({__name__=~"wakey[._].*"})
+```
+
+With Alloy's Prometheus exporter, service names normally become `job`, and agent
+instance IDs become `instance`. Inspect the returned labels before adding filters.
+For example, append `{job="wakey-agent"}` to a metric selector only if that is
+the job value you actually see. Metrics export every 30 seconds; use a time range
+that includes recent activity and allow at least two exports for rate queries.
+
+## The Three Types
+
+| Type | What it records | How to read it |
+| --- | --- | --- |
+| Counter | Cumulative events or bytes since process startup | `rate` for per-second throughput; `increase` for a window's estimated total |
+| UpDownCounter | Adds and subtracts from a current quantity | Read the current value directly; do not use `rate` for active sessions |
+| Histogram | Distribution of individual measurements | `_sum / _count` for an average; `_bucket` for estimated percentiles |
+
+Counter resets on restart are handled by `rate` and `increase`. Apply those
+functions **before** summing across instances so individual resets remain visible.
+
+## Operations
+
+Completed attempts per second, including errors and timeouts:
+
+```promql
+sum by (job, operation, outcome) (rate(wakey_operations_total[5m]))
+```
+
+Estimated attempts over the last hour:
+
+```promql
+sum by (job, operation, outcome) (increase(wakey_operations_total[1h]))
+```
+
+Average duration over the last five minutes:
+
+```promql
+sum by (job, operation) (rate(wakey_operation_duration_seconds_sum[5m]))
+/
+sum by (job, operation) (rate(wakey_operation_duration_seconds_count[5m]))
+```
+
+Set Grafana's unit to **seconds**. This is a weighted average, not an average of
+instance averages. Both successful and unsuccessful attempts are included;
+add `outcome="ok"` to both selectors for successful attempts only.
+
+Estimated p95 duration:
+
+```promql
+histogram_quantile(
+  0.95,
+  sum by (le, job, operation) (
+    rate(wakey_operation_duration_seconds_bucket[5m])
+  )
+)
+```
+
+This needs classic histogram bucket series. Keep `le` in the grouping: it is
+the bucket boundary, in seconds. p95 is an estimate from those boundaries,
+not an exact sort of recorded durations. Few operations make percentiles noisy.
+
+CC relay duration includes waiting for the agent; agent execution duration
+measures its dispatch separately. Do not sum both and call that user latency.
+
+## Terminals
+
+Active PTY workers by agent, including detached sessions:
+
+```promql
+sum by (job, instance) (wakey_terminal_active)
+```
+
+Read this directly as a count. It is not a count of browser tabs or attachments.
+
+PTY output bytes per second:
+
+```promql
+sum by (job, instance) (rate(wakey_terminal_output_bytes_total[5m]))
+```
+
+Input bytes per second:
+
+```promql
+sum by (job, instance) (rate(wakey_terminal_input_bytes_total[5m]))
+```
+
+Set the unit to **bytes/sec**. Output is measured at the PTY read, even while
+detached; it is not WebSocket wire traffic or confirmed browser delivery.
+
+Relay queue-full encounters over the last fifteen minutes:
+
+```promql
+sum by (job, instance) (increase(wakey_terminal_backpressure_total[15m]))
+```
+
+This counts encounters, not their duration and not dropped chunks. A burst of
+image output can raise it while output still arrives intact through backpressure.
+
+## Snapshots
+
+Device entries in successfully sent snapshots over the last hour:
+
+```promql
+sum by (job, instance) (increase(wakey_snapshot_devices_total[1h]))
+```
+
+Repeated snapshots count the same devices again. This is collection volume,
+**not unique devices, current fleet size, or confirmed CC persistence**.
+
+## Empty or Strange Results
+
+- No series: generate the relevant activity, wait for export, and check the metrics pipeline.
+- Different names: use the discovery query; collector translation can change suffixes and labels.
+- `NaN` average: no operations occurred in the window, so the denominator is zero. Show a gap, not a fake zero latency.
+- Empty `rate`: it needs multiple samples. Try a longer window and check the time range.
+- Fractional `increase`: Prometheus extrapolates to the window boundaries; event counts can therefore appear fractional.
+- Stale active count: the last exported value is not an immediate liveness check after a crash or network loss.
+- No `instance` label: CC does not currently set a service instance ID; agent telemetry does.
+
+## References
+
+- [Wakey telemetry setup and limitations](telemetry.md)
+- [Prometheus rate, increase, and histogram_quantile](https://prometheus.io/docs/prometheus/latest/querying/functions/)
+- [Alloy OTLP-to-Prometheus conversion](https://grafana.com/docs/alloy/latest/reference/components/otelcol/otelcol.exporter.prometheus/)

docs/telemetry.md

@@ -87,9 +87,22 @@ existing debug events too; inspect them before enabling it broadly.
   persistence separately. Snapshot traces are not yet linked across the wire.
 - Terminal create/attach/close/resume spans describe lifecycle operations, not
   output chunks. Trace-linked OTLP logs use the SDK's tracing-context bridge.
+- 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.
+
+Instrument operation boundaries rather than individual rows or terminal chunks.
+Use `skip_all`, explicitly allow count fields, and instrument async futures rather
+than holding entered span guards across awaits. Do not record query arguments,
+SQL parameters, device payloads, or credentials. Span names are fixed across requests.
 
 Metrics are exported every 30 seconds and at orderly shutdown:
 
+The table uses OpenTelemetry instrument names. Prometheus conversion typically
+replaces dots with underscores and adds type/unit suffixes; histograms produce
+multiple series. See the [metrics recipe book](telemetry-recipes.md) for discovery
+queries, rates, averages, percentiles, and Grafana units.
+
 | Metric | Meaning |
 | --- | --- |
 | `wakey.operations` | Completed command/snapshot attempts, labeled by operation and outcome |

src/service/inventory.rs

@@ -1,5 +1,5 @@
 use anyhow::Result;
-use tracing::{debug, warn};
+use tracing::{Instrument, debug, warn};
 use wakey_core::{
     Device, DeviceInventory, DeviceObservationFact, DhcpLease, DhcpLeaseWithState, InventoryQuery,
     NeighborEntry, Presence, Query,
@@ -19,10 +19,28 @@ pub async fn resolve_devices(input: impl Into<String>) -> Result<Vec<Device>> {
 /// This is the current center of gravity for the service layer. Higher-level
 /// status and wake flows should prefer deriving from inventory rather than
 /// directly from raw Linux source rows.
+#[tracing::instrument(name = "inventory", skip_all, fields(devices = tracing::field::Empty))]
 pub async fn inventory(query: InventoryQuery) -> Result<DeviceInventory> {
-    let neighbors = wakey_linux::devices::query_neighbors(&query).await?;
-    let leases = get_leases().await?;
-    let observations = match wakey_linux::observations::list_local_observations().await {
+    // Instrument futures, not entered guards: each source remains correctly
+    // parented when Tokio moves the task between workers during an await.
+    let neighbors = async {
+        let rows = wakey_linux::devices::query_neighbors(&query).await?;
+        debug!(rows = rows.len(), "neighbor source collected");
+        Ok::<_, anyhow::Error>(rows)
+    }
+    .instrument(tracing::info_span!("inventory_neighbors"))
+    .await?;
+    let leases = async {
+        let rows = get_leases().await?;
+        debug!(rows = rows.len(), "DHCP source collected");
+        Ok::<_, anyhow::Error>(rows)
+    }
+    .instrument(tracing::info_span!("inventory_dhcp_leases"))
+    .await?;
+    let observations = match wakey_linux::observations::list_local_observations()
+        .instrument(tracing::info_span!("inventory_hook_store"))
+        .await
+    {
         Ok(observations) => observations
             .into_iter()
             .filter_map(local_observation_to_fact)
@@ -39,6 +57,7 @@ pub async fn inventory(query: InventoryQuery) -> Result<DeviceInventory> {
         "building device inventory"
     );
     let devices = merge_devices_with_observations(neighbors, leases, observations, &query);
+    tracing::Span::current().record("devices", devices.len());
     debug!(devices = devices.len(), "merged device inventory");
     Ok(DeviceInventory { devices })
 }
@@ -57,6 +76,7 @@ pub fn merge_devices(
 
 /// Merge raw neighbor entries, DHCP leases, and hook observations into device
 /// aggregates.
+#[tracing::instrument(name = "inventory_merge", skip_all, fields(neighbors = neighbors.len(), leases = leases.len(), observations = observations.len()))]
 pub fn merge_devices_with_observations(
     neighbors: Vec<NeighborEntry>,
     leases: Vec<DhcpLeaseWithState>,

wakey-control-plane/src/state/store/agent_devices.rs

@@ -10,6 +10,7 @@ impl Store {
     ///
     /// This is the only write path for agent device state. Every sync source
     /// (WebSocket snapshot, fleet refresh, HTTP upload) must go through here.
+    #[tracing::instrument(name = "db_replace_agent_snapshot", skip_all, fields(devices = devices.len(), accepted = tracing::field::Empty))]
     pub async fn replace_agent_device_snapshot(
         &self,
         agent_id: &str,
@@ -222,10 +223,12 @@ impl Store {
             .await
             .context("failed committing device snapshot")?;
 
+        tracing::Span::current().record("accepted", incoming_keys.len());
         Ok(incoming_keys.len())
     }
 
     /// List all agent device rows with their child MAC/IP/hostname/fact rows.
+    #[tracing::instrument(name = "db_list_fleet_devices", skip_all)]
     pub async fn list_agent_device_rows(&self) -> Result<Vec<AgentDeviceWithChildren>> {
         let devices = sqlx::query_as!(
             AgentDeviceRow,
@@ -290,6 +293,7 @@ impl Store {
 
     /// List agent device rows for a single agent.
     #[allow(dead_code)]
+    #[tracing::instrument(name = "db_list_agent_devices", skip_all)]
     pub async fn list_agent_device_rows_for_agent(
         &self,
         agent_id: &str,

wakey-control-plane/src/state/store/devices.rs

@@ -1,5 +1,6 @@
 use super::*;
 impl Store {
+    #[tracing::instrument(name = "db_create_known_device", skip_all, fields(identifiers = input.identifiers.len()))]
     pub async fn create_known_device(&self, input: KnownDeviceInput) -> Result<KnownDevice> {
         let display_name = normalize_required_text(&input.display_name, "display_name")?;
         let notes = normalize_optional_text(input.notes.as_deref());
@@ -41,6 +42,7 @@ impl Store {
             .ok_or_else(|| anyhow::anyhow!("created known device disappeared"))
     }
 
+    #[tracing::instrument(name = "db_list_known_devices", skip_all)]
     pub async fn list_known_devices(&self) -> Result<Vec<KnownDevice>> {
         let rows = list_known_device_rows(&self.pool).await?;
 
@@ -52,6 +54,7 @@ impl Store {
         Ok(out)
     }
 
+    #[tracing::instrument(name = "db_get_known_device", skip_all)]
     pub async fn get_known_device(&self, device_id: &str) -> Result<Option<KnownDevice>> {
         let row = get_known_device_row(&self.pool, device_id).await?;
 
@@ -61,6 +64,7 @@ impl Store {
         }
     }
 
+    #[tracing::instrument(name = "db_forget_known_device", skip_all)]
     pub async fn forget_known_device(&self, device_id: &str) -> Result<bool> {
         let result = sqlx::query!("DELETE FROM known_devices WHERE device_id = ?1", device_id)
             .execute(&self.pool)
@@ -69,6 +73,7 @@ impl Store {
         Ok(result.rows_affected() > 0)
     }
 
+    #[tracing::instrument(name = "db_merge_known_devices", skip_all)]
     pub async fn merge_known_devices(
         &self,
         target_device_id: &str,
@@ -134,6 +139,7 @@ impl Store {
         self.get_known_device(target_device_id).await
     }
 
+    #[tracing::instrument(name = "db_attach_device_identifier", skip_all)]
     pub async fn attach_device_identifier(
         &self,
         device_id: &str,
@@ -169,6 +175,7 @@ impl Store {
         self.get_known_device(device_id).await
     }
 
+    #[tracing::instrument(name = "db_detach_device_identifier", skip_all)]
     pub async fn detach_device_identifier(
         &self,
         device_id: &str,