diff --git a/Cargo.lock b/Cargo.lock index 4d6927ecb..53bee022d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1396,6 +1396,15 @@ version = "2.7.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f52b00d39961fc5b2736ea853c9cc86238e165017a493d1d5c8eac6bdc4cc273" +[[package]] +name = "memmap2" +version = "0.9.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "714098028fe011992e1c3962653c96b2d578c4b4bce9036e15ff220319b1e0e3" +dependencies = [ + "libc", +] + [[package]] name = "memoffset" version = "0.9.1" @@ -1789,6 +1798,17 @@ dependencies = [ "zerocopy", ] +[[package]] +name = "probing-memtable" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d6b153e11fbce92a05d067c191f88b14d3162c15d0f90679619b38435465b548" +dependencies = [ + "libc", + "memmap2", + "xxhash-rust", +] + [[package]] name = "proc-macro2" version = "1.0.103" @@ -1839,6 +1859,7 @@ dependencies = [ "opentelemetry", "opentelemetry-otlp", "opentelemetry_sdk", + "probing-memtable", "rand 0.9.2", "rcgen", "rustls", @@ -1918,7 +1939,6 @@ dependencies = [ "thiserror 2.0.17", "tokio", "tracing", - "tracing-subscriber", "uuid", ] @@ -3691,6 +3711,12 @@ version = "0.6.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9edde0db4769d2dc68579893f2306b26c6ecfbe0ef499b013d731b7b9247e0b9" +[[package]] +name = "xxhash-rust" +version = "0.8.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fdd20c5420375476fbd4394763288da7eb0cc0b8c11deed431a91562af7335d3" + [[package]] name = "yasna" version = "0.5.2" diff --git a/Cargo.toml b/Cargo.toml index 72eb95ffd..72ca43ede 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -96,5 +96,6 @@ hf-hub = { version = "0.4", features = ["tokio"] } vergen-gitcl = "1.0" pkcs8 = { version = "0.10", features = ["alloc"] } sha2 = "0.10" +probing-memtable = "0.2.4" ed25519-dalek = { version = "2.0", features = ["pkcs8"] } der = "0.7" diff --git a/crates/pulsing-actor/Cargo.toml b/crates/pulsing-actor/Cargo.toml index 5b67ed392..a6f866108 100644 --- a/crates/pulsing-actor/Cargo.toml +++ b/crates/pulsing-actor/Cargo.toml @@ -69,6 +69,9 @@ rustls-pemfile = { workspace = true, optional = true } time = { workspace = true, optional = true } sha2 = { version = "0.10", optional = true } +# Span storage (mmap'd ring buffer, also discoverable by probing) +probing-memtable = { workspace = true } + [dev-dependencies] tokio-test = { workspace = true } tracing-subscriber = { workspace = true } @@ -98,5 +101,9 @@ name = "behavior_fsm" path = "../../examples/rust/behavior_fsm.rs" +[[bench]] +name = "tracing_overhead" +harness = false + [lints] workspace = true diff --git a/crates/pulsing-actor/benches/tracing_overhead.rs b/crates/pulsing-actor/benches/tracing_overhead.rs new file mode 100644 index 000000000..26f0cb460 --- /dev/null +++ b/crates/pulsing-actor/benches/tracing_overhead.rs @@ -0,0 +1,333 @@ +//! Benchmark: memtable write overhead for each tracing store. +//! +//! Measures the **per-operation cost** of writing to the four memtable-backed +//! stores, so we know how much overhead tracing adds to actor messaging. +//! +//! Run: `cargo bench -p pulsing-actor --bench tracing_overhead` + +use std::hint::black_box; +use std::time::Instant; + +use probing_memtable::discover::ExposedHashTable; +use probing_memtable::discover::ExposedTable; +use probing_memtable::{DType, Schema, Value}; + +// ── Helpers ─────────────────────────────────────────────────────────── + +fn bench_loop(name: &str, iterations: u64, mut f: impl FnMut(u64)) { + // Warmup + for i in 0..iterations.min(1000) { + f(i); + } + + let start = Instant::now(); + for i in 0..iterations { + f(i); + } + let elapsed = start.elapsed(); + + let per_op = elapsed / iterations as u32; + let ops_sec = if elapsed.as_secs_f64() > 0.0 { + iterations as f64 / elapsed.as_secs_f64() + } else { + f64::INFINITY + }; + + println!(" {name:<40} {elapsed:>10.3?} {per_op:>8.0?}/op {ops_sec:>12.0} ops/s",); +} + +// ── Span store (MEMT ring buffer, wide row) ────────────────────────── + +fn bench_span_write() { + let mut schema = Schema::new() + .col("trace_id", DType::Str) + .col("span_id", DType::Str) + .col("parent_span_id", DType::Str) + .col("name", DType::Str) + .col("kind", DType::Str) + .col("start_us", DType::I64) + .col("end_us", DType::I64) + .col("duration_us", DType::I64) + .col("status_code", DType::Str) + .col("instrumentation_scope", DType::Str); + for name in &[ + "attr_actor_name", + "attr_message_type", + "attr_target_addr", + "attr_target_path", + "attr_pulsing_op", + "attr_http_method", + "attr_http_route", + "attr_http_url", + "attr_http_peer", + ] { + schema = schema.col(name, DType::Str); + } + schema = schema + .col("events_json", DType::Str) + .col("links_json", DType::Str); + + let mut table = ExposedTable::create("__bench_spans", &schema, 65536, 16).unwrap(); + + let n = 100_000u64; + bench_loop("span_write (MEMT, 21-col row)", n, |i| { + let tid = format!("trace-{i:08x}"); + let sid = format!("span-{i:08x}"); + let row: Vec = vec![ + Value::Str(&tid), + Value::Str(&sid), + Value::Str(""), + Value::Str("actor.ask"), + Value::Str("INTERNAL"), + Value::I64(1_700_000_000_000_000 + i as i64), + Value::I64(1_700_000_000_001_000 + i as i64), + Value::I64(1000), + Value::Str("OK"), + Value::Str("pulsing"), + Value::Str("actors/counter_0"), + Value::Str("ask"), + Value::Str(""), + Value::Str(""), + Value::Str("ask"), + Value::Str(""), + Value::Str(""), + Value::Str(""), + Value::Str(""), + Value::Str("[]"), + Value::Str("[]"), + ]; + let mut w = table.writer(); + w.push_row(black_box(&row)); + }); +} + +// ── Actor store (MEMH hash table) ──────────────────────────────────── + +fn bench_actor_upsert() { + let mut table = ExposedHashTable::create("__bench_actors", 256, 65536, 0).unwrap(); + + let names: Vec = (0..256).map(|i| format!("actors/worker_{i}")).collect(); + let n = 100_000u64; + + bench_loop("actor_upsert (MEMH, 256 actors)", n, |i| { + let key = &names[(i % 256) as usize]; + let val = format!("{i}|1|Worker|mymod"); + let mut w = table.writer(); + let _ = w.insert(black_box(key), black_box(&Value::Str(&val))); + }); +} + +fn bench_actor_remove() { + let mut table = ExposedHashTable::create("__bench_actors_rm", 256, 65536, 0).unwrap(); + + let names: Vec = (0..256).map(|i| format!("actors/worker_{i}")).collect(); + + // Pre-populate + { + let mut w = table.writer(); + for name in &names { + let _ = w.insert(name, &Value::Str("0|1|Worker|mymod")); + } + } + + let n = 100_000u64; + bench_loop("actor_remove (MEMH, 256 actors)", n, |i| { + let key = &names[(i % 256) as usize]; + let mut w = table.writer(); + w.remove(black_box(key)); + // Re-insert so next remove has something to do + let _ = w.insert(key, &Value::Str("0|1|Worker|mymod")); + }); +} + +// ── Metrics store (MEMT ring buffer) ───────────────────────────────── + +fn bench_metrics_write() { + let schema = Schema::new() + .col("timestamp_us", DType::I64) + .col("node_id", DType::Str) + .col("actors_count", DType::I64) + .col("messages_total", DType::I64) + .col("actors_created", DType::I64) + .col("actors_stopped", DType::I64) + .col("uptime_secs", DType::I64); + + let mut table = ExposedTable::create("__bench_metrics", &schema, 65536, 16).unwrap(); + + let n = 100_000u64; + bench_loop("metrics_write (MEMT, 7-col row)", n, |i| { + let row: Vec = vec![ + Value::I64(1_700_000_000_000_000 + i as i64), + Value::Str("node-1"), + Value::I64(42), + Value::I64(i as i64), + Value::I64(100), + Value::I64(5), + Value::I64(3600), + ]; + let mut w = table.writer(); + w.push_row(black_box(&row)); + }); +} + +// ── Members store (MEMH hash table) ────────────────────────────────── + +fn bench_member_upsert() { + let mut table = ExposedHashTable::create("__bench_members", 64, 16384, 0).unwrap(); + + let keys: Vec = (0..32).map(|i| format!("{i}")).collect(); + let n = 100_000u64; + + bench_loop("member_upsert (MEMH, 32 nodes)", n, |i| { + let key = &keys[(i % 32) as usize]; + let val = format!("10.0.0.{}:9000|online|{}", i % 32, i); + let mut w = table.writer(); + let _ = w.insert(black_box(key), black_box(&Value::Str(&val))); + }); +} + +// ── Span write with u64 IDs (no format!) ───────────────────────────── + +fn bench_span_write_u64_ids() { + let mut schema = Schema::new() + .col("trace_id_hi", DType::U64) + .col("trace_id_lo", DType::U64) + .col("span_id", DType::U64) + .col("parent_span_id", DType::U64) + .col("name", DType::Str) + .col("kind", DType::Str) + .col("start_us", DType::I64) + .col("end_us", DType::I64) + .col("duration_us", DType::I64) + .col("status_code", DType::Str) + .col("instrumentation_scope", DType::Str); + for name in &[ + "attr_actor_name", + "attr_message_type", + "attr_target_addr", + "attr_target_path", + "attr_pulsing_op", + "attr_http_method", + "attr_http_route", + "attr_http_url", + "attr_http_peer", + ] { + schema = schema.col(name, DType::Str); + } + schema = schema + .col("events_json", DType::Str) + .col("links_json", DType::Str); + + let mut table = ExposedTable::create("__bench_spans_u64", &schema, 65536, 16).unwrap(); + + let n = 100_000u64; + bench_loop("span_write (MEMT, u64 IDs)", n, |i| { + let row: Vec = vec![ + Value::U64(0xdeadbeef00000000 + i), // trace_id_hi + Value::U64(i), // trace_id_lo + Value::U64(i * 7), // span_id + Value::U64(0), // parent_span_id + Value::Str("actor.ask"), + Value::Str("INTERNAL"), + Value::I64(1_700_000_000_000_000 + i as i64), + Value::I64(1_700_000_000_001_000 + i as i64), + Value::I64(1000), + Value::Str("OK"), + Value::Str("pulsing"), + Value::Str("actors/counter_0"), + Value::Str("ask"), + Value::Str(""), + Value::Str(""), + Value::Str("ask"), + Value::Str(""), + Value::Str(""), + Value::Str(""), + Value::Str(""), + Value::Str("[]"), + Value::Str("[]"), + ]; + let mut w = table.writer(); + w.push_row(black_box(&row)); + }); +} + +// ── Isolate: format! cost alone ────────────────────────────────────── + +fn bench_format_overhead() { + let n = 100_000u64; + bench_loop("format!(trace_id + span_id) only", n, |i| { + let _tid = black_box(format!("trace-{i:08x}")); + let _sid = black_box(format!("span-{i:08x}")); + }); +} + +// ── Isolate: raw MEMT push_row (no alloc) ──────────────────────────── + +fn bench_memt_raw_push() { + let schema = Schema::new() + .col("a", DType::I64) + .col("b", DType::I64) + .col("c", DType::Str); + + let mut table = ExposedTable::create("__bench_raw_push", &schema, 65536, 16).unwrap(); + let n = 100_000u64; + bench_loop("memt_push (new writer each time)", n, |i| { + let row = [Value::I64(i as i64), Value::I64(42), Value::Str("hello")]; + let mut w = table.writer(); + w.push_row(black_box(&row)); + }); +} + +fn bench_memt_direct_push() { + let schema = Schema::new() + .col("a", DType::I64) + .col("b", DType::I64) + .col("c", DType::Str); + + let mut table = ExposedTable::create("__bench_direct_push", &schema, 65536, 16).unwrap(); + let n = 100_000u64; + bench_loop("memt_push (ExposedTable::push_row)", n, |i| { + let row = [Value::I64(i as i64), Value::I64(42), Value::Str("hello")]; + table.push_row(black_box(&row)); + }); +} + +// ── Baseline: mutex acquire/release ────────────────────────────────── + +fn bench_mutex_baseline() { + let m = std::sync::Mutex::new(0u64); + let n = 1_000_000u64; + bench_loop("mutex lock/unlock (baseline)", n, |_| { + let mut g = m.lock().unwrap(); + *g = black_box(*g + 1); + }); +} + +// ── Main ───────────────────────────────────────────────────────────── + +fn main() { + println!(); + println!("Pulsing tracing overhead benchmark"); + println!("==================================="); + println!(); + println!( + " {:<40} {:>10} {:>10} {:>12}", + "test", "total", "latency", "throughput" + ); + println!(" {}", "-".repeat(76)); + + bench_mutex_baseline(); + bench_format_overhead(); + bench_memt_raw_push(); + bench_memt_direct_push(); + println!(); + bench_span_write(); + bench_span_write_u64_ids(); + bench_metrics_write(); + println!(); + bench_actor_upsert(); + bench_actor_remove(); + bench_member_upsert(); + + println!(); +} diff --git a/crates/pulsing-actor/src/actor/context.rs b/crates/pulsing-actor/src/actor/context.rs index 315cacf07..53710fcfb 100644 --- a/crates/pulsing-actor/src/actor/context.rs +++ b/crates/pulsing-actor/src/actor/context.rs @@ -3,6 +3,9 @@ use super::mailbox::Envelope; use super::reference::ActorRef; use super::traits::{ActorId, Message, NodeId}; +use crate::tracing::{ + capture_linked_traceparent_for_mailbox, capture_linked_tracestate_for_mailbox, +}; use lru::LruCache; use serde::Serialize; use std::num::NonZeroUsize; @@ -118,12 +121,16 @@ impl ActorContext { msg: M, delay: Duration, ) -> crate::error::Result<()> { + let tp = capture_linked_traceparent_for_mailbox(); + let ts = capture_linked_tracestate_for_mailbox(); let sender = self.self_sender.clone(); let message = Message::pack(&msg)?; tokio::spawn(async move { tokio::time::sleep(delay).await; - let envelope = Envelope::tell(message); + let envelope = Envelope::tell(message) + .with_linked_traceparent(tp) + .with_linked_tracestate(ts); if let Err(e) = sender.send(envelope).await { tracing::warn!("Failed to deliver scheduled message: {}", e); } diff --git a/crates/pulsing-actor/src/actor/mailbox.rs b/crates/pulsing-actor/src/actor/mailbox.rs index 8724d1195..250525b10 100644 --- a/crates/pulsing-actor/src/actor/mailbox.rs +++ b/crates/pulsing-actor/src/actor/mailbox.rs @@ -22,6 +22,11 @@ impl Responder { pub struct Envelope { message: Message, respond_to: Option, + /// W3C traceparent captured at send time so `actor.receive` can parent to the dispatcher span + /// (e.g. `http.request`) on a different task. + linked_traceparent: Option, + /// W3C `tracestate` paired with [`Self::linked_traceparent`]. + linked_tracestate: Option, } impl Envelope { @@ -29,6 +34,8 @@ impl Envelope { Self { message, respond_to: None, + linked_traceparent: None, + linked_tracestate: None, } } @@ -36,15 +43,34 @@ impl Envelope { Self { message, respond_to: Some(respond_to), + linked_traceparent: None, + linked_tracestate: None, } } + /// Attach trace link from [`crate::tracing::capture_linked_traceparent_for_mailbox`]. + pub fn with_linked_traceparent(mut self, tp: Option) -> Self { + self.linked_traceparent = tp; + self + } + + /// Attach W3C `tracestate` from [`crate::tracing::capture_linked_tracestate_for_mailbox`]. + pub fn with_linked_tracestate(mut self, ts: Option) -> Self { + self.linked_tracestate = ts; + self + } + pub fn msg_type(&self) -> &str { self.message.msg_type() } - pub fn into_parts(self) -> (Message, Responder) { - (self.message, Responder(self.respond_to)) + pub fn into_parts(self) -> (Message, Responder, Option, Option) { + ( + self.message, + Responder(self.respond_to), + self.linked_traceparent, + self.linked_tracestate, + ) } pub fn expects_response(&self) -> bool { @@ -57,6 +83,8 @@ impl std::fmt::Debug for Envelope { f.debug_struct("Envelope") .field("msg_type", &self.message.msg_type()) .field("expects_response", &self.respond_to.is_some()) + .field("linked_trace", &self.linked_traceparent.is_some()) + .field("linked_tracestate", &self.linked_tracestate.is_some()) .finish() } } @@ -154,7 +182,7 @@ mod tests { let envelope = Envelope::ask(msg, tx); assert!(envelope.expects_response()); - let (_, responder) = envelope.into_parts(); + let (_, responder, _, _) = envelope.into_parts(); responder.send(Ok(Message::single("", b"world"))); let result = rx.await.unwrap().unwrap(); diff --git a/crates/pulsing-actor/src/actor/reference.rs b/crates/pulsing-actor/src/actor/reference.rs index f11b61c27..9ff5a870e 100644 --- a/crates/pulsing-actor/src/actor/reference.rs +++ b/crates/pulsing-actor/src/actor/reference.rs @@ -4,6 +4,9 @@ use super::address::ActorPath; use super::mailbox::Envelope; use super::traits::{ActorId, Message}; use crate::error::{PulsingError, Result, RuntimeError}; +use crate::tracing::{ + capture_linked_traceparent_for_mailbox, capture_linked_tracestate_for_mailbox, +}; use serde::{de::DeserializeOwned, Serialize}; use std::net::SocketAddr; use std::sync::Arc; @@ -200,12 +203,33 @@ impl ActorRef { /// Use this when you need direct access to `Message`, e.g., for streaming. /// For type-safe communication, prefer `ask()` and `tell()`. pub async fn send(&self, msg: Message) -> Result { + let tp = capture_linked_traceparent_for_mailbox(); + let ts = capture_linked_tracestate_for_mailbox(); + self.send_with_trace(msg, tp, ts).await + } + + /// Like [`send`](Self::send) but uses pre-captured W3C trace context instead + /// of reading from the current span. Use when the send happens on a different + /// tokio task from the one that owns the parent span (e.g. PyO3 bridge). + pub async fn send_with_trace( + &self, + msg: Message, + tp: Option, + ts: Option, + ) -> Result { match &self.inner { ActorRefInner::Local(sender) => { let (tx, rx) = oneshot::channel(); - sender.send(Envelope::ask(msg, tx)).await.map_err(|_| { - PulsingError::from(RuntimeError::Other("Actor mailbox closed".into())) - })?; + sender + .send( + Envelope::ask(msg, tx) + .with_linked_traceparent(tp) + .with_linked_tracestate(ts), + ) + .await + .map_err(|_| { + PulsingError::from(RuntimeError::Other("Actor mailbox closed".into())) + })?; rx.await .map_err(|_| PulsingError::from(RuntimeError::Other("Actor dropped".into())))? } @@ -213,28 +237,43 @@ impl ActorRef { remote.transport.send_message(&self.actor_id, msg).await } ActorRefInner::Lazy(lazy) => { - // Resolve and delegate to the resolved reference let resolved = lazy.get().await?; - // Box the recursive future to avoid infinite size - Box::pin(resolved.send(msg)).await + Box::pin(resolved.send_with_trace(msg, tp, ts)).await } } } /// Send a raw message without waiting for response (low-level fire-and-forget) pub async fn send_oneway(&self, msg: Message) -> Result<()> { + let tp = capture_linked_traceparent_for_mailbox(); + let ts = capture_linked_tracestate_for_mailbox(); + self.send_oneway_with_trace(msg, tp, ts).await + } + + /// Like [`send_oneway`](Self::send_oneway) but with pre-captured trace context. + pub async fn send_oneway_with_trace( + &self, + msg: Message, + tp: Option, + ts: Option, + ) -> Result<()> { match &self.inner { - ActorRefInner::Local(sender) => sender.send(Envelope::tell(msg)).await.map_err(|_| { - PulsingError::from(RuntimeError::Other("Actor mailbox closed".into())) - }), + ActorRefInner::Local(sender) => sender + .send( + Envelope::tell(msg) + .with_linked_traceparent(tp) + .with_linked_tracestate(ts), + ) + .await + .map_err(|_| { + PulsingError::from(RuntimeError::Other("Actor mailbox closed".into())) + }), ActorRefInner::Remote(remote) => { remote.transport.send_oneway(&self.actor_id, msg).await } ActorRefInner::Lazy(lazy) => { - // Resolve and delegate to the resolved reference let resolved = lazy.get().await?; - // Box the recursive future to avoid infinite size - Box::pin(resolved.send_oneway(msg)).await + Box::pin(resolved.send_oneway_with_trace(msg, tp, ts)).await } } } @@ -328,7 +367,7 @@ mod tests { let reply_handle = tokio::spawn(async move { let envelope = rx.recv().await.unwrap(); - let (msg, responder) = envelope.into_parts(); + let (msg, responder, _, _) = envelope.into_parts(); let req: TestMsg = msg.unpack().unwrap(); responder.send(Ok(Message::pack(&TestReply { result: req.value * 2, @@ -473,7 +512,7 @@ mod tests { let Some(envelope) = rx.recv().await else { break; }; - let (msg, responder) = envelope.into_parts(); + let (msg, responder, _, _) = envelope.into_parts(); let req: TestMsg = msg.unpack().unwrap(); responder.send(Ok(Message::pack(&TestReply { result: req.value + 1, diff --git a/crates/pulsing-actor/src/actor_store.rs b/crates/pulsing-actor/src/actor_store.rs new file mode 100644 index 000000000..28edeb826 --- /dev/null +++ b/crates/pulsing-actor/src/actor_store.rs @@ -0,0 +1,60 @@ +//! Live actor registry backed by an MEMH hash table (mmap'd file). +//! +//! Key = actor name, value = `actor_id|node_id|type|module`. +//! On-disk file `pulsing.actors`; probing SQL: `SELECT * FROM pulsing.actors`. + +use crate::actor::{ActorId, NodeId, StopReason}; +use probing_memtable::discover::ExposedHashTable; +use probing_memtable::Value; +use std::collections::HashMap; +use std::sync::{Mutex, Once, OnceLock}; + +static ACTOR_HT: OnceLock> = OnceLock::new(); +static INIT_ACTOR_MEMTABLE: Once = Once::new(); + +pub(crate) fn init_actor_memtable() { + // Only one create per process: repeated `ActorSystem::new` (e.g. parallel + // integration tests) must not call `ExposedHashTable::create` again — it + // truncates the same path and can race mmap to length 0. + INIT_ACTOR_MEMTABLE.call_once(|| { + match ExposedHashTable::create("pulsing.actors", 256, 65536, 0) { + Ok(table) => { + let _ = ACTOR_HT.set(Mutex::new(table)); + } + Err(e) => { + eprintln!("pulsing: failed to create actor hash table: {e}"); + } + } + }); +} + +pub(crate) fn write_actor_spawned( + name: &str, + actor_id: ActorId, + node_id: NodeId, + metadata: &HashMap, +) { + let Some(ht) = ACTOR_HT.get() else { return }; + let Ok(mut table) = ht.lock() else { return }; + + let actor_type = metadata + .get("class") + .or_else(|| metadata.get("type")) + .map(|s| s.as_str()) + .unwrap_or(""); + let module = metadata.get("module").map(|s| s.as_str()).unwrap_or(""); + + let val = format!("{}|{}|{}|{}", actor_id, node_id, actor_type, module); + let _ = table.writer().insert(name, &Value::Str(&val)); +} + +pub(crate) fn write_actor_stopped( + name: &str, + _actor_id: ActorId, + _node_id: NodeId, + _reason: &StopReason, +) { + let Some(ht) = ACTOR_HT.get() else { return }; + let Ok(mut table) = ht.lock() else { return }; + table.writer().remove(name); +} diff --git a/crates/pulsing-actor/src/cluster/gossip.rs b/crates/pulsing-actor/src/cluster/gossip.rs index a6fa72ae5..279dca099 100644 --- a/crates/pulsing-actor/src/cluster/gossip.rs +++ b/crates/pulsing-actor/src/cluster/gossip.rs @@ -205,6 +205,7 @@ impl ClusterState { if remote.epoch > existing.epoch || (remote.epoch == existing.epoch && remote.status > existing.status) { + let old = existing.status; nodes.insert( remote.node_id, ClusterNode { @@ -215,6 +216,14 @@ impl ClusterState { last_seen: remote.last_seen, }, ); + if old != remote.status { + crate::members_store::upsert_member( + remote.node_id, + fixed_addr, + remote.status, + remote.epoch, + ); + } } } None => { @@ -228,6 +237,12 @@ impl ClusterState { last_seen: remote.last_seen, }, ); + crate::members_store::upsert_member( + remote.node_id, + fixed_addr, + remote.status, + remote.epoch, + ); tracing::debug!(node_id = %remote.node_id, "Added new node from gossip"); } } @@ -240,6 +255,12 @@ impl ClusterState { if failure.epoch >= node.epoch && failure.status > node.status { node.status = failure.status; node.epoch = failure.epoch; + crate::members_store::upsert_member( + failure.node_id, + node.addr, + failure.status, + failure.epoch, + ); } } } @@ -262,11 +283,12 @@ impl ClusterState { let mut nodes = self.cluster_nodes.write().await; if let Some(node) = nodes.get_mut(&node_id) { node.last_seen = now_millis(); - // Recover from any failed/tombstone state if node is seen alive if node.status.can_recover() { let old_status = node.status; node.status = NodeStatus::Online; - node.epoch = self.increment_epoch(); + let epoch = self.increment_epoch(); + node.epoch = epoch; + crate::members_store::upsert_member(node_id, node.addr, NodeStatus::Online, epoch); tracing::info!( node_id = %node_id, old_status = ?old_status, @@ -433,9 +455,9 @@ impl GossipCluster { let fixed_addr = fix_addr(from_addr, peer_addr.ip()); let mut nodes = self.state.cluster_nodes.write().await; - nodes - .entry(from) - .and_modify(|n| { + match nodes.entry(from) { + std::collections::hash_map::Entry::Occupied(mut e) => { + let n = e.get_mut(); n.addr = fixed_addr; n.last_seen = now_millis(); if n.epoch < current_epoch { @@ -443,15 +465,30 @@ impl GossipCluster { } if n.status != NodeStatus::Online { n.status = NodeStatus::Online; + crate::members_store::upsert_member( + from, + fixed_addr, + NodeStatus::Online, + n.epoch, + ); } - }) - .or_insert_with(|| ClusterNode { - node_id: from, - addr: fixed_addr, - status: NodeStatus::Online, - epoch: current_epoch, - last_seen: now_millis(), - }); + } + std::collections::hash_map::Entry::Vacant(e) => { + e.insert(ClusterNode { + node_id: from, + addr: fixed_addr, + status: NodeStatus::Online, + epoch: current_epoch, + last_seen: now_millis(), + }); + crate::members_store::upsert_member( + from, + fixed_addr, + NodeStatus::Online, + current_epoch, + ); + } + } drop(nodes); // Return full cluster info to new node @@ -843,7 +880,9 @@ impl GossipCluster { if let Some(node) = nodes.get_mut(&node_id) { if node.status == NodeStatus::Online { node.status = NodeStatus::PFail; - node.epoch = self.state.increment_epoch(); + let epoch = self.state.increment_epoch(); + node.epoch = epoch; + crate::members_store::upsert_member(node_id, node.addr, NodeStatus::PFail, epoch); tracing::warn!(node_id = %node_id, "Marked node as PFail"); } } @@ -854,7 +893,9 @@ impl GossipCluster { if let Some(node) = nodes.get_mut(&node_id) { if node.status != NodeStatus::Fail { node.status = NodeStatus::Fail; - node.epoch = self.state.increment_epoch(); + let epoch = self.state.increment_epoch(); + node.epoch = epoch; + crate::members_store::upsert_member(node_id, node.addr, NodeStatus::Fail, epoch); tracing::error!(node_id = %node_id, "Marked node as Fail"); } } @@ -1070,7 +1111,9 @@ async fn detect_failures(state: &ClusterState, config: &GossipConfig) { if node.status.can_recover() { let old_status = node.status; node.status = NodeStatus::Online; - node.epoch = state.increment_epoch(); + let epoch = state.increment_epoch(); + node.epoch = epoch; + crate::members_store::upsert_member(node_id, node.addr, NodeStatus::Online, epoch); tracing::info!( node_id = %node_id, old_status = ?old_status, @@ -1085,10 +1128,11 @@ async fn detect_failures(state: &ClusterState, config: &GossipConfig) { for node_id in pfail_nodes { let mut nodes = state.cluster_nodes.write().await; if let Some(node) = nodes.get_mut(&node_id) { - // Only mark as PFail if still Online (may have been updated by another thread) if node.status == NodeStatus::Online { node.status = NodeStatus::PFail; - node.epoch = state.increment_epoch(); + let epoch = state.increment_epoch(); + node.epoch = epoch; + crate::members_store::upsert_member(node_id, node.addr, NodeStatus::PFail, epoch); tracing::debug!( node_id = %node_id, elapsed_ms = now.saturating_sub(node.last_seen), @@ -1102,12 +1146,11 @@ async fn detect_failures(state: &ClusterState, config: &GossipConfig) { for node_id in fail_nodes { let mut nodes = state.cluster_nodes.write().await; if let Some(node) = nodes.get_mut(&node_id) { - // Only mark as Fail if still in PFail state (may have recovered) if node.status == NodeStatus::PFail { node.status = NodeStatus::Fail; - node.epoch = state.increment_epoch(); - // Use warn level instead of error to reduce log noise in high-load scenarios - // This is often a false positive during stress tests + let epoch = state.increment_epoch(); + node.epoch = epoch; + crate::members_store::upsert_member(node_id, node.addr, NodeStatus::Fail, epoch); tracing::warn!( node_id = %node_id, elapsed_ms = now.saturating_sub(node.last_seen), @@ -1126,13 +1169,19 @@ async fn detect_failures(state: &ClusterState, config: &GossipConfig) { if let Some(node) = nodes.get_mut(node_id) { if node.status == NodeStatus::Fail { node.status = NodeStatus::Tombstone; - node.epoch = state.increment_epoch(); + let epoch = state.increment_epoch(); + node.epoch = epoch; + crate::members_store::upsert_member( + *node_id, + node.addr, + NodeStatus::Tombstone, + epoch, + ); tracing::info!( node_id = %node_id, "Tombstoned failed node (named actors cleared, node info retained)" ); - // Clean up named actors on this node (but keep node info) for info in named.values_mut() { info.remove_instance(node_id); } @@ -1147,16 +1196,15 @@ async fn detect_failures(state: &ClusterState, config: &GossipConfig) { let mut nodes = state.cluster_nodes.write().await; for node_id in &remove_nodes { - if nodes - .get(node_id) - .map(|n| n.status == NodeStatus::Tombstone) - .unwrap_or(false) - { - tracing::info!( - node_id = %node_id, - "Permanently removing tombstoned node after retention period" - ); - nodes.remove(node_id); + if let Some(removed) = nodes.get(node_id) { + if removed.status == NodeStatus::Tombstone { + tracing::info!( + node_id = %node_id, + "Permanently removing tombstoned node after retention period" + ); + crate::members_store::remove_member(*node_id); + nodes.remove(node_id); + } } } } diff --git a/crates/pulsing-actor/src/lib.rs b/crates/pulsing-actor/src/lib.rs index ca3cadf02..edb6aaf86 100644 --- a/crates/pulsing-actor/src/lib.rs +++ b/crates/pulsing-actor/src/lib.rs @@ -134,13 +134,18 @@ //! ``` pub mod actor; +pub(crate) mod actor_store; pub mod behavior; pub mod circuit_breaker; pub mod cluster; pub mod connect; pub mod error; +pub(crate) mod members_store; pub mod metrics; +pub(crate) mod metrics_store; +pub mod performance_store; pub mod policies; +pub mod span_store; pub mod supervision; pub mod system; pub mod system_actor; @@ -184,6 +189,8 @@ pub mod test_helper; /// /// For advanced usage (ActorPath, ActorAddress, NodeId, etc.), /// import from `pulsing_actor::actor::*`. +pub use performance_store::{PerformanceSnapshot, PerformanceStore}; + pub mod prelude { pub use crate::actor::{Actor, ActorContext, ActorRef, IntoActor, Message}; pub use crate::supervision::{BackoffStrategy, RestartPolicy, SupervisionSpec}; diff --git a/crates/pulsing-actor/src/members_store.rs b/crates/pulsing-actor/src/members_store.rs new file mode 100644 index 000000000..453cbfada --- /dev/null +++ b/crates/pulsing-actor/src/members_store.rs @@ -0,0 +1,58 @@ +//! Cluster membership state backed by an MEMH hash table (mmap'd file). +//! +//! Key = `node_id`, value = pipe-separated `addr|status|epoch`. +//! Gossip state changes call `upsert_member` / `remove_member`. +//! Probing SQL: `SELECT * FROM pulsing.members`. + +use crate::actor::NodeId; +use crate::cluster::NodeStatus; +use probing_memtable::discover::ExposedHashTable; +use probing_memtable::Value; +use std::net::SocketAddr; +use std::sync::{Mutex, Once, OnceLock}; + +static MEMBERS_HT: OnceLock> = OnceLock::new(); +static INIT_MEMBERS_MEMTABLE: Once = Once::new(); + +pub(crate) fn init_members_memtable() { + // 64 buckets, 16 KiB arena — plenty for a cluster of dozens of nodes + INIT_MEMBERS_MEMTABLE.call_once(|| { + match ExposedHashTable::create("pulsing.members", 64, 16384, 0) { + Ok(table) => { + let _ = MEMBERS_HT.set(Mutex::new(table)); + } + Err(e) => { + eprintln!("pulsing: failed to create members hash table: {e}"); + } + } + }); +} + +fn status_str(s: NodeStatus) -> &'static str { + match s { + NodeStatus::Online => "online", + NodeStatus::PFail => "suspect", + NodeStatus::Fail => "fail", + NodeStatus::Handshake => "handshake", + NodeStatus::Tombstone => "tombstone", + } +} + +/// Insert or update a member's current state in the hash table. +pub(crate) fn upsert_member(node_id: NodeId, addr: SocketAddr, new_status: NodeStatus, epoch: u64) { + let Some(ht) = MEMBERS_HT.get() else { return }; + let Ok(mut table) = ht.lock() else { return }; + + let key = node_id.to_string(); + let val = format!("{}|{}|{}", addr, status_str(new_status), epoch,); + let _ = table.writer().insert(&key, &Value::Str(&val)); +} + +/// Remove a member from the hash table (permanent removal after tombstone). +pub(crate) fn remove_member(node_id: NodeId) { + let Some(ht) = MEMBERS_HT.get() else { return }; + let Ok(mut table) = ht.lock() else { return }; + + let key = node_id.to_string(); + table.writer().remove(&key); +} diff --git a/crates/pulsing-actor/src/metrics_store.rs b/crates/pulsing-actor/src/metrics_store.rs new file mode 100644 index 000000000..79259978f --- /dev/null +++ b/crates/pulsing-actor/src/metrics_store.rs @@ -0,0 +1,64 @@ +//! System metrics time-series backed by a MEMT ring buffer (mmap'd file). +//! +//! Each `GetMetrics` call appends one row to `/probing//pulsing.metrics`. +//! Older rows are overwritten when the ring buffer wraps. + +use probing_memtable::discover::ExposedTable; +use probing_memtable::Value; +use std::sync::{Mutex, Once, OnceLock}; + +static METRICS_MEMTABLE: OnceLock> = OnceLock::new(); +static INIT_METRICS_MEMTABLE: Once = Once::new(); + +pub(crate) fn init_metrics_memtable() { + use probing_memtable::{DType, Schema}; + + INIT_METRICS_MEMTABLE.call_once(|| { + let schema = Schema::new() + .col("timestamp_us", DType::I64) + .col("node_id", DType::Str) + .col("actors_count", DType::I64) + .col("messages_total", DType::I64) + .col("actors_created", DType::I64) + .col("actors_stopped", DType::I64) + .col("uptime_secs", DType::I64); + + match ExposedTable::create("pulsing.metrics", &schema, 65536, 16) { + Ok(table) => { + let _ = METRICS_MEMTABLE.set(Mutex::new(table)); + } + Err(e) => { + eprintln!("pulsing: failed to create metrics memtable: {e}"); + } + } + }); +} + +pub(crate) fn write_metrics_snapshot( + ts_unix_micros: u64, + node_id: &str, + actors_count: u64, + messages_total: u64, + actors_created: u64, + actors_stopped: u64, + uptime_secs: u64, +) { + let Some(mt) = METRICS_MEMTABLE.get() else { + return; + }; + let Ok(mut table) = mt.lock() else { + return; + }; + + let row: Vec = vec![ + Value::I64(ts_unix_micros as i64), + Value::Str(node_id), + Value::I64(actors_count as i64), + Value::I64(messages_total as i64), + Value::I64(actors_created as i64), + Value::I64(actors_stopped as i64), + Value::I64(uptime_secs as i64), + ]; + + table.push_row(&row); +} diff --git a/crates/pulsing-actor/src/performance_store.rs b/crates/pulsing-actor/src/performance_store.rs new file mode 100644 index 000000000..ecc397c37 --- /dev/null +++ b/crates/pulsing-actor/src/performance_store.rs @@ -0,0 +1,95 @@ +//! In-memory ring buffer of recent system metric snapshots. +//! +//! **Deprecated**: prefer `SELECT * FROM pulsing.metrics` via probing's DataFusion engine. + +use serde::Serialize; +use std::collections::VecDeque; +use std::sync::Mutex; + +/// Default max snapshots retained per node. +pub const DEFAULT_PERFORMANCE_HISTORY_CAPACITY: usize = 4096; + +/// One point-in-time row aligned with `GetMetrics` / dashboard metrics. +#[derive(Debug, Clone, Serialize, PartialEq, Eq)] +pub struct PerformanceSnapshot { + /// Unix time in microseconds (UTC). + pub ts_unix_micros: u64, + pub node_id: String, + pub actors_count: u64, + pub messages_total: u64, + pub actors_created: u64, + pub actors_stopped: u64, + pub uptime_secs: u64, +} + +/// Bounded FIFO of [`PerformanceSnapshot`] (thread-safe). +pub struct PerformanceStore { + max: usize, + inner: Mutex>, +} + +impl PerformanceStore { + pub fn new(capacity: usize) -> Self { + let max = capacity.max(1); + Self { + max, + inner: Mutex::new(VecDeque::new()), + } + } + + pub fn capacity(&self) -> usize { + self.max + } + + pub fn len(&self) -> usize { + self.inner.lock().map(|g| g.len()).unwrap_or(0) + } + + pub fn is_empty(&self) -> bool { + self.inner.lock().map(|g| g.is_empty()).unwrap_or(true) + } + + pub fn record(&self, snapshot: PerformanceSnapshot) { + if let Ok(mut g) = self.inner.lock() { + if g.len() >= self.max { + g.pop_front(); + } + g.push_back(snapshot); + } + } + + /// Newest-first, at most `limit` items. + pub fn recent(&self, limit: usize) -> Vec { + let Ok(g) = self.inner.lock() else { + return Vec::new(); + }; + let n = limit.min(g.len()); + g.iter().rev().take(n).cloned().collect() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn ring_drops_oldest() { + let s = PerformanceStore::new(2); + for i in 0..3 { + s.record(PerformanceSnapshot { + ts_unix_micros: i, + node_id: "n".into(), + actors_count: i, + messages_total: 0, + actors_created: 0, + actors_stopped: 0, + uptime_secs: 0, + }); + } + assert_eq!(s.len(), 2); + let r = s.recent(10); + assert_eq!(r.len(), 2); + assert_eq!(r[0].ts_unix_micros, 2); + assert_eq!(r[1].ts_unix_micros, 1); + } +} diff --git a/crates/pulsing-actor/src/span_store.rs b/crates/pulsing-actor/src/span_store.rs new file mode 100644 index 000000000..5bf0df689 --- /dev/null +++ b/crates/pulsing-actor/src/span_store.rs @@ -0,0 +1,316 @@ +//! Span storage backed by a memtable (mmap'd ring buffer). +//! +//! When tracing is initialized, an [`ExposedTable`] is created at +//! `/probing//pulsing.spans`. Any tool that understands the memtable +//! format (e.g. probing's DataFusion engine) can read spans in real time via mmap. + +use futures::future::BoxFuture; +use opentelemetry::trace::{SpanId, Status}; +use opentelemetry::KeyValue; +use opentelemetry_sdk::export::trace::{ExportResult, SpanData, SpanExporter}; +use probing_memtable::discover::ExposedTable; +use probing_memtable::Value; +use std::sync::{Mutex, Once, OnceLock}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +const MAX_ATTR_ENTRIES: usize = 128; +const MAX_ATTR_VALUE_BYTES: usize = 1024; +const MAX_EVENTS_SERIALIZED: usize = 32; +const MAX_EVENT_ATTR_ENTRIES: usize = 16; +const MAX_LINKS_SERIALIZED: usize = 32; + +const SPAN_ATTR_KEYS: &[(&str, &str)] = &[ + ("actor.name", "attr_actor_name"), + ("message.type", "attr_message_type"), + ("target.addr", "attr_target_addr"), + ("target.path", "attr_target_path"), + ("pulsing.op", "attr_pulsing_op"), + ("http.method", "attr_http_method"), + ("http.route", "attr_http_route"), + ("http.url", "attr_http_url"), + ("http.peer", "attr_http_peer"), +]; + +// ── memtable storage ─────────────────────────────────────────────────── + +static SPAN_MEMTABLE: OnceLock> = OnceLock::new(); +static INIT_SPAN_MEMTABLE: Once = Once::new(); + +pub(crate) fn init_span_memtable() { + use probing_memtable::{DType, Schema}; + + INIT_SPAN_MEMTABLE.call_once(|| { + let mut schema = Schema::new() + .col("trace_id", DType::Str) + .col("span_id", DType::Str) + .col("parent_span_id", DType::Str) + .col("name", DType::Str) + .col("kind", DType::Str) + .col("start_us", DType::I64) + .col("end_us", DType::I64) + .col("duration_us", DType::I64) + .col("status_code", DType::Str) + .col("instrumentation_scope", DType::Str); + + for &(_, col_name) in SPAN_ATTR_KEYS { + schema = schema.col(col_name, DType::Str); + } + schema = schema + .col("events_json", DType::Str) + .col("links_json", DType::Str); + + match ExposedTable::create("pulsing.spans", &schema, 65536, 16) { + Ok(table) => { + let _ = SPAN_MEMTABLE.set(Mutex::new(table)); + } + Err(e) => { + eprintln!("pulsing: failed to create span memtable: {e}"); + } + } + }); +} + +fn write_span_to_memtable(record: &SpanRecord) { + let Some(mt) = SPAN_MEMTABLE.get() else { + return; + }; + let Ok(mut table) = mt.lock() else { + return; + }; + + let attrs: serde_json::Map = + serde_json::from_str(&record.attributes_json).unwrap_or_default(); + + let mut attr_vals: Vec = Vec::with_capacity(SPAN_ATTR_KEYS.len()); + for &(src_key, _) in SPAN_ATTR_KEYS { + let v = attrs.get(src_key).and_then(|v| v.as_str()).unwrap_or(""); + attr_vals.push(v.to_string()); + } + + let mut row: Vec = Vec::with_capacity(10 + SPAN_ATTR_KEYS.len() + 2); + row.push(Value::Str(&record.trace_id)); + row.push(Value::Str(&record.span_id)); + row.push(Value::Str(&record.parent_span_id)); + row.push(Value::Str(&record.name)); + row.push(Value::Str(&record.kind)); + row.push(Value::I64((record.start_unix_nanos / 1000) as i64)); + row.push(Value::I64((record.end_unix_nanos / 1000) as i64)); + row.push(Value::I64(record.duration_nanos as i64 / 1000)); + row.push(Value::Str(&record.status_code)); + row.push(Value::Str(&record.instrumentation_scope)); + for v in &attr_vals { + row.push(Value::Str(v)); + } + row.push(Value::Str(&record.events_json)); + row.push(Value::Str(&record.links_json)); + + table.push_row(&row); +} + +// ── exporter ─────────────────────────────────────────────────────────── + +#[derive(Debug)] +pub(crate) struct InMemorySpanExporter; + +impl SpanExporter for InMemorySpanExporter { + fn export(&mut self, batch: Vec) -> BoxFuture<'static, ExportResult> { + for sd in batch { + write_span_to_memtable(&span_data_to_record(sd)); + } + Box::pin(std::future::ready(Ok(()))) + } + + fn shutdown(&mut self) {} +} + +// ── internal SpanRecord (used by span_data_to_record / write_span_to_memtable) ── + +#[derive(Debug, Clone)] +#[allow(dead_code)] +pub(crate) struct SpanRecord { + pub trace_id: String, + pub span_id: String, + pub parent_span_id: String, + pub sampled: bool, + pub name: String, + pub kind: String, + pub start_unix_nanos: u128, + pub end_unix_nanos: u128, + pub duration_nanos: u64, + pub status_code: String, + pub status_message: String, + pub instrumentation_scope: String, + pub instrumentation_version: String, + pub dropped_attributes_count: u32, + pub events_dropped_count: u32, + pub links_dropped_count: u32, + pub attributes_json: String, + pub events_json: String, + pub links_json: String, +} + +// ── helpers ───────────────────────────────────────────────────────────── + +fn system_time_unix_nanos(st: SystemTime) -> u128 { + st.duration_since(UNIX_EPOCH) + .map(|d| d.as_nanos()) + .unwrap_or(0) +} + +fn truncate_bytes(s: &str, max: usize) -> String { + if s.len() <= max { + return s.to_string(); + } + let mut end = max.saturating_sub(1); + while end > 0 && !s.is_char_boundary(end) { + end -= 1; + } + format!("{}…", &s[..end]) +} + +fn kv_slice_to_json_map(attrs: &[KeyValue], max_entries: usize, max_val_bytes: usize) -> String { + let mut m = serde_json::Map::new(); + for kv in attrs.iter().take(max_entries) { + let k = kv.key.to_string(); + let v = truncate_bytes(&kv.value.to_string(), max_val_bytes); + m.insert(k, serde_json::Value::String(v)); + } + serde_json::Value::Object(m).to_string() +} + +fn span_data_to_record(sd: SpanData) -> SpanRecord { + let sc = sd.span_context; + let trace_id = sc.trace_id().to_string(); + let span_id = sc.span_id().to_string(); + let parent = sd.parent_span_id; + let parent_span_id = if parent == SpanId::INVALID { + String::new() + } else { + parent.to_string() + }; + + let start_n = system_time_unix_nanos(sd.start_time); + let end_n = system_time_unix_nanos(sd.end_time); + let duration_nanos = sd + .end_time + .duration_since(sd.start_time) + .map(|d: Duration| d.as_nanos().min(u128::from(u64::MAX)) as u64) + .unwrap_or(0); + + let (status_code, status_message) = match &sd.status { + Status::Unset => ("unset".to_string(), String::new()), + Status::Ok => ("ok".to_string(), String::new()), + Status::Error { description } => ( + "error".to_string(), + truncate_bytes(description.as_ref(), 4096), + ), + }; + + let scope = &sd.instrumentation_scope; + let instrumentation_version = scope.version().map(|s| s.to_string()).unwrap_or_default(); + + let kind = format!("{:?}", sd.span_kind); + let name = sd.name.into_owned(); + + let attributes_json = + kv_slice_to_json_map(&sd.attributes, MAX_ATTR_ENTRIES, MAX_ATTR_VALUE_BYTES); + + let events: Vec = sd + .events + .iter() + .take(MAX_EVENTS_SERIALIZED) + .map(|ev| { + let ts = system_time_unix_nanos(ev.timestamp); + let attrs = + kv_slice_to_json_map(&ev.attributes, MAX_EVENT_ATTR_ENTRIES, MAX_ATTR_VALUE_BYTES); + serde_json::json!({ + "name": ev.name.as_ref(), + "timestamp_unix_nanos": ts.to_string(), + "attributes": serde_json::from_str::(&attrs).unwrap_or_default(), + }) + }) + .collect(); + let events_json = serde_json::to_string(&events).unwrap_or_else(|_| "[]".to_string()); + + let links: Vec = sd + .links + .iter() + .take(MAX_LINKS_SERIALIZED) + .map(|lk| { + let lsc = lk.span_context.clone(); + serde_json::json!({ + "trace_id": lsc.trace_id().to_string(), + "span_id": lsc.span_id().to_string(), + }) + }) + .collect(); + let links_json = serde_json::to_string(&links).unwrap_or_else(|_| "[]".to_string()); + + SpanRecord { + trace_id, + span_id, + parent_span_id, + sampled: sc.is_sampled(), + name, + kind, + start_unix_nanos: start_n, + end_unix_nanos: end_n, + duration_nanos, + status_code, + status_message, + instrumentation_scope: scope.name().to_string(), + instrumentation_version, + dropped_attributes_count: sd.dropped_attributes_count, + events_dropped_count: sd.events.dropped_count, + links_dropped_count: sd.links.dropped_count, + attributes_json, + events_json, + links_json, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use opentelemetry::trace::{SpanContext, SpanKind, TraceFlags, TraceId, TraceState}; + use opentelemetry::InstrumentationScope; + use opentelemetry_sdk::trace::{SpanEvents, SpanLinks}; + use std::borrow::Cow; + + fn sample_span_data(name: &str) -> SpanData { + let trace_id = TraceId::from_bytes([9u8; 16]); + let span_id = SpanId::from_bytes([8u8; 8]); + let sc = SpanContext::new( + trace_id, + span_id, + TraceFlags::SAMPLED, + true, + TraceState::default(), + ); + SpanData { + span_context: sc, + parent_span_id: SpanId::INVALID, + span_kind: SpanKind::Internal, + name: Cow::Owned(name.to_string()), + start_time: UNIX_EPOCH + Duration::from_secs(1), + end_time: UNIX_EPOCH + Duration::from_secs(2), + attributes: vec![KeyValue::new("actor.name", "test_actor")], + dropped_attributes_count: 0, + events: SpanEvents::default(), + links: SpanLinks::default(), + status: Status::Ok, + instrumentation_scope: InstrumentationScope::builder("test.scope").build(), + } + } + + #[tokio::test] + async fn exporter_writes_to_memtable() { + init_span_memtable(); + let mut exp = InMemorySpanExporter; + let _ = exp.export(vec![sample_span_data("x")]).await; + + let mt = SPAN_MEMTABLE.get().unwrap().lock().unwrap(); + let view = mt.view(); + assert!(view.num_rows(view.write_chunk()) >= 1); + } +} diff --git a/crates/pulsing-actor/src/system/handler.rs b/crates/pulsing-actor/src/system/handler.rs index 6b278cf75..56cd2b03a 100644 --- a/crates/pulsing-actor/src/system/handler.rs +++ b/crates/pulsing-actor/src/system/handler.rs @@ -6,12 +6,15 @@ use crate::cluster::{GossipBackend, GossipMessage, HeadNodeBackend, NamingBacken use crate::error::{PulsingError, Result, RuntimeError}; use crate::metrics::{metrics, SystemMetrics as PrometheusMetrics}; use crate::system::registry::ActorRegistry; +use crate::tracing::{ + capture_linked_traceparent_for_mailbox, capture_linked_tracestate_for_mailbox, +}; use crate::transport::Http2ServerHandler; use serde::{Deserialize, Serialize}; use std::collections::HashMap; use std::net::SocketAddr; use std::sync::atomic::Ordering; -use std::sync::Arc; +use std::sync::{Arc, OnceLock}; use tokio::sync::{mpsc, RwLock}; /// Unified message handler for HTTP/2 transport. @@ -23,6 +26,13 @@ pub(crate) struct SystemMessageHandler { registry: Arc, /// Cluster backend cluster: Arc>>>, + /// Same state as [`SystemActor`] metrics registry (after `system/core` starts). + system_monitor: Arc< + OnceLock<( + Arc, + Arc, + )>, + >, } impl SystemMessageHandler { @@ -30,11 +40,18 @@ impl SystemMessageHandler { node_id: NodeId, registry: Arc, cluster: Arc>>>, + system_monitor: Arc< + OnceLock<( + Arc, + Arc, + )>, + >, ) -> Self { Self { node_id, registry, cluster, + system_monitor, } } @@ -71,7 +88,11 @@ impl SystemMessageHandler { let sender = self.find_actor_sender(actor_name)?; let (tx, rx) = tokio::sync::oneshot::channel(); - let envelope = Envelope::ask(msg, tx); + let tp = capture_linked_traceparent_for_mailbox(); + let ts = capture_linked_tracestate_for_mailbox(); + let envelope = Envelope::ask(msg, tx) + .with_linked_traceparent(tp) + .with_linked_tracestate(ts); sender .send(envelope) @@ -84,7 +105,11 @@ impl SystemMessageHandler { async fn tell_local_actor(&self, actor_name: &str, msg: Message) -> Result<()> { let sender = self.find_actor_sender(actor_name)?; - let envelope = Envelope::tell(msg); + let tp = capture_linked_traceparent_for_mailbox(); + let ts = capture_linked_tracestate_for_mailbox(); + let envelope = Envelope::tell(msg) + .with_linked_traceparent(tp) + .with_linked_tracestate(ts); sender .send(envelope) @@ -254,13 +279,20 @@ impl Http2ServerHandler for SystemMessageHandler { total_messages += entry.value().stats.message_count.load(Ordering::Relaxed); } + let (actors_count, actors_created, actors_stopped) = + if let Some((m, r)) = self.system_monitor.get() { + (r.count(), m.actors_created(), m.actors_stopped()) + } else { + (self.registry.actor_count(), 0u64, 0u64) + }; + // Build system metrics let system_metrics = PrometheusMetrics { node_id: self.node_id.0, - actors_count: self.registry.actor_count(), + actors_count, messages_total: total_messages, - actors_created: self.registry.actor_count() as u64, - actors_stopped: 0, + actors_created, + actors_stopped, cluster_members, }; @@ -542,6 +574,16 @@ mod tests { use super::*; use crate::system::handle::{ActorStats, LocalActorHandle}; use std::collections::HashMap; + use std::sync::OnceLock; + + fn empty_monitor() -> Arc< + OnceLock<( + Arc, + Arc, + )>, + > { + Arc::new(OnceLock::new()) + } fn make_mock_actor( _actor_id: ActorId, @@ -549,7 +591,7 @@ mod tests { let (tx, mut rx) = mpsc::channel::(8); let join = tokio::spawn(async move { while let Some(env) = rx.recv().await { - let (_, responder) = env.into_parts(); + let (_, responder, _, _) = env.into_parts(); responder.send(Ok(Message::single("pong", vec![1, 2, 3]))); } }); @@ -560,7 +602,8 @@ mod tests { async fn test_dispatch_message_invalid_path() { let registry = Arc::new(ActorRegistry::new()); let cluster = Arc::new(RwLock::new(None)); - let handler = SystemMessageHandler::new(NodeId::generate(), registry, cluster); + let handler = + SystemMessageHandler::new(NodeId::generate(), registry, cluster, empty_monitor()); let msg = Message::single("ping", vec![]); let err = handler @@ -575,7 +618,8 @@ mod tests { async fn test_dispatch_message_actor_not_found() { let registry = Arc::new(ActorRegistry::new()); let cluster = Arc::new(RwLock::new(None)); - let handler = SystemMessageHandler::new(NodeId::generate(), registry, cluster); + let handler = + SystemMessageHandler::new(NodeId::generate(), registry, cluster, empty_monitor()); let msg = Message::single("ping", vec![]); let err = handler @@ -590,7 +634,8 @@ mod tests { async fn test_dispatch_message_named_actor_not_found() { let registry = Arc::new(ActorRegistry::new()); let cluster = Arc::new(RwLock::new(None)); - let handler = SystemMessageHandler::new(NodeId::generate(), registry, cluster); + let handler = + SystemMessageHandler::new(NodeId::generate(), registry, cluster, empty_monitor()); let msg = Message::single("ping", vec![]); let err = handler @@ -623,7 +668,8 @@ mod tests { registry.register_name("test-actor".to_string(), actor_id); let cluster = Arc::new(RwLock::new(None)); - let handler = SystemMessageHandler::new(NodeId::generate(), registry, cluster); + let handler = + SystemMessageHandler::new(NodeId::generate(), registry, cluster, empty_monitor()); let msg = Message::single("ping", vec![]); let response = handler @@ -652,7 +698,8 @@ mod tests { registry.register_name("simple".to_string(), actor_id); let cluster = Arc::new(RwLock::new(None)); - let handler = SystemMessageHandler::new(NodeId::generate(), registry, cluster); + let handler = + SystemMessageHandler::new(NodeId::generate(), registry, cluster, empty_monitor()); let response = handler .handle_message_simple("/actors/simple", "req", vec![1, 2, 3]) @@ -665,7 +712,8 @@ mod tests { async fn test_dispatch_tell_actor_not_found() { let registry = Arc::new(ActorRegistry::new()); let cluster = Arc::new(RwLock::new(None)); - let handler = SystemMessageHandler::new(NodeId::generate(), registry, cluster); + let handler = + SystemMessageHandler::new(NodeId::generate(), registry, cluster, empty_monitor()); let err = handler .handle_tell("/actors/missing", "msg", vec![]) diff --git a/crates/pulsing-actor/src/system/lifecycle.rs b/crates/pulsing-actor/src/system/lifecycle.rs index e0dfaa91b..afe32b98b 100644 --- a/crates/pulsing-actor/src/system/lifecycle.rs +++ b/crates/pulsing-actor/src/system/lifecycle.rs @@ -195,11 +195,13 @@ impl ActorSystem { } // 3. Handle lifecycle cleanup + let reason_copy = reason.clone(); + let actor_id = handle.actor_id; let registry = self.registry.clone(); self.registry .lifecycle .handle_termination( - &handle.actor_id, + &actor_id, named_path, reason, ®istry.named_actor_paths, @@ -210,5 +212,7 @@ impl ActorSystem { }, ) .await; + + self.notify_monitor_actor_stopped(actor_name, actor_id, &reason_copy); } } diff --git a/crates/pulsing-actor/src/system/mod.rs b/crates/pulsing-actor/src/system/mod.rs index c436d5f9d..8daeead95 100644 --- a/crates/pulsing-actor/src/system/mod.rs +++ b/crates/pulsing-actor/src/system/mod.rs @@ -91,16 +91,21 @@ pub use load_balancer::NodeLoadTracker; pub use registry::ActorRegistry; pub use traits::{ActorSystemCoreExt, ActorSystemOpsExt}; -use crate::actor::{ActorId, ActorPath, ActorRef, ActorResolver, ActorSystemRef, Envelope, NodeId}; +use crate::actor::{ + ActorId, ActorPath, ActorRef, ActorResolver, ActorSystemRef, Envelope, NodeId, StopReason, +}; use crate::cluster::{GossipBackend, HeadNodeBackend, NamingBackend}; use crate::error::{PulsingError, Result, RuntimeError}; +use crate::performance_store::{ + PerformanceSnapshot, PerformanceStore, DEFAULT_PERFORMANCE_HISTORY_CAPACITY, +}; use crate::policies::{LoadBalancingPolicy, RoundRobinPolicy}; use crate::system_actor::{BoxedActorFactory, SystemActor, SystemRef, SYSTEM_ACTOR_PATH}; use crate::transport::Http2Transport; use dashmap::DashMap; use handler::SystemMessageHandler; use std::net::SocketAddr; -use std::sync::Arc; +use std::sync::{Arc, OnceLock}; use tokio::sync::mpsc; use tokio::sync::RwLock; use tokio_util::sync::CancellationToken; @@ -136,6 +141,17 @@ pub struct ActorSystem { /// Per-node load tracking for remote nodes pub(crate) node_load: Arc>>, + + /// Shared [`SystemActor`] monitoring registry + metrics (filled when `system/core` starts). + pub(crate) system_monitor: Arc< + OnceLock<( + Arc, + Arc, + )>, + >, + + /// Ring buffer of recent `GetMetrics` snapshots (for local analysis / SQL adapters). + pub(crate) performance_store: Arc, } impl ActorSystem { @@ -157,9 +173,22 @@ impl ActorSystem { let registry = Arc::new(ActorRegistry::new()); let cluster_holder: Arc>>> = Arc::new(RwLock::new(None)); + let system_monitor: Arc< + OnceLock<( + Arc, + Arc, + )>, + > = Arc::new(OnceLock::new()); + let performance_store = + Arc::new(PerformanceStore::new(DEFAULT_PERFORMANCE_HISTORY_CAPACITY)); // Create message handler (needs registry and cluster reference) - let handler = SystemMessageHandler::new(node_id, registry.clone(), cluster_holder.clone()); + let handler = SystemMessageHandler::new( + node_id, + registry.clone(), + cluster_holder.clone(), + system_monitor.clone(), + ); // Clone http2_config before moving it to transport let http2_config_for_backend = config.http2_config.clone(); @@ -213,6 +242,16 @@ impl ActorSystem { backend.join(Vec::new()).await?; } + crate::actor_store::init_actor_memtable(); + crate::metrics_store::init_metrics_memtable(); + crate::members_store::init_members_memtable(); + crate::members_store::upsert_member( + node_id, + actual_addr, + crate::cluster::NodeStatus::Online, + 0, + ); + let system = Arc::new(Self { node_id, addr: actual_addr, @@ -223,6 +262,8 @@ impl ActorSystem { cancel_token, default_lb_policy: Arc::new(RoundRobinPolicy::new()), node_load: Arc::new(DashMap::new()), + system_monitor, + performance_store, }); // Start SystemActor @@ -261,8 +302,15 @@ impl ActorSystem { named_actor_paths, }); - // Create SystemActor with default factory - let system_actor = SystemActor::with_default_factory(system_ref); + let metrics = Arc::new(crate::system_actor::SystemMetrics::new()); + let sa_registry = Arc::new(crate::system_actor::ActorRegistry::new()); + let system_actor = SystemActor::with_default_factory_shared( + system_ref, + sa_registry.clone(), + metrics.clone(), + self.performance_store.clone(), + ) + .with_system_registry(self.registry.clone()); // Spawn as named actor with path "system" (use new_system to bypass namespace check) let system_path = ActorPath::new_system(SYSTEM_ACTOR_PATH)?; @@ -271,6 +319,8 @@ impl ActorSystem { .spawn(system_actor) .await?; + let _ = self.system_monitor.set((metrics, sa_registry)); + // Note: The local_actors_ref and actor_names_ref are used internally, // SystemRef snapshot may become stale for new actors but that's acceptable // since SystemActor doesn't need real-time actor list @@ -303,8 +353,16 @@ impl ActorSystem { named_actor_paths: named_paths_snapshot, }); - // Create SystemActor with custom factory - let system_actor = SystemActor::new(system_ref, factory); + let metrics = Arc::new(crate::system_actor::SystemMetrics::new()); + let sa_registry = Arc::new(crate::system_actor::ActorRegistry::new()); + let system_actor = SystemActor::new_shared( + system_ref, + factory, + sa_registry.clone(), + metrics.clone(), + self.performance_store.clone(), + ) + .with_system_registry(self.registry.clone()); // Spawn as named actor (use new_system to bypass namespace check) let system_path = ActorPath::new_system(SYSTEM_ACTOR_PATH)?; @@ -313,6 +371,8 @@ impl ActorSystem { .spawn(system_actor) .await?; + let _ = self.system_monitor.set((metrics, sa_registry)); + tracing::debug!( path = SYSTEM_ACTOR_PATH, "SystemActor started with custom factory" @@ -348,6 +408,65 @@ impl ActorSystem { pub fn local_actor_ref_by_name(&self, name: &str) -> Option { self.registry.local_actor_ref_by_name(name) } + + /// Recent metric snapshots (newest first), recorded on each `GetMetrics` to `system/core`. + pub fn performance_recent(&self, limit: usize) -> Vec { + self.performance_store.recent(limit) + } + + /// In-memory performance history store (same instance as [`SystemActor`] uses). + pub fn performance_store(&self) -> &Arc { + &self.performance_store + } + + /// Keep [`SystemActor`] `GetMetrics` / list in sync with real spawns (excludes `system/core`). + pub(crate) fn notify_monitor_actor_spawned( + &self, + name_str: &Option, + actor_id: ActorId, + actor_metadata: &std::collections::HashMap, + ) { + if name_str.as_deref() == Some(SYSTEM_ACTOR_PATH) { + return; + } + let key = match name_str { + Some(n) => n.clone(), + None => actor_id.to_string(), + }; + + crate::actor_store::write_actor_spawned(&key, actor_id, self.node_id, actor_metadata); + + let Some((metrics, reg)) = self.system_monitor.get() else { + return; + }; + let actor_type = actor_metadata + .get("class") + .or_else(|| actor_metadata.get("type")) + .cloned() + .unwrap_or_else(|| "Actor".to_string()); + reg.register_with_metadata(&key, actor_id, &actor_type, actor_metadata.clone()); + metrics.inc_actor_created(); + } + + pub(crate) fn notify_monitor_actor_stopped( + &self, + actor_name: &str, + actor_id: ActorId, + reason: &StopReason, + ) { + if actor_name == SYSTEM_ACTOR_PATH { + return; + } + + crate::actor_store::write_actor_stopped(actor_name, actor_id, self.node_id, reason); + + let Some((metrics, reg)) = self.system_monitor.get() else { + return; + }; + if reg.unregister(actor_name).is_some() { + metrics.inc_actor_stopped(); + } + } } #[async_trait::async_trait] diff --git a/crates/pulsing-actor/src/system/resolve.rs b/crates/pulsing-actor/src/system/resolve.rs index 8af097a0d..11d13a73b 100644 --- a/crates/pulsing-actor/src/system/resolve.rs +++ b/crates/pulsing-actor/src/system/resolve.rs @@ -347,10 +347,11 @@ impl ActorSystem { /// Get cluster member information pub async fn members(&self) -> Vec { - match self.cluster_opt().await { + let members = match self.cluster_opt().await { Some(cluster) => cluster.all_members().await, None => Vec::new(), - } + }; + members } /// Get all named actors in the cluster diff --git a/crates/pulsing-actor/src/system/runtime.rs b/crates/pulsing-actor/src/system/runtime.rs index fe4d0f646..d1f4df694 100644 --- a/crates/pulsing-actor/src/system/runtime.rs +++ b/crates/pulsing-actor/src/system/runtime.rs @@ -3,9 +3,11 @@ use super::handle::ActorStats; use crate::actor::{Actor, ActorContext, Envelope, StopReason}; use crate::supervision::SupervisionSpec; +use crate::tracing::actor_receive_span; use std::sync::Arc; use tokio::sync::mpsc; use tokio_util::sync::CancellationToken; +use tracing::Instrument; /// Actor instance loop - runs a single instance of an actor pub(crate) async fn run_actor_instance( @@ -28,14 +30,29 @@ pub(crate) async fn run_actor_instance( match msg { Some(envelope) => { stats.inc_message(); - let (message, responder) = envelope.into_parts(); - - match actor.receive(message, ctx).await { + let (message, responder, linked_tp, linked_ts) = envelope.into_parts(); + + let actor_name = ctx + .named_path() + .map(str::to_string) + .unwrap_or_else(|| ctx.id().to_string()); + let recv_span = actor_receive_span( + &actor_name, + message.msg_type(), + linked_tp.as_deref(), + linked_ts.as_deref(), + ); + + let result = actor + .receive(message, ctx) + .instrument(recv_span) + .await; + + match result { Ok(response) => { responder.send(Ok(response)); } Err(e) => { - // Business error: receive returns Err, only return error to caller, actor continues processing next message tracing::warn!(actor_id = ?ctx.id(), error = %e, "Receive returned error (returned to caller)"); responder.send(Err(e)); } diff --git a/crates/pulsing-actor/src/system/spawn.rs b/crates/pulsing-actor/src/system/spawn.rs index 83961fcf5..6f3913560 100644 --- a/crates/pulsing-actor/src/system/spawn.rs +++ b/crates/pulsing-actor/src/system/spawn.rs @@ -96,7 +96,7 @@ impl ActorSystem { cluster.register_named_actor(path.clone()).await; } else { cluster - .register_named_actor_full(path.clone(), actor_id, metadata) + .register_named_actor_full(path.clone(), actor_id, metadata.clone()) .await; } // So refer(actor_id) / lookup_actor can resolve this actor on any node @@ -108,6 +108,8 @@ impl ActorSystem { self.registry.register_name(actor_id.to_string(), actor_id); } + self.notify_monitor_actor_spawned(&name_str, actor_id, &metadata); + Ok(ActorRef::local(actor_id, sender)) } diff --git a/crates/pulsing-actor/src/system_actor/mod.rs b/crates/pulsing-actor/src/system_actor/mod.rs index 28fa6ba79..931589daa 100644 --- a/crates/pulsing-actor/src/system_actor/mod.rs +++ b/crates/pulsing-actor/src/system_actor/mod.rs @@ -27,11 +27,12 @@ pub use messages::{ActorInfo, ActorStatusInfo, SystemMessage, SystemResponse}; use crate::actor::{Actor, ActorContext, ActorId, Message}; use crate::error::{PulsingError, Result, RuntimeError}; use crate::metrics::SystemMetrics as PrometheusSystemMetrics; +use crate::performance_store::{PerformanceSnapshot, PerformanceStore}; use dashmap::DashMap; use std::collections::HashMap; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; -use std::time::Instant; +use std::time::{Instant, SystemTime, UNIX_EPOCH}; use tokio::sync::mpsc; /// Named path for SystemActor (system/core satisfies namespace/name format requirement) @@ -83,6 +84,8 @@ struct ActorEntry { actor_id: ActorId, actor_type: String, created_at: Instant, + /// Spawn-time metadata (e.g. Python ``class`` / ``module`` / ``file``). + metadata: HashMap, } /// Actor registry @@ -99,12 +102,23 @@ impl ActorRegistry { } pub fn register(&self, name: &str, actor_id: ActorId, actor_type: &str) { + self.register_with_metadata(name, actor_id, actor_type, HashMap::new()); + } + + pub fn register_with_metadata( + &self, + name: &str, + actor_id: ActorId, + actor_type: &str, + metadata: HashMap, + ) { self.actors.insert( name.to_string(), ActorEntry { actor_id, actor_type: actor_type.to_string(), created_at: Instant::now(), + metadata, }, ); } @@ -133,7 +147,7 @@ impl ActorRegistry { actor_id: e.actor_id.0, actor_type: e.actor_type.clone(), uptime_secs: e.created_at.elapsed().as_secs(), - metadata: std::collections::HashMap::new(), // TODO: get from actor + metadata: e.metadata.clone(), }) .collect() } @@ -144,7 +158,7 @@ impl ActorRegistry { actor_id: e.actor_id.0, actor_type: e.actor_type.clone(), uptime_secs: e.created_at.elapsed().as_secs(), - metadata: std::collections::HashMap::new(), // TODO: get from actor + metadata: e.metadata.clone(), }) } } @@ -183,23 +197,80 @@ pub struct SystemActor { /// Start time start_time: Instant, + + /// Ring buffer of recent `GetMetrics` samples + performance_store: Arc, + + /// System-level registry with per-actor stats (for accurate messages_total) + system_registry: Option>, } impl SystemActor { - /// Create a new SystemActor + /// Create a new SystemActor (private registry + metrics). pub fn new(system_ref: Arc, factory: BoxedActorFactory) -> Self { + Self::new_shared( + system_ref, + factory, + Arc::new(ActorRegistry::new()), + Arc::new(SystemMetrics::new()), + Arc::new(PerformanceStore::new(256)), + ) + } + + /// Create SystemActor with default factory + pub fn with_default_factory(system_ref: Arc) -> Self { + Self::with_default_factory_shared( + system_ref, + Arc::new(ActorRegistry::new()), + Arc::new(SystemMetrics::new()), + Arc::new(PerformanceStore::new(256)), + ) + } + + /// Shared registry + metrics (must match [`crate::system::ActorSystem::system_monitor`]). + pub fn with_default_factory_shared( + system_ref: Arc, + registry: Arc, + metrics: Arc, + performance_store: Arc, + ) -> Self { Self { - registry: Arc::new(ActorRegistry::new()), + registry, + factory: Box::new(DefaultActorFactory), + metrics, + system_ref, + start_time: Instant::now(), + performance_store, + system_registry: None, + } + } + + /// Shared registry + metrics with a custom factory. + pub fn new_shared( + system_ref: Arc, + factory: BoxedActorFactory, + registry: Arc, + metrics: Arc, + performance_store: Arc, + ) -> Self { + Self { + registry, factory, - metrics: Arc::new(SystemMetrics::new()), + metrics, system_ref, start_time: Instant::now(), + performance_store, + system_registry: None, } } - /// Create SystemActor with default factory - pub fn with_default_factory(system_ref: Arc) -> Self { - Self::new(system_ref, Box::new(DefaultActorFactory)) + /// Attach the system-level registry for accurate global message counting. + pub fn with_system_registry( + mut self, + reg: Arc, + ) -> Self { + self.system_registry = Some(reg); + self } /// Get registry (for Python bindings) @@ -231,12 +302,44 @@ impl SystemActor { /// Handle GetMetrics request fn handle_get_metrics(&self) -> SystemResponse { + let actors_count = self.registry.count(); + let messages_total = self.total_messages(); + let actors_created = self.metrics.actors_created(); + let actors_stopped = self.metrics.actors_stopped(); + let uptime_secs = self.start_time.elapsed().as_secs(); + + let ts_unix_micros = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_micros() as u64) + .unwrap_or(0); + let node_id_str = self.system_ref.node_id.to_string(); + + self.performance_store.record(PerformanceSnapshot { + ts_unix_micros, + node_id: node_id_str.clone(), + actors_count: actors_count as u64, + messages_total, + actors_created, + actors_stopped, + uptime_secs, + }); + + crate::metrics_store::write_metrics_snapshot( + ts_unix_micros, + &node_id_str, + actors_count as u64, + messages_total, + actors_created, + actors_stopped, + uptime_secs, + ); + SystemResponse::Metrics { - actors_count: self.registry.count(), - messages_total: self.metrics.messages_total(), - actors_created: self.metrics.actors_created(), - actors_stopped: self.metrics.actors_stopped(), - uptime_secs: self.start_time.elapsed().as_secs(), + actors_count, + messages_total, + actors_created, + actors_stopped, + uptime_secs, } } @@ -287,13 +390,27 @@ impl SystemActor { PrometheusSystemMetrics { node_id: self.system_ref.node_id.0, actors_count: self.registry.count(), - messages_total: self.metrics.messages_total(), + messages_total: self.total_messages(), actors_created: self.metrics.actors_created(), actors_stopped: self.metrics.actors_stopped(), cluster_members: HashMap::new(), // Will be filled by caller with cluster info } } + /// Sum message counts from all actors in the system-level registry, + /// falling back to `SystemMetrics::messages_total` (system-actor only) if unavailable. + fn total_messages(&self) -> u64 { + if let Some(reg) = &self.system_registry { + let mut total = 0u64; + for entry in reg.iter_actors() { + total += entry.value().stats.message_count.load(Ordering::Relaxed); + } + total + } else { + self.metrics.messages_total() + } + } + /// Generate JSON error response fn json_error_response(&self, message: &str) -> Result { let response = SystemResponse::Error { diff --git a/crates/pulsing-actor/src/tracing/mod.rs b/crates/pulsing-actor/src/tracing/mod.rs index f45486540..0f1e41b42 100644 --- a/crates/pulsing-actor/src/tracing/mod.rs +++ b/crates/pulsing-actor/src/tracing/mod.rs @@ -30,11 +30,15 @@ use opentelemetry::trace::{ use opentelemetry::{global, Context, KeyValue}; use opentelemetry_sdk::trace::{RandomIdGenerator, Sampler, TracerProvider}; use opentelemetry_sdk::Resource; +use std::str::FromStr; + +use crate::span_store::InMemorySpanExporter; // Re-export for external use pub use opentelemetry; use std::sync::OnceLock; use tracing_opentelemetry::OpenTelemetryLayer; +pub use tracing_opentelemetry::OpenTelemetrySpanExt; use tracing_subscriber::layer::SubscriberExt; use tracing_subscriber::util::SubscriberInitExt; use tracing_subscriber::EnvFilter; @@ -47,9 +51,6 @@ static TRACING_INITIALIZED: OnceLock = OnceLock::new(); pub struct TracingConfig { /// Service name for traces pub service_name: String, - /// OTLP endpoint (e.g., "http://jaeger:4317") - /// If None, traces are only logged to console - pub otlp_endpoint: Option, /// Sampling ratio (0.0 - 1.0) pub sampling_ratio: f64, /// Enable console output @@ -62,11 +63,8 @@ impl Default for TracingConfig { fn default() -> Self { Self { service_name: "pulsing".to_string(), - otlp_endpoint: None, sampling_ratio: 1.0, console_output: true, - // Filter out HTTP request/client logs and gossip logs by default to avoid span ID spam - // Users can enable with RUST_LOG=pulsing_actor::transport=debug,pulsing_actor::cluster=debug log_filter: "info,pulsing_actor::transport::http2=warn,pulsing_actor::cluster::gossip=warn" .to_string(), @@ -97,10 +95,13 @@ pub fn init_tracing(config: TracingConfig) -> anyhow::Result<()> { config.service_name.clone(), )]); + let span_exporter = InMemorySpanExporter; + let provider = TracerProvider::builder() .with_sampler(sampler) .with_id_generator(RandomIdGenerator::default()) .with_resource(resource) + .with_simple_exporter(span_exporter) .build(); global::set_tracer_provider(provider.clone()); @@ -140,6 +141,8 @@ pub fn init_tracing(config: TracingConfig) -> anyhow::Result<()> { .map_err(|e| anyhow::anyhow!("Failed to init tracing: {}", e))?; } + crate::span_store::init_span_memtable(); + TRACING_INITIALIZED.set(true).ok(); tracing::info!(service = %config.service_name, "Distributed tracing initialized"); @@ -178,11 +181,15 @@ impl TraceContext { return None; } + let ts = span_ctx.trace_state(); + let h = ts.header(); + let trace_state = if h.is_empty() { None } else { Some(h) }; + Some(Self { trace_id: span_ctx.trace_id().to_string(), span_id: span_ctx.span_id().to_string(), trace_flags: span_ctx.trace_flags().to_u8(), - trace_state: None, // TraceState propagation handled separately if needed + trace_state, }) } @@ -215,6 +222,20 @@ impl TraceContext { }) } + /// Parse `traceparent` plus optional W3C `tracestate` header (vendor / tenant state). + /// + /// Invalid `tracestate` is ignored; a valid `traceparent` is still returned. + pub fn from_w3c_headers(traceparent: &str, tracestate: Option<&str>) -> Option { + let mut tc = Self::from_traceparent(traceparent)?; + if let Some(raw) = tracestate { + let t = raw.trim(); + if !t.is_empty() && TraceState::from_str(t).is_ok() { + tc.trace_state = Some(t.to_string()); + } + } + Some(tc) + } + /// Convert to W3C traceparent header value pub fn to_traceparent(&self) -> String { format!( @@ -225,13 +246,26 @@ impl TraceContext { /// Create OpenTelemetry Context from this TraceContext pub fn to_otel_context(&self) -> Context { + self.remote_parent_context() + } + + /// Context for an incoming W3C parent: remote span only (no merge with `Context::current()`). + /// + /// Use with [`OpenTelemetrySpanExt::set_parent`] on a new server span so OTLP exports a proper + /// child-of-remote link. + pub fn remote_parent_context(&self) -> Context { let trace_id = TraceId::from_hex(&self.trace_id).unwrap_or(TraceId::INVALID); let span_id = SpanId::from_hex(&self.span_id).unwrap_or(SpanId::INVALID); let trace_flags = TraceFlags::new(self.trace_flags); - let trace_state = TraceState::default(); + let trace_state = self + .trace_state + .as_deref() + .filter(|s| !s.is_empty()) + .and_then(|s| TraceState::from_str(s).ok()) + .unwrap_or_default(); let span_context = SpanContext::new(trace_id, span_id, trace_flags, true, trace_state); - Context::current().with_remote_span_context(span_context) + Context::new().with_remote_span_context(span_context) } /// Generate a new child span ID @@ -279,6 +313,140 @@ pub const TRACEPARENT_HEADER: &str = "traceparent"; /// W3C Trace State header name pub const TRACESTATE_HEADER: &str = "tracestate"; +/// W3C traceparent for the current span when it is a valid OpenTelemetry context. +/// +/// Use when building an [`crate::actor::Envelope`] so the actor task can parent +/// `actor.receive` to the span active at send time (e.g. `http.request`). +pub fn capture_linked_traceparent_for_mailbox() -> Option { + let cx = tracing::Span::current().context(); + let span = cx.span(); + let sc = span.span_context(); + if !sc.is_valid() { + return None; + } + Some(format!( + "00-{}-{}-{:02x}", + sc.trace_id(), + sc.span_id(), + sc.trace_flags().to_u8() + )) +} + +/// W3C `tracestate` for the current span when non-empty (paired with [`capture_linked_traceparent_for_mailbox`]). +pub fn capture_linked_tracestate_for_mailbox() -> Option { + let cx = tracing::Span::current().context(); + let span = cx.span(); + let sc = span.span_context(); + if !sc.is_valid() { + return None; + } + let h = sc.trace_state().header(); + if h.is_empty() { + None + } else { + Some(h) + } +} + +/// W3C `traceparent` for the active `tracing` span when it has a valid OpenTelemetry context. +/// +/// Returns [`None`] when there is no sampled/valid span — **do not** send a `traceparent` header +/// in that case (avoids inventing a fake trace id on the wire). +/// +/// Call after creating the client span and [`OpenTelemetrySpanExt::set_parent`], typically with +/// [`tracing::Instrument`](tracing::Instrument) so the outbound request runs inside that span. +pub fn active_span_traceparent() -> Option { + let cx = tracing::Span::current().context(); + let span = cx.span(); + let sc = span.span_context(); + if !sc.is_valid() { + return None; + } + Some(format!( + "00-{}-{}-{:02x}", + sc.trace_id(), + sc.span_id(), + sc.trace_flags().to_u8() + )) +} + +/// W3C `tracestate` for the active span when non-empty. Omit the header when [`None`]. +pub fn active_span_tracestate() -> Option { + let cx = tracing::Span::current().context(); + let span = cx.span(); + let sc = span.span_context(); + if !sc.is_valid() { + return None; + } + let h = sc.trace_state().header(); + if h.is_empty() { + None + } else { + Some(h) + } +} + +/// One message-handling span for [`Actor::receive`](crate::actor::Actor::receive). +/// +/// When `linked_traceparent` is [`Some`], parents to that W3C context (remote parent), typically +/// captured into [`crate::actor::Envelope`] at send time. Otherwise uses [`Context::current`]. +/// +/// For propagating an external W3C parent into Rust `tracing` (e.g. HTTP `traceparent`). +/// When `traceparent` is invalid or missing, returns [`tracing::Span::none`]. +pub fn span_linked_to_traceparent(traceparent: Option<&str>) -> tracing::Span { + let Some(tp) = traceparent.filter(|s| !s.is_empty()) else { + return tracing::Span::none(); + }; + let Some(tc) = TraceContext::from_traceparent(tp) else { + return tracing::Span::none(); + }; + let parent = tc.remote_parent_context(); + let span = tracing::debug_span!("pulsing.boundary", otel.name = "pulsing.boundary",); + OpenTelemetrySpanExt::set_parent(&span, parent); + span +} + +/// Child of [`tracing::Span::current()`] for Python → Rust actor calls (PyO3), without Python OTel. +pub fn span_python_actor_boundary(operation: &'static str) -> tracing::Span { + let parent = tracing::Span::current(); + tracing::debug_span!( + parent: &parent, + "pulsing.py.boundary", + otel.name = %format!("pulsing.python {}", operation), + pulsing.op = operation, + ) +} + +pub fn actor_receive_span( + actor_name: &str, + msg_type: &str, + linked_traceparent: Option<&str>, + linked_tracestate: Option<&str>, +) -> tracing::Span { + let parent = match linked_traceparent { + Some(tp) => match TraceContext::from_traceparent(tp) { + Some(mut t) => { + if let Some(ts) = linked_tracestate.filter(|s| !s.is_empty()) { + if TraceState::from_str(ts).is_ok() { + t.trace_state = Some(ts.to_string()); + } + } + t.remote_parent_context() + } + None => Context::current(), + }, + None => Context::current(), + }; + let span = tracing::info_span!( + "actor.receive", + otel.name = %format!("actor.receive {}", actor_name), + actor.name = %actor_name, + message.type = %msg_type, + ); + OpenTelemetrySpanExt::set_parent(&span, parent); + span +} + // ============================================================================ // Span Helpers // ============================================================================ @@ -305,19 +473,6 @@ macro_rules! actor_span { }; } -/// Create a span for remote actor call -#[macro_export] -macro_rules! remote_call_span { - ($target_addr:expr, $path:expr) => { - tracing::info_span!( - "actor.remote_call", - otel.name = %format!("actor.remote_call {}", $path), - target.addr = %$target_addr, - target.path = %$path, - ) - }; -} - // ============================================================================ // Tests // ============================================================================ @@ -380,4 +535,38 @@ mod tests { assert!(TraceContext::from_traceparent("invalid").is_none()); assert!(TraceContext::from_traceparent("00-short-id-01").is_none()); } + + #[test] + fn test_actor_receive_span_with_linked_traceparent() { + let tp = "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01"; + let span = super::actor_receive_span("a", "m", Some(tp), None); + drop(span); + } + + #[test] + fn test_actor_receive_span_with_linked_tracestate() { + let tp = "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01"; + let ts = "vendor1=opaque1"; + let span = super::actor_receive_span("a", "m", Some(tp), Some(ts)); + drop(span); + } + + #[test] + fn test_span_linked_to_traceparent_none_on_invalid() { + assert!(super::span_linked_to_traceparent(None).is_disabled()); + assert!(super::span_linked_to_traceparent(Some("bad")).is_disabled()); + } + + #[test] + fn test_span_linked_to_traceparent_valid() { + let tp = "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01"; + let span = super::span_linked_to_traceparent(Some(tp)); + let _g = span.enter(); + } + + #[test] + fn test_span_python_actor_boundary() { + let span = super::span_python_actor_boundary("test.op"); + let _g = span.enter(); + } } diff --git a/crates/pulsing-actor/src/transport/http2/client.rs b/crates/pulsing-actor/src/transport/http2/client.rs index 75b81cc02..2408c384a 100644 --- a/crates/pulsing-actor/src/transport/http2/client.rs +++ b/crates/pulsing-actor/src/transport/http2/client.rs @@ -7,18 +7,23 @@ use super::stream::{BinaryFrameParser, StreamFrame, StreamHandle}; use super::{headers, MessageMode, RequestType}; use crate::actor::{Message, MessageStream}; use crate::error::{PulsingError, Result, RuntimeError}; -use crate::tracing::{TraceContext, TRACEPARENT_HEADER}; +use crate::tracing::{ + active_span_traceparent, active_span_tracestate, OpenTelemetrySpanExt, TRACEPARENT_HEADER, + TRACESTATE_HEADER, +}; use bytes::Bytes; use futures::{Stream, StreamExt, TryStreamExt}; use http_body_util::{BodyExt, Full, StreamBody}; use hyper::body::{Frame, Incoming}; use hyper::{Method, Request}; +use opentelemetry::Context; use std::convert::Infallible; use std::net::SocketAddr; use std::sync::Arc; use std::time::Duration; use tokio::sync::Mutex; use tokio_util::sync::CancellationToken; +use tracing::Instrument; /// Context for fault injection (testing / chaos). #[derive(Clone, Debug)] @@ -346,20 +351,16 @@ impl Http2Client { use hyper_util::rt::{TokioExecutor, TokioIo}; use tokio::net::TcpStream; - // Create trace context for outgoing streaming request - let trace_ctx = TraceContext::from_current() - .map(|p| p.child()) - .unwrap_or_default(); - + let parent_otel = Context::current(); let span = tracing::info_span!( "http.client.stream", otel.name = %format!("POST {} (stream)", path), http.method = "POST", http.url = %format!("http://{}{}", addr, path), - trace_id = %trace_ctx.trace_id, - span_id = %trace_ctx.span_id, ); - let _enter = span.enter(); + span.set_parent(parent_otel); + let (trace_header, trace_state_header) = + span.in_scope(|| (active_span_traceparent(), active_span_tracestate())); // Create a dedicated connection for streaming request let tcp_stream = @@ -430,18 +431,22 @@ impl Http2Client { let body = StreamBody::new(body_stream); let uri = format!("http://{}{}", addr, path); - let request = Request::builder() + let mut req_b = Request::builder() .method(Method::POST) .uri(&uri) .header(headers::MESSAGE_MODE, MessageMode::Ask.as_str()) .header(headers::MESSAGE_TYPE, msg_type) .header(headers::REQUEST_TYPE, RequestType::Stream.as_str()) - .header(TRACEPARENT_HEADER, trace_ctx.to_traceparent()) - .header("content-type", "application/octet-stream") - .body(body) - .map_err(|e| { - RuntimeError::protocol_error(format!("Failed to build request: {}", e)) - })?; + .header("content-type", "application/octet-stream"); + if let Some(ref h) = trace_header { + req_b = req_b.header(TRACEPARENT_HEADER, h); + } + if let Some(ref h) = trace_state_header { + req_b = req_b.header(TRACESTATE_HEADER, h); + } + let request = req_b.body(body).map_err(|e| { + RuntimeError::protocol_error(format!("Failed to build request: {}", e)) + })?; let send_future = sender.send_request(request); let response = tokio::time::timeout(self.config.stream_timeout, send_future) @@ -510,14 +515,20 @@ impl Http2Client { // Build request with streaming body and trace context let uri = format!("http://{}{}", addr, path); - let request = Request::builder() + let mut req_b = Request::builder() .method(Method::POST) .uri(&uri) .header(headers::MESSAGE_MODE, MessageMode::Ask.as_str()) .header(headers::MESSAGE_TYPE, msg_type) .header(headers::REQUEST_TYPE, RequestType::Stream.as_str()) - .header(TRACEPARENT_HEADER, trace_ctx.to_traceparent()) - .header("content-type", "application/octet-stream") + .header("content-type", "application/octet-stream"); + if let Some(ref h) = trace_header { + req_b = req_b.header(TRACEPARENT_HEADER, h); + } + if let Some(ref h) = trace_state_header { + req_b = req_b.header(TRACESTATE_HEADER, h); + } + let request = req_b .body(body) .map_err(|e| RuntimeError::protocol_error(format!("Failed to build request: {}", e)))?; @@ -641,45 +652,53 @@ impl Http2Client { payload: Vec, mode: MessageMode, ) -> Result> { - let trace_ctx = TraceContext::from_current() - .map(|p| p.child()) - .unwrap_or_default(); - + let this = self.clone(); + let path = path.to_string(); + let msg_type = msg_type.to_string(); + let parent_otel = Context::current(); let span = tracing::info_span!( "http.client", otel.name = %format!("POST {}", path), http.method = "POST", http.url = %format!("http://{}{}", addr, path), - trace_id = %trace_ctx.trace_id, - span_id = %trace_ctx.span_id, ); - let _enter = span.enter(); + span.set_parent(parent_otel); - let conn_guard = self.pool.get_connection(addr).await?; - let mut conn = conn_guard.get().await; + async move { + let trace_header = active_span_traceparent(); + let trace_state_header = active_span_tracestate(); + let conn_guard = this.pool.get_connection(addr).await?; + let mut conn = conn_guard.get().await; - // Build request with trace context header - let uri = format!("http://{}{}", addr, path); - let request = Request::builder() - .method(Method::POST) - .uri(&uri) - .header(headers::MESSAGE_MODE, mode.as_str()) - .header(headers::MESSAGE_TYPE, msg_type) - .header(TRACEPARENT_HEADER, trace_ctx.to_traceparent()) - .header("content-type", "application/octet-stream") - .body(Full::new(Bytes::from(payload))) - .map_err(|e| RuntimeError::protocol_error(format!("Failed to build request: {}", e)))?; + let uri = format!("http://{}{}", addr, path); + let mut builder = Request::builder() + .method(Method::POST) + .uri(&uri) + .header(headers::MESSAGE_MODE, mode.as_str()) + .header(headers::MESSAGE_TYPE, &msg_type) + .header("content-type", "application/octet-stream"); + if let Some(ref h) = trace_header { + builder = builder.header(TRACEPARENT_HEADER, h); + } + if let Some(ref h) = trace_state_header { + builder = builder.header(TRACESTATE_HEADER, h); + } + let request = builder.body(Full::new(Bytes::from(payload))).map_err(|e| { + RuntimeError::protocol_error(format!("Failed to build request: {}", e)) + })?; - // Send request with timeout - let send_future = conn.sender.send_request(request); - let response = tokio::time::timeout(self.config.request_timeout, send_future) - .await - .map_err(|_| { - RuntimeError::request_timeout(self.config.request_timeout.as_millis() as u64) - })? - .map_err(|e| RuntimeError::protocol_error(format!("Request failed: {}", e)))?; + let send_future = conn.sender.send_request(request); + let response = tokio::time::timeout(this.config.request_timeout, send_future) + .await + .map_err(|_| { + RuntimeError::request_timeout(this.config.request_timeout.as_millis() as u64) + })? + .map_err(|e| RuntimeError::protocol_error(format!("Request failed: {}", e)))?; - Ok(response) + Ok(response) + } + .instrument(span) + .await } } diff --git a/crates/pulsing-actor/src/transport/http2/server.rs b/crates/pulsing-actor/src/transport/http2/server.rs index 51d46f878..a9c3b4270 100644 --- a/crates/pulsing-actor/src/transport/http2/server.rs +++ b/crates/pulsing-actor/src/transport/http2/server.rs @@ -5,7 +5,7 @@ use super::stream::{BinaryFrameParser, StreamFrame}; use super::{headers, MessageMode, RequestType}; use crate::actor::Message; use crate::error::{PulsingError, Result, RuntimeError}; -use crate::tracing::{TraceContext, TRACEPARENT_HEADER}; +use crate::tracing::{OpenTelemetrySpanExt, TraceContext, TRACEPARENT_HEADER, TRACESTATE_HEADER}; use bytes::Bytes; use futures::StreamExt; use http_body_util::{BodyExt, Full, StreamBody}; @@ -21,6 +21,7 @@ use tokio::io::{AsyncRead, AsyncWrite}; use tokio::net::TcpListener; use tokio::sync::mpsc; use tokio_util::sync::CancellationToken; +use tracing::Instrument; /// Handler trait for HTTP/2 server. #[async_trait::async_trait] @@ -275,30 +276,32 @@ impl Http2Server { let path = req.uri().path().to_string(); let method = req.method().clone(); - // Extract trace context from incoming request - let parent_trace = req + // Link to upstream trace: W3C traceparent + optional tracestate → remote parent → child span + let tp = req .headers() .get(TRACEPARENT_HEADER) - .and_then(|v| v.to_str().ok()) - .and_then(TraceContext::from_traceparent); + .and_then(|v| v.to_str().ok()); + let ts = req + .headers() + .get(TRACESTATE_HEADER) + .and_then(|v| v.to_str().ok()); + let parent_trace = tp.and_then(|tp| TraceContext::from_w3c_headers(tp, ts)); - // Create a child span ID for this request - let trace_ctx = parent_trace.map(|p| p.child()).unwrap_or_default(); + let parent_otel = parent_trace + .as_ref() + .map(TraceContext::remote_parent_context) + .unwrap_or_default(); - // Create span with trace context let span = tracing::info_span!( "http.request", otel.name = %format!("{} {}", method, path), http.method = %method, http.route = %path, http.peer = %peer_addr, - trace_id = %trace_ctx.trace_id, - span_id = %trace_ctx.span_id, ); + span.set_parent(parent_otel); - // Enter the span for the duration of request handling - let _enter = span.enter(); - + async move { // Health check endpoint if path == "/health" && method == Method::GET { let health = handler.health_check().await; @@ -483,6 +486,9 @@ impl Http2Server { } } } + } + .instrument(span) + .await } /// Parse a streaming request body (binary frames) into Message::Stream diff --git a/crates/pulsing-actor/tests/unit/actor/mailbox_tests.rs b/crates/pulsing-actor/tests/unit/actor/mailbox_tests.rs index 68e6be413..9bbf7a9b5 100644 --- a/crates/pulsing-actor/tests/unit/actor/mailbox_tests.rs +++ b/crates/pulsing-actor/tests/unit/actor/mailbox_tests.rs @@ -39,7 +39,7 @@ async fn test_mailbox_send_receive() { assert_eq!(received.msg_type(), "test"); // Extract payload from message - let (message, _) = received.into_parts(); + let (message, _, _, _) = received.into_parts(); if let Message::Single { data, .. } = message { assert_eq!(data, vec![1, 2, 3]); } else { @@ -57,7 +57,7 @@ async fn test_mailbox_ask_pattern() { let receiver_task = tokio::spawn(async move { if let Some(envelope) = receiver.recv().await { assert!(envelope.expects_response()); - let (_, responder) = envelope.into_parts(); + let (_, responder, _, _) = envelope.into_parts(); responder.send(Ok(Message::single("", vec![42, 43, 44]))); } }); @@ -78,6 +78,26 @@ async fn test_mailbox_ask_pattern() { receiver_task.await.unwrap(); } +#[tokio::test] +async fn test_envelope_linked_traceparent_roundtrip() { + let tp = "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01".to_string(); + let msg = Message::single("t", vec![]); + let env = Envelope::tell(msg).with_linked_traceparent(Some(tp.clone())); + let (_, _, got_tp, got_ts) = env.into_parts(); + assert_eq!(got_tp.as_deref(), Some(tp.as_str())); + assert!(got_ts.is_none()); +} + +#[tokio::test] +async fn test_envelope_linked_tracestate_roundtrip() { + let ts = "vendor1=opaque1".to_string(); + let msg = Message::single("t", vec![]); + let env = Envelope::tell(msg).with_linked_tracestate(Some(ts.clone())); + let (_, _, got_tp, got_ts) = env.into_parts(); + assert!(got_tp.is_none()); + assert_eq!(got_ts.as_deref(), Some(ts.as_str())); +} + #[tokio::test] async fn test_mailbox_closed_detection() { let mut mailbox = Mailbox::new(); @@ -275,7 +295,7 @@ async fn test_mailbox_no_message_loss() { // Receiver collects all message IDs let receiver_task = tokio::spawn(async move { while let Some(envelope) = receiver.recv().await { - let (message, _) = envelope.into_parts(); + let (message, _, _, _) = envelope.into_parts(); if let Message::Single { data, .. } = message { if data.len() >= 4 { let id = u32::from_le_bytes([data[0], data[1], data[2], data[3]]); @@ -321,7 +341,7 @@ async fn test_mailbox_response_delivery() { let receiver_task = tokio::spawn(async move { while let Some(envelope) = receiver.recv().await { let expects = envelope.expects_response(); - let (message, responder) = envelope.into_parts(); + let (message, responder, _, _) = envelope.into_parts(); if expects { // Echo back the payload doubled if let Message::Single { ref data, .. } = message { @@ -371,7 +391,7 @@ async fn test_mailbox_empty_payload() { let received = receiver.recv().await.unwrap(); assert_eq!(received.msg_type(), "empty"); - let (message, _) = received.into_parts(); + let (message, _, _, _) = received.into_parts(); if let Message::Single { data, .. } = message { assert!(data.is_empty()); } else { @@ -391,7 +411,7 @@ async fn test_mailbox_large_payload() { sender.send(envelope).await.unwrap(); let received = receiver.recv().await.unwrap(); - let (message, _) = received.into_parts(); + let (message, _, _, _) = received.into_parts(); if let Message::Single { data, .. } = message { assert_eq!(data.len(), 1_000_000); diff --git a/crates/pulsing-actor/tests/unit/system/system_actor_tests.rs b/crates/pulsing-actor/tests/unit/system/system_actor_tests.rs index d02b7c47d..8637e78f1 100644 --- a/crates/pulsing-actor/tests/unit/system/system_actor_tests.rs +++ b/crates/pulsing-actor/tests/unit/system/system_actor_tests.rs @@ -295,6 +295,23 @@ async fn test_system_actor_get_metrics() { system.shutdown().await.unwrap(); } +#[tokio::test] +async fn test_performance_history_records_get_metrics() { + let system = create_test_system().await; + let sys_ref = system.system().await.unwrap(); + + for _ in 0..3 { + let msg = create_system_message(&SystemMessage::GetMetrics); + let _ = sys_ref.send(msg).await.unwrap(); + } + + let hist = system.performance_recent(10); + assert!(hist.len() >= 3); + assert_eq!(hist[0].actors_count as usize, hist[1].actors_count as usize); + + system.shutdown().await.unwrap(); +} + #[tokio::test] async fn test_system_actor_list_actors() { let system = create_test_system().await; @@ -451,6 +468,19 @@ fn test_actor_registry_list_all() { assert_eq!(actors.len(), 2); } +#[test] +fn test_actor_registry_metadata_roundtrip() { + let registry = ActorRegistry::new(); + let mut meta = std::collections::HashMap::new(); + meta.insert("class".to_string(), "MyActor".to_string()); + meta.insert("module".to_string(), "my.mod".to_string()); + registry.register_with_metadata("a1", ActorId::generate(), "MyActor", meta.clone()); + + let info = registry.get_info("a1").unwrap(); + assert_eq!(info.metadata.get("class"), Some(&"MyActor".to_string())); + assert_eq!(info.metadata.get("module"), Some(&"my.mod".to_string())); +} + #[test] fn test_actor_registry_get_not_found() { let registry = ActorRegistry::new(); diff --git a/crates/pulsing-py/Cargo.toml b/crates/pulsing-py/Cargo.toml index 4bf0e8f48..4e4d6d9f5 100644 --- a/crates/pulsing-py/Cargo.toml +++ b/crates/pulsing-py/Cargo.toml @@ -25,7 +25,6 @@ bincode = { workspace = true } thiserror = { workspace = true } tokio = { workspace = true } tracing = { workspace = true } -tracing-subscriber = { workspace = true } reqwest = { workspace = true } pythonize = "0.23" uuid = { workspace = true } diff --git a/crates/pulsing-py/src/actor.rs b/crates/pulsing-py/src/actor.rs index 1d4039deb..e8da2eaa7 100644 --- a/crates/pulsing-py/src/actor.rs +++ b/crates/pulsing-py/src/actor.rs @@ -6,7 +6,7 @@ use pulsing_actor::prelude::*; use pulsing_actor::supervision::{BackoffStrategy, RestartPolicy, SupervisionSpec}; use pyo3::exceptions::{PyRuntimeError, PyStopAsyncIteration, PyValueError}; use pyo3::prelude::*; -use pyo3::types::PyBytes; +use pyo3::types::{PyBytes, PyDict}; use serde::{Deserialize, Serialize}; use std::cmp::min; use std::net::SocketAddr; @@ -18,6 +18,18 @@ use tokio::sync::Mutex as TokioMutex; use crate::errors::pulsing_error_to_py_err; use crate::python_error_converter::convert_python_exception_to_actor_error; use crate::python_executor::python_executor; +use pulsing_actor::tracing::{ + capture_linked_traceparent_for_mailbox, capture_linked_tracestate_for_mailbox, +}; + +/// Read a ContextVar from `pulsing.core.remote` by attribute name. +/// Returns `None` on any failure (module not loaded, var not set, etc.). +fn read_python_contextvar(py: Python<'_>, var_name: &str) -> Option { + let module = py.import("pulsing.core.remote").ok()?; + let cv = module.getattr(var_name).ok()?; + let val = cv.call_method0("get").ok()?; + val.extract::>().ok().flatten() +} /// Special message type identifier for pickle-encoded Python objects const SEALED_PY_MSG_TYPE: &str = "__sealed_py_message__"; @@ -1066,7 +1078,6 @@ impl PyActorRef { fn ask<'py>(&self, py: Python<'py>, msg: PyObject) -> PyResult> { let actor_ref = self.inner.clone(); - // Check if msg is already a PyMessage let msg_bound = msg.bind(py); let actor_msg = if msg_bound.is_instance_of::() { let py_msg: PyMessage = msg_bound.extract()?; @@ -1075,8 +1086,18 @@ impl PyActorRef { encode_python_payload(py, &msg)? }; + // Read trace context from Python ContextVar (set by _WrappedActor.receive), + // fall back to Rust tracing span for pure-Rust callers. + let tp = read_python_contextvar(py, "_current_traceparent") + .or_else(capture_linked_traceparent_for_mailbox); + let ts = read_python_contextvar(py, "_current_tracestate") + .or_else(capture_linked_tracestate_for_mailbox); + pyo3_async_runtimes::tokio::future_into_py(py, async move { - let response = actor_ref.send(actor_msg).await.map_err(to_pyerr)?; + let response = actor_ref + .send_with_trace(actor_msg, tp, ts) + .await + .map_err(to_pyerr)?; decode_message_to_pyobject(response).await }) } @@ -1089,7 +1110,6 @@ impl PyActorRef { fn tell<'py>(&self, py: Python<'py>, msg: PyObject) -> PyResult> { let actor_ref = self.inner.clone(); - // Check if msg is already a PyMessage let msg_bound = msg.bind(py); let actor_msg = if msg_bound.is_instance_of::() { let py_msg: PyMessage = msg_bound.extract()?; @@ -1098,8 +1118,16 @@ impl PyActorRef { encode_python_payload(py, &msg)? }; + let tp = read_python_contextvar(py, "_current_traceparent") + .or_else(capture_linked_traceparent_for_mailbox); + let ts = read_python_contextvar(py, "_current_tracestate") + .or_else(capture_linked_tracestate_for_mailbox); + pyo3_async_runtimes::tokio::future_into_py(py, async move { - actor_ref.send_oneway(actor_msg).await.map_err(to_pyerr)?; + actor_ref + .send_oneway_with_trace(actor_msg, tp, ts) + .await + .map_err(to_pyerr)?; Ok(()) }) } @@ -1377,10 +1405,14 @@ impl Actor for PythonActorWrapper { msg: Message, _ctx: &mut ActorContext, ) -> pulsing_actor::error::Result { + // Capture trace context BEFORE any .await — at this point recv_span is + // still entered by Instrument::poll on the tokio thread. + let captured_tp = capture_linked_traceparent_for_mailbox(); + let captured_ts = capture_linked_tracestate_for_mailbox(); + let (handler, event_loop) = Python::with_gil(|py| (self.handler.clone_ref(py), self.event_loop.clone_ref(py))); - // Decode-first: convert any message format to a Python object let call_arg = decode_message_to_pyobject(msg).await.map_err(|e| { pulsing_actor::error::PulsingError::from(pulsing_actor::error::RuntimeError::Other( e.to_string(), @@ -1390,6 +1422,17 @@ impl Actor for PythonActorWrapper { let response: Result = python_executor() .execute(move || { Python::with_gil(|py| -> PyResult { + // Inject trace context as handler attributes so the Python + // coroutine (on the asyncio event loop) can read them. + match &captured_tp { + Some(tp) => handler.setattr(py, "__pulsing_tp__", tp.as_str())?, + None => handler.setattr(py, "__pulsing_tp__", py.None())?, + } + match &captured_ts { + Some(ts) => handler.setattr(py, "__pulsing_ts__", ts.as_str())?, + None => handler.setattr(py, "__pulsing_ts__", py.None())?, + } + let receive_method = handler.getattr(py, "receive")?; let result = match receive_method.call1(py, (&call_arg,)) { Ok(value) => value, @@ -1611,6 +1654,31 @@ impl PyActorSystem { self.inner.addr().to_string() } + /// Recent ``GetMetrics`` snapshots (newest first), for local analysis / DataFusion adapters. + /// + /// Rows are appended whenever ``system/core`` handles ``GetMetrics`` (e.g. ``get_metrics()``). + #[pyo3(signature = (limit=None))] + fn performance_history(&self, limit: Option) -> Vec> { + let limit = limit.unwrap_or(256).min(10_000); + let snaps = self.inner.performance_recent(limit); + Python::with_gil(|py| { + snaps + .into_iter() + .map(|s| { + let d = PyDict::new(py); + let _ = d.set_item("ts_unix_micros", s.ts_unix_micros); + let _ = d.set_item("node_id", s.node_id); + let _ = d.set_item("actors_count", s.actors_count); + let _ = d.set_item("messages_total", s.messages_total); + let _ = d.set_item("actors_created", s.actors_created); + let _ = d.set_item("actors_stopped", s.actors_stopped); + let _ = d.set_item("uptime_secs", s.uptime_secs); + d.unbind() + }) + .collect() + }) + } + #[pyo3(signature = ( actor, *, diff --git a/crates/pulsing-py/src/connect.rs b/crates/pulsing-py/src/connect.rs index ac3f90684..7ff6397ca 100644 --- a/crates/pulsing-py/src/connect.rs +++ b/crates/pulsing-py/src/connect.rs @@ -7,6 +7,8 @@ use std::net::SocketAddr; use std::sync::Arc; use crate::actor::PyMessage; +use pulsing_actor::tracing::span_python_actor_boundary; +use tracing::Instrument; fn to_pyerr(err: pulsing_actor::error::PulsingError) -> PyErr { crate::errors::pulsing_error_to_py_err(err) @@ -129,10 +131,15 @@ impl PyConnectActorRef { crate::actor::encode_python_payload(py, &msg)? }; - pyo3_async_runtimes::tokio::future_into_py(py, async move { - let response = actor_ref.send_message(actor_msg).await.map_err(to_pyerr)?; - crate::actor::decode_message_to_pyobject(response).await - }) + let span = span_python_actor_boundary("ConnectActorRef.ask"); + pyo3_async_runtimes::tokio::future_into_py( + py, + async move { + let response = actor_ref.send_message(actor_msg).await.map_err(to_pyerr)?; + crate::actor::decode_message_to_pyobject(response).await + } + .instrument(span), + ) } /// Fire-and-forget message. @@ -147,17 +154,22 @@ impl PyConnectActorRef { crate::actor::encode_python_payload(py, &msg)? }; - pyo3_async_runtimes::tokio::future_into_py(py, async move { - match actor_msg { - pulsing_actor::actor::Message::Single { msg_type, data } => { - actor_ref.tell(&msg_type, data).await.map_err(to_pyerr)?; - } - _ => { - return Err(PyRuntimeError::new_err("Streaming not supported for tell")); + let span = span_python_actor_boundary("ConnectActorRef.tell"); + pyo3_async_runtimes::tokio::future_into_py( + py, + async move { + match actor_msg { + pulsing_actor::actor::Message::Single { msg_type, data } => { + actor_ref.tell(&msg_type, data).await.map_err(to_pyerr)?; + } + _ => { + return Err(PyRuntimeError::new_err("Streaming not supported for tell")); + } } + Ok(()) } - Ok(()) - }) + .instrument(span), + ) } /// Get the actor path. diff --git a/crates/pulsing-py/src/lib.rs b/crates/pulsing-py/src/lib.rs index 8249c44d0..82fe95fb2 100644 --- a/crates/pulsing-py/src/lib.rs +++ b/crates/pulsing-py/src/lib.rs @@ -11,6 +11,7 @@ mod errors; mod policies; mod python_error_converter; mod python_executor; +mod tracing_py; pub use python_executor::{init_python_executor, python_executor, ExecutorError}; @@ -24,18 +25,6 @@ pub use python_executor::{init_python_executor, python_executor, ExecutorError}; /// - Load balancing policies: Random, RoundRobin, PowerOfTwo, ConsistentHash, CacheAware #[pymodule] fn _core(m: &Bound<'_, PyModule>) -> PyResult<()> { - // Initialize tracing for logging (only if PULSING_INIT_TRACING is set) - // This allows applications to control their own tracing configuration - if std::env::var("PULSING_INIT_TRACING").is_ok() { - tracing_subscriber::fmt() - .with_env_filter( - tracing_subscriber::EnvFilter::from_default_env() - .add_directive(tracing::Level::INFO.into()), - ) - .try_init() - .ok(); - } - // Add error classes errors::add_to_module(m)?; @@ -48,6 +37,8 @@ fn _core(m: &Bound<'_, PyModule>) -> PyResult<()> { // Add out-cluster connect connect::add_to_module(m)?; + tracing_py::add_to_module(m)?; + // Add version m.add("__version__", env!("CARGO_PKG_VERSION"))?; diff --git a/crates/pulsing-py/src/tracing_py.rs b/crates/pulsing-py/src/tracing_py.rs new file mode 100644 index 000000000..e1d4cd46c --- /dev/null +++ b/crates/pulsing-py/src/tracing_py.rs @@ -0,0 +1,33 @@ +//! Rust-side OpenTelemetry / `tracing` init for Python processes. + +use pulsing_actor::tracing::{init_tracing, shutdown_tracing, TracingConfig}; +use pyo3::exceptions::PyRuntimeError; +use pyo3::prelude::*; + +#[pyfunction] +#[pyo3(signature = (service_name=None, console_output=None))] +fn init_distributed_tracing( + service_name: Option, + console_output: Option, +) -> PyResult<()> { + let mut cfg = TracingConfig::default(); + if let Some(s) = service_name { + cfg.service_name = s; + } + if let Some(c) = console_output { + cfg.console_output = c; + } + init_tracing(cfg).map_err(|e| PyRuntimeError::new_err(format!("{e}")))?; + Ok(()) +} + +#[pyfunction] +fn shutdown_distributed_tracing() { + shutdown_tracing(); +} + +pub(crate) fn add_to_module(m: &Bound<'_, PyModule>) -> PyResult<()> { + m.add_function(wrap_pyfunction!(init_distributed_tracing, m)?)?; + m.add_function(wrap_pyfunction!(shutdown_distributed_tracing, m)?)?; + Ok(()) +} diff --git a/examples/python/probing_dashboard_smoke.py b/examples/python/probing_dashboard_smoke.py new file mode 100644 index 000000000..af78d41ca --- /dev/null +++ b/examples/python/probing_dashboard_smoke.py @@ -0,0 +1,252 @@ +#!/usr/bin/env python3 +""" +Pulsing + probing dashboard demo with cascading actor messages. + +Actors randomly forward messages to other actors, creating variable-length +call chains visible in the Trace Timeline. + +Quick SQL check (no HTTP server):: + + uv run python examples/python/probing_dashboard_smoke.py --once + +Live demo:: + + uv run python examples/python/probing_dashboard_smoke.py --port 8765 + +Then open http://127.0.0.1:8765/pulsing to see the Trace Timeline. +Press Ctrl+C to stop. +""" + +from __future__ import annotations + +import argparse +import asyncio +import json +import os +import random +import sys + + +def _row_estimate(payload: dict) -> int: + size = payload.get("size") + if isinstance(size, int) and size > 0: + return size + cols = payload.get("cols") or [] + if not cols: + return 0 + c0 = cols[0] + if isinstance(c0, dict): + for _tag, seq in c0.items(): + if isinstance(seq, list): + return len(seq) + return 0 + + +def _summarize_df(payload: dict) -> str: + names = payload.get("names") or [] + return f"columns={names!r} rows≈{_row_estimate(payload)}" + + +def _parse_args() -> argparse.Namespace: + p = argparse.ArgumentParser(description=__doc__) + p.add_argument( + "--once", + action="store_true", + help="Run a quick SQL smoke test and exit (no HTTP server)", + ) + p.add_argument( + "--port", type=int, default=8765, help="Probing HTTP port. Default: 8765" + ) + p.add_argument("--actors", type=int, default=6, help="Number of actors. Default: 6") + p.add_argument( + "--interval", + type=float, + default=2.0, + help="Seconds between message bursts. Default: 2", + ) + p.add_argument( + "--max-depth", type=int, default=4, help="Max forwarding depth. Default: 4" + ) + return p.parse_args() + + +async def _run_once() -> int: + try: + import probing # noqa: F401 + except ImportError: + print("error: probing is not installed.", file=sys.stderr) + return 1 + + import pulsing as pul + from pulsing.integrations.probing import ( + refresh_probing_tables_async, + stop_probing_integration, + ) + + @pul.remote + class _SmokeActor: + def __init__(self) -> None: + pass + + def ping(self) -> str: + return "ok" + + stop_probing_integration() + await pul.init() + try: + await _SmokeActor.spawn(name="actors/smoke_demo") + ok = await refresh_probing_tables_async() + if not ok: + print( + "error: refresh_probing_tables_async() returned False", file=sys.stderr + ) + return 2 + + from probing import _core + + queries = [ + ("pulsing.members", "SELECT * FROM pulsing.members LIMIT 20"), + ("pulsing.metrics", "SELECT * FROM pulsing.metrics LIMIT 30"), + ("pulsing.actors", "SELECT * FROM pulsing.actors LIMIT 30"), + ("pulsing.spans", "SELECT * FROM pulsing.spans LIMIT 30"), + ] + + print("Pulsing ↔ probing smoke: OK\n") + for title, sql in queries: + raw = _core.query_json(sql) + payload = json.loads(raw) + print(f"--- {title} ---") + print(_summarize_df(payload)) + cols = payload.get("cols") or [] + if cols and isinstance(cols[0], dict): + for tag, seq in cols[0].items(): + if isinstance(seq, list) and seq: + print(f"sample (first column {tag!r}): {seq[:3]!r}") + break + print() + return 0 + finally: + await pul.shutdown() + + +async def _run_live(args: argparse.Namespace) -> int: + try: + import probing # noqa: F401 + except ImportError: + print("error: probing is not installed.", file=sys.stderr) + return 1 + + import pulsing as pul + from pulsing.integrations.probing import stop_probing_integration + + # Registry: all actor proxies indexed by name, shared with the actor class. + actor_names: list[str] = [] + actor_proxies: dict[str, object] = {} + + @pul.remote + class CascadeActor: + """Actor that randomly forwards messages to peers, creating call chains.""" + + def __init__(self, label: str = "", max_depth: int = 4) -> None: + self.label = label + self.max_depth = max_depth + + async def cascade(self, depth: int = 0, origin: str = "") -> str: + """Receive a message and optionally forward to random peers.""" + # Simulate own computation (counted as self time) + await asyncio.sleep(random.uniform(0, 0.5)) + + if depth >= self.max_depth: + return f"{self.label}: leaf (depth={depth})" + + peers = [n for n in actor_names if n != self.label] + fan_out = random.randint(0, min(2, len(peers))) + targets = random.sample(peers, fan_out) if fan_out > 0 else [] + + results = [] + for t in targets: + proxy = actor_proxies.get(t) + if proxy is not None: + try: + r = await proxy.cascade(depth=depth + 1, origin=self.label) + results.append(r) + except Exception: + pass + + return f"{self.label}: forwarded to {len(results)} peers at depth={depth}" + + def ping(self) -> str: + return f"{self.label}: ok" + + stop_probing_integration() + await pul.init() + + try: + # Spawn actors + for i in range(args.actors): + name = f"actors/node_{i}" + proxy = await CascadeActor.spawn( + name=name, + label=name, + max_depth=args.max_depth, + ) + actor_names.append(name) + actor_proxies[name] = proxy + + async def _message_burst() -> None: + """Periodically start cascading messages from random actors.""" + while True: + # Pick 1–3 random actors to initiate a cascade + initiators = random.sample( + actor_names, min(random.randint(1, 3), len(actor_names)) + ) + for name in initiators: + proxy = actor_proxies.get(name) + if proxy is not None: + try: + await proxy.cascade(depth=0, origin="external") + except Exception: + pass + await asyncio.sleep(args.interval) + + burst_task = asyncio.create_task(_message_burst(), name="cascade-burst") + + host = "127.0.0.1" + print() + print("Pulsing + probing cascade demo running.") + print(f" Actors: {args.actors}, max chain depth: {args.max_depth}") + print(f" Web UI: http://{host}:{args.port}/pulsing") + print(f" SQL: http://{host}:{args.port}/analytics") + print() + print("Press Ctrl+C to stop.") + print() + + try: + await asyncio.Event().wait() + except asyncio.CancelledError: + pass + return 0 + finally: + burst_task.cancel() + try: + await burst_task + except asyncio.CancelledError: + pass + await pul.shutdown() + + +def main() -> None: + args = _parse_args() + if args.once: + raise SystemExit(asyncio.run(_run_once())) + + os.environ.setdefault("PROBING_PORT", str(args.port)) + try: + raise SystemExit(asyncio.run(_run_live(args))) + except KeyboardInterrupt: + print("\nStopped.") + raise SystemExit(0) from None + + +if __name__ == "__main__": + main() diff --git a/python/pulsing/core/__init__.py b/python/pulsing/core/__init__.py index 0ae6b3395..45093556c 100644 --- a/python/pulsing/core/__init__.py +++ b/python/pulsing/core/__init__.py @@ -18,6 +18,7 @@ def incr(self): self.value += 1; return self.value """ import asyncio +import os from pulsing._core import ( ActorId, @@ -28,6 +29,8 @@ def incr(self): self.value += 1; return self.value StreamWriter, SystemConfig, ZeroCopyDescriptor, + init_distributed_tracing, + shutdown_distributed_tracing, ) from .messaging import ( Message, @@ -41,6 +44,22 @@ def incr(self): self.value += 1; return self.value _global_system: ActorSystem = None +def _init_rust_tracing_from_env() -> None: + """Enable Rust `tracing` + OpenTelemetry layer when ``PULSING_TRACING`` is truthy.""" + if os.environ.get("PULSING_TRACING", "").lower() not in ("1", "true", "yes"): + return + console = os.environ.get("PULSING_TRACING_CONSOLE", "1").lower() not in ( + "0", + "false", + "no", + ) + service = os.environ.get("PULSING_SERVICE_NAME") or None + init_distributed_tracing( + service_name=service, + console_output=console, + ) + + async def init( addr: str = None, *, @@ -58,6 +77,14 @@ async def init( head_addr: Address of head node (worker mode). Mutually exclusive with is_head_node. is_head_node: If True, this node runs as head. Mutually exclusive with head_addr. + Tracing: + Set ``PULSING_TRACING=1`` to install the Rust subscriber (console + OTLP-ready layer). + Optional: ``PULSING_OTLP_ENDPOINT``, ``PULSING_SERVICE_NAME``, ``PULSING_TRACING_CONSOLE``, + ``PULSING_SPAN_HISTORY_CAPACITY`` (max retained completed spans, default 8192). + + If the ``probing`` package is installed, ``PULSING_PROBING_AUTO_TRACING`` (default ``1``) may + install silent Rust tracing so ``pulsing.spans`` gets populated. + Returns: ActorSystem instance """ @@ -66,6 +93,17 @@ async def init( if _global_system is not None: return _global_system + _init_rust_tracing_from_env() + + try: + import probing # noqa: F401 + + from pulsing.integrations.probing import ensure_tracing_for_span_capture + + ensure_tracing_for_span_capture() + except ImportError: + pass + if is_head_node and head_addr: raise ValueError("Cannot set both is_head_node and head_addr") @@ -98,6 +136,13 @@ async def init( except ImportError: pass + try: + from pulsing.integrations.probing import start_probing_integration + + start_probing_integration() + except ImportError: + pass + return _global_system @@ -105,10 +150,19 @@ async def shutdown() -> None: """Shutdown the global actor system""" global _global_system + try: + from pulsing.integrations.probing import stop_probing_integration + + stop_probing_integration() + except ImportError: + pass + if _global_system is not None: await _global_system.shutdown() _global_system = None + shutdown_distributed_tracing() + def get_system() -> ActorSystem: """Get the global actor system (must call init() first)""" @@ -153,6 +207,8 @@ def is_initialized() -> bool: __all__ = [ "init", "shutdown", + "init_distributed_tracing", + "shutdown_distributed_tracing", "remote", "resolve", "mount", diff --git a/python/pulsing/core/remote.py b/python/pulsing/core/remote.py index 98551a015..f03e8ff81 100644 --- a/python/pulsing/core/remote.py +++ b/python/pulsing/core/remote.py @@ -1,6 +1,7 @@ """@remote decorator, ActorClass, and actor lifecycle management.""" import asyncio +import contextvars import inspect import logging import random @@ -24,6 +25,16 @@ T = TypeVar("T") +# Trace context ContextVars — set by _WrappedActor.receive (from Rust handler +# attributes), read by PyO3 ask()/tell() to propagate W3C traceparent across +# actor-to-actor calls within the same asyncio Task. +_current_traceparent: contextvars.ContextVar[str | None] = contextvars.ContextVar( + "_pulsing_tp", default=None +) +_current_tracestate: contextvars.ContextVar[str | None] = contextvars.ContextVar( + "_pulsing_ts", default=None +) + # ============================================================================ # Actor base class # ============================================================================ @@ -164,6 +175,10 @@ def metadata(self) -> dict[str, str]: return {} async def receive(self, msg) -> Any: + # Propagate trace context injected by Rust PythonActorWrapper::receive + _current_traceparent.set(getattr(self, "__pulsing_tp__", None)) + _current_tracestate.set(getattr(self, "__pulsing_ts__", None)) + if isinstance(msg, dict): method, args, kwargs, is_async_call = _unwrap_call(msg) diff --git a/python/pulsing/integrations/__init__.py b/python/pulsing/integrations/__init__.py index e653774d4..3149ee541 100644 --- a/python/pulsing/integrations/__init__.py +++ b/python/pulsing/integrations/__init__.py @@ -1 +1,15 @@ """Third-party framework integrations.""" + +from pulsing.integrations.probing import ( + ensure_tracing_for_span_capture, + refresh_probing_tables_async, + start_probing_integration, + stop_probing_integration, +) + +__all__ = [ + "ensure_tracing_for_span_capture", + "refresh_probing_tables_async", + "start_probing_integration", + "stop_probing_integration", +] diff --git a/python/pulsing/integrations/probing.py b/python/pulsing/integrations/probing.py new file mode 100644 index 000000000..dd9052571 --- /dev/null +++ b/python/pulsing/integrations/probing.py @@ -0,0 +1,138 @@ +"""Bridge Pulsing runtime telemetry to probing via memtable (mmap'd ring buffers). + +All data is written by Rust directly to mmap files under ``/probing//``, +discovered automatically by probing's DataFusion engine: + +- ``pulsing.actors`` — live actor registry (MEMH hash table) +- ``pulsing.spans`` — completed OTel spans (MEMT ring buffer) +- ``pulsing.metrics`` — periodic system metric snapshots (MEMT ring buffer) +- ``pulsing.members`` — cluster membership state (MEMH hash table) + +The Python background thread triggers periodic ``get_metrics()`` calls so +the Rust side records fresh metric snapshots. Member events are written +automatically by the gossip protocol on every state transition. + +Environment: + +- ``PULSING_PROBING_INTEGRATION``: set to ``0`` / ``false`` / ``no`` to disable. +- ``PULSING_PROBING_REFRESH_SEC``: refresh interval for the background thread (default 10). +- ``PULSING_PROBING_AUTO_TRACING``: default ``1`` — if ``PULSING_TRACING`` is not set, still + install a **silent** Rust OTel subscriber so span / actor memtables get populated. + Set to ``0`` to disable. +""" + +from __future__ import annotations + +import logging +import os +import threading + +logger = logging.getLogger(__name__) + +_refresh_thread: threading.Thread | None = None +_stop = threading.Event() + + +def _integration_disabled() -> bool: + v = os.environ.get("PULSING_PROBING_INTEGRATION", "1").strip().lower() + return v in ("0", "false", "no") + + +def _refresh_interval_sec() -> float: + try: + return float(os.environ.get("PULSING_PROBING_REFRESH_SEC", "10")) + except ValueError: + return 10.0 + + +def ensure_tracing_for_span_capture() -> None: + """Install Rust OTel subscriber so memtable span/actor data gets populated. + + No-op when ``PULSING_TRACING=1`` (tracing already active) or + ``PULSING_PROBING_AUTO_TRACING=0``. + """ + v = os.environ.get("PULSING_PROBING_AUTO_TRACING", "1").strip().lower() + if v in ("0", "false", "no"): + return + if os.environ.get("PULSING_TRACING", "").strip().lower() in ("1", "true", "yes"): + return + try: + from pulsing._core import init_distributed_tracing + + init_distributed_tracing( + service_name=os.environ.get("PULSING_SERVICE_NAME") or None, + console_output=False, + ) + except Exception as e: + logger.debug("ensure_tracing_for_span_capture failed: %s", e) + + +async def _trigger_snapshots() -> None: + """Call Rust APIs that record metric snapshots to memtable.""" + from pulsing.core import get_system, get_system_actor, is_initialized + + if not is_initialized(): + return + + ensure_tracing_for_span_capture() + + system = get_system() + proxy = await get_system_actor(system) + await proxy.get_metrics() + + +def _background_loop() -> None: + import asyncio + + interval = _refresh_interval_sec() + while True: + try: + asyncio.run(_trigger_snapshots()) + except Exception as e: + logger.debug("Pulsing probing integration refresh failed: %s", e) + if _stop.wait(timeout=interval): + break + + +def start_probing_integration() -> bool: + """Start a daemon thread that periodically triggers memtable snapshot writes.""" + global _refresh_thread + + if _integration_disabled(): + return False + if _refresh_thread is not None and _refresh_thread.is_alive(): + return True + + ensure_tracing_for_span_capture() + + _stop.clear() + _refresh_thread = threading.Thread( + target=_background_loop, + name="pulsing-probing-integration", + daemon=True, + ) + _refresh_thread.start() + return True + + +def stop_probing_integration() -> None: + """Stop the background refresh thread.""" + global _refresh_thread + + _stop.set() + t = _refresh_thread + if t is not None and t.is_alive(): + t.join(timeout=3.0) + _refresh_thread = None + + +async def refresh_probing_tables_async() -> bool: + """Trigger one metric snapshot refresh cycle (metrics → memtable).""" + if _integration_disabled(): + return False + try: + await _trigger_snapshots() + except Exception as e: + logger.debug("Pulsing probing async refresh failed: %s", e) + return False + return True diff --git a/tests/python/test_probing_integration.py b/tests/python/test_probing_integration.py new file mode 100644 index 000000000..25f91aa15 --- /dev/null +++ b/tests/python/test_probing_integration.py @@ -0,0 +1,36 @@ +"""Optional integration with probing (skip if probing not installed).""" + +import pytest + +pytest.importorskip("probing") + +import pulsing as pul +from pulsing.integrations.probing import ( + refresh_probing_tables_async, + start_probing_integration, + stop_probing_integration, +) + + +@pytest.mark.asyncio +async def test_refresh_triggers_snapshots(): + """refresh_probing_tables_async triggers Rust-side memtable writes.""" + stop_probing_integration() + + await pul.init() + try: + stop_probing_integration() + assert await refresh_probing_tables_async() is True + finally: + await pul.shutdown() + + +@pytest.mark.asyncio +async def test_start_probing_integration_idempotent(): + stop_probing_integration() + await pul.init() + try: + assert start_probing_integration() is True + assert start_probing_integration() is True + finally: + await pul.shutdown()