diff --git a/Cargo.lock b/Cargo.lock index e01221b..5cd52bb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3692,6 +3692,7 @@ dependencies = [ "serde_json", "serial_test", "tokio", + "tracing", "wakey-core", ] diff --git a/docs/telemetry-recipes.md b/docs/telemetry-recipes.md index 3ea105b..d2c86c7 100644 --- a/docs/telemetry-recipes.md +++ b/docs/telemetry-recipes.md @@ -103,6 +103,23 @@ 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. +Average duration of completed backpressure episodes, in seconds: + +```promql +sum by (job, instance, outcome) ( + rate(wakey_operation_duration_seconds_sum{operation="terminal_backpressure_wait"}[5m]) +) +/ +sum by (job, instance, outcome) ( + rate(wakey_operation_duration_seconds_count{operation="terminal_backpressure_wait"}[5m]) +) +``` + +An episode starts when the output channel is full and ends when the pending +backlog drains or the relay disconnects. It is not per-frame network latency; +unfinished episodes do not appear until they end. Keep the outcome grouping so +disconnected episodes are not confused with successfully drained ones. + ## Snapshots Device entries in successfully sent snapshots over the last hour: diff --git a/docs/telemetry.md b/docs/telemetry.md index 86c726d..53139f6 100644 --- a/docs/telemetry.md +++ b/docs/telemetry.md @@ -100,6 +100,12 @@ existing debug events too; inspect them before enabling it broadly. with backoff and CC disconnect duration. Sending credentials is not an auth ACK; CC's authentication event is the authoritative result. DNS diagnostics are a separate lookup, not a measurement of the resolver inside WebSocket connect. +- Terminal recovery uses `terminal_snapshot_generate` (encoded byte count), + transport-attachment spans, and `terminal_snapshot_request`. These measure + local work, not browser restoration or end-to-end image replay. Resume operation + success means credentials were queued for the worker. `terminal_backpressure_wait` + operation durations cover one full-queue backlog episode until it drains (`ok`) + or disconnects (`disconnected`); cancellation can retain the default `error` outcome. 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-agent/src/terminal.rs b/wakey-agent/src/terminal.rs index 87bad61..6613393 100644 --- a/wakey-agent/src/terminal.rs +++ b/wakey-agent/src/terminal.rs @@ -35,6 +35,7 @@ struct RelayOutputBuffer { sender: Option>, pending: VecDeque, snapshot_pending: bool, + backpressure_wait: Option, } impl RelayOutputBuffer { @@ -44,6 +45,9 @@ impl RelayOutputBuffer { } fn disconnect(&mut self) { + if let Some(mut wait) = self.backpressure_wait.take() { + wait.outcome("disconnected"); + } self.sender = None; self.pending.clear(); self.snapshot_pending = false; @@ -80,6 +84,10 @@ impl RelayOutputBuffer { match sender.try_send(frame) { Ok(()) => {} Err(mpsc::error::TrySendError::Full(frame)) => { + // One measurement per backlog episode, not per select poll. + self.backpressure_wait = Some(wakey_core::telemetry::Operation::new( + "terminal_backpressure_wait", + )); wakey_core::telemetry::count( wakey_core::telemetry::Counter::TerminalBackpressure, 1, @@ -111,6 +119,9 @@ impl RelayOutputBuffer { permit.send(frame); if self.pending.is_empty() { self.snapshot_pending = false; + if let Some(mut wait) = self.backpressure_wait.take() { + wait.outcome("ok"); + } } } @@ -134,10 +145,14 @@ impl TerminalState { self.parser.screen_mut().set_size(rows, cols); } + #[tracing::instrument(name = "terminal_snapshot_generate", skip_all, fields(bytes = tracing::field::Empty))] fn snapshot(&self) -> Vec { - self.parser + let snapshot = self + .parser .screen() - .snapshot_formatted(TERMINAL_SCROLLBACK_ROWS) + .snapshot_formatted(TERMINAL_SCROLLBACK_ROWS); + tracing::Span::current().record("bytes", snapshot.len()); + snapshot } } @@ -262,6 +277,7 @@ impl TerminalManager { #[tracing::instrument(name = "terminal_resume", skip_all, fields(terminal_id = %terminal_id))] pub fn resume(&self, terminal_id: &TerminalId, relay_token: String) -> Result<()> { + let mut operation = wakey_core::telemetry::Operation::new("terminal_resume"); let active = self.active.lock().expect("terminal manager poisoned"); let session = active .get(terminal_id.as_str()) @@ -269,6 +285,8 @@ impl TerminalManager { if session.relay_credentials.send(relay_token).is_err() { anyhow::bail!("terminal session {terminal_id} has stopped"); } + operation.outcome("ok"); + info!("terminal resume credentials queued for worker"); Ok(()) } @@ -1190,6 +1208,7 @@ mod tests { assert!(relay.is_connected()); assert!(relay.has_pending()); + assert!(relay.backpressure_wait.is_some()); assert_eq!( rx.recv().await.expect("original queued frame").into_data(), vec![1] @@ -1203,6 +1222,20 @@ mod tests { vec![2] ); assert!(!relay.has_pending()); + assert!(relay.backpressure_wait.is_none()); + } + + #[test] + fn disconnect_finishes_backpressure_episode() { + let (tx, _rx) = mpsc::channel(1); + let mut relay = RelayOutputBuffer::default(); + relay.connect(tx); + relay.queue_live(Message::Binary(vec![1].into())); + relay.queue_live(Message::Binary(vec![2].into())); + assert!(relay.backpressure_wait.is_some()); + relay.disconnect(); + assert!(relay.backpressure_wait.is_none()); + assert!(!relay.has_pending()); } #[tokio::test] diff --git a/wakey-control-plane/src/api/terminals.rs b/wakey-control-plane/src/api/terminals.rs index a207a2a..98f12d2 100644 --- a/wakey-control-plane/src/api/terminals.rs +++ b/wakey-control-plane/src/api/terminals.rs @@ -7,7 +7,7 @@ use futures_util::{SinkExt, StreamExt}; use serde::{Deserialize, Serialize}; use std::time::Duration; use tokio::sync::broadcast; -use tracing::{info, warn}; +use tracing::{Instrument, info, warn}; use wakey_agent::protocol::{ AgentCapability, DEFAULT_TERMINAL_MAX_SESSIONS, DEFAULT_TERMINAL_SESSION_TTL_SECONDS, ServerMessage, TerminalAgentHandshake, TerminalControl, TerminalId, TerminalOperatorHandshake, @@ -307,10 +307,17 @@ async fn handle_agent_terminal_socket(state: AppState, terminal_id: String, mut let (mut outbound, pending) = match state .terminals .attach_agent(&terminal_id, &auth.0, &auth.1) + .instrument( + tracing::info_span!("terminal_agent_transport_attach", terminal_id = %terminal_id), + ) .await { Ok(outbound) => outbound, Err(code) => { + warn!( + terminal_id, + code, "agent terminal transport attachment rejected" + ); close_socket(&mut socket, code).await; return; } @@ -393,10 +400,17 @@ async fn handle_operator_terminal_socket( let mut outbound = match state .terminals .attach_operator(&terminal_id, &auth.0, &auth.1) + .instrument( + tracing::info_span!("terminal_operator_transport_attach", terminal_id = %terminal_id), + ) .await { Ok(attached) => attached, Err(code) => { + warn!( + terminal_id, + code, "operator terminal transport attachment rejected" + ); close_socket(&mut socket, code).await; return; } @@ -407,6 +421,7 @@ async fn handle_operator_terminal_socket( if let Err(code) = state .terminals .relay_to_agent(&terminal_id, TerminalRelayFrame::Text(snapshot)) + .instrument(tracing::info_span!("terminal_snapshot_request", terminal_id = %terminal_id)) .await { warn!(terminal_id, code, "failed to request terminal snapshot"); diff --git a/wakey-linux/Cargo.toml b/wakey-linux/Cargo.toml index d805791..2d9bf88 100644 --- a/wakey-linux/Cargo.toml +++ b/wakey-linux/Cargo.toml @@ -11,6 +11,7 @@ macaddr = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } tokio = { workspace = true, features = ["net", "sync"] } +tracing = "0.1" wakey-core = { path = "../wakey-core", registry = "gitea", version = "0"} [dev-dependencies] diff --git a/wakey-linux/src/devices/neighbors.rs b/wakey-linux/src/devices/neighbors.rs index d888b24..52c2a9b 100644 --- a/wakey-linux/src/devices/neighbors.rs +++ b/wakey-linux/src/devices/neighbors.rs @@ -3,9 +3,11 @@ use futures::future::try_join_all; use lda_ipjs::subcommands::neighbor; use std::collections::HashSet; use std::net::IpAddr; +use tracing::Instrument; use wakey_core::{InventoryQuery, NeighborEntry, NeighborState, Query}; /// Resolve a hostname through the local resolver and return all reported IPs. +#[tracing::instrument(name = "neighbor_hostname_resolve", skip_all)] pub async fn get_ips(machine_name: &str) -> Result> { Ok(tokio::net::lookup_host((machine_name, 0)) .await @@ -54,6 +56,10 @@ pub async fn get_neighbors( #[cfg(unix)] { let results = neighbor::nl::get(&ip_filter, &dev_strs, &nud_filter, macs) + .instrument(tracing::info_span!( + "neighbor_table_read", + backend = "netlink" + )) .await .context("rtnetlink failed")? .into_iter() @@ -68,6 +74,7 @@ pub async fn get_neighbors( if ip_filter.is_empty() { let dev = dev_strs.first().copied(); results = neighbor::get_with_backend(neighbor::Backend::Json, None, dev, &nud_filter) + .instrument(tracing::info_span!("neighbor_table_read", backend = "json")) .await .context("ip -j neigh failed")? .into_iter() @@ -82,6 +89,7 @@ pub async fn get_neighbors( dev, &nud_filter, ) + .instrument(tracing::info_span!("neighbor_table_read", backend = "json")) .await .with_context(|| format!("ip -j neigh failed for {ip}"))? .into_iter()