moq_stats/

lib.rs

//! Publish and consume MoQ traffic stats.
//!
//! `moq-net` collects per-session traffic counters in a
//! [`stats::Registry`](moq_net::stats::Registry); this crate turns that
//! registry into MoQ broadcasts and back:
//!
//! - [`Producer`] drains a registry on an interval and publishes the counters
//!   as JSON tracks on an origin.
//! - [`Consumer`] subscribes to one published stats broadcast and yields typed
//!   frames, for aggregators, dashboards, and billing meters.
//! - [`aggregate::Consumer`] folds a whole group's per-node broadcasts into one
//!   merged view, so a downstream sees a project's total live traffic as if it
//!   came from a single node.
//!
//! # Wire format
//!
//! A [`Producer`] publishes one broadcast per node at `<prefix>/node/<node>`
//! (default prefix `.stats`), or one per group of leading broadcast-path
//! segments at `<prefix>/<group>/node/<node>`; parse announce paths back with
//! [`parse_node_path`]. Each [`Tier`] carries `publisher.json`,
//! `subscriber.json`, and `sessions.json` tracks of cumulative [`Traffic`] and
//! [`Presence`] counters, plus `.json.z` siblings encoded with
//! [`moq_json::snapshot`]; compute names with [`traffic_track`] /
//! [`sessions_track`]. The full contract (paths, tracks, both encodings, and
//! counter semantics) is at <https://doc.moq.dev/concept/stats>.

pub mod aggregate;
pub mod consume;
pub mod produce;

pub use consume::Consumer;
pub use produce::Producer;

use std::collections::BTreeMap;

/// Counter collection, re-exported from [`moq_net::stats`] so stats consumers
/// can depend on this crate alone.
pub use moq_net::stats::{Handle, Presence, Registry, Role, Tier, Traffic};

use moq_net::{AsPath, Path, PathOwned};

/// One frame off a traffic track: cumulative counters keyed by broadcast path.
pub type TrafficFrame = BTreeMap<String, Traffic>;

/// One frame off a sessions track: connect/disconnect gauges keyed by auth root.
pub type SessionsFrame = BTreeMap<String, Presence>;

/// Suffix appended to a plain track name for its compressed sibling.
pub const COMPRESSED_SUFFIX: &str = ".z";

/// The traffic track name for a tier and role: `<role>.json` at the prefix root
/// on the default tier (`publisher.json` / `subscriber.json`), `<tier>/<role>.json`
/// on a named one, plus [`COMPRESSED_SUFFIX`] when `compressed`.
pub fn traffic_track(tier: &Tier, role: Role, compressed: bool) -> String {
	let mut name = tier.track_name(&format!("{}.json", role.as_str()));
	if compressed {
		name.push_str(COMPRESSED_SUFFIX);
	}
	name
}

/// The sessions track name for a tier: `sessions.json` on the default tier,
/// `<tier>/sessions.json` on a named one, plus [`COMPRESSED_SUFFIX`] when
/// `compressed`.
pub fn sessions_track(tier: &Tier, compressed: bool) -> String {
	let mut name = tier.track_name("sessions.json");
	if compressed {
		name.push_str(COMPRESSED_SUFFIX);
	}
	name
}

/// A parsed stats broadcast path: `<prefix>[/<group>]/node[/<node>]`.
/// See [`parse_node_path`].
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub struct NodePath {
	/// The grouping key: the leading broadcast-path segments selected by the
	/// producer's `depth`, empty at depth 0.
	pub group: PathOwned,
	/// The node suffix, empty when the producer has no node configured.
	pub node: PathOwned,
}

/// Parse a stats broadcast announce path published under `prefix` with the
/// given grouping `depth`, splitting it into its group and node parts.
///
/// Returns `None` when the path is not under `prefix` or has no `node`
/// category segment where one is expected (which also filters out sibling
/// categories another producer may publish under the same prefix). A group
/// segment literally named `node` is ambiguous and will mis-parse; don't name
/// groups that.
pub fn parse_node_path(prefix: impl AsPath, depth: usize, path: impl AsPath) -> Option<NodePath> {
	let prefix = prefix.as_path();
	let path = path.as_path();
	let rest = if prefix.is_empty() {
		path.as_str()
	} else {
		path.as_str().strip_prefix(prefix.as_str())?.strip_prefix('/')?
	};

	// The group is at most `depth` segments (fewer when the broadcast path was
	// shorter), so `node` is the first literal "node" segment at or before
	// index `depth`.
	let mut segments = rest.split('/');
	let mut group: Vec<&str> = Vec::new();
	loop {
		let segment = segments.next()?;
		if segment == "node" {
			break;
		}
		if group.len() >= depth {
			return None;
		}
		group.push(segment);
	}

	let node = segments.collect::<Vec<_>>().join("/");
	Some(NodePath {
		group: Path::new(&group.join("/")).to_owned(),
		node: Path::new(&node).to_owned(),
	})
}

/// Errors produced while publishing or consuming stats.
#[derive(thiserror::Error, Debug, Clone)]
#[non_exhaustive]
pub enum Error {
	/// An error from the underlying track or broadcast.
	#[error(transparent)]
	Net(#[from] moq_net::Error),

	/// An error decoding or encoding a stats frame.
	#[error(transparent)]
	Json(#[from] moq_json::Error),
}

/// A [`Result`](std::result::Result) using this crate's [`Error`].
pub type Result<T> = std::result::Result<T, Error>;

#[cfg(test)]
mod tests {
	use super::*;

	#[test]
	fn parse_node_path_variants() {
		let parse = |depth, path| parse_node_path(".stats", depth, path);

		// Depth 0: no group segment.
		assert_eq!(
			parse(0, ".stats/node/sjc"),
			Some(NodePath {
				group: Path::empty().to_owned(),
				node: Path::new("sjc").to_owned(),
			})
		);
		assert_eq!(
			parse(0, ".stats/node/sjc/1").unwrap().node,
			Path::new("sjc/1").to_owned(),
			"multi-segment node"
		);
		assert_eq!(
			parse(0, ".stats/node"),
			Some(NodePath {
				group: Path::empty().to_owned(),
				node: Path::empty().to_owned(),
			}),
			"nodeless path"
		);

		// Depth 1: one group segment, as published per tenant.
		assert_eq!(
			parse(1, ".stats/acme/node/sjc"),
			Some(NodePath {
				group: Path::new("acme").to_owned(),
				node: Path::new("sjc").to_owned(),
			})
		);
		// A shorter broadcast path yields a shorter group; still parses.
		assert_eq!(parse(1, ".stats/node/sjc").unwrap().group, Path::empty().to_owned());

		// Not ours: wrong prefix, sibling category, group deeper than depth.
		assert_eq!(parse(0, "other/node/sjc"), None);
		assert_eq!(parse(1, ".stats/acme/vod/sjc"), None, "sibling category filtered");
		assert_eq!(parse(0, ".stats/acme/node/sjc"), None, "group deeper than depth");
	}

	#[test]
	fn track_names() {
		let default = Tier::default();
		let regional = Tier::new("region/sjc");
		assert_eq!(traffic_track(&default, Role::Publisher, false), "publisher.json");
		assert_eq!(traffic_track(&default, Role::Subscriber, true), "subscriber.json.z");
		assert_eq!(
			traffic_track(&regional, Role::Publisher, false),
			"region/sjc/publisher.json"
		);
		assert_eq!(sessions_track(&default, false), "sessions.json");
		assert_eq!(sessions_track(&regional, true), "region/sjc/sessions.json.z");
	}
}