diff --git a/crates/common/src/config/telemetry.rs b/crates/common/src/config/telemetry.rs index 1884e82..fa65202 100644 --- a/crates/common/src/config/telemetry.rs +++ b/crates/common/src/config/telemetry.rs @@ -31,6 +31,10 @@ pub struct TelemetrySubscriber { pub interests: Interests, pub typ: TelemetrySubscriberType, pub lossy: bool, + /// inbuxa: a hash of the settings the running tracer is built from + /// (everything but its events, level and lossiness, which change in + /// place), so a reload can tell which tracers to start over. + pub settings: u64, } #[allow(clippy::large_enum_variant)] @@ -167,6 +171,7 @@ impl Tracers { for tracer in bp.list_infallible::().await { let id = tracer.id; let tracer = tracer.object; + let settings = tracer_settings(&tracer); let level; let lossy; let events; @@ -379,6 +384,7 @@ impl Tracers { interests: Default::default(), lossy, typ, + settings, }; // Parse disabled events @@ -426,6 +432,7 @@ impl Tracers { for hook in bp.list_infallible::().await { let id = hook.id; let hook = hook.object; + let settings = webhook_settings(&hook); if !hook.enable { continue; @@ -448,6 +455,7 @@ impl Tracers { id: format!("w_{}", id.id()), interests: Default::default(), lossy: hook.lossy, + settings, typ: TelemetrySubscriberType::Webhook(WebhookTracer { url: hook.url, timeout: hook.timeout.into_inner(), @@ -516,6 +524,8 @@ impl Tracers { data: storage.data.clone(), }), lossy: true, + // Stores take a restart + settings: 0, }); } @@ -541,6 +551,7 @@ impl Tracers { buffered: true, }), lossy: false, + settings: 0, }); } } else { @@ -568,6 +579,7 @@ impl Tracers { buffered: true, }), lossy: false, + settings: 0, }); } @@ -701,6 +713,42 @@ impl Metrics { } } +// inbuxa: what a tracer is built from, less what changes in place +macro_rules! in_place_reset { + ($tracer:expr) => {{ + $tracer.enable = true; + $tracer.level = Default::default(); + $tracer.lossy = false; + $tracer.events = Default::default(); + $tracer.events_policy = Default::default(); + }}; +} + +fn settings_hash(settings: &impl std::fmt::Debug) -> u64 { + use std::hash::{Hash, Hasher}; + let mut hasher = std::collections::hash_map::DefaultHasher::new(); + format!("{settings:?}").hash(&mut hasher); + hasher.finish() +} + +fn tracer_settings(tracer: &Tracer) -> u64 { + let mut tracer = tracer.clone(); + match &mut tracer { + Tracer::Log(tracer) => in_place_reset!(tracer), + Tracer::Stdout(tracer) => in_place_reset!(tracer), + Tracer::Journal(tracer) => in_place_reset!(tracer), + Tracer::OtelHttp(tracer) => in_place_reset!(tracer), + Tracer::OtelGrpc(tracer) => in_place_reset!(tracer), + } + settings_hash(&tracer) +} + +fn webhook_settings(hook: &WebHook) -> u64 { + let mut hook = hook.clone(); + in_place_reset!(hook); + settings_hash(&hook) +} + fn apply_events( event_types: impl IntoIterator, policy: EventPolicy, diff --git a/crates/common/src/telemetry/mod.rs b/crates/common/src/telemetry/mod.rs index 0f1a404..f75ce64 100644 --- a/crates/common/src/telemetry/mod.rs +++ b/crates/common/src/telemetry/mod.rs @@ -14,15 +14,26 @@ pub mod webhooks; use tracers::log::spawn_log_tracer; use tracers::otel::spawn_otel_tracer; use tracers::stdout::spawn_console_tracer; +use ahash::AHashMap; +use parking_lot::Mutex; use trc::{Collector, ipc::subscriber::SubscriberBuilder}; use webhooks::spawn_webhook_tracer; use crate::config::telemetry::{Telemetry, TelemetrySubscriberType}; +/// inbuxa: the tracers this server started, by subscriber id, with the +/// settings each was built from. Live-tracing streams and other subscribers +/// registered elsewhere aren't listed, so a reload leaves them running. +static RUNNING_TRACERS: Mutex>> = Mutex::new(None); + impl Telemetry { pub fn enable(self) { + let mut running = RUNNING_TRACERS.lock(); + let running = running.get_or_insert_with(AHashMap::new); + // Spawn tracers for tracer in self.tracers.subscribers { + running.insert(tracer.id.clone(), tracer.settings); tracer.typ.spawn( SubscriberBuilder::new(tracer.id) .with_interests(tracer.interests) @@ -37,25 +48,39 @@ impl Telemetry { Collector::reload(); } + // inbuxa: upstream only refreshed the events, level and lossiness of a + // tracer that was already running, so a Log tracer moved to another + // path (or any tracer whose own settings changed) kept going as it was + // built until a restart, while the reload reported the change applied. + // A tracer whose settings changed is now started over: the new one is + // registered under the same id and the collector swaps it in at an + // event boundary, so no event is lost or written twice (see + // Update::RegisterSubscriber); the old one writes what it has queued + // and stops. pub fn update(self) { + let mut running = RUNNING_TRACERS.lock(); + let running = running.get_or_insert_with(AHashMap::new); + // Remove tracers that are no longer active - let active_subscribers = Collector::get_subscribers(); - for subscribed_id in &active_subscribers { - if !self + running.retain(|id, _| { + let keep = self .tracers .subscribers .iter() - .any(|tracer| tracer.id == *subscribed_id) - { - Collector::remove_subscriber(subscribed_id.clone()); + .any(|tracer| tracer.id == *id); + if !keep { + Collector::remove_subscriber(id.clone()); } - } + keep + }); - // Activate new tracers or update existing ones + // Start new tracers, start over those whose settings changed and + // update the rest in place for tracer in self.tracers.subscribers { - if active_subscribers.contains(&tracer.id) { + if running.get(&tracer.id) == Some(&tracer.settings) { Collector::update_subscriber(tracer.id, tracer.interests, tracer.lossy); } else { + running.insert(tracer.id.clone(), tracer.settings); tracer.typ.spawn( SubscriberBuilder::new(tracer.id) .with_interests(tracer.interests) diff --git a/crates/common/src/telemetry/tracers/log.rs b/crates/common/src/telemetry/tracers/log.rs index 455db0f..930fdbe 100644 --- a/crates/common/src/telemetry/tracers/log.rs +++ b/crates/common/src/telemetry/tracers/log.rs @@ -2,6 +2,8 @@ * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC * * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + * + * Modified by Coffey Labs in 2026 for INBUXA. */ use std::{path::PathBuf, time::SystemTime}; @@ -15,9 +17,27 @@ use tokio::{ }; use trc::{TelemetryEvent, ipc::subscriber::SubscriberBuilder, serializers::text::FmtWriter}; +// inbuxa: when a Log tracer is started over on the same files (its rotation +// or format changed), the new one waits for the old one to write what it +// has queued, so their lines don't interleave. Keyed by path and prefix; +// each entry is the last tracer's "done" signal, sent when it ends. +type LogFileOwners = ahash::AHashMap<(String, String), tokio::sync::oneshot::Receiver<()>>; +static LOG_FILE_OWNERS: parking_lot::Mutex> = parking_lot::Mutex::new(None); + pub(crate) fn spawn_log_tracer(builder: SubscriberBuilder, settings: LogTracer) { + let (done_tx, done_rx) = tokio::sync::oneshot::channel::<()>(); + let previous = LOG_FILE_OWNERS + .lock() + .get_or_insert_with(Default::default) + .insert((settings.path.clone(), settings.prefix.clone()), done_rx); let (_, mut rx) = builder.register(); tokio::spawn(async move { + // Dropped when this tracer ends, however it ends + let _done = done_tx; + if let Some(previous) = previous { + let _ = previous.await; + } + if let Some(writer) = settings.build_writer().await { let mut buf = FmtWriter::new(writer) .with_ansi(settings.ansi) diff --git a/crates/common/src/telemetry/tracers/otel.rs b/crates/common/src/telemetry/tracers/otel.rs index cf2685f..af1a3c4 100644 --- a/crates/common/src/telemetry/tracers/otel.rs +++ b/crates/common/src/telemetry/tracers/otel.rs @@ -47,6 +47,10 @@ pub(crate) fn spawn_otel_tracer(builder: SubscriberBuilder, mut otel: OtelTracer let mut pending_spans = Vec::new(); let mut active_spans = AHashMap::new(); + let mut closing = false; + let started = std::time::SystemTime::now() + .duration_since(std::time::SystemTime::UNIX_EPOCH) + .map_or(0, |d| d.as_secs()); loop { // Wait for the next event or timeout @@ -75,12 +79,26 @@ pub(crate) fn spawn_otel_tracer(builder: SubscriberBuilder, mut otel: OtelTracer events.iter().chain(std::iter::once(&event)), &instrumentation, )); + } else if span.inner.timestamp < started { + // inbuxa: a span that was open when this + // tracer replaced another one (its settings + // changed) is exported with its end event + // rather than dropped + pending_spans.push(build_span_data( + span, + &event, + std::iter::once(&event), + &instrumentation, + )); } } } } Ok(None) => { - break; + // inbuxa: the tracer was removed or replaced; export + // what is pending now rather than drop it + closing = true; + next_delivery = Instant::now(); } Err(_) => (), } @@ -131,6 +149,9 @@ pub(crate) fn spawn_otel_tracer(builder: SubscriberBuilder, mut otel: OtelTracer } } } + if closing { + break; + } wakeup_time = next_retry.unwrap_or(LONG_1Y_SLUMBER); } }); diff --git a/crates/common/src/telemetry/webhooks/mod.rs b/crates/common/src/telemetry/webhooks/mod.rs index 33f966a..218b039 100644 --- a/crates/common/src/telemetry/webhooks/mod.rs +++ b/crates/common/src/telemetry/webhooks/mod.rs @@ -2,6 +2,8 @@ * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC * * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + * + * Modified by Coffey Labs in 2026 for INBUXA. */ use crate::{LONG_1Y_SLUMBER, config::telemetry::WebhookTracer}; @@ -25,6 +27,11 @@ use trc::{ pub(crate) fn spawn_webhook_tracer(builder: SubscriberBuilder, settings: WebhookTracer) { let (tx, mut rx) = builder.register(); + // inbuxa: failed deliveries come back through a weak sender, so the + // channel closes when the collector drops this webhook (removed, or + // replaced after a settings change) and the task ends; upstream held a + // sender here and the task outlived its subscription + let tx = tx.downgrade(); tokio::spawn(async move { let settings = Arc::new(settings); let mut wakeup_time = LONG_1Y_SLUMBER; @@ -58,6 +65,15 @@ pub(crate) fn spawn_webhook_tracer(builder: SubscriberBuilder, settings: Webhook } } Ok(None) => { + // inbuxa: deliver what is pending rather than drop it + if !pending_events.is_empty() { + spawn_webhook_handler( + settings.clone(), + in_flight.clone(), + std::mem::take(&mut pending_events), + tx.clone(), + ); + } break; } Err(_) => (), @@ -102,7 +118,7 @@ fn spawn_webhook_handler( settings: Arc, in_flight: Arc, events: EventBatch, - webhook_tx: mpsc::Sender, + webhook_tx: mpsc::WeakSender, ) { tokio::spawn(async move { in_flight.store(true, Ordering::Relaxed); @@ -113,7 +129,11 @@ fn spawn_webhook_handler( if let Err(err) = post_webhook_events(&settings, &wrapper).await { trc::event!(Telemetry(TelemetryEvent::WebhookError), Details = err); - if webhook_tx.send(wrapper.events.into_inner()).await.is_err() { + let sent = match webhook_tx.upgrade() { + Some(webhook_tx) => webhook_tx.send(wrapper.events.into_inner()).await.is_ok(), + None => false, + }; + if !sent { trc::event!( Server(ServerEvent::ThreadError), Details = "Failed to send failed webhook events back to main thread", diff --git a/crates/trc/src/ipc/collector.rs b/crates/trc/src/ipc/collector.rs index 0b30de3..e8c8af9 100644 --- a/crates/trc/src/ipc/collector.rs +++ b/crates/trc/src/ipc/collector.rs @@ -245,9 +245,27 @@ impl Collector { Update::RegisterReceiver { receiver } => { self.receivers.push(receiver); } - Update::RegisterSubscriber { subscriber } => { - ACTIVE_SUBSCRIBERS.lock().push(subscriber.id.clone()); - self.subscribers.push(subscriber); + Update::RegisterSubscriber { mut subscriber } => { + // inbuxa: a subscriber registered under the id of a + // running one replaces it (a tracer whose settings + // changed). Every event collected so far went to the old + // one, every later event goes to the new one: the old + // one's batch is sent first (anything its full channel + // can't take moves over, rather than being dropped), and + // dropping it closes its channel, so its task writes + // what is queued and ends. + if let Some(old) = self.subscribers.iter_mut().find(|s| s.id == subscriber.id) { + let _ = old.send_batch(); + if !old.batch.is_empty() { + let mut batch = std::mem::take(&mut old.batch); + batch.append(&mut subscriber.batch); + subscriber.batch = batch; + } + *old = subscriber; + } else { + ACTIVE_SUBSCRIBERS.lock().push(subscriber.id.clone()); + self.subscribers.push(subscriber); + } } Update::UnregisterSubscriber { id } => { ACTIVE_SUBSCRIBERS.lock().retain(|s| s != &id); diff --git a/crates/trc/src/ipc/subscriber.rs b/crates/trc/src/ipc/subscriber.rs index 7a9c588..f72623f 100644 --- a/crates/trc/src/ipc/subscriber.rs +++ b/crates/trc/src/ipc/subscriber.rs @@ -2,6 +2,8 @@ * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC * * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + * + * Modified by Coffey Labs in 2026 for INBUXA. */ use std::sync::Arc; @@ -105,6 +107,9 @@ impl SubscriberBuilder { self } + /// Registers the subscriber with the collector. inbuxa: one registered + /// under the id of a running subscriber replaces it, handing over at an + /// event boundary; the old one's channel then closes. pub fn register(self) -> (mpsc::Sender, mpsc::Receiver) { let (tx, rx) = mpsc::channel(8192); diff --git a/tests/src/system/mod.rs b/tests/src/system/mod.rs index b48db1c..5e1856c 100644 --- a/tests/src/system/mod.rs +++ b/tests/src/system/mod.rs @@ -24,6 +24,7 @@ pub mod quota; pub mod reload; // inbuxa: reloads and build errors pub mod security; pub mod task; +pub mod tracer_reload; // inbuxa: tracers start over when their settings change pub mod tenant; pub mod undelete; diff --git a/tests/src/system/tracer_reload.rs b/tests/src/system/tracer_reload.rs new file mode 100644 index 0000000..eb3c586 --- /dev/null +++ b/tests/src/system/tracer_reload.rs @@ -0,0 +1,204 @@ +/* + * SPDX-FileCopyrightText: 2026 Coffey Labs + * + * SPDX-License-Identifier: AGPL-3.0-only + */ + +// inbuxa: a tracer whose own settings change is started over by the reload +// that follows the write: a Log tracer moved to another directory writes +// there from then on, and no event is lost or written twice on the way. + +use crate::utils::{ + jmap::JmapResponse, + server::{TestServer, TestServerBuilder}, +}; +use registry::{ + schema::{ + enums::{EventPolicy, LogRotateFrequency, TracingLevel}, + prelude::ObjectType, + structs::{Expression, MtaStageAuth, Tracer, TracerLog}, + }, + types::map::Map, +}; +use serde_json::json; +use std::{ + path::{Path, PathBuf}, + time::{Duration, Instant}, +}; +use trc::{EventType, ServerEvent}; + +const PREFIX: &str = "tracer-reload"; + +#[tokio::test(flavor = "multi_thread")] +pub async fn tracer_reload_tests() { + let mut test = TestServerBuilder::new("tracer_reload_tests") + .await + .with_default_listeners() + .await + .with_object(MtaStageAuth { + require: Expression { + else_: "false".to_string(), + ..Default::default() + }, + ..Default::default() + }) + .await + .build() + .await; + + let admin = test + .create_user_account( + "admin", + "admin@example.org", + "these_pretzels_are_making_me_thirsty", + &[], + "Admin", + ) + .await; + test.account("admin") + .assign_roles_to_account(admin.id(), &["user", "system"]) + .await; + test.insert_account(admin); + + test_log_tracer_moves(&test).await; + + if test.is_reset() { + test.temp_dir.delete(); + } +} + +async fn test_log_tracer_moves(test: &TestServer) { + println!("Running Log tracer path change..."); + let admin = test.account("admin@example.org"); + let old_dir = test.temp_dir.path.join("tracer-old"); + let new_dir = test.temp_dir.path.join("tracer-new"); + for dir in [&old_dir, &new_dir] { + let _ = std::fs::remove_dir_all(dir); + std::fs::create_dir_all(dir).unwrap(); + } + let old_file = old_dir.join(PREFIX); + let new_file = new_dir.join(PREFIX); + + // A Log tracer for one event type, written to the old directory + let response = admin + .registry_create([Tracer::Log(TracerLog { + path: old_dir.to_string_lossy().into_owned(), + prefix: PREFIX.into(), + rotate: LogRotateFrequency::Never, + ansi: false, + multiline: false, + enable: true, + level: TracingLevel::Trace, + lossy: false, + events: Map::new(vec![EventType::Server(ServerEvent::Licensing)]), + events_policy: EventPolicy::Include, + })]) + .await; + assert_applied(&response); + let tracer_id = response.created_id(0); + emit("marker-before"); + wait_for(&old_file, "marker-before").await; + + // Events keep coming while the path changes + let stream = tokio::spawn(async { + for i in 0..2000u32 { + emit(&format!("seq-{i:05}-end")); + if i % 50 == 0 { + tokio::time::sleep(Duration::from_millis(1)).await; + } + } + }); + tokio::time::sleep(Duration::from_millis(5)).await; + let response = admin + .registry_update( + ObjectType::Tracer, + [(tracer_id, json!({"path": new_dir.to_string_lossy()}))], + ) + .await; + assert_applied(&response); + stream.await.unwrap(); + + // Once the reload has run, events go to the new file only + tokio::time::sleep(Duration::from_millis(200)).await; + emit("marker-after"); + wait_for(&new_file, "marker-after").await; + wait_for(&new_file, "seq-01999-end").await; + tokio::time::sleep(Duration::from_millis(200)).await; + let old = read(&old_file); + let new = read(&new_file); + assert!(!old.contains("marker-after"), "old file still written to"); + assert!(!new.contains("marker-before")); + + // Every event written once, in one file or the other + let old_seq = count_seq(&old); + let new_seq = count_seq(&new); + println!( + "{} events in the old file, {} in the new one", + old_seq.iter().filter(|c| **c > 0).count(), + new_seq.iter().filter(|c| **c > 0).count() + ); + for i in 0..2000 { + assert_eq!( + old_seq[i] + new_seq[i], + 1, + "seq-{i:05} written {} + {} times", + old_seq[i], + new_seq[i] + ); + } + assert!( + new_seq.iter().any(|c| *c > 0), + "no event of the stream reached the new file" + ); + + // Removing the tracer stops it + let response = admin + .registry_destroy(ObjectType::Tracer, [tracer_id]) + .await; + assert_applied(&response); + tokio::time::sleep(Duration::from_millis(200)).await; + emit("marker-removed"); + tokio::time::sleep(Duration::from_millis(300)).await; + assert!(!read(&new_file).contains("marker-removed")); + assert!(!read(&old_file).contains("marker-removed")); +} + +fn emit(marker: &str) { + trc::event!(Server(ServerEvent::Licensing), Details = marker.to_string()); +} + +fn read(path: &Path) -> String { + std::fs::read_to_string(path).unwrap_or_default() +} + +fn count_seq(text: &str) -> Vec { + let mut counts = vec![0u32; 2000]; + for part in text.split("seq-").skip(1) { + if let Some(n) = part.get(..5).and_then(|n| n.parse::().ok()) + && part[5..].starts_with("-end") + { + counts[n] += 1; + } + } + counts +} + +async fn wait_for(path: &PathBuf, marker: &str) { + let started = Instant::now(); + while !read(path).contains(marker) { + assert!( + started.elapsed() < Duration::from_secs(10), + "{marker} not in {}", + path.display() + ); + tokio::time::sleep(Duration::from_millis(20)).await; + } +} + +fn assert_applied(response: &JmapResponse) { + assert_eq!( + response.pointer("/methodResponses/0/1/x:settingsReload"), + Some(&json!({"applied": true})), + "{response:?}" + ); +}