mirror of
https://github.com/ruvnet/RuView
synced 2026-08-08 20:11:43 +00:00
feat(sensing-server): add secure opt-in OTLP log export
Import and harden the OpenTelemetry logging work from #1382. Preserve default stderr behavior, register and validate RuView semantic conventions, attach a published schema, enable TLS roots, pin demo images, and fix first-CSI node lifecycle reporting. Supersedes #1382 Closes #1460 Co-authored-by: Jens Holdgaard Pedersen <jens@holdgaard.org>
This commit is contained in:
ruv
Generated
+168
@@ -3976,6 +3976,19 @@ dependencies = [
|
||||
"webpki-roots 1.0.7",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "hyper-timeout"
|
||||
version = "0.5.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "2b90d566bffbce6a75bd8b09a05aa8c2cb1fabb6cb348f8840c9e4c90a0d83b0"
|
||||
dependencies = [
|
||||
"hyper 1.8.1",
|
||||
"hyper-util",
|
||||
"pin-project-lite",
|
||||
"tokio",
|
||||
"tower-service",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "hyper-tls"
|
||||
version = "0.6.0"
|
||||
@@ -5752,6 +5765,73 @@ dependencies = [
|
||||
"vcpkg",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "opentelemetry"
|
||||
version = "0.32.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b0142c63252a9e054e68a4c61a5778f7b14f576274d593f8ce883d191a099682"
|
||||
dependencies = [
|
||||
"js-sys",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "opentelemetry-appender-tracing"
|
||||
version = "0.32.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "2c0080f0dc1d7c786f467cd85a4e395fcab11ee852004f39a29a18ab7c25d837"
|
||||
dependencies = [
|
||||
"opentelemetry",
|
||||
"tracing",
|
||||
"tracing-core",
|
||||
"tracing-subscriber",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "opentelemetry-otlp"
|
||||
version = "0.32.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "9966929966d17620d7c316c643ba62631826e10021409357772d5eea84f62c35"
|
||||
dependencies = [
|
||||
"http 1.4.0",
|
||||
"opentelemetry",
|
||||
"opentelemetry-proto",
|
||||
"opentelemetry_sdk",
|
||||
"prost",
|
||||
"thiserror 2.0.18",
|
||||
"tokio",
|
||||
"tonic",
|
||||
"tonic-types",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "opentelemetry-proto"
|
||||
version = "0.32.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "56d658ba1faf63f7b9c492cfbe6e0ec365440a16132d3270c1065f7b33f1b638"
|
||||
dependencies = [
|
||||
"opentelemetry",
|
||||
"opentelemetry_sdk",
|
||||
"prost",
|
||||
"tonic",
|
||||
"tonic-prost",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "opentelemetry_sdk"
|
||||
version = "0.32.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "9b59f80e1ac4d5ff7a2db8fb6c80badb7f0f3f858211fba08dd9aaec750894f9"
|
||||
dependencies = [
|
||||
"futures-channel",
|
||||
"futures-executor",
|
||||
"futures-util",
|
||||
"opentelemetry",
|
||||
"portable-atomic",
|
||||
"thiserror 2.0.18",
|
||||
"tokio",
|
||||
"tokio-stream",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "option-ext"
|
||||
version = "0.2.0"
|
||||
@@ -6502,6 +6582,38 @@ dependencies = [
|
||||
"unarray",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "prost"
|
||||
version = "0.14.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "528ac67416ff8646872a3c02cad9cc4ee5dc9f9540c9b10771855c95cb2e5ae1"
|
||||
dependencies = [
|
||||
"bytes",
|
||||
"prost-derive",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "prost-derive"
|
||||
version = "0.14.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b570b25f7617e43d59005d0990ccb79e950a423952cea19671b7a876da390adf"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"itertools 0.14.0",
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn 2.0.117",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "prost-types"
|
||||
version = "0.14.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f94967dc7688f3054c7fac87473ffae4cc4c3904800e2d9f5b857246d8963b0a"
|
||||
dependencies = [
|
||||
"prost",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "ptr_meta"
|
||||
version = "0.3.1"
|
||||
@@ -9881,6 +9993,56 @@ version = "1.1.1+spec-1.1.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "756daf9b1013ebe47a8776667b466417e2d4c5679d441c26230efd9ef78692db"
|
||||
|
||||
[[package]]
|
||||
name = "tonic"
|
||||
version = "0.14.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ac2a5518c70fa84342385732db33fb3f44bc4cc748936eb5833d2df34d6445ef"
|
||||
dependencies = [
|
||||
"async-trait",
|
||||
"base64 0.22.1",
|
||||
"bytes",
|
||||
"http 1.4.0",
|
||||
"http-body 1.0.1",
|
||||
"http-body-util",
|
||||
"hyper 1.8.1",
|
||||
"hyper-timeout",
|
||||
"hyper-util",
|
||||
"percent-encoding",
|
||||
"pin-project",
|
||||
"rustls-native-certs 0.8.3",
|
||||
"sync_wrapper 1.0.2",
|
||||
"tokio",
|
||||
"tokio-rustls 0.26.4",
|
||||
"tokio-stream",
|
||||
"tower 0.5.3",
|
||||
"tower-layer",
|
||||
"tower-service",
|
||||
"tracing",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tonic-prost"
|
||||
version = "0.14.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "50849f68853be452acf590cde0b146665b8d507b3b8af17261df47e02c209ea0"
|
||||
dependencies = [
|
||||
"bytes",
|
||||
"prost",
|
||||
"tonic",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tonic-types"
|
||||
version = "0.14.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "73ab1b02061f83d519bba3caa167f88f261ef05720ab8ebc954ade70de3348e8"
|
||||
dependencies = [
|
||||
"prost",
|
||||
"prost-types",
|
||||
"tonic",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "torch-sys"
|
||||
version = "0.24.0"
|
||||
@@ -9922,9 +10084,12 @@ checksum = "ebe5ef63511595f1344e2d5cfa636d973292adc0eec1f0ad45fae9f0851ab1d4"
|
||||
dependencies = [
|
||||
"futures-core",
|
||||
"futures-util",
|
||||
"indexmap 2.13.0",
|
||||
"pin-project-lite",
|
||||
"slab",
|
||||
"sync_wrapper 1.0.2",
|
||||
"tokio",
|
||||
"tokio-util",
|
||||
"tower-layer",
|
||||
"tower-service",
|
||||
"tracing",
|
||||
@@ -11476,6 +11641,9 @@ dependencies = [
|
||||
"jsonwebtoken",
|
||||
"midstreamer-attractor",
|
||||
"midstreamer-temporal-compare",
|
||||
"opentelemetry-appender-tracing",
|
||||
"opentelemetry-otlp",
|
||||
"opentelemetry_sdk",
|
||||
"p256",
|
||||
"proptest",
|
||||
"rand 0.8.5",
|
||||
|
||||
@@ -110,8 +110,22 @@ rand = "0.8"
|
||||
# uses rustls).
|
||||
rumqttc = { version = "0.24", default-features = false, features = ["use-rustls"], optional = true }
|
||||
|
||||
# `otel` feature — OTLP log export (`telemetry` module). Same gating
|
||||
# principle as `mqtt`: the heavy exporter stack (opentelemetry SDK +
|
||||
# tonic) stays out of the default binary; with the feature built,
|
||||
# export still only activates when OTEL_EXPORTER_OTLP_ENDPOINT is set.
|
||||
# Curated event names / attribute keys live in `src/semconv.rs`,
|
||||
# generated from the repo-root `semconv/registry/` by weaver.
|
||||
opentelemetry_sdk = { version = "0.32", default-features = false, features = ["logs", "rt-tokio"], optional = true }
|
||||
opentelemetry-otlp = { version = "0.32", default-features = false, features = ["logs", "grpc-tonic", "tls", "tls-roots"], optional = true }
|
||||
opentelemetry-appender-tracing = { version = "0.32", default-features = false, optional = true }
|
||||
|
||||
[features]
|
||||
default = []
|
||||
# Enables OTLP log export from the `telemetry` module (dogfooding into an
|
||||
# OTLP-native log backend). Without this feature the module falls back to
|
||||
# the plain stderr fmt subscriber.
|
||||
otel = ["dep:opentelemetry_sdk", "dep:opentelemetry-otlp", "dep:opentelemetry-appender-tracing"]
|
||||
# Enables the ADR-115 §2 MQTT auto-discovery publisher. Without this feature
|
||||
# all `--mqtt-*` CLI flags still parse (cli.rs declares them unconditionally),
|
||||
# but enabling `--mqtt` at runtime logs a `WARN` and the publisher is a no-op.
|
||||
|
||||
@@ -28,6 +28,8 @@ pub mod semantic;
|
||||
pub mod rufield_surface;
|
||||
pub mod rvf_container;
|
||||
pub mod rvf_pipeline;
|
||||
pub mod semconv;
|
||||
pub mod telemetry;
|
||||
#[allow(dead_code)]
|
||||
pub mod trainer;
|
||||
pub mod vital_signs;
|
||||
|
||||
@@ -35,7 +35,8 @@ mod vital_signs;
|
||||
|
||||
// Training pipeline modules (exposed via lib.rs)
|
||||
use wifi_densepose_sensing_server::{
|
||||
dataset, embedding, error_response, graph_transformer, rufield_surface, trainer,
|
||||
dataset, embedding, error_response, graph_transformer, rufield_surface, semconv, telemetry,
|
||||
trainer,
|
||||
};
|
||||
|
||||
use ruvector_mincut::{DynamicMinCut, MinCutBuilder};
|
||||
@@ -759,7 +760,11 @@ impl NodeState {
|
||||
})
|
||||
}
|
||||
|
||||
pub(crate) fn observe_csi_frame_arrival(&mut self, now: std::time::Instant) {
|
||||
/// Record a CSI data-frame arrival and return whether it was the node's
|
||||
/// first sensing frame. Sync packets deliberately do not change this
|
||||
/// result, so a sync-before-CSI sequence still produces `node.online`.
|
||||
pub(crate) fn observe_csi_frame_arrival(&mut self, now: std::time::Instant) -> bool {
|
||||
let first_sensing_frame = self.last_frame_time.is_none();
|
||||
if let Some(prev) = self.last_frame_time {
|
||||
let dt = now.duration_since(prev).as_secs_f64();
|
||||
// Burst arrivals (sub-floor dt, issue #1180): do NOT re-anchor on
|
||||
@@ -769,7 +774,7 @@ impl NodeState {
|
||||
// frames arrive in 36 µs bursts every 25 ms still reads ~40 fps,
|
||||
// not 27 kHz.
|
||||
if dt < MIN_PLAUSIBLE_CSI_DT_SEC {
|
||||
return;
|
||||
return false;
|
||||
}
|
||||
if let Some(new_ema) = update_csi_fps_ema(self.csi_fps_ema, dt) {
|
||||
self.csi_fps_ema = new_ema;
|
||||
@@ -777,6 +782,7 @@ impl NodeState {
|
||||
}
|
||||
}
|
||||
self.last_frame_time = Some(now);
|
||||
first_sensing_frame
|
||||
}
|
||||
|
||||
pub(crate) fn new() -> Self {
|
||||
@@ -2841,6 +2847,7 @@ async fn windows_wifi_task(state: SharedState, tick_ms: u64) {
|
||||
if let Ok(json) = serde_json::to_string(&update) {
|
||||
let _ = s.tx.send(json);
|
||||
}
|
||||
observe_sensing_update(s.latest_update.as_ref(), &update);
|
||||
s.latest_update = Some(update);
|
||||
|
||||
debug!(
|
||||
@@ -2996,6 +3003,7 @@ async fn windows_wifi_fallback_tick(state: &SharedState, seq: u32) {
|
||||
if let Ok(json) = serde_json::to_string(&update) {
|
||||
let _ = s.tx.send(json);
|
||||
}
|
||||
observe_sensing_update(s.latest_update.as_ref(), &update);
|
||||
s.latest_update = Some(update);
|
||||
}
|
||||
|
||||
@@ -4768,7 +4776,11 @@ async fn load_model(
|
||||
let mut s = state.write().await;
|
||||
s.active_model_id = Some(model_id.clone());
|
||||
s.model_loaded = true;
|
||||
info!("Model loaded: {model_id}");
|
||||
if telemetry::curated_events_enabled() {
|
||||
info!(name: semconv::EVENT_RUVIEW_MODEL_LOADED, { "ruview.model.id" = %model_id }, "Model loaded: {model_id}");
|
||||
} else {
|
||||
info!("Model loaded: {model_id}");
|
||||
}
|
||||
Json(serde_json::json!({ "success": true, "model_id": model_id }))
|
||||
}
|
||||
|
||||
@@ -5824,7 +5836,18 @@ async fn udp_receiver_task(state: SharedState, udp_port: u16) {
|
||||
// ── Per-node state for edge vitals (issue #249) ──────
|
||||
let node_id = vitals.node_id;
|
||||
let ns = s.node_states.entry(node_id).or_insert_with(NodeState::new);
|
||||
let first_sensing_frame = ns.last_frame_time.is_none();
|
||||
ns.last_frame_time = Some(std::time::Instant::now());
|
||||
if first_sensing_frame && telemetry::curated_events_enabled() {
|
||||
info!(name: semconv::EVENT_RUVIEW_NODE_ONLINE, { "ruview.node.id" = node_id }, "node {node_id} online (edge vitals)");
|
||||
}
|
||||
// Edge-triggered on the fall flag's rising edge (against
|
||||
// the node's previous edge-vitals frame), so a persisting
|
||||
// flag does not re-emit every frame.
|
||||
let prev_fall = ns.edge_vitals.as_ref().is_some_and(|v| v.fall_detected);
|
||||
if vitals.fall_detected && !prev_fall && telemetry::curated_events_enabled() {
|
||||
warn!(name: semconv::EVENT_RUVIEW_FALL_DETECTED, { "ruview.node.id" = node_id }, "fall detected by node {node_id}");
|
||||
}
|
||||
ns.edge_vitals = Some(vitals.clone());
|
||||
ns.rssi_history.push_back(vitals.rssi as f64);
|
||||
if ns.rssi_history.len() > 60 {
|
||||
@@ -6037,6 +6060,7 @@ async fn udp_receiver_task(state: SharedState, udp_port: u16) {
|
||||
if let Ok(json) = serde_json::to_string(&update) {
|
||||
let _ = s.tx.send(json);
|
||||
}
|
||||
observe_sensing_update(s.latest_update.as_ref(), &update);
|
||||
s.latest_update = Some(update);
|
||||
s.edge_vitals = Some(vitals);
|
||||
continue;
|
||||
@@ -6216,7 +6240,11 @@ async fn udp_receiver_task(state: SharedState, udp_port: u16) {
|
||||
// ADR-110 iter 19 — feed the per-node fps EMA from real
|
||||
// CSI arrivals. The helper sets `last_frame_time` as a
|
||||
// side effect, so the previous bare assignment is gone.
|
||||
ns.observe_csi_frame_arrival(std::time::Instant::now());
|
||||
let first_sensing_frame =
|
||||
ns.observe_csi_frame_arrival(std::time::Instant::now());
|
||||
if first_sensing_frame && telemetry::curated_events_enabled() {
|
||||
info!(name: semconv::EVENT_RUVIEW_NODE_ONLINE, { "ruview.node.id" = node_id }, "node {node_id} online (CSI)");
|
||||
}
|
||||
|
||||
// ADR-084 Pass 3: cluster-Pi novelty sensor.
|
||||
// Score this frame's feature vector against the per-node
|
||||
@@ -6481,21 +6509,31 @@ async fn udp_receiver_task(state: SharedState, udp_port: u16) {
|
||||
// held edge-local. `presence == false` ⇒ no phantom event.
|
||||
emit_rufield_event(&s, &update, node_id);
|
||||
|
||||
observe_sensing_update(s.latest_update.as_ref(), &update);
|
||||
s.latest_update = Some(update);
|
||||
|
||||
// Evict stale nodes every 100 ticks to prevent memory leak.
|
||||
if tick % 100 == 0 {
|
||||
let stale = Duration::from_secs(60);
|
||||
let before = s.node_states.len();
|
||||
s.node_states.retain(|_id, ns| {
|
||||
ns.last_frame_time
|
||||
.is_some_and(|t| now.duration_since(t) < stale)
|
||||
});
|
||||
let evicted = before - s.node_states.len();
|
||||
if evicted > 0 {
|
||||
let stale_ids: Vec<u8> = s
|
||||
.node_states
|
||||
.iter()
|
||||
.filter(|(_, ns)| {
|
||||
!ns.last_frame_time
|
||||
.is_some_and(|t| now.duration_since(t) < stale)
|
||||
})
|
||||
.map(|(&id, _)| id)
|
||||
.collect();
|
||||
for id in &stale_ids {
|
||||
s.node_states.remove(id);
|
||||
if telemetry::curated_events_enabled() {
|
||||
info!(name: semconv::EVENT_RUVIEW_NODE_OFFLINE, { "ruview.node.id" = *id }, "node {id} offline (no frames for 60s)");
|
||||
}
|
||||
}
|
||||
if !stale_ids.is_empty() {
|
||||
info!(
|
||||
"Evicted {} stale node(s), {} active",
|
||||
evicted,
|
||||
stale_ids.len(),
|
||||
s.node_states.len()
|
||||
);
|
||||
}
|
||||
@@ -6510,6 +6548,67 @@ async fn udp_receiver_task(state: SharedState, udp_port: u16) {
|
||||
}
|
||||
}
|
||||
|
||||
/// Cadence, in sensing ticks, of the periodic `ruview.csi.stats` and
|
||||
/// `ruview.vitals.estimate` telemetry snapshots.
|
||||
const TELEMETRY_SNAPSHOT_TICKS: u64 = 100;
|
||||
|
||||
/// Emit the curated telemetry events for a finished sensing cycle
|
||||
/// (names and attribute keys from `semconv/registry/` at the repo root):
|
||||
/// `ruview.presence.changed` on presence transitions against the
|
||||
/// previously published update, plus the cadenced `ruview.csi.stats` and
|
||||
/// `ruview.vitals.estimate` snapshots. Called from every path that
|
||||
/// publishes a `SensingUpdate`, right before it lands in `latest_update`
|
||||
/// — never per frame at full rate.
|
||||
fn observe_sensing_update(prev: Option<&SensingUpdate>, update: &SensingUpdate) {
|
||||
if !telemetry::curated_events_enabled() {
|
||||
return;
|
||||
}
|
||||
|
||||
let presence = update.classification.presence;
|
||||
if prev.map(|u| u.classification.presence) != Some(presence) {
|
||||
let state = if presence { "present" } else { "absent" };
|
||||
info!(
|
||||
name: semconv::EVENT_RUVIEW_PRESENCE_CHANGED,
|
||||
{
|
||||
"ruview.presence.state" = state,
|
||||
"ruview.motion.level" = %update.classification.motion_level,
|
||||
"ruview.inference.confidence" = update.classification.confidence,
|
||||
"ruview.persons.count" = update.estimated_persons.unwrap_or(0) as u64,
|
||||
"ruview.csi.source" = %update.source,
|
||||
},
|
||||
"presence changed: {state}"
|
||||
);
|
||||
}
|
||||
if update.tick % TELEMETRY_SNAPSHOT_TICKS == 0 {
|
||||
info!(
|
||||
name: semconv::EVENT_RUVIEW_CSI_STATS,
|
||||
{
|
||||
"ruview.csi.frames_total" = update.tick,
|
||||
"ruview.csi.nodes_active" = update.nodes.len() as u64,
|
||||
"ruview.csi.source" = %update.source,
|
||||
},
|
||||
"csi stats: {} frames processed, {} active node(s)",
|
||||
update.tick,
|
||||
update.nodes.len()
|
||||
);
|
||||
if let Some(v) = &update.vital_signs {
|
||||
if v.breathing_rate_bpm.is_some() || v.heart_rate_bpm.is_some() {
|
||||
info!(
|
||||
name: semconv::EVENT_RUVIEW_VITALS_ESTIMATE,
|
||||
{
|
||||
"ruview.vitals.breathing_rate_bpm" = v.breathing_rate_bpm.unwrap_or(0.0),
|
||||
"ruview.vitals.heart_rate_bpm" = v.heart_rate_bpm.unwrap_or(0.0),
|
||||
"ruview.vitals.breathing_confidence" = v.breathing_confidence,
|
||||
"ruview.vitals.heartbeat_confidence" = v.heartbeat_confidence,
|
||||
"ruview.csi.source" = %update.source,
|
||||
},
|
||||
"vitals estimate"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ── Simulated data task ──────────────────────────────────────────────────────
|
||||
|
||||
async fn simulated_data_task(state: SharedState, tick_ms: u64) {
|
||||
@@ -6659,6 +6758,7 @@ async fn simulated_data_task(state: SharedState, tick_ms: u64) {
|
||||
if let Ok(json) = serde_json::to_string(&update) {
|
||||
let _ = s.tx.send(json);
|
||||
}
|
||||
observe_sensing_update(s.latest_update.as_ref(), &update);
|
||||
s.latest_update = Some(update);
|
||||
}
|
||||
}
|
||||
@@ -7060,13 +7160,11 @@ fn coalesce_ui_path(initial: std::path::PathBuf) -> std::path::PathBuf {
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() {
|
||||
// Initialize tracing
|
||||
tracing_subscriber::fmt()
|
||||
.with_env_filter(
|
||||
tracing_subscriber::EnvFilter::try_from_default_env()
|
||||
.unwrap_or_else(|_| "info,tower_http=debug".into()),
|
||||
)
|
||||
.init();
|
||||
// Initialize tracing; with the `otel` feature and
|
||||
// OTEL_EXPORTER_OTLP_ENDPOINT set, logs also export over OTLP
|
||||
// (service.name = "ruview") — see telemetry.rs. The guard flushes
|
||||
// pending log records on exit.
|
||||
let _telemetry = telemetry::init();
|
||||
|
||||
let mut args = Args::parse();
|
||||
args.ui_path = coalesce_ui_path(args.ui_path);
|
||||
@@ -8577,6 +8675,22 @@ mod sync_snapshot_helper_tests {
|
||||
assert!(ns.sync_snapshot().is_some());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn sync_before_first_csi_still_marks_csi_as_first_sensing_frame() {
|
||||
let mut ns = NodeState::new();
|
||||
let now = std::time::Instant::now();
|
||||
ns.apply_sync_packet(populated_sync(9), now);
|
||||
|
||||
assert!(
|
||||
ns.observe_csi_frame_arrival(now + std::time::Duration::from_millis(20)),
|
||||
"a sync packet must not consume the first sensing-frame transition"
|
||||
);
|
||||
assert!(
|
||||
!ns.observe_csi_frame_arrival(now + std::time::Duration::from_millis(40)),
|
||||
"subsequent CSI frames must not re-emit node.online"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn apply_sync_packet_overwrites_older_data() {
|
||||
// Subsequent packets must replace, not accumulate. Otherwise the
|
||||
|
||||
@@ -35,6 +35,26 @@ use tokio::sync::broadcast;
|
||||
use tokio::task::JoinHandle;
|
||||
use tracing::{error, info, warn};
|
||||
|
||||
macro_rules! otel_error {
|
||||
($($arg:tt)*) => {
|
||||
if crate::telemetry::curated_events_enabled() {
|
||||
error!(name: crate::semconv::EVENT_RUVIEW_MQTT_ERROR, $($arg)*);
|
||||
} else {
|
||||
error!($($arg)*);
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
macro_rules! otel_warn {
|
||||
($($arg:tt)*) => {
|
||||
if crate::telemetry::curated_events_enabled() {
|
||||
warn!(name: crate::semconv::EVENT_RUVIEW_MQTT_ERROR, $($arg)*);
|
||||
} else {
|
||||
warn!($($arg)*);
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
use super::config::{MqttConfig, TlsConfig};
|
||||
use super::discovery::{DiscoveryBuilder, EntityKind};
|
||||
use super::state::{RateLimiter, StateEncoder, StateMessage, VitalsSnapshot};
|
||||
@@ -184,7 +204,7 @@ async fn run(
|
||||
match ev {
|
||||
Ok(_) => {}
|
||||
Err(e) => {
|
||||
error!("[mqtt] event loop error, will reconnect: {e}");
|
||||
otel_error!("[mqtt] event loop error, will reconnect: {e}");
|
||||
rate_limiter.reset();
|
||||
// Brief backoff before next poll attempt.
|
||||
tokio::time::sleep(Duration::from_millis(500)).await;
|
||||
@@ -197,7 +217,7 @@ async fn run(
|
||||
if last_heartbeat.elapsed() >= AVAILABILITY_HEARTBEAT {
|
||||
for (_, na) in nodes.values() {
|
||||
if let Err(e) = publish_availability(&client, na, "online").await {
|
||||
warn!("[mqtt] heartbeat publish failed: {e}");
|
||||
otel_warn!("[mqtt] heartbeat publish failed: {e}");
|
||||
}
|
||||
}
|
||||
last_heartbeat = Instant::now();
|
||||
@@ -207,7 +227,7 @@ async fn run(
|
||||
if let Err(e) =
|
||||
publish_all_discovery(&client, &nb.as_borrowed(), &entities).await
|
||||
{
|
||||
warn!("[mqtt] discovery refresh failed: {e}");
|
||||
otel_warn!("[mqtt] discovery refresh failed: {e}");
|
||||
}
|
||||
}
|
||||
last_refresh = Instant::now();
|
||||
@@ -228,11 +248,11 @@ async fn run(
|
||||
if let Err(e) =
|
||||
publish_all_discovery(&client, &borrowed, &entities).await
|
||||
{
|
||||
warn!("[mqtt] node {} discovery failed: {e}", snap.node_id);
|
||||
otel_warn!("[mqtt] node {} discovery failed: {e}", snap.node_id);
|
||||
}
|
||||
let na = NodeAvailability::for_builder(&borrowed, &entities);
|
||||
if let Err(e) = publish_availability(&client, &na, "online").await {
|
||||
warn!("[mqtt] node {} availability failed: {e}", snap.node_id);
|
||||
otel_warn!("[mqtt] node {} availability failed: {e}", snap.node_id);
|
||||
}
|
||||
nodes.insert(snap.node_id.clone(), (nb, na));
|
||||
}
|
||||
|
||||
@@ -0,0 +1,146 @@
|
||||
//! Generated OpenTelemetry semantic-convention name constants for
|
||||
//! RuView's curated telemetry (event names and attribute keys).
|
||||
//!
|
||||
//! GENERATED from `semconv/registry/` by `weaver registry generate`.
|
||||
//! Do not edit by hand: change the registry or the template at
|
||||
//! `templates/registry/rust/`, then regenerate (the exact command CI
|
||||
//! runs — note `--future`, matching `weaver registry check --future`)
|
||||
//! from the repository root and commit the result:
|
||||
//!
|
||||
//! ```text
|
||||
//! weaver registry generate rust v2/crates/wifi-densepose-sensing-server/src \
|
||||
//! -t templates -r semconv/registry --future
|
||||
//! cargo fmt -p wifi-densepose-sensing-server
|
||||
//! ```
|
||||
//!
|
||||
//! The CI `semconv` workflow fails if this file drifts from the registry.
|
||||
|
||||
/// The semantic-conventions schema URL these constants were generated from —
|
||||
/// the registry manifest's `schema_url`, which carries the conventions
|
||||
/// version. Attach it to a telemetry resource so consumers can resolve the
|
||||
/// schema.
|
||||
pub const SCHEMA_URL: &str =
|
||||
"https://raw.githubusercontent.com/ruvnet/RuView/main/semconv/schema/ruview-0.1.0.yaml";
|
||||
|
||||
// Attribute keys.
|
||||
|
||||
/// `ruview.csi.frames_total` attribute key.
|
||||
pub const RUVIEW_CSI_FRAMES_TOTAL: &str = "ruview.csi.frames_total";
|
||||
|
||||
/// `ruview.csi.nodes_active` attribute key.
|
||||
pub const RUVIEW_CSI_NODES_ACTIVE: &str = "ruview.csi.nodes_active";
|
||||
|
||||
/// `ruview.csi.source` attribute key.
|
||||
pub const RUVIEW_CSI_SOURCE: &str = "ruview.csi.source";
|
||||
|
||||
/// `ruview.inference.confidence` attribute key.
|
||||
pub const RUVIEW_INFERENCE_CONFIDENCE: &str = "ruview.inference.confidence";
|
||||
|
||||
/// `ruview.model.id` attribute key.
|
||||
pub const RUVIEW_MODEL_ID: &str = "ruview.model.id";
|
||||
|
||||
/// `ruview.motion.level` attribute key.
|
||||
pub const RUVIEW_MOTION_LEVEL: &str = "ruview.motion.level";
|
||||
|
||||
/// `ruview.node.id` attribute key.
|
||||
pub const RUVIEW_NODE_ID: &str = "ruview.node.id";
|
||||
|
||||
/// `ruview.persons.count` attribute key.
|
||||
pub const RUVIEW_PERSONS_COUNT: &str = "ruview.persons.count";
|
||||
|
||||
/// `ruview.presence.state` attribute key.
|
||||
pub const RUVIEW_PRESENCE_STATE: &str = "ruview.presence.state";
|
||||
|
||||
/// `ruview.vitals.breathing_confidence` attribute key.
|
||||
pub const RUVIEW_VITALS_BREATHING_CONFIDENCE: &str = "ruview.vitals.breathing_confidence";
|
||||
|
||||
/// `ruview.vitals.breathing_rate_bpm` attribute key.
|
||||
pub const RUVIEW_VITALS_BREATHING_RATE_BPM: &str = "ruview.vitals.breathing_rate_bpm";
|
||||
|
||||
/// `ruview.vitals.heart_rate_bpm` attribute key.
|
||||
pub const RUVIEW_VITALS_HEART_RATE_BPM: &str = "ruview.vitals.heart_rate_bpm";
|
||||
|
||||
/// `ruview.vitals.heartbeat_confidence` attribute key.
|
||||
pub const RUVIEW_VITALS_HEARTBEAT_CONFIDENCE: &str = "ruview.vitals.heartbeat_confidence";
|
||||
|
||||
/// Every attribute key registered for curated RuView events.
|
||||
pub const ATTRIBUTE_KEYS: &[&str] = &[
|
||||
RUVIEW_CSI_FRAMES_TOTAL,
|
||||
RUVIEW_CSI_NODES_ACTIVE,
|
||||
RUVIEW_CSI_SOURCE,
|
||||
RUVIEW_INFERENCE_CONFIDENCE,
|
||||
RUVIEW_MODEL_ID,
|
||||
RUVIEW_MOTION_LEVEL,
|
||||
RUVIEW_NODE_ID,
|
||||
RUVIEW_PERSONS_COUNT,
|
||||
RUVIEW_PRESENCE_STATE,
|
||||
RUVIEW_VITALS_BREATHING_CONFIDENCE,
|
||||
RUVIEW_VITALS_BREATHING_RATE_BPM,
|
||||
RUVIEW_VITALS_HEART_RATE_BPM,
|
||||
RUVIEW_VITALS_HEARTBEAT_CONFIDENCE,
|
||||
];
|
||||
|
||||
// Log event names (each instrumented `tracing` call site names its event
|
||||
// with one of these so the exported Logs signal stays registry-backed).
|
||||
|
||||
/// `ruview.csi.stats` log event name.
|
||||
pub const EVENT_RUVIEW_CSI_STATS: &str = "ruview.csi.stats";
|
||||
|
||||
/// `ruview.fall.detected` log event name.
|
||||
pub const EVENT_RUVIEW_FALL_DETECTED: &str = "ruview.fall.detected";
|
||||
|
||||
/// `ruview.model.loaded` log event name.
|
||||
pub const EVENT_RUVIEW_MODEL_LOADED: &str = "ruview.model.loaded";
|
||||
|
||||
/// `ruview.mqtt.error` log event name.
|
||||
pub const EVENT_RUVIEW_MQTT_ERROR: &str = "ruview.mqtt.error";
|
||||
|
||||
/// `ruview.node.offline` log event name.
|
||||
pub const EVENT_RUVIEW_NODE_OFFLINE: &str = "ruview.node.offline";
|
||||
|
||||
/// `ruview.node.online` log event name.
|
||||
pub const EVENT_RUVIEW_NODE_ONLINE: &str = "ruview.node.online";
|
||||
|
||||
/// `ruview.presence.changed` log event name.
|
||||
pub const EVENT_RUVIEW_PRESENCE_CHANGED: &str = "ruview.presence.changed";
|
||||
|
||||
/// `ruview.vitals.estimate` log event name.
|
||||
pub const EVENT_RUVIEW_VITALS_ESTIMATE: &str = "ruview.vitals.estimate";
|
||||
|
||||
/// Every curated event name in the generated registry.
|
||||
pub const EVENT_NAMES: &[&str] = &[
|
||||
EVENT_RUVIEW_CSI_STATS,
|
||||
EVENT_RUVIEW_FALL_DETECTED,
|
||||
EVENT_RUVIEW_MODEL_LOADED,
|
||||
EVENT_RUVIEW_MQTT_ERROR,
|
||||
EVENT_RUVIEW_NODE_OFFLINE,
|
||||
EVENT_RUVIEW_NODE_ONLINE,
|
||||
EVENT_RUVIEW_PRESENCE_CHANGED,
|
||||
EVENT_RUVIEW_VITALS_ESTIMATE,
|
||||
];
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::{ATTRIBUTE_KEYS, EVENT_NAMES};
|
||||
|
||||
#[test]
|
||||
fn instrumentation_uses_only_registered_ruview_literals() {
|
||||
let sources = [include_str!("main.rs"), include_str!("mqtt/publisher.rs")];
|
||||
|
||||
for source in sources {
|
||||
let mut rest = source;
|
||||
while let Some(start) = rest.find("\"ruview.") {
|
||||
let value = &rest[start + 1..];
|
||||
let end = value
|
||||
.find('"')
|
||||
.expect("ruview string literal must have a closing quote");
|
||||
let literal = &value[..end];
|
||||
assert!(
|
||||
ATTRIBUTE_KEYS.contains(&literal) || EVENT_NAMES.contains(&literal),
|
||||
"instrumentation literal `{literal}` is absent from semconv/registry"
|
||||
);
|
||||
rest = &value[end + 1..];
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,155 @@
|
||||
//! Tracing bootstrap, with optional OTLP log export.
|
||||
//!
|
||||
//! Without the `otel` cargo feature (the default) this is exactly the
|
||||
//! stderr `tracing_subscriber::fmt()` setup the server has always had.
|
||||
//! With the feature, and only when `OTEL_EXPORTER_OTLP_ENDPOINT` is set
|
||||
//! in the environment, [`init`] additionally installs an
|
||||
//! `opentelemetry-appender-tracing` bridge so every `tracing` event is
|
||||
//! exported as an OpenTelemetry log record (`service.name = "ruview"`)
|
||||
//! to that endpoint over OTLP/gRPC. Endpoint unset means the OTLP
|
||||
//! pipeline is never constructed and no curated events are emitted.
|
||||
//!
|
||||
//! The curated events named in [`crate::semconv`] carry registry-backed
|
||||
//! event names and attribute keys (see `semconv/registry/` at the repo
|
||||
//! root); everything else exports under tracing's default event names.
|
||||
|
||||
#[cfg(feature = "otel")]
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
|
||||
/// Service name reported in the OTLP resource. Log backends that derive
|
||||
/// a tenant from `service.name` file all RuView logs under it.
|
||||
#[cfg(feature = "otel")]
|
||||
const SERVICE_NAME: &str = "ruview";
|
||||
|
||||
/// True only after an explicitly configured OTLP logger has been installed.
|
||||
///
|
||||
/// Curated sensing events are new output, so call sites consult this flag
|
||||
/// before emitting them. This preserves the server's pre-OTel stderr behavior
|
||||
/// both in default builds and in `otel` builds without a working exporter.
|
||||
#[cfg(feature = "otel")]
|
||||
static CURATED_EVENTS_ENABLED: AtomicBool = AtomicBool::new(false);
|
||||
|
||||
/// The server's long-standing default log filter.
|
||||
const DEFAULT_FILTER: &str = "info,tower_http=debug";
|
||||
|
||||
/// Whether the explicitly configured OTLP pipeline is ready to receive curated
|
||||
/// `ruview.*` events.
|
||||
#[cfg(feature = "otel")]
|
||||
pub fn curated_events_enabled() -> bool {
|
||||
CURATED_EVENTS_ENABLED.load(Ordering::Acquire)
|
||||
}
|
||||
|
||||
/// Default builds cannot emit curated OTLP events.
|
||||
#[cfg(not(feature = "otel"))]
|
||||
pub const fn curated_events_enabled() -> bool {
|
||||
false
|
||||
}
|
||||
|
||||
/// Owns the OTLP logger pipeline when one was installed. Hold it for the
|
||||
/// process lifetime; dropping it flushes pending log records.
|
||||
pub struct TelemetryGuard {
|
||||
#[cfg(feature = "otel")]
|
||||
logger: Option<opentelemetry_sdk::logs::SdkLoggerProvider>,
|
||||
}
|
||||
|
||||
impl Drop for TelemetryGuard {
|
||||
fn drop(&mut self) {
|
||||
#[cfg(feature = "otel")]
|
||||
if let Some(logger) = &self.logger {
|
||||
CURATED_EVENTS_ENABLED.store(false, Ordering::Release);
|
||||
let _ = logger.shutdown();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Install the global `tracing` subscriber. Call once, at start-up,
|
||||
/// inside the tokio runtime (the OTLP exporter runs on it).
|
||||
#[must_use = "dropping the guard tears the OTLP log pipeline down"]
|
||||
pub fn init() -> TelemetryGuard {
|
||||
#[cfg(feature = "otel")]
|
||||
if std::env::var_os("OTEL_EXPORTER_OTLP_ENDPOINT").is_some() {
|
||||
match init_with_otlp() {
|
||||
Ok(logger) => {
|
||||
CURATED_EVENTS_ENABLED.store(true, Ordering::Release);
|
||||
return TelemetryGuard {
|
||||
logger: Some(logger),
|
||||
};
|
||||
}
|
||||
Err(e) => eprintln!("OTLP log export disabled (exporter build failed): {e}"),
|
||||
}
|
||||
}
|
||||
|
||||
init_fmt_only();
|
||||
TelemetryGuard {
|
||||
#[cfg(feature = "otel")]
|
||||
logger: None,
|
||||
}
|
||||
}
|
||||
|
||||
fn init_fmt_only() {
|
||||
tracing_subscriber::fmt()
|
||||
.with_env_filter(
|
||||
tracing_subscriber::EnvFilter::try_from_default_env()
|
||||
.unwrap_or_else(|_| DEFAULT_FILTER.into()),
|
||||
)
|
||||
.init();
|
||||
}
|
||||
|
||||
#[cfg(feature = "otel")]
|
||||
fn init_with_otlp(
|
||||
) -> Result<opentelemetry_sdk::logs::SdkLoggerProvider, opentelemetry_otlp::ExporterBuildError> {
|
||||
use opentelemetry_appender_tracing::layer::OpenTelemetryTracingBridge;
|
||||
use opentelemetry_otlp::LogExporter;
|
||||
use opentelemetry_sdk::logs::SdkLoggerProvider;
|
||||
use opentelemetry_sdk::Resource;
|
||||
use tracing_subscriber::layer::SubscriberExt as _;
|
||||
use tracing_subscriber::util::SubscriberInitExt as _;
|
||||
use tracing_subscriber::{EnvFilter, Layer as _};
|
||||
|
||||
// The exporter reads OTEL_EXPORTER_OTLP_ENDPOINT (and the other
|
||||
// OTEL_EXPORTER_* variables) from the environment itself.
|
||||
let exporter = LogExporter::builder().with_tonic().build()?;
|
||||
let resource = Resource::builder()
|
||||
.with_service_name(SERVICE_NAME)
|
||||
.with_schema_url([], crate::semconv::SCHEMA_URL)
|
||||
.build();
|
||||
let logger = SdkLoggerProvider::builder()
|
||||
.with_batch_exporter(exporter)
|
||||
.with_resource(resource)
|
||||
.build();
|
||||
|
||||
// Telemetry-induced-telemetry loop guard: the OTLP exporter is itself
|
||||
// a tonic/hyper client, so its internal tracing events must not
|
||||
// re-enter the bridge (a failed export would emit records that
|
||||
// trigger more exports). The `off` directives win over RUST_LOG.
|
||||
let mut bridge_filter =
|
||||
EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new(DEFAULT_FILTER));
|
||||
for directive in [
|
||||
"hyper=off",
|
||||
"tonic=off",
|
||||
"h2=off",
|
||||
"tower=off",
|
||||
"opentelemetry=off",
|
||||
"opentelemetry_sdk=off",
|
||||
"opentelemetry_otlp=off",
|
||||
] {
|
||||
if let Ok(directive) = directive.parse() {
|
||||
bridge_filter = bridge_filter.add_directive(directive);
|
||||
}
|
||||
}
|
||||
let bridge = OpenTelemetryTracingBridge::new(&logger).with_filter(bridge_filter);
|
||||
let fmt = tracing_subscriber::fmt::layer().with_filter(
|
||||
EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new(DEFAULT_FILTER)),
|
||||
);
|
||||
tracing_subscriber::registry().with(bridge).with(fmt).init();
|
||||
Ok(logger)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
#[cfg(not(feature = "otel"))]
|
||||
#[test]
|
||||
fn default_build_never_enables_curated_events() {
|
||||
assert!(!super::curated_events_enabled());
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user