From a89166714908a8ba91828dfe7243268fef933d28 Mon Sep 17 00:00:00 2001 From: John Coffey Date: Thu, 24 Sep 2026 20:46:09 -0700 Subject: [PATCH] Tracers whose settings change start over on reload A cluster rehearsal moved a Log tracer to another directory: the write was reported x:settingsReload applied:true, but the tracer kept writing to the old file until a restart. Telemetry::update only refreshed each running tracer's events, level and lossiness; a tracer's own settings (path, prefix, rotation, format, endpoint, headers, ...) stayed as built. Each tracer now carries a hash of the registry object it was built from, less the fields that change in place. The reload compares it with the running tracer's: unchanged ones are updated in place as before, changed ones are started over, new ones started and removed ones stopped. Only tracers this server started are removed; upstream removed every subscriber not in the settings, which also cut off live-tracing streams on each reload. Starting over is a swap in the collector, so no event is lost or written twice: a subscriber registered under a running one's id replaces it between two collection passes. The old one's batch is sent first (what its full channel can't take moves to the new one), and dropping it closes its channel, so its task writes what is queued and ends. Per tracer kind: - Log: a tracer started over on the same files (rotation or format changed) waits for the old one to finish, so lines don't interleave. - Webhook: the task held a sender of its own channel for retries, so it never ended; retries now use a weak sender, and pending events are posted when the channel closes. - OpenTelemetry: pending logs and spans are exported when the channel closes instead of dropped, and a span that was open across the swap is exported by the new tracer with the events it saw. - Console and journal: nothing kept between batches. - Trace history: built from the tracing store, which takes a restart, so it is never started over. No kind needs a restart, so x:settingsReload doesn't gain one. system::tracer_reload::tracer_reload_tests (new): a Log tracer created over JMAP writes to its directory; its path is changed over JMAP while 2000 numbered events are emitted; after the reload, events land in the new file and not the old one, each numbered event is in exactly one of the two files, and a destroyed tracer writes nothing. On main the new file never appears. --- crates/common/src/config/telemetry.rs | 48 +++++ crates/common/src/telemetry/mod.rs | 43 ++++- crates/common/src/telemetry/tracers/log.rs | 20 ++ crates/common/src/telemetry/tracers/otel.rs | 23 ++- crates/common/src/telemetry/webhooks/mod.rs | 24 ++- crates/trc/src/ipc/collector.rs | 24 ++- crates/trc/src/ipc/subscriber.rs | 5 + tests/src/system/mod.rs | 1 + tests/src/system/tracer_reload.rs | 204 ++++++++++++++++++++ 9 files changed, 377 insertions(+), 15 deletions(-) create mode 100644 tests/src/system/tracer_reload.rs 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:?}" + ); +}