From a4029e20bf953bb810190b555edb26b7876961de Mon Sep 17 00:00:00 2001 From: lda Date: Sun, 11 Oct 2026 18:49:46 +0700 Subject: [PATCH] feat(telemetry): expose current fleet device and endpoint counts --- docs/roadmap.md | 2 +- docs/telemetry-recipes.md | 58 ++++++++++++ docs/telemetry.md | 9 ++ wakey-control-plane/src/api/control.rs | 1 + .../src/api/control/fleet/mod.rs | 29 ++++++ .../src/api/control/fleet/tests.rs | 55 +++++++++++ wakey-control-plane/src/api/mod.rs | 1 + .../src/runtime/fleet_metrics.rs | 34 +++++++ wakey-control-plane/src/runtime/mod.rs | 2 + wakey-core/src/telemetry.rs | 2 + wakey-core/src/telemetry/fleet.rs | 91 +++++++++++++++++++ wakey-core/tests/telemetry.rs | 72 +++++++++++++++ 12 files changed, 355 insertions(+), 1 deletion(-) create mode 100644 wakey-control-plane/src/runtime/fleet_metrics.rs create mode 100644 wakey-core/src/telemetry/fleet.rs diff --git a/docs/roadmap.md b/docs/roadmap.md index 5a68529..893a993 100644 --- a/docs/roadmap.md +++ b/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 diff --git a/docs/telemetry-recipes.md b/docs/telemetry-recipes.md index d2c86c7..60329d8 100644 --- a/docs/telemetry-recipes.md +++ b/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. diff --git a/docs/telemetry.md b/docs/telemetry.md index b8c13db..8424068 100644 --- a/docs/telemetry.md +++ b/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 diff --git a/wakey-control-plane/src/api/control.rs b/wakey-control-plane/src/api/control.rs index c3875c0..4c8e3a0 100644 --- a/wakey-control-plane/src/api/control.rs +++ b/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}; diff --git a/wakey-control-plane/src/api/control/fleet/mod.rs b/wakey-control-plane/src/api/control/fleet/mod.rs index bcf8cc5..4566a55 100644 --- a/wakey-control-plane/src/api/control/fleet/mod.rs +++ b/wakey-control-plane/src/api/control/fleet/mod.rs @@ -49,6 +49,35 @@ async fn fleet_view(state: &AppState) -> anyhow::Result anyhow::Result { + 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 { diff --git a/wakey-control-plane/src/api/control/fleet/tests.rs b/wakey-control-plane/src/api/control/fleet/tests.rs index 412fe3e..815081c 100644 --- a/wakey-control-plane/src/api/control/fleet/tests.rs +++ b/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(); } diff --git a/wakey-control-plane/src/api/mod.rs b/wakey-control-plane/src/api/mod.rs index 4579ceb..0d0530d 100644 --- a/wakey-control-plane/src/api/mod.rs +++ b/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; diff --git a/wakey-control-plane/src/runtime/fleet_metrics.rs b/wakey-control-plane/src/runtime/fleet_metrics.rs new file mode 100644 index 0000000..63837cd --- /dev/null +++ b/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 { + 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(); + } +} diff --git a/wakey-control-plane/src/runtime/mod.rs b/wakey-control-plane/src/runtime/mod.rs index c286e12..82fffc1 100644 --- a/wakey-control-plane/src/runtime/mod.rs +++ b/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 diff --git a/wakey-core/src/telemetry.rs b/wakey-core/src/telemetry.rs index a7be4ae..7903e3a 100644 --- a/wakey-core/src/telemetry.rs +++ b/wakey-core/src/telemetry.rs @@ -291,6 +291,8 @@ fn local_timer() -> fmt::time::OffsetTime = OnceLock::new(); +mod fleet; +pub use fleet::{AgentDeviceCounts, FleetCounts, FleetMetrics}; struct Instruments { operations: opentelemetry::metrics::Counter, duration: opentelemetry::metrics::Histogram, diff --git a/wakey-core/src/telemetry/fleet.rs b/wakey-core/src/telemetry/fleet.rs new file mode 100644 index 0000000..e5ee716 --- /dev/null +++ b/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, + 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>>, + _devices: ObservableGauge, + _endpoints: ObservableGauge, + _fleet: ObservableGauge, +} + +impl FleetMetrics { + /// Avoid polling the DB at all when OTLP metrics are disabled. + pub fn new() -> Option { + super::METRICS.get()?; + let meter = opentelemetry::global::meter("wakey"); + let sample = Arc::new(RwLock::new(None::)); + 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) { + if let Ok(mut current) = self.sample.write() { + *current = sample; + } + } +} diff --git a/wakey-core/tests/telemetry.rs b/wakey-core/tests/telemetry.rs index 9602515..cbe6a4f 100644 --- a/wakey-core/tests/telemetry.rs +++ b/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"); + } + } }