From 7b6959155b52e025e0b9a5a54a3d07c97c426aa9 Mon Sep 17 00:00:00 2001 From: John Coffey Date: Sat, 19 Sep 2026 08:34:51 -0700 Subject: [PATCH] Monitoring: live tracing and live metrics streams, and their tokens (MON-20 to MON-24) GET /api/token/tracing and /api/token/metrics issue a 60-second token for holders of liveTracing or liveMetrics outside a tenant. /api/live/tracing streams event: trace frames of x:TraceEvents, never raw I/O, filtered by text or by key, with a ping while idle; /api/live/metrics streams event: metrics frames of current totals every interval. At most eight streams run per node, each for 30 minutes. Upstream's documented paths are aliases. --- crates/http/src/api/mod.rs | 68 ++++++-- crates/http/src/lib.rs | 1 + crates/http/src/live.rs | 266 ++++++++++++++++++++++++++++++++ crates/jmap/src/registry/get.rs | 1 + 4 files changed, 326 insertions(+), 10 deletions(-) create mode 100644 crates/http/src/live.rs diff --git a/crates/http/src/api/mod.rs b/crates/http/src/api/mod.rs index f1f1c4a..8e40c96 100644 --- a/crates/http/src/api/mod.rs +++ b/crates/http/src/api/mod.rs @@ -146,9 +146,44 @@ impl ManagementApi for Server { .await?, )) } - Some("tracing") | Some("metrics") => { - Err(trc::ResourceEvent::NotFound - .ctx(trc::Key::Details, "Enterprise feature")) + // inbuxa: MON-23: a live telemetry token, valid 60 seconds + Some(kind @ ("tracing" | "metrics")) => { + let (grant, permission) = if kind == "tracing" { + (GrantType::LiveTracing, Permission::LiveTracing) + } else { + (GrantType::LiveMetrics, Permission::LiveMetrics) + }; + access_token.enforce_permission(permission)?; + crate::live::assert_server_level(&access_token)?; + Ok(HttpResponse::new(StatusCode::OK) + .with_no_cache() + .with_text_body( + self.encode_access_token( + grant, + account_id, + self.account(account_id).await?.name(), + 60, + None, + None, + ) + .await?, + )) + } + _ => Err(trc::ResourceEvent::NotFound.into_err()), + } + } + // inbuxa: the paths upstream's docs name, as aliases (MON-20, MON-22) + "telemetry" if req.method() == Method::GET && path.get(2).copied() == Some("live") => { + let access_token = self.management_access_token(req, session).await?; + crate::live::assert_server_level(&access_token)?; + match path.get(1).copied() { + Some("traces") => { + access_token.enforce_permission(Permission::LiveTracing)?; + crate::live::live_tracing(req.uri().query()) + } + Some("metrics") => { + access_token.enforce_permission(Permission::LiveMetrics)?; + crate::live::live_metrics(&UrlParams::new(req.uri().query())) } _ => Err(trc::ResourceEvent::NotFound.into_err()), } @@ -192,9 +227,16 @@ impl ManagementApi for Server { }, )))) } - ("tracing" | "metrics", _, &Method::GET) => { - Err(trc::ResourceEvent::NotFound - .ctx(trc::Key::Details, "Enterprise feature")) + // inbuxa: MON-20 to MON-22: live telemetry + ("tracing", _, &Method::GET) => { + access_token.enforce_permission(Permission::LiveTracing)?; + crate::live::assert_server_level(&access_token)?; + crate::live::live_tracing(req.uri().query()) + } + ("metrics", _, &Method::GET) => { + access_token.enforce_permission(Permission::LiveMetrics)?; + crate::live::assert_server_level(&access_token)?; + crate::live::live_metrics(¶ms) } _ => Err(trc::ResourceEvent::NotFound.into_err()), } @@ -213,11 +255,17 @@ impl ManagementApi for Server { let path = req.uri().path(); let grant = if path.starts_with("/api/live/delivery") { Some((GrantType::LiveDelivery, Permission::LiveDeliveryTest)) + } else if path.starts_with("/api/live/tracing") + || path.starts_with("/api/telemetry/traces/live") + { + // inbuxa: MON-23 + Some((GrantType::LiveTracing, Permission::LiveTracing)) + } else if path.starts_with("/api/live/metrics") + || path.starts_with("/api/telemetry/metrics/live") + { + Some((GrantType::LiveMetrics, Permission::LiveMetrics)) } else { - #[cfg(not(feature = "enterprise"))] - { - None - } + None }; if let Some((grant_type, permission)) = grant { diff --git a/crates/http/src/lib.rs b/crates/http/src/lib.rs index e8bb3be..7696b7d 100644 --- a/crates/http/src/lib.rs +++ b/crates/http/src/lib.rs @@ -10,6 +10,7 @@ pub mod api; pub mod auth; pub mod branding; // inbuxa: branding pub mod form; +pub mod live; // inbuxa: monitoring (MON-20 to MON-24) pub mod request; use common::Inner; diff --git a/crates/http/src/live.rs b/crates/http/src/live.rs new file mode 100644 index 0000000..5659297 --- /dev/null +++ b/crates/http/src/live.rs @@ -0,0 +1,266 @@ +/* + * SPDX-FileCopyrightText: 2026 Coffey Labs + * + * SPDX-License-Identifier: AGPL-3.0-only + */ + +//! Live tracing and live metrics (monitoring spec MON-20 to MON-24), in the +//! shapes INBUXA Admin reads: `event: trace` frames of `x:TraceEvent`s, and +//! `event: metrics` frames of `{@type, metric, count[, sum]}` totals. + +use common::telemetry::tracers::TraceEvents; +use http_body_util::{StreamBody, combinators::BoxBody}; +use http_proto::HttpResponse; +use hyper::{StatusCode, body::Bytes}; +use hyper::body::Frame; +use registry::{jmap::IntoValue, schema::structs::Trace}; +use std::{ + sync::atomic::{AtomicUsize, Ordering}, + time::{Duration, Instant}, +}; +use trc::{ + Collector, EventType, Key, MetricType, + ipc::subscriber::{Interests, SubscriberBuilder}, +}; +use utils::url_params::UrlParams; + +/// Live streams at once per node (MON-24). +const MAX_STREAMS: usize = 8; +/// How long one stream lasts (MON-24). +const STREAM_LIFETIME: Duration = Duration::from_secs(30 * 60); +/// The idle keep-alive (observed upstream: 30 s). +const PING_INTERVAL: Duration = Duration::from_secs(30); + +static STREAMS: AtomicUsize = AtomicUsize::new(0); + +/// A stream slot, freed when the stream ends. +struct Slot; + +impl Slot { + fn take() -> Option { + STREAMS + .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |n| { + (n < MAX_STREAMS).then_some(n + 1) + }) + .ok() + .map(|_| Slot) + } +} + +impl Drop for Slot { + fn drop(&mut self) { + STREAMS.fetch_sub(1, Ordering::SeqCst); + } +} + +/// Removes the live subscriber when the stream ends. +struct Subscription(String); + +impl Drop for Subscription { + fn drop(&mut self) { + Collector::remove_subscriber(std::mem::take(&mut self.0)); + } +} + +/// Live telemetry is server-level: nobody inside a tenant sees it (MON-31). +pub fn assert_server_level(access_token: &common::auth::AccessToken) -> trc::Result<()> { + if access_token.tenant_id().is_some() { + Err(trc::SecurityEvent::Unauthorized + .into_err() + .details("Live telemetry is server-level")) + } else { + Ok(()) + } +} + +fn too_many() -> trc::Error { + trc::LimitEvent::ConcurrentRequest + .into_err() + .details("Too many live telemetry streams") +} + +fn frame(event: &str, data: String) -> Frame { + Frame::data(Bytes::from(format!("event: {event}\ndata: {data}\n\n"))) +} + +fn event_stream(body: BoxBody) -> HttpResponse { + HttpResponse::new(StatusCode::OK) + .with_content_type("text/event-stream") + .with_cache_control("no-store") + .with_stream_body(body) +} + +/// A `Key` from its camel-case name, or the hyphenated one upstream's docs +/// use (`remote-ip`), MON-20. +fn parse_key(name: &str) -> Option { + Key::parse(name).or_else(|| { + let mut camel = String::with_capacity(name.len()); + let mut upper = false; + for c in name.chars() { + if c == '-' || c == '_' { + upper = true; + } else if upper { + camel.extend(c.to_uppercase()); + upper = false; + } else { + camel.push(c); + } + } + Key::parse(&camel) + }) +} + +fn value_text(value: &trc::Value) -> String { + match value { + trc::Value::String(v) => v.to_string(), + trc::Value::UInt(v) => v.to_string(), + trc::Value::Int(v) => v.to_string(), + trc::Value::Ipv4(v) => v.to_string(), + trc::Value::Ipv6(v) => v.to_string(), + trc::Value::Bool(v) => v.to_string(), + trc::Value::Bytes(v) => String::from_utf8_lossy(v).into_owned(), + trc::Value::Array(values) => values.iter().map(value_text).collect::>().join(","), + _ => String::new(), + } +} + +/// `GET /api/live/tracing` (MON-20, MON-21). +pub fn live_tracing(query: Option<&str>) -> trc::Result { + let slot = Slot::take().ok_or_else(too_many)?; + + // Filters: `filter` anywhere, a key name for that key only; all must hold + let mut anywhere = None; + let mut by_key = Vec::new(); + for (name, value) in http_proto::form_urlencoded::parse(query.unwrap_or_default().as_bytes()) { + let value = value.to_lowercase(); + let name = name.as_ref(); + if name == "filter" { + anywhere = Some(value); + } else if name != "token" + && let Some(key) = parse_key(name) + { + by_key.push((key, value)); + } + } + + // Every event but raw I/O (MON-21) + let mut interests = Interests::default(); + for event in EventType::variants() { + if !event.is_raw_io() { + interests.set(event.to_id() as usize); + } + } + let id = format!("live-tracing-{}", store::rand::random::()); + let (_, mut rx) = SubscriberBuilder::new(id.clone()) + .with_interests(interests.clone()) + .with_lossy(true) + .register(); + Collector::union_interests(interests); + Collector::reload(); + let subscription = Subscription(id); + + Ok(event_stream(BoxBody::new(StreamBody::new(async_stream::stream! { + let _slot = slot; + let _subscription = subscription; + let ends = Instant::now() + STREAM_LIFETIME; + loop { + let wait = PING_INTERVAL.min(ends.saturating_duration_since(Instant::now())); + if wait.is_zero() { + break; + } + match tokio::time::timeout(wait, rx.recv()).await { + Ok(Some(events)) => { + let matching = events + .iter() + .filter(|event| { + let texts = event + .keys + .iter() + .map(|(key, value)| (*key, value_text(value).to_lowercase())) + .collect::>(); + anywhere.as_ref().is_none_or(|needle| { + texts.iter().any(|(_, text)| text.contains(needle.as_str())) + }) && by_key.iter().all(|(key, needle)| { + texts.iter().any(|(k, text)| k == key && text == needle) + }) + }) + .map(|event| event.as_ref()) + .collect::>(); + if !matching.is_empty() { + let events = Trace::build_trace_events(matching.iter().copied(), matching.len()) + .into_iter() + .map(|event| event.into_value()) + .collect::>(); + yield Ok(frame("trace", serde_json::to_string(&events).unwrap_or_default())); + } + } + Ok(None) => break, + Err(_) => { + if Instant::now() >= ends { + break; + } + yield Ok(frame("ping", format!("{{\"interval\": {}}}", PING_INTERVAL.as_millis()))); + } + } + } + })))) +} + +/// The current totals of the named metrics, all when `names` is empty, in +/// the `{@type, metric, count[, sum]}` shape (MON-22). +fn metric_frame(names: &[MetricType]) -> String { + let wants = |metric: MetricType| names.is_empty() || names.contains(&metric); + let mut out = Vec::new(); + for counter in Collector::collect_counters() { + if let Some(metric) = MetricType::parse(counter.id().as_str()) + && wants(metric) + { + out.push(serde_json::json!({ + "@type": "Counter", "metric": metric.as_str(), "count": counter.value() + })); + } + } + for gauge in Collector::collect_gauges() { + if wants(gauge.id()) { + out.push(serde_json::json!({ + "@type": "Gauge", "metric": gauge.id().as_str(), "count": gauge.get() + })); + } + } + for histogram in Collector::collect_histograms() { + if wants(histogram.id()) { + out.push(serde_json::json!({ + "@type": "Histogram", "metric": histogram.id().as_str(), + "count": histogram.count(), "sum": histogram.sum() + })); + } + } + serde_json::to_string(&out).unwrap_or_default() +} + +/// `GET /api/live/metrics` (MON-22). +pub fn live_metrics(params: &UrlParams<'_>) -> trc::Result { + let slot = Slot::take().ok_or_else(too_many)?; + let names = params + .get("metrics") + .map(|list| { + list.split(',') + .filter_map(|name| MetricType::parse(name.trim())) + .collect::>() + }) + .unwrap_or_default(); + let interval = Duration::from_secs( + params + .parse::("interval") + .filter(|interval| *interval >= 1) + .unwrap_or(30), + ); + Ok(event_stream(BoxBody::new(StreamBody::new(async_stream::stream! { + let _slot = slot; + let ends = Instant::now() + STREAM_LIFETIME; + while Instant::now() < ends { + yield Ok(frame("metrics", metric_frame(&names))); + tokio::time::sleep(interval).await; + } + })))) +} diff --git a/crates/jmap/src/registry/get.rs b/crates/jmap/src/registry/get.rs index 451c38f..d9a4f48 100644 --- a/crates/jmap/src/registry/get.rs +++ b/crates/jmap/src/registry/get.rs @@ -398,6 +398,7 @@ impl RegistryGet for Server { .await .map(|get| get.into_response()), #[cfg(not(feature = "enterprise"))] + #[allow(unreachable_patterns)] // inbuxa: every object type has an arm now _ => Ok(get.not_found_any().into_response()), } }