feat(telemetry): expose current fleet device and endpoint counts

docs/roadmap.md

@@ -45,7 +45,7 @@ of remembered identifiers based on observation expiry.
 - [ ] Record hostname-resolution success/failure without exporting hostnames or resolved addresses. Current spans show timing but do not explicitly classify the result.
 - [ ] Investigate the slow resolver with a reproducible device-side check. Timing isolated the phase, not the cause; do not treat instrumentation as a fix.
 - [x] Propagate optional W3C trace context for periodic/requested WS snapshots so agent collection and CC persistence can be inspected together. Wire compatibility and context linkage tested; live validation pending.
-- [ ] Add current-state device gauges: stored rows per agent and grouped fleet devices split known/unknown. Define scope and update lifecycle first; never derive gauges from the last filtered UI query.
+- [x] Add current-state gauges: retained device/endpoint rows per agent and unfiltered grouped fleet identities split saved/unsaved. Sample independently every 30 seconds with bounded revision checks; replace labels and suppress failed samples. Collector and state lifecycle tests included; dashboard recipes ready.
 - [ ] Decide whether snapshots need persistence acknowledgements. The existing sent-device counter does not prove CC storage; any confirmed-storage metric needs an authoritative measurement point.
 
 The existing snapshot counter is cumulative collection volume, not unique devices

docs/telemetry-recipes.md

@@ -20,6 +20,7 @@ that includes recent activity and allow at least two exports for rate queries.
 | --- | --- | --- |
 | 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 |
+| Gauge | Measures a current value, such as stored device rows | Read directly; do not use `rate` or `increase` |
 | 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
@@ -131,6 +132,63 @@ 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**.
 
+## Current Devices And Endpoints
+
+These gauges are emitted by CC. Counts are sampled at startup and every 30 seconds,
+independently of UI requests, and exported on the normal metrics schedule. Allow
+up to two intervals for a change to appear. The snapshot counter above cannot
+substitute for these gauges.
+
+A **device** is a grouped device identity. An **endpoint** is one observed address
+entry with an IP, optional MAC/interface, source, and presence. One PC can have
+multiple IPv4/IPv6 endpoints, and different sources or agents can observe the
+same address. An endpoint here is not an HTTP API route.
+
+| Panel | Scope | Suggested visualization |
+| --- | --- | --- |
+| Stored devices by agent | Device rows currently retained in CC for each agent; not globally unique identities | Bar gauge, one value per agent |
+| Stored endpoints by agent | Endpoint rows currently retained in CC for each agent, including inactive entries; not unique IPs | Bar gauge, one value per agent |
+| Fleet devices | Grouped fleet identities, split into saved and unsaved | Stat, one value per category |
+
+Stored device rows by agent:
+
+```promql
+sum by (job, agent_id) (wakey_stored_devices{job="wakey-control-plane"})
+```
+
+Stored endpoint rows by agent:
+
+```promql
+sum by (job, agent_id) (wakey_stored_endpoints{job="wakey-control-plane"})
+```
+
+Grouped fleet devices by manual identity status:
+
+```promql
+sum by (job, identity) (wakey_fleet_devices{job="wakey-control-plane"})
+```
+
+Adjust the `job` selector to your actual service label. These queries assume one
+CC owns the fleet; do not sum replicas of the same database as distinct devices.
+Enrolled agents with no stored rows report zero. After revocation, an agent can
+remain represented while its device rows are retained. When neither enrollment
+nor stored rows exist, the sampler stops reporting that agent label.
+
+Use Grafana's **short** unit with zero decimal places. Read the gauges directly;
+do not apply `rate` or `increase`. Counts must reflect unfiltered CC state, not
+the last UI search or visibility filter. An agent going offline does not by
+itself delete its stored rows, so these counts are not online-device counts.
+
+Do not sum per-agent device counts and label that value "unique fleet devices":
+the same device can be observed by more than one agent. Likewise, an endpoint
+count is not a count of distinct IP addresses. Saved/unsaved describes manual
+identity management, not online/offline/unknown presence.
+
+Failed refreshes suppress observations until sampling recovers; they do not report
+zero or keep exporting an old sample. Prometheus may still show its last value
+until that series becomes stale. Export callbacks only read the in-memory sample;
+they never wait on database work. Sampling is disabled when OTLP is disabled.
+
 ## Empty or Strange Results
 
 - No series: generate the relevant activity, wait for export, and check the metrics pipeline.

docs/telemetry.md

@@ -140,6 +140,9 @@ queries, rates, averages, percentiles, and Grafana units.
 | `wakey.operations` | Completed command/snapshot attempts, labeled by operation and outcome |
 | `wakey.operation.duration` | Corresponding duration histogram in seconds |
 | `wakey.snapshot.devices` | Cumulative devices in snapshots successfully sent by the agent |
+| `wakey.stored.devices` | CC gauge of retained device rows, labeled by `agent_id` |
+| `wakey.stored.endpoints` | CC gauge of retained endpoint rows, including inactive entries, labeled by `agent_id` |
+| `wakey.fleet.devices` | CC gauge of unfiltered grouped identities, labeled by `identity` (`saved` or `unsaved`) |
 | `wakey.terminal.active` | Agent PTY workers currently alive |
 | `wakey.terminal.input.bytes` | Input bytes received by the terminal relay |
 | `wakey.terminal.output.bytes` | Output bytes read from the PTY, including while detached |
@@ -150,6 +153,12 @@ Metric labels do not include request/session IDs, MACs, or IPs. New spans use
 not attached. Profiles, host CPU/RSS metrics, connected-agent gauges, and export-loss
 dashboards are not part of this initial slice.
 
+Current-state fleet gauges sample independently every 30 seconds, with bounded
+revision retries to avoid mixed-state counts. Enrolled empty agents report zero;
+removed labels disappear when no enrollment or retained rows remain. Failed samples
+are omitted, not converted to zeros. No per-device labels are emitted. See the
+recipe book for scopes, queries, and the distinction between devices and endpoints.
+
 ## Failure policy
 
 Trace and log queues are capped at 256 records each, with export batches of 64

wakey-control-plane/src/api/control.rs

@@ -13,5 +13,6 @@ pub use enroll::{
     set_agent_nickname,
 };
 pub(crate) use fleet::FleetCache;
+pub(crate) use fleet::sample_fleet_counts;
 pub use fleet::{list_fleet_devices, refresh_fleet_devices, wake_fleet_device};
 pub use stats::{StateStatsResponse, state_stats};

wakey-control-plane/src/api/control/fleet/mod.rs

@@ -49,6 +49,35 @@ async fn fleet_view(state: &AppState) -> anyhow::Result<std::sync::Arc<Vec<Fleet
         .await
 }
 
+/// Count stored rows separately from grouped identities. Reject mixed revisions;
+/// filtered UI results must never become the source of current-state metrics.
+pub(crate) async fn sample_fleet_counts(
+    state: &AppState,
+) -> anyhow::Result<wakey_core::telemetry::FleetCounts> {
+    for _ in 0..3 {
+        let revision = state.store.fleet_revision();
+        let mut counts = wakey_core::telemetry::FleetCounts::default();
+        for (agent, _) in state.store.list_agents_with_nicknames().await? {
+            counts.agents.entry(agent).or_default();
+        }
+        for row in state.store.list_agent_device_rows().await? {
+            let agent = counts.agents.entry(row.device.agent_id).or_default();
+            agent.devices += 1;
+            agent.endpoints += row.endpoints.len() as u64;
+        }
+        let fleet = fleet_view(state).await?;
+        counts.saved = fleet
+            .iter()
+            .filter(|device| device.known_device.is_some())
+            .count() as u64;
+        counts.unsaved = fleet.len() as u64 - counts.saved;
+        if state.store.fleet_revision() == revision {
+            return Ok(counts);
+        }
+    }
+    Err(crate::runtime::view_cache::RebuildLimit.into())
+}
+
 const REFRESH_CONCURRENCY: usize = 8;
 
 fn fleet_error(err: anyhow::Error, fallback_code: &str) -> ApiError {

wakey-control-plane/src/api/control/fleet/tests.rs

@@ -53,11 +53,18 @@ async fn fleet_cache_tracks_store_edits_and_connections_without_caching_filters(
         fleet: Arc::new(super::FleetCache::default()),
     };
     let store = &state.store;
+    assert_eq!(
+        super::sample_fleet_counts(&state).await.unwrap(),
+        Default::default()
+    );
     let token = store
         .issue_enroll_token(Duration::from_secs(60))
         .await
         .unwrap();
     let agent = store.enroll(&token.enroll_token).await.unwrap();
+    let empty = super::sample_fleet_counts(&state).await.unwrap();
+    assert_eq!(empty.agents[&agent.agent_id].devices, 0);
+    assert_eq!(empty.agents[&agent.agent_id].endpoints, 0);
     let observed = wakey_core::Device::from_parts(
         vec![wakey_core::NeighborEntry {
             ip: "192.168.1.10".parse().unwrap(),
@@ -73,6 +80,10 @@ async fn fleet_cache_tracks_store_edits_and_connections_without_caching_filters(
         .unwrap();
     let initial = super::fleet_view(&state).await.unwrap();
     assert_eq!(initial.len(), 1);
+    let counts = super::sample_fleet_counts(&state).await.unwrap();
+    assert_eq!(counts.agents[&agent.agent_id].devices, 1);
+    assert_eq!(counts.agents[&agent.agent_id].endpoints, 1);
+    assert_eq!((counts.saved, counts.unsaved), (0, 1));
     assert!(Arc::ptr_eq(
         &initial,
         &super::fleet_view(&state).await.unwrap()
@@ -148,6 +159,8 @@ async fn fleet_cache_tracks_store_edits_and_connections_without_caching_filters(
             .unwrap()
             .is_empty()
     );
+    let counts = super::sample_fleet_counts(&state).await.unwrap();
+    assert_eq!((counts.saved, counts.unsaved), (1, 0));
     let all = ListFleetDevicesQuery {
         visibility: Some("all".into()),
         ..Default::default()
@@ -162,6 +175,8 @@ async fn fleet_cache_tracks_store_edits_and_connections_without_caching_filters(
         .await
         .unwrap();
     assert_eq!(super::fleet_view(&state).await.unwrap().len(), 2);
+    let counts = super::sample_fleet_counts(&state).await.unwrap();
+    assert_eq!((counts.saved, counts.unsaved), (1, 1));
     store
         .attach_device_identifier(
             &saved.device_id,
@@ -195,6 +210,8 @@ async fn fleet_cache_tracks_store_edits_and_connections_without_caching_filters(
             .len(),
         1
     );
+    let counts = super::sample_fleet_counts(&state).await.unwrap();
+    assert_eq!((counts.saved, counts.unsaved), (2, 0));
     assert_eq!(
         super::load_fleet_devices(&state, &all).await.unwrap().len(),
         2
@@ -212,6 +229,9 @@ async fn fleet_cache_tracks_store_edits_and_connections_without_caching_filters(
     );
 
     state.sessions.write().await.clear();
+    let counts = super::sample_fleet_counts(&state).await.unwrap();
+    assert_eq!(counts.agents[&agent.agent_id].devices, 1);
+    assert_eq!((counts.saved, counts.unsaved), (0, 1));
     assert!(
         super::fleet_view(&state).await.unwrap()[0]
             .recommended_route
@@ -234,6 +254,41 @@ async fn fleet_cache_tracks_store_edits_and_connections_without_caching_filters(
         .await
         .unwrap();
     assert!(super::fleet_view(&state).await.unwrap().is_empty());
+    assert_eq!(
+        super::sample_fleet_counts(&state).await.unwrap(),
+        Default::default()
+    );
+    // Identical devices on two agents count twice as stored rows, once as a fleet identity.
+    // Offline endpoints are retained too; these metrics are not reachability counters.
+    let offline = wakey_core::Device::from_parts(
+        vec![wakey_core::NeighborEntry {
+            ip: "192.168.1.10".parse().unwrap(),
+            mac: Some("aa:bb:cc:dd:ee:01".parse().unwrap()),
+            dev: Some("br-lan".into()),
+            state: wakey_core::NeighborState::Failed,
+        }],
+        vec![],
+    );
+    for id in ["agent-a", "agent-b"] {
+        store
+            .replace_agent_device_snapshot(id, std::slice::from_ref(&offline))
+            .await
+            .unwrap();
+    }
+    let counts = super::sample_fleet_counts(&state).await.unwrap();
+    assert_eq!(counts.agents.len(), 2);
+    for count in counts.agents.values() {
+        assert_eq!(count.devices, 1);
+        assert_eq!(count.endpoints, 1);
+    }
+    assert_eq!((counts.saved, counts.unsaved), (0, 1));
+    store
+        .replace_agent_device_snapshot("agent-a", &[])
+        .await
+        .unwrap();
+    let counts = super::sample_fleet_counts(&state).await.unwrap();
+    assert!(!counts.agents.contains_key("agent-a"));
+    assert_eq!(counts.agents["agent-b"].devices, 1);
     drop(state);
     std::fs::remove_dir_all(dir).unwrap();
 }

wakey-control-plane/src/api/mod.rs

@@ -7,6 +7,7 @@ mod commands;
 mod control;
 mod terminals;
 pub(crate) use control::FleetCache;
+pub(crate) use control::sample_fleet_counts;
 
 pub use alerts::{AlertsHub, active_alerts, alert_history, alerts_stream, run_alert_evaluator};
 pub use audit::list_audit_events;

wakey-control-plane/src/runtime/fleet_metrics.rs

@@ -0,0 +1,34 @@
+use std::time::Duration;
+
+use wakey_core::telemetry::FleetMetrics;
+
+use super::AppState;
+
+/// Tie polling to the daemon lifetime, including early server errors.
+pub(super) struct FleetMetricsTask(tokio::task::JoinHandle<()>);
+
+impl FleetMetricsTask {
+    pub(super) fn start(state: AppState) -> Option<Self> {
+        let metrics = FleetMetrics::new()?;
+        Some(Self(tokio::spawn(async move {
+            let mut tick = tokio::time::interval(Duration::from_secs(30));
+            tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+            loop {
+                tick.tick().await;
+                match crate::api::sample_fleet_counts(&state).await {
+                    Ok(sample) => metrics.replace(Some(sample)),
+                    Err(error) => {
+                        metrics.replace(None);
+                        tracing::warn!(%error, "failed to refresh fleet metrics");
+                    }
+                }
+            }
+        })))
+    }
+}
+
+impl Drop for FleetMetricsTask {
+    fn drop(&mut self) {
+        self.0.abort();
+    }
+}

wakey-control-plane/src/runtime/mod.rs

@@ -24,6 +24,7 @@ use crate::state;
 use crate::ws;
 
 mod admin;
+mod fleet_metrics;
 mod process;
 pub mod terminals;
 pub(crate) mod view_cache;
@@ -196,6 +197,7 @@ pub async fn serve(daemon: config::DaemonConfig) -> Result<()> {
     };
 
     tokio::spawn(api::run_alert_evaluator(app_state.clone()));
+    let _fleet_metrics = fleet_metrics::FleetMetricsTask::start(app_state.clone());
 
     // Keep route classes explicit so edge policy can map directly:
     // - public_api_routes: intended internet-facing agent endpoints

wakey-core/src/telemetry.rs

@@ -291,6 +291,8 @@ fn local_timer() -> fmt::time::OffsetTime<time::format_description::well_known::
 // Construct instruments once, after the real provider exists. Disabled telemetry
 // therefore does no per-PTY-read allocations or instrument registration.
 static METRICS: OnceLock<Instruments> = OnceLock::new();
+mod fleet;
+pub use fleet::{AgentDeviceCounts, FleetCounts, FleetMetrics};
 struct Instruments {
     operations: opentelemetry::metrics::Counter<u64>,
     duration: opentelemetry::metrics::Histogram<f64>,

wakey-core/src/telemetry/fleet.rs

@@ -0,0 +1,91 @@
+use std::collections::BTreeMap;
+use std::sync::{Arc, RwLock};
+
+use opentelemetry::{KeyValue, metrics::ObservableGauge};
+
+/// Retained CC rows, not unique addresses or online-device counts.
+#[derive(Debug, Default, PartialEq, Eq)]
+pub struct AgentDeviceCounts {
+    pub devices: u64,
+    pub endpoints: u64,
+}
+
+/// One unfiltered, revision-checked CC sample. Agent IDs are the only identity labels.
+#[derive(Debug, Default, PartialEq, Eq)]
+pub struct FleetCounts {
+    pub agents: BTreeMap<String, AgentDeviceCounts>,
+    pub saved: u64,
+    pub unsaved: u64,
+}
+
+/// Callbacks only read memory; async DB work stays outside the export thread.
+/// Replacing the entire sample removes vanished labels instead of retaining their series.
+pub struct FleetMetrics {
+    sample: Arc<RwLock<Option<FleetCounts>>>,
+    _devices: ObservableGauge<u64>,
+    _endpoints: ObservableGauge<u64>,
+    _fleet: ObservableGauge<u64>,
+}
+
+impl FleetMetrics {
+    /// Avoid polling the DB at all when OTLP metrics are disabled.
+    pub fn new() -> Option<Self> {
+        super::METRICS.get()?;
+        let meter = opentelemetry::global::meter("wakey");
+        let sample = Arc::new(RwLock::new(None::<FleetCounts>));
+        let devices = sample.clone();
+        let endpoints = sample.clone();
+        let fleet = sample.clone();
+        Some(Self {
+            sample,
+            _devices: meter
+                .u64_observable_gauge("wakey.stored.devices")
+                .with_callback(move |observer| {
+                    if let Ok(sample) = devices.read()
+                        && let Some(sample) = sample.as_ref()
+                    {
+                        for (agent, counts) in &sample.agents {
+                            observer.observe(
+                                counts.devices,
+                                &[KeyValue::new("agent_id", agent.clone())],
+                            );
+                        }
+                    }
+                })
+                .build(),
+            _endpoints: meter
+                .u64_observable_gauge("wakey.stored.endpoints")
+                .with_callback(move |observer| {
+                    if let Ok(sample) = endpoints.read()
+                        && let Some(sample) = sample.as_ref()
+                    {
+                        for (agent, counts) in &sample.agents {
+                            observer.observe(
+                                counts.endpoints,
+                                &[KeyValue::new("agent_id", agent.clone())],
+                            );
+                        }
+                    }
+                })
+                .build(),
+            _fleet: meter
+                .u64_observable_gauge("wakey.fleet.devices")
+                .with_callback(move |observer| {
+                    if let Ok(sample) = fleet.read()
+                        && let Some(sample) = sample.as_ref()
+                    {
+                        observer.observe(sample.saved, &[KeyValue::new("identity", "saved")]);
+                        observer.observe(sample.unsaved, &[KeyValue::new("identity", "unsaved")]);
+                    }
+                })
+                .build(),
+        })
+    }
+
+    /// None suppresses observations after a failed refresh, rather than reporting stale counts.
+    pub fn replace(&self, sample: Option<FleetCounts>) {
+        if let Ok(mut current) = self.sample.write() {
+            *current = sample;
+        }
+    }
+}

wakey-core/tests/telemetry.rs

@@ -3,6 +3,7 @@
 use opentelemetry_proto::tonic::collector::{
     logs::v1 as logs, metrics::v1 as metrics, trace::v1 as traces,
 };
+use opentelemetry_proto::tonic::metrics::v1::{metric, number_data_point};
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::{Arc, Mutex};
 use tonic::{Request, Response, Status};
@@ -212,6 +213,36 @@ async fn exports_all_signals_and_preserves_remote_command_parent() {
     )
     .unwrap();
 
+    let fleet_metrics = telemetry::FleetMetrics::new().unwrap();
+    fleet_metrics.replace(Some(telemetry::FleetCounts {
+        agents: [(
+            "removed-agent".into(),
+            telemetry::AgentDeviceCounts {
+                devices: 9,
+                endpoints: 20,
+            },
+        )]
+        .into(),
+        saved: 9,
+        unsaved: 0,
+    }));
+    fleet_metrics.replace(None);
+    fleet_metrics.replace(Some(telemetry::FleetCounts {
+        agents: [
+            (
+                "router".into(),
+                telemetry::AgentDeviceCounts {
+                    devices: 2,
+                    endpoints: 7,
+                },
+            ),
+            ("empty-agent".into(), Default::default()),
+        ]
+        .into(),
+        saved: 1,
+        unsaved: 1,
+    }));
+
     let parent = tracing::info_span!("relay_command");
     let context = parent.in_scope(telemetry::current_trace_context);
     assert!(context.contains_key("traceparent"));
@@ -306,7 +337,48 @@ async fn exports_all_signals_and_preserves_remote_command_parent() {
         "wakey.operation.duration",
         "wakey.snapshot.devices",
         "wakey.terminal.active",
+        "wakey.stored.devices",
+        "wakey.stored.endpoints",
+        "wakey.fleet.devices",
     ] {
         assert!(names.contains(&name), "missing {name}");
     }
+    let latest = metrics.last().unwrap();
+    for (name, label, expected) in [
+        (
+            "wakey.stored.devices",
+            "agent_id",
+            vec![("router", 2), ("empty-agent", 0)],
+        ),
+        (
+            "wakey.stored.endpoints",
+            "agent_id",
+            vec![("router", 7), ("empty-agent", 0)],
+        ),
+        (
+            "wakey.fleet.devices",
+            "identity",
+            vec![("saved", 1), ("unsaved", 1)],
+        ),
+    ] {
+        let metric = latest
+            .resource_metrics
+            .iter()
+            .flat_map(|r| &r.scope_metrics)
+            .flat_map(|s| &s.metrics)
+            .find(|m| m.name == name)
+            .unwrap();
+        let Some(metric::Data::Gauge(gauge)) = &metric.data else {
+            panic!("{name} must be a gauge");
+        };
+        assert_eq!(gauge.data_points.len(), expected.len());
+        for (value, count) in expected {
+            let point = gauge.data_points.iter().find(|p| p.attributes.iter().any(|a| {
+                a.key == label && a.value.as_ref().and_then(|v| v.value.as_ref())
+                    == Some(&opentelemetry_proto::tonic::common::v1::any_value::Value::StringValue(value.into()))
+            })).unwrap();
+            assert_eq!(point.value, Some(number_data_point::Value::AsInt(count)));
+            assert_eq!(point.attributes.len(), 1, "no device/address labels");
+        }
+    }
 }