diff --git a/changelog.d/rust-tracing.feature b/changelog.d/rust-tracing.feature new file mode 100644 index 0000000000..c09fb045be --- /dev/null +++ b/changelog.d/rust-tracing.feature @@ -0,0 +1 @@ +Report opentracing spans from Synapse's Rust code to the same Jaeger agent as the Python spans. diff --git a/docs/opentracing.md b/docs/opentracing.md index bf48874160..c82c6affba 100644 --- a/docs/opentracing.md +++ b/docs/opentracing.md @@ -97,3 +97,17 @@ to two problems, namely: Sampling strategies can be set as in this document: . + +## Spans from Rust code + +Synapse reports spans for work done in its Rust extension through the same +Jaeger agent, under the same service name, as the Python spans. It uses the +agent address from `jaeger_config.local_agent` (or the `JAEGER_AGENT_HOST` and +`JAEGER_AGENT_PORT` environment variables), so there is nothing extra to +configure. A Rust span with a Python parent follows the parent's sampling +decision. A trace that starts in Rust is sampled according to the `const` or +`probabilistic` sampler. Under any other sampler, such traces are not sampled. + +The Rust reporter's counters are exported as Prometheus metrics named +`synapse_rust_tracing_*`. A growing `synapse_rust_tracing_spans_dropped_total` +means spans are finishing faster than they can be sent. diff --git a/rust/src/logging/opentracing/mod.rs b/rust/src/logging/opentracing/mod.rs index 4fdc77eb9c..bb3e7a0f4c 100644 --- a/rust/src/logging/opentracing/mod.rs +++ b/rust/src/logging/opentracing/mod.rs @@ -23,14 +23,18 @@ //! Tracing is off unless a homeserver calls `RustRuntime.enable_tracing`. //! While it is off, a capture costs one relaxed atomic load. +pub mod reporter; pub mod thrift; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; +use std::time::Duration; use pyo3::prelude::*; use pyo3::types::{PyDict, PyTuple}; use pyo3::{intern, IntoPyObjectExt}; +use reporter::Reporter; +use thrift::{FinishedSpan, Process, Tag}; use crate::logging::context::{LogContextHandle, LoggingContext}; use crate::runtime::RustRuntime; @@ -142,6 +146,46 @@ pub struct TracerConfig { pub sampler: SamplerConfig, } +/// A homeserver's tracing: its settings and the reporter sending its spans. +pub struct Tracer { + config: TracerConfig, + reporter: Reporter, +} + +impl Tracer { + /// Start reporting to the agent `config` names. + pub fn start(config: TracerConfig) -> std::io::Result { + let process = Process { + service_name: config.service_name.clone(), + tags: config + .process_tags + .iter() + .map(|(key, value)| Tag::string(key.clone(), value.clone())) + .collect(), + }; + let reporter = Reporter::start(process, &config.reporting_host, config.reporting_port)?; + Ok(Tracer { config, reporter }) + } + + pub fn config(&self) -> &TracerConfig { + &self.config + } + + /// Queue a finished span for the agent. Never blocks. + pub fn report(&self, span: FinishedSpan) { + self.reporter.report(span); + } + + pub fn reporter(&self) -> &Reporter { + &self.reporter + } + + /// Send the queued spans and stop the reporter, waiting up to `timeout`. + pub fn shutdown(&self, timeout: Duration) { + self.reporter.shutdown(timeout); + } +} + /// How to decide whether to sample a trace that starts in Rust. Spans with a /// parent never consult this: they copy the parent's flags. #[derive(Clone, Debug, PartialEq)] diff --git a/rust/src/logging/opentracing/reporter.rs b/rust/src/logging/opentracing/reporter.rs new file mode 100644 index 0000000000..3a69953662 --- /dev/null +++ b/rust/src/logging/opentracing/reporter.rs @@ -0,0 +1,655 @@ +/* + * This file is licensed under the Affero General Public License (AGPL) version 3. + * + * Copyright (C) 2026 Element Creations Ltd + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * See the GNU Affero General Public License for more details: + * . + * + */ + +//! Send finished spans to the Jaeger agent over UDP. +//! +//! [`Reporter::report`] never blocks: it queues the span on a bounded channel +//! and drops it if the queue is full. A dedicated `jaeger-reporter` thread +//! encodes the spans and sends them in `emitBatch` datagrams. A batch goes out +//! when the next span would not fit in one datagram, when it holds +//! [`MAX_BATCH_SPANS`] spans, or [`FLUSH_INTERVAL`] after its first span was +//! queued. +//! +//! The thread is a plain std thread rather than a tokio task, so reporting +//! works before the runtime starts and does not compete with request work. +//! +//! Like jaeger-client, the reporter sends from an unconnected socket and +//! tolerates an agent host name that does not resolve yet. It resolves the +//! name in the reporter thread, again every [`RESOLVE_INTERVAL`], and again +//! after a failed send, so spans follow an agent that moves to a new address. + +use std::io; +use std::net::{SocketAddr, ToSocketAddrs, UdpSocket}; +use std::panic::{catch_unwind, AssertUnwindSafe}; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::mpsc::{self, Receiver, RecvTimeoutError, SyncSender, TrySendError}; +use std::sync::{Arc, Mutex}; +use std::thread; +use std::time::{Duration, Instant}; + +use log::{error, warn}; + +use super::thrift::{ + emit_batch_len, encode_emit_batch, encode_span, encoded_process_len, FinishedSpan, Process, +}; + +/// How many spans may wait for the reporter thread before new ones are +/// dropped. +const QUEUE_CAPACITY: usize = 1000; + +/// The most spans sent in one batch. +pub const MAX_BATCH_SPANS: usize = 100; + +/// The largest datagram sent. The agent reads UDP packets of up to 65000 +/// bytes. +pub const MAX_PACKET_SIZE: usize = 65_000; + +/// How long a span may wait in a part-filled batch. +pub const FLUSH_INTERVAL: Duration = Duration::from_secs(1); + +/// How long a resolved agent address is used before resolving the host name +/// again. +const RESOLVE_INTERVAL: Duration = Duration::from_secs(30); + +/// Counters for the reporter, read by Python's metrics collector. +#[derive(Debug, Default)] +pub struct ReporterStats { + /// Spans queued but not yet taken by the reporter thread. + pub queue_depth: AtomicU64, + pub spans_sent: AtomicU64, + pub batches_sent: AtomicU64, + /// Spans dropped because the queue was full. + pub dropped_queue_full: AtomicU64, + /// Spans dropped because a batch of one would not fit in a datagram. + pub dropped_oversize: AtomicU64, + /// Spans dropped because the reporter has shut down. + pub dropped_closed: AtomicU64, + /// Batches whose send failed, including when the agent's address did + /// not resolve. Their spans are lost. + pub send_errors: AtomicU64, + /// The last send error, logged when it differs from the one before. + last_error: Mutex>, +} + +impl ReporterStats { + fn record_error(&self, err: &io::Error) { + self.send_errors.fetch_add(1, Ordering::Relaxed); + + let message = err.to_string(); + let Ok(mut last) = self.last_error.lock() else { + return; + }; + if last.as_deref() != Some(message.as_str()) { + warn!("Failed to send spans to the Jaeger agent: {message}"); + *last = Some(message); + } + } + + /// Note a successful send, so that the next error is logged again. + fn record_success(&self) { + if let Ok(mut last) = self.last_error.lock() { + *last = None; + } + } +} + +enum Message { + Span(Box), + /// Send whatever is queued now, then signal. + Flush(SyncSender<()>), +} + +/// Reports spans to one Jaeger agent. See the module docs. +pub struct Reporter { + /// `None` once shut down. + sender: Mutex>>, + /// Signalled by the thread when it exits. + finished: Mutex>>, + stats: Arc, +} + +impl Reporter { + /// Start the reporter thread, sending to the agent at `host:port`. The + /// host name is resolved by the thread, so a name that does not resolve + /// yet is not an error here. + pub fn start(process: Process, host: &str, port: u16) -> io::Result { + let agent = AgentSocket::new(host.to_owned(), port); + let (sender, receiver) = mpsc::sync_channel(QUEUE_CAPACITY); + let (finished_tx, finished_rx) = mpsc::sync_channel(1); + let stats = Arc::new(ReporterStats::default()); + + let thread_stats = Arc::clone(&stats); + thread::Builder::new() + .name("jaeger-reporter".to_owned()) + .spawn(move || { + let batcher = Batcher::new(process, agent, thread_stats); + // A panic would otherwise end reporting silently: spans would + // count as dropped because the reporter had shut down. + if catch_unwind(AssertUnwindSafe(|| batcher.run(receiver))).is_err() { + error!( + "The Jaeger reporter thread panicked; spans from Rust are no longer sent" + ); + } + let _ = finished_tx.send(()); + })?; + + Ok(Reporter { + sender: Mutex::new(Some(sender)), + finished: Mutex::new(Some(finished_rx)), + stats, + }) + } + + /// Queue a span for sending. Never blocks: drops the span if the queue is + /// full or the reporter has shut down. + pub fn report(&self, span: FinishedSpan) { + let sender = match self.sender.lock() { + Ok(sender) => sender.clone(), + Err(_) => None, + }; + let Some(sender) = sender else { + self.stats.dropped_closed.fetch_add(1, Ordering::Relaxed); + return; + }; + + // Count before sending, so that the thread's decrement never runs + // first and wraps the counter. + self.stats.queue_depth.fetch_add(1, Ordering::Relaxed); + match sender.try_send(Message::Span(Box::new(span))) { + Ok(()) => {} + Err(TrySendError::Full(_)) => { + self.stats.queue_depth.fetch_sub(1, Ordering::Relaxed); + self.stats + .dropped_queue_full + .fetch_add(1, Ordering::Relaxed); + } + Err(TrySendError::Disconnected(_)) => { + self.stats.queue_depth.fetch_sub(1, Ordering::Relaxed); + self.stats.dropped_closed.fetch_add(1, Ordering::Relaxed); + } + } + } + + /// Send everything queued so far, waiting up to `timeout`. Returns + /// whether the reporter thread handled it in time, which it does even if + /// the send fails (see [`ReporterStats::send_errors`]). + pub fn flush(&self, timeout: Duration) -> bool { + let sender = match self.sender.lock() { + Ok(sender) => sender.clone(), + Err(_) => None, + }; + let Some(sender) = sender else { + return false; + }; + + let (done_tx, done_rx) = mpsc::sync_channel(1); + let deadline = Instant::now() + timeout; + // Unlike spans, a flush waits for room in the queue. + let mut message = Message::Flush(done_tx); + loop { + match sender.try_send(message) { + Ok(()) => break, + Err(TrySendError::Full(returned)) if Instant::now() < deadline => { + message = returned; + thread::sleep(Duration::from_millis(1)); + } + Err(_) => return false, + } + } + + done_rx + .recv_timeout(deadline.saturating_duration_since(Instant::now())) + .is_ok() + } + + /// Stop accepting spans, send the ones queued, and wait up to `timeout` + /// for the reporter thread to exit. A thread that does not exit in time + /// is left to finish on its own. + pub fn shutdown(&self, timeout: Duration) { + // Dropping the last sender ends the thread's loop once it has drained + // the queue. + if let Ok(mut sender) = self.sender.lock() { + sender.take(); + } + + let finished = match self.finished.lock() { + Ok(mut finished) => finished.take(), + Err(_) => None, + }; + if let Some(finished) = finished { + if finished.recv_timeout(timeout).is_err() { + warn!("The Jaeger reporter thread did not finish within {timeout:?}"); + } + } + } + + pub fn stats(&self) -> &ReporterStats { + &self.stats + } +} + +impl Drop for Reporter { + fn drop(&mut self) { + // Let the thread drain and exit without waiting for it. + if let Ok(mut sender) = self.sender.lock() { + sender.take(); + } + } +} + +/// The reporter thread's state: the batch being filled. +struct Batcher { + process: Process, + process_len: usize, + agent: AgentSocket, + stats: Arc, + spans: Vec>, + spans_len: usize, + /// When the first span of the current batch arrived. + batch_started: Option, + seq_id: u32, +} + +impl Batcher { + fn new(process: Process, agent: AgentSocket, stats: Arc) -> Self { + let process_len = encoded_process_len(&process); + if emit_batch_len(process_len, 0, 0) > MAX_PACKET_SIZE { + warn!( + "The Jaeger process tags take {process_len} bytes, too many for a datagram; \ + no spans from Rust will be sent" + ); + } + + Batcher { + process_len, + process, + agent, + stats, + spans: Vec::new(), + spans_len: 0, + batch_started: None, + seq_id: 0, + } + } + + fn run(mut self, receiver: Receiver) { + loop { + let message = match self.batch_started { + Some(started) => { + let wait = (started + FLUSH_INTERVAL).saturating_duration_since(Instant::now()); + receiver.recv_timeout(wait) + } + None => receiver.recv().map_err(|_| RecvTimeoutError::Disconnected), + }; + + match message { + Ok(Message::Span(span)) => { + self.stats.queue_depth.fetch_sub(1, Ordering::Relaxed); + self.add(&span); + } + Ok(Message::Flush(done)) => { + self.send(); + let _ = done.send(()); + } + Err(RecvTimeoutError::Timeout) => self.send(), + Err(RecvTimeoutError::Disconnected) => { + self.send(); + return; + } + } + } + } + + fn add(&mut self, span: &FinishedSpan) { + let encoded = encode_span(span); + if emit_batch_len(self.process_len, 1, encoded.len()) > MAX_PACKET_SIZE { + self.stats.dropped_oversize.fetch_add(1, Ordering::Relaxed); + return; + } + + let len_with_span = emit_batch_len( + self.process_len, + self.spans.len() + 1, + self.spans_len + encoded.len(), + ); + if len_with_span > MAX_PACKET_SIZE { + self.send(); + } + + self.spans_len += encoded.len(); + self.spans.push(encoded); + self.batch_started.get_or_insert_with(Instant::now); + + if self.spans.len() >= MAX_BATCH_SPANS { + self.send(); + } + } + + fn send(&mut self) { + self.batch_started = None; + if self.spans.is_empty() { + return; + } + + self.seq_id = self.seq_id.wrapping_add(1); + let message = encode_emit_batch(&self.process, &self.spans, self.seq_id); + let count = self.spans.len() as u64; + self.spans.clear(); + self.spans_len = 0; + + match self.agent.send(&message) { + Ok(()) => { + self.stats.batches_sent.fetch_add(1, Ordering::Relaxed); + self.stats.spans_sent.fetch_add(count, Ordering::Relaxed); + self.stats.record_success(); + } + Err(err) => self.stats.record_error(&err), + } + } +} + +/// Sends datagrams to the agent's host name, resolving it as needed. +struct AgentSocket { + host: String, + port: u16, + /// The resolved address and when it was resolved. + address: Option<(SocketAddr, Instant)>, + /// Unconnected sockets, one per address family, bound on first use. + v4: Option, + v6: Option, +} + +impl AgentSocket { + fn new(host: String, port: u16) -> Self { + AgentSocket { + host, + port, + address: None, + v4: None, + v6: None, + } + } + + fn send(&mut self, datagram: &[u8]) -> io::Result<()> { + let address = self.address()?; + let slot = if address.is_ipv4() { + &mut self.v4 + } else { + &mut self.v6 + }; + let socket = match slot { + Some(socket) => socket, + None => { + let local: SocketAddr = if address.is_ipv4() { + ([0, 0, 0, 0], 0).into() + } else { + ([0u16; 8], 0).into() + }; + slot.insert(UdpSocket::bind(local)?) + } + }; + + let result = socket.send_to(datagram, address).map(|_| ()); + if result.is_err() { + // Resolve again next time, in case the agent has moved. + self.address = None; + } + result + } + + /// The agent's address, resolving the host name if it has not been + /// resolved for [`RESOLVE_INTERVAL`]. A failed re-resolution keeps the + /// old address. + fn address(&mut self) -> io::Result { + if let Some((address, resolved)) = self.address { + if resolved.elapsed() < RESOLVE_INTERVAL { + return Ok(address); + } + } + + match (self.host.as_str(), self.port).to_socket_addrs() { + Ok(mut addresses) => { + if let Some(address) = addresses.next() { + self.address = Some((address, Instant::now())); + return Ok(address); + } + } + Err(err) => { + if self.address.is_none() { + return Err(err); + } + } + } + + match &mut self.address { + Some((address, resolved)) => { + // Keep using the old address, and try again later. + *resolved = Instant::now(); + Ok(*address) + } + None => Err(io::Error::new( + io::ErrorKind::NotFound, + format!("no address for {}:{}", self.host, self.port), + )), + } + } +} + +#[cfg(test)] +mod tests { + use std::borrow::Cow; + use std::time::UNIX_EPOCH; + + use super::super::thrift::Tag; + use super::*; + + fn process() -> Process { + Process { + service_name: "test".to_owned(), + tags: vec![Tag::string("hostname", "host")], + } + } + + fn span(span_id: u64, tag_len: usize) -> FinishedSpan { + FinishedSpan { + trace_id: 1, + span_id, + parent_id: 0, + flags: 1, + operation: Cow::Borrowed("op"), + start: UNIX_EPOCH + Duration::from_secs(1_760_000_000), + duration: Duration::from_micros(5), + tags: if tag_len > 0 { + vec![Tag::string("big", "x".repeat(tag_len))] + } else { + vec![] + }, + logs: vec![], + references: vec![], + } + } + + /// A reporter sending to a socket the test reads. + fn reporter() -> (Reporter, UdpSocket) { + let agent = UdpSocket::bind("127.0.0.1:0").unwrap(); + agent + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + let port = agent.local_addr().unwrap().port(); + ( + Reporter::start(process(), "127.0.0.1", port).unwrap(), + agent, + ) + } + + fn receive(agent: &UdpSocket) -> Vec { + let mut buf = vec![0; 70_000]; + let len = agent.recv(&mut buf).expect("expected a datagram"); + buf.truncate(len); + buf + } + + fn assert_nothing_more(agent: &UdpSocket) { + agent + .set_read_timeout(Some(Duration::from_millis(50))) + .unwrap(); + let mut buf = vec![0; 70_000]; + assert!(agent.recv(&mut buf).is_err(), "unexpected datagram"); + } + + #[test] + fn flush_sends_one_batch() { + let (reporter, agent) = reporter(); + let spans = [span(1, 0), span(2, 10)]; + for span in &spans { + reporter.report(span.clone()); + } + assert!(reporter.flush(Duration::from_secs(5))); + + let encoded: Vec<_> = spans.iter().map(encode_span).collect(); + assert_eq!(receive(&agent), encode_emit_batch(&process(), &encoded, 1)); + assert_nothing_more(&agent); + + let stats = reporter.stats(); + assert_eq!(stats.spans_sent.load(Ordering::Relaxed), 2); + assert_eq!(stats.batches_sent.load(Ordering::Relaxed), 1); + assert_eq!(stats.queue_depth.load(Ordering::Relaxed), 0); + } + + #[test] + fn batches_are_capped_by_span_count() { + let (reporter, agent) = reporter(); + for id in 0..(MAX_BATCH_SPANS as u64 + 10) { + reporter.report(span(id, 0)); + } + assert!(reporter.flush(Duration::from_secs(5))); + + let first: Vec<_> = (0..MAX_BATCH_SPANS as u64) + .map(|id| encode_span(&span(id, 0))) + .collect(); + assert_eq!(receive(&agent), encode_emit_batch(&process(), &first, 1)); + let rest: Vec<_> = (MAX_BATCH_SPANS as u64..MAX_BATCH_SPANS as u64 + 10) + .map(|id| encode_span(&span(id, 0))) + .collect(); + assert_eq!(receive(&agent), encode_emit_batch(&process(), &rest, 2)); + assert_nothing_more(&agent); + } + + #[test] + fn batches_fit_in_a_datagram() { + let (reporter, agent) = reporter(); + // Three spans of ~30 KB: two fit in one datagram, the third does not. + for id in 0..3 { + reporter.report(span(id, 30_000)); + } + assert!(reporter.flush(Duration::from_secs(5))); + + let first = receive(&agent); + let second = receive(&agent); + assert!(first.len() <= MAX_PACKET_SIZE); + let encoded: Vec<_> = (0..2).map(|id| encode_span(&span(id, 30_000))).collect(); + assert_eq!(first, encode_emit_batch(&process(), &encoded, 1)); + assert_eq!( + second, + encode_emit_batch(&process(), &[encode_span(&span(2, 30_000))], 2) + ); + } + + #[test] + fn oversize_span_is_dropped() { + let (reporter, agent) = reporter(); + reporter.report(span(1, MAX_PACKET_SIZE)); + reporter.report(span(2, 0)); + assert!(reporter.flush(Duration::from_secs(5))); + + assert_eq!( + receive(&agent), + encode_emit_batch(&process(), &[encode_span(&span(2, 0))], 1) + ); + assert_eq!(reporter.stats().dropped_oversize.load(Ordering::Relaxed), 1); + } + + #[test] + fn partial_batch_goes_after_the_flush_interval() { + let (reporter, agent) = reporter(); + let started = Instant::now(); + reporter.report(span(1, 0)); + + receive(&agent); + let waited = started.elapsed(); + assert!(waited >= FLUSH_INTERVAL, "sent after {waited:?}"); + } + + #[test] + fn shutdown_sends_queued_spans_then_drops_new_ones() { + let (reporter, agent) = reporter(); + reporter.report(span(1, 0)); + reporter.shutdown(Duration::from_secs(5)); + + assert_eq!( + receive(&agent), + encode_emit_batch(&process(), &[encode_span(&span(1, 0))], 1) + ); + + reporter.report(span(2, 0)); + assert_eq!(reporter.stats().dropped_closed.load(Ordering::Relaxed), 1); + assert!(!reporter.flush(Duration::from_millis(10))); + } + + #[test] + fn unresolvable_agent_counts_send_errors() { + // `.invalid` never resolves (RFC 6761). Starting still succeeds: the + // name is resolved when a batch is sent. + let reporter = Reporter::start(process(), "agent.invalid", 6831).unwrap(); + for id in 0..2 { + reporter.report(span(id, 0)); + assert!(reporter.flush(Duration::from_secs(30))); + } + + let stats = reporter.stats(); + assert_eq!(stats.send_errors.load(Ordering::Relaxed), 2); + assert_eq!(stats.spans_sent.load(Ordering::Relaxed), 0); + assert!(stats.last_error.lock().unwrap().is_some()); + } + + #[test] + fn agent_restart_loses_no_batch() { + // On a connected socket, the ICMP error from a send to a closed port + // fails the next send. The reporter's socket is unconnected, so once + // an agent listens again its first batch arrives. + let closed = UdpSocket::bind("127.0.0.1:0").unwrap(); + let port = closed.local_addr().unwrap().port(); + drop(closed); + + let reporter = Reporter::start(process(), "127.0.0.1", port).unwrap(); + reporter.report(span(1, 0)); + assert!(reporter.flush(Duration::from_secs(5))); + + let Ok(agent) = UdpSocket::bind(("127.0.0.1", port)) else { + // Another test took the port in between: nothing to check. + return; + }; + agent + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + reporter.report(span(2, 0)); + assert!(reporter.flush(Duration::from_secs(5))); + + assert_eq!( + receive(&agent), + encode_emit_batch(&process(), &[encode_span(&span(2, 0))], 2) + ); + } +} diff --git a/rust/src/runtime.rs b/rust/src/runtime.rs index 0f891536c5..ae5ecca5a2 100644 --- a/rust/src/runtime.rs +++ b/rust/src/runtime.rs @@ -40,7 +40,8 @@ use crate::caches::registry::CacheRegistry; use crate::clock::{set_virtual_time_msec, Clock}; use crate::homeserver::HomeServer; use crate::logging::context::LogContextHandle; -use crate::logging::opentracing::{set_tracing_enabled, SamplerConfig, TracerConfig}; +use crate::logging::opentracing::thrift::{FinishedSpan, Tag}; +use crate::logging::opentracing::{set_tracing_enabled, SamplerConfig, Tracer, TracerConfig}; use crate::reactor::Reactor; use crate::twisted_dispatch::{self, TwistedDispatchReader, TwistedDispatcher}; @@ -53,6 +54,9 @@ use crate::twisted_dispatch::{self, TwistedDispatchReader, TwistedDispatcher}; /// See [`tokio::runtime`] for details. const SHUTDOWN_TIMEOUT: Duration = Duration::from_millis(100); +/// How long shutdown waits for the Jaeger reporter to send its last spans. +const REPORTER_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(1); + /// State of the lazily-started tokio runtime. enum TokioState { /// Not started yet; the runtime is built on first use. @@ -81,7 +85,7 @@ pub struct RustRuntimeInner { /// Set by `enable_tracing` when opentracing is enabled for this /// homeserver. See [`crate::logging::opentracing`]. - tracing: OnceLock>, + tracing: OnceLock>, } impl RustRuntimeInner { @@ -165,8 +169,8 @@ impl RustRuntimeInner { } } - /// The tracing settings, if opentracing is enabled for this homeserver. - pub fn tracer_config(&self) -> Option<&Arc> { + /// The tracer, if opentracing is enabled for this homeserver. + pub fn tracer(&self) -> Option<&Arc> { self.tracing.get() } @@ -200,11 +204,20 @@ impl RustRuntimeInner { // All tokio tasks should be stopped by now, so only long running // blocking threads may still be active. If they try to dispatch new // work they get an error and should stop. - self.reactor - .remove_reader(py, self.dispatch_reader.bind(py).as_any())?; - self.dispatch_reader.get().close_and_drain(py); + let removed = self + .reactor + .remove_reader(py, self.dispatch_reader.bind(py).as_any()); + if removed.is_ok() { + self.dispatch_reader.get().close_and_drain(py); + } - Ok(()) + // Last, so that spans finished by the work above still go out, and + // even if removing the reader failed. + if let Some(tracer) = self.tracing.get() { + py.detach(|| tracer.shutdown(REPORTER_SHUTDOWN_TIMEOUT)); + } + + removed } } @@ -298,7 +311,9 @@ impl RustRuntime { /// other type never samples such a trace. Rust spans with a Python parent /// follow the parent's decision. /// - /// A second call is ignored. + /// The agent's host name is resolved when spans are sent, so one that + /// does not resolve yet is not an error. Raises `OSError` only if the + /// reporter thread cannot start. A second call is ignored. #[pyo3(signature = (service_name, process_tags, reporting_host, reporting_port, sampler_type, sampler_param))] fn enable_tracing( &self, @@ -308,7 +323,12 @@ impl RustRuntime { reporting_port: u16, sampler_type: &str, sampler_param: f64, - ) { + ) -> PyResult<()> { + if self.inner.tracing.get().is_some() { + log::warn!("Rust tracing was already enabled for this homeserver"); + return Ok(()); + } + let config = TracerConfig { service_name, process_tags, @@ -317,12 +337,15 @@ impl RustRuntime { sampler: SamplerConfig::from_type_and_param(sampler_type, sampler_param), }; - if self.inner.tracing.set(Arc::new(config)).is_err() { + let tracer = Tracer::start(config)?; + + if self.inner.tracing.set(Arc::new(tracer)).is_err() { log::warn!("Rust tracing was already enabled for this homeserver"); - return; + return Ok(()); } set_tracing_enabled(true); + Ok(()) } /// The settings `enable_tracing` stored, as `(service_name, @@ -332,7 +355,8 @@ impl RustRuntime { fn tracing_config_for_test( &self, ) -> Option<(String, Vec<(String, String)>, String, u16, String)> { - self.inner.tracing.get().map(|config| { + self.inner.tracing.get().map(|tracer| { + let config = tracer.config(); ( config.service_name.clone(), config.process_tags.clone(), @@ -343,6 +367,79 @@ impl RustRuntime { }) } + /// The Jaeger reporter's counters, as `(queue_depth, spans_sent, + /// batches_sent, dropped_queue_full, dropped_oversize, dropped_closed, + /// send_errors)`, or `None` if tracing is not enabled. + #[allow(clippy::type_complexity)] + fn tracing_stats(&self) -> Option<(u64, u64, u64, u64, u64, u64, u64)> { + use std::sync::atomic::Ordering::Relaxed; + + self.inner.tracing.get().map(|tracer| { + let stats = tracer.reporter().stats(); + ( + stats.queue_depth.load(Relaxed), + stats.spans_sent.load(Relaxed), + stats.batches_sent.load(Relaxed), + stats.dropped_queue_full.load(Relaxed), + stats.dropped_oversize.load(Relaxed), + stats.dropped_closed.load(Relaxed), + stats.send_errors.load(Relaxed), + ) + }) + } + + /// Report a span with the given IDs and string tags, started now and one + /// millisecond long. For tests. + #[pyo3(signature = (operation, trace_id, span_id, parent_id, flags, tags))] + fn report_span_for_test( + &self, + operation: String, + trace_id: u128, + span_id: u64, + parent_id: u64, + flags: u8, + tags: Vec<(String, String)>, + ) -> PyResult<()> { + let tracer = self + .inner + .tracing + .get() + .ok_or_else(|| PyRuntimeError::new_err("tracing is not enabled"))?; + tracer.report(FinishedSpan { + trace_id, + span_id, + parent_id, + flags, + operation: operation.into(), + start: std::time::SystemTime::now(), + duration: Duration::from_millis(1), + tags: tags + .into_iter() + .map(|(key, value)| Tag::string(key, value)) + .collect(), + logs: vec![], + references: vec![], + }); + Ok(()) + } + + /// Send the queued spans now, waiting up to `timeout_secs`. Returns + /// whether they went in time. For tests. + fn flush_tracing_for_test(&self, py: Python<'_>, timeout_secs: f64) -> bool { + let Some(tracer) = self.inner.tracing.get() else { + return false; + }; + let timeout = Duration::from_secs_f64(timeout_secs.max(0.0)); + py.detach(|| tracer.reporter().flush(timeout)) + } + + /// Shut the Jaeger reporter down as homeserver shutdown does. For tests. + fn shutdown_tracing_for_test(&self, py: Python<'_>) { + if let Some(tracer) = self.inner.tracing.get() { + py.detach(|| tracer.shutdown(REPORTER_SHUTDOWN_TIMEOUT)); + } + } + /// Evict entries not read for `expiry_secs` from every Rust cache of this /// homeserver. Returns how many went. See [`crate::caches::registry`]. fn expire_caches(&self, expiry_secs: f64) -> PyResult { diff --git a/scripts-dev/mypy_synapse_plugin.py b/scripts-dev/mypy_synapse_plugin.py index 2c52bcb2b3..0da97cf442 100644 --- a/scripts-dev/mypy_synapse_plugin.py +++ b/scripts-dev/mypy_synapse_plugin.py @@ -152,6 +152,7 @@ prometheus_metric_fullname_to_label_arg_map: Mapping[str, ArgLocation | None] = "synapse.util.metrics.DynamicCollectorRegistry": None, "synapse.metrics.background_process_metrics._Collector": None, "synapse.storage.rust_pool._RustPoolStatsCollector": None, + "synapse.logging.rust_tracing_metrics._RustTracingCollector": None, # # `Metric` subclasses: "prometheus_client.metrics_core.Metric": None, diff --git a/synapse/logging/opentracing.py b/synapse/logging/opentracing.py index 9a9d3978cd..c55690ba7d 100644 --- a/synapse/logging/opentracing.py +++ b/synapse/logging/opentracing.py @@ -525,7 +525,10 @@ def _enable_rust_tracing( # never starts a sampled trace of its own under them. sampler_type, sampler_param = type(sampler).__name__, 0.0 - hs.get_rust_runtime().enable_tracing( + from synapse.logging.rust_tracing_metrics import rust_tracing_collector + + runtime = hs.get_rust_runtime() + runtime.enable_tracing( service_name=tracer.service_name, process_tags=[(str(k), str(v)) for k, v in tracer.tags.items()], reporting_host=config.local_agent_reporting_host, @@ -533,6 +536,7 @@ def _enable_rust_tracing( sampler_type=sampler_type, sampler_param=sampler_param, ) + rust_tracing_collector.add(runtime, hs.hostname) # Whitelisting diff --git a/synapse/logging/rust_tracing_metrics.py b/synapse/logging/rust_tracing_metrics.py new file mode 100644 index 0000000000..fd7dd79eef --- /dev/null +++ b/synapse/logging/rust_tracing_metrics.py @@ -0,0 +1,111 @@ +# +# This file is licensed under the Affero General Public License (AGPL) version 3. +# +# Copyright (C) 2026 Element Creations Ltd +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as +# published by the Free Software Foundation, either version 3 of the +# License, or (at your option) any later version. +# +# See the GNU Affero General Public License for more details: +# . +# + +"""Prometheus metrics for the Rust Jaeger reporter.""" + +import threading +import weakref +from typing import Iterable + +from prometheus_client.core import ( + REGISTRY, + CounterMetricFamily, + GaugeMetricFamily, + Metric, +) +from prometheus_client.registry import Collector + +from synapse.metrics import SERVER_NAME_LABEL +from synapse.synapse_rust.runtime import RustRuntime + + +class _RustTracingCollector(Collector): + """Reports the reporter counters of every `RustRuntime` with tracing + enabled. Holds the runtimes weakly.""" + + def __init__(self) -> None: + self._runtimes: "weakref.WeakKeyDictionary[RustRuntime, str]" = ( + weakref.WeakKeyDictionary() + ) + # Runtimes are added on the reactor thread and read on the metrics + # thread. + self._lock = threading.Lock() + + def add(self, runtime: RustRuntime, server_name: str) -> None: + with self._lock: + self._runtimes[runtime] = server_name + + def collect(self) -> Iterable[Metric]: + with self._lock: + runtimes = list(self._runtimes.items()) + + queue_depth = GaugeMetricFamily( + "synapse_rust_tracing_queue_depth", + "Spans from Rust code waiting to be sent to the Jaeger agent", + labels=[SERVER_NAME_LABEL], + ) + spans_sent = CounterMetricFamily( + "synapse_rust_tracing_spans_sent", + "Spans from Rust code sent to the Jaeger agent", + labels=[SERVER_NAME_LABEL], + ) + batches_sent = CounterMetricFamily( + "synapse_rust_tracing_batches_sent", + "UDP datagrams of spans from Rust code sent to the Jaeger agent", + labels=[SERVER_NAME_LABEL], + ) + dropped = CounterMetricFamily( + "synapse_rust_tracing_spans_dropped", + "Spans from Rust code dropped without being sent, by reason: the queue " + "was full, the span alone was too big for a datagram, or the reporter " + "had shut down", + labels=["reason", SERVER_NAME_LABEL], + ) + send_errors = CounterMetricFamily( + "synapse_rust_tracing_send_errors", + "Datagrams of spans from Rust code that failed to send", + labels=[SERVER_NAME_LABEL], + ) + + for runtime, server_name in runtimes: + stats = runtime.tracing_stats() + if stats is None: + continue + ( + depth, + sent, + batches, + dropped_queue_full, + dropped_oversize, + dropped_closed, + errors, + ) = stats + queue_depth.add_metric([server_name], depth) + spans_sent.add_metric([server_name], sent) + batches_sent.add_metric([server_name], batches) + dropped.add_metric(["queue_full", server_name], dropped_queue_full) + dropped.add_metric(["oversize", server_name], dropped_oversize) + dropped.add_metric(["closed", server_name], dropped_closed) + send_errors.add_metric([server_name], errors) + + yield queue_depth + yield spans_sent + yield batches_sent + yield dropped + yield send_errors + + +# Its metrics carry `SERVER_NAME_LABEL`, which the collector adds per runtime. +rust_tracing_collector = _RustTracingCollector() # type: ignore[missing-server-name-label] +REGISTRY.register(rust_tracing_collector) diff --git a/synapse/synapse_rust/runtime.pyi b/synapse/synapse_rust/runtime.pyi index a8f34667fc..6dcf939a7d 100644 --- a/synapse/synapse_rust/runtime.pyi +++ b/synapse/synapse_rust/runtime.pyi @@ -54,7 +54,9 @@ class RustRuntime: `sampler_type` is `const` or `probabilistic`, with `sampler_param` the decision or rate. Any other type means Rust never starts a sampled - trace of its own. A second call is ignored.""" + trace of its own. The agent's host name is resolved when spans are + sent. Raises `OSError` only if the reporter thread cannot start. A + second call is ignored.""" def tracing_config_for_test( self, @@ -63,6 +65,30 @@ class RustRuntime: process_tags, reporting_host, reporting_port, sampler)`, or None. For tests.""" + def tracing_stats(self) -> tuple[int, int, int, int, int, int, int] | None: + """The Jaeger reporter's counters, as `(queue_depth, spans_sent, + batches_sent, dropped_queue_full, dropped_oversize, dropped_closed, + send_errors)`, or None if tracing is not enabled.""" + + def report_span_for_test( + self, + operation: str, + trace_id: int, + span_id: int, + parent_id: int, + flags: int, + tags: list[tuple[str, str]], + ) -> None: + """Report a span started now and one millisecond long. For tests.""" + + def flush_tracing_for_test(self, timeout_secs: float) -> bool: + """Send the queued spans now. Returns whether they went within the + timeout. For tests.""" + + def shutdown_tracing_for_test(self) -> None: + """Shut the Jaeger reporter down as homeserver shutdown does. For + tests.""" + def set_virtual_time_msec(millis: int) -> None: """Pin the Rust clock to the given time. Only for tests, which run against a virtual reactor clock.""" diff --git a/tests/logging/test_rust_tracing.py b/tests/logging/test_rust_tracing.py index abef984541..7747d053db 100644 --- a/tests/logging/test_rust_tracing.py +++ b/tests/logging/test_rust_tracing.py @@ -14,12 +14,14 @@ """Tests for reporting opentracing spans from Rust.""" +import socket from typing import cast from twisted.internet.testing import MemoryReactor from synapse.logging.context import LoggingContext from synapse.logging.opentracing import _enable_rust_tracing, start_active_span +from synapse.logging.rust_tracing_metrics import rust_tracing_collector from synapse.server import HomeServer from synapse.synapse_rust.opentracing import ( captured_span_for_test, @@ -34,6 +36,14 @@ try: import jaeger_client import opentracing from jaeger_client.sampler import ConstSampler, ProbabilisticSampler + from jaeger_client.thrift_gen.agent.Agent import emitBatch_args + from jaeger_client.thrift_gen.jaeger import ttypes + from thrift.protocol.TCompactProtocol import ( # type: ignore[import-untyped] + TCompactProtocol, + ) + from thrift.transport.TTransport import ( # type: ignore[import-untyped] + TMemoryBuffer, + ) from synapse.logging.scopecontextmanager import LogContextScopeManager except ImportError: @@ -209,3 +219,109 @@ class EnableRustTracingTestCase(HomeserverTestCase): settings = self.hs.get_rust_runtime().tracing_config_for_test() assert settings is not None self.assertEqual(settings[4], "Never") + + +def _decode_emit_batch(data: bytes) -> "ttypes.Batch": + """Decode an `emitBatch` datagram the way the Jaeger agent does.""" + protocol = TCompactProtocol(TMemoryBuffer(data)) + name, _message_type, _seq_id = protocol.readMessageBegin() + assert name == "emitBatch", name + args = emitBatch_args() + args.read(protocol) + return args.batch + + +class ReporterRoundTripTestCase(HomeserverTestCase): + """Spans reported from Rust reach the Jaeger agent's UDP port in a form + jaeger-client's own thrift code decodes.""" + + if jaeger_client is None: + skip = "Requires jaeger_client" # type: ignore[unreachable] + + def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: + self.addCleanup(set_tracing_enabled_for_test, False) + + self.agent = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + self.addCleanup(self.agent.close) + self.agent.bind(("127.0.0.1", 0)) + self.agent.settimeout(5) + port = self.agent.getsockname()[1] + + config = jaeger_client.config.Config( + config={ + "local_agent": {"reporting_host": "127.0.0.1", "reporting_port": port}, + "tags": {"instance": "master"}, + }, + service_name="test service", + ) + tracer = config.create_tracer( + sampler=ConstSampler(True), + reporter=jaeger_client.reporter.InMemoryReporter(), + ) + _enable_rust_tracing(hs, config, tracer) + self.runtime = hs.get_rust_runtime() + + def _receive(self) -> "ttypes.Batch": + data, _ = self.agent.recvfrom(70_000) + return _decode_emit_batch(data) + + def test_round_trip(self) -> None: + trace_id = (1 << 100) + 5 + span_id = (1 << 63) + 7 + self.runtime.report_span_for_test( + "db.query", trace_id, span_id, 42, 1, [("db.statement", "SELECT 1")] + ) + self.assertTrue(self.runtime.flush_tracing_for_test(5)) + + batch = self._receive() + self.assertEqual(batch.process.serviceName, "test service") + process_tags = {tag.key: tag.vStr for tag in batch.process.tags} + self.assertEqual(process_tags["instance"], "master") + + (span,) = batch.spans + self.assertEqual(span.operationName, "db.query") + # Thrift's i64s are signed, so the high span ID arrives negative, + # exactly as jaeger-client would send it. + self.assertEqual(span.traceIdLow, 5) + self.assertEqual(span.traceIdHigh, 1 << 36) + self.assertEqual(span.spanId, span_id - (1 << 64)) + self.assertEqual(span.parentSpanId, 42) + self.assertEqual(span.flags, 1) + self.assertEqual(span.duration, 1000) + self.assertGreater(span.startTime, 0) + self.assertEqual( + [(tag.key, tag.vType, tag.vStr) for tag in span.tags], + [("db.statement", ttypes.TagType.STRING, "SELECT 1")], + ) + + def test_metrics(self) -> None: + for span_id in range(1, 4): + self.runtime.report_span_for_test("op", 1, span_id, 0, 1, []) + self.assertTrue(self.runtime.flush_tracing_for_test(5)) + self.assertEqual(len(self._receive().spans), 3) + + self.assertEqual(self.runtime.tracing_stats(), (0, 3, 1, 0, 0, 0, 0)) + + metrics = { + (metric.name, sample.labels.get("reason")): sample.value + for metric in rust_tracing_collector.collect() + for sample in metric.samples + if sample.labels.get("server_name") == self.hs.hostname + } + self.assertEqual(metrics[("synapse_rust_tracing_spans_sent", None)], 3) + self.assertEqual(metrics[("synapse_rust_tracing_batches_sent", None)], 1) + self.assertEqual( + metrics[("synapse_rust_tracing_spans_dropped", "queue_full")], 0 + ) + + def test_shutdown_sends_queued_spans(self) -> None: + self.runtime.report_span_for_test("op", 1, 2, 0, 1, []) + self.runtime.shutdown_tracing_for_test() + + self.assertEqual(len(self._receive().spans), 1) + + # Spans reported after shutdown are dropped and counted. + self.runtime.report_span_for_test("op", 1, 3, 0, 1, []) + stats = self.runtime.tracing_stats() + assert stats is not None + self.assertEqual(stats[5], 1)