atomr_telemetry/
bus.rs

1//! Typed telemetry bus backed by a `tokio::sync::broadcast` channel.
2//!
3//! Probes publish `TelemetryEvent`s through this bus. Dashboard clients
4//! (WebSocket subscribers), exporters (Prometheus/OpenTelemetry), and
5//! the in-memory `DeadLetterFeed` ring buffer all receive copies.
6
7use std::sync::Arc;
8
9use parking_lot::RwLock;
10use serde::{Deserialize, Serialize};
11use tokio::sync::broadcast;
12
13use crate::dto::{
14    ActorStatus, ClusterMembershipDiff, DeadLetterRecord, JournalWriteInfo, RemoteAssociationInfo,
15    ShardingEvent,
16};
17use crate::exporters::Exporter;
18
19/// A single telemetry event. Kept deliberately wide so clients can filter
20/// by the `topic` field without deserializing every payload variant.
21#[derive(Debug, Clone, Serialize, Deserialize)]
22#[serde(tag = "kind", rename_all = "snake_case")]
23pub enum TelemetryEvent {
24    ActorSpawned(ActorStatus),
25    ActorStopped { path: String },
26    MailboxSampled { path: String, depth: u64 },
27    DeadLetter(DeadLetterRecord),
28    ClusterChanged(ClusterMembershipDiff),
29    ShardingChanged(ShardingEvent),
30    JournalWrite(JournalWriteInfo),
31    RemoteAssociation(RemoteAssociationInfo),
32    StreamsGraphStarted { id: u64, name: String },
33    StreamsGraphFinished { id: u64 },
34    DDataUpdated { key: String },
35}
36
37impl TelemetryEvent {
38    /// A short, stable topic string used by WebSocket clients to filter
39    /// the event stream.
40    pub fn topic(&self) -> &'static str {
41        match self {
42            Self::ActorSpawned(_) | Self::ActorStopped { .. } | Self::MailboxSampled { .. } => "actors",
43            Self::DeadLetter(_) => "dead_letters",
44            Self::ClusterChanged(_) => "cluster",
45            Self::ShardingChanged(_) => "sharding",
46            Self::JournalWrite(_) => "persistence",
47            Self::RemoteAssociation(_) => "remote",
48            Self::StreamsGraphStarted { .. } | Self::StreamsGraphFinished { .. } => "streams",
49            Self::DDataUpdated { .. } => "ddata",
50        }
51    }
52
53    /// All telemetry topics the bus emits. Used by the dashboard /
54    /// spec parity tests to ensure every probe surface is wired.
55    pub const ALL_TOPICS: &'static [&'static str] =
56        &["actors", "dead_letters", "cluster", "sharding", "persistence", "remote", "streams", "ddata"];
57}
58
59/// Cheap-to-clone broadcast bus. Wraps a `tokio::sync::broadcast` sender
60/// plus a slot for attached exporters (synchronous callbacks).
61#[derive(Clone)]
62pub struct TelemetryBus {
63    tx: broadcast::Sender<TelemetryEvent>,
64    exporters: Arc<RwLock<Vec<Arc<dyn Exporter>>>>,
65    /// Optional span (trace) exporter for the FR-11 OTel span model. Spans
66    /// are emitted out-of-band (via the message interceptor and direct
67    /// `record_*_span` calls), not through the `TelemetryEvent` fan-out, so
68    /// this lives in its own slot rather than the `Exporter` list.
69    #[cfg(feature = "otel")]
70    span_exporter: Arc<RwLock<Option<Arc<crate::exporters::otel_tracer::OtelTracerExporter>>>>,
71}
72
73impl TelemetryBus {
74    pub fn new(capacity: usize) -> Self {
75        let (tx, _rx) = broadcast::channel(capacity.max(16));
76        Self {
77            tx,
78            exporters: Arc::new(RwLock::new(Vec::new())),
79            #[cfg(feature = "otel")]
80            span_exporter: Arc::new(RwLock::new(None)),
81        }
82    }
83
84    pub fn publish(&self, event: TelemetryEvent) {
85        // Fan out to in-process exporters first (synchronous, cheap).
86        let exporters = self.exporters.read().clone();
87        for exp in &exporters {
88            exp.on_event(&event);
89        }
90        // Then broadcast to async subscribers (WS clients, tests).
91        let _ = self.tx.send(event);
92    }
93
94    pub fn subscribe(&self) -> broadcast::Receiver<TelemetryEvent> {
95        self.tx.subscribe()
96    }
97
98    /// Subscribe to a single topic. The returned receiver yields only
99    /// events whose `topic()` matches `wanted`. Backed by a forwarder
100    /// task that filters the broadcast stream — drop the receiver to
101    /// stop the forwarder.
102    pub fn subscribe_topic(
103        &self,
104        wanted: &'static str,
105    ) -> tokio::sync::mpsc::UnboundedReceiver<TelemetryEvent> {
106        let mut src = self.tx.subscribe();
107        let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
108        tokio::spawn(async move {
109            while let Ok(ev) = src.recv().await {
110                if ev.topic() == wanted && tx.send(ev).is_err() {
111                    return;
112                }
113            }
114        });
115        rx
116    }
117
118    pub fn receiver_count(&self) -> usize {
119        self.tx.receiver_count()
120    }
121
122    pub(crate) fn attach_exporter(&self, exporter: Arc<dyn Exporter>) {
123        self.exporters.write().push(exporter);
124    }
125
126    /// Install the FR-11 span exporter and return a `TraceContextInterceptor`
127    /// wired to it. Install the interceptor on actor `Props` via
128    /// `Props::with_interceptor` so handled messages open `actor.handle` spans
129    /// and outgoing messages carry child trace context. The metrics path is
130    /// untouched.
131    #[cfg(feature = "otel")]
132    pub fn attach_span_exporter(
133        &self,
134        exporter: Arc<crate::exporters::otel_tracer::OtelTracerExporter>,
135        probe: crate::exporters::otel_tracer::SpanProbeConfig,
136    ) -> Arc<crate::exporters::otel_tracer::TraceContextInterceptor> {
137        *self.span_exporter.write() = Some(exporter.clone());
138        Arc::new(crate::exporters::otel_tracer::TraceContextInterceptor::new(exporter, probe))
139    }
140
141    /// Builder-style variant of [`attach_span_exporter`](Self::attach_span_exporter).
142    /// Returns `(self, interceptor)` so the bus can be threaded fluently.
143    #[cfg(feature = "otel")]
144    pub fn with_span_exporter(
145        self,
146        exporter: Arc<crate::exporters::otel_tracer::OtelTracerExporter>,
147        probe: crate::exporters::otel_tracer::SpanProbeConfig,
148    ) -> (Self, Arc<crate::exporters::otel_tracer::TraceContextInterceptor>) {
149        let interceptor = self.attach_span_exporter(exporter, probe);
150        (self, interceptor)
151    }
152
153    /// The installed span exporter, if any. Lets application code call
154    /// `record_handle_span` / `record_restart_span` directly.
155    #[cfg(feature = "otel")]
156    pub fn span_exporter(&self) -> Option<Arc<crate::exporters::otel_tracer::OtelTracerExporter>> {
157        self.span_exporter.read().clone()
158    }
159}
160
161#[cfg(test)]
162mod tests {
163    use super::*;
164    use crate::dto::ActorStatus;
165
166    #[tokio::test]
167    async fn publish_and_subscribe_roundtrip() {
168        let bus = TelemetryBus::new(8);
169        let mut rx = bus.subscribe();
170        bus.publish(TelemetryEvent::ActorStopped { path: "/user/a".into() });
171        let got = rx.recv().await.unwrap();
172        assert_eq!(got.topic(), "actors");
173    }
174
175    #[tokio::test]
176    async fn topic_labels_correct() {
177        let e = TelemetryEvent::ActorSpawned(ActorStatus {
178            path: "/user/x".into(),
179            parent: Some("/user".into()),
180            actor_type: "Test".into(),
181            mailbox_depth: 0,
182            spawned_at: "now".into(),
183            host: None,
184        });
185        assert_eq!(e.topic(), "actors");
186    }
187
188    #[tokio::test]
189    async fn subscribe_topic_filters_by_topic() {
190        let bus = TelemetryBus::new(16);
191        let mut rx = bus.subscribe_topic("ddata");
192        bus.publish(TelemetryEvent::ActorStopped { path: "/x".into() }); // actors — skipped
193        bus.publish(TelemetryEvent::DDataUpdated { key: "k".into() });
194        bus.publish(TelemetryEvent::DDataUpdated { key: "j".into() });
195        let first = rx.recv().await.unwrap();
196        let second = rx.recv().await.unwrap();
197        match (first, second) {
198            (TelemetryEvent::DDataUpdated { key: k1 }, TelemetryEvent::DDataUpdated { key: k2 }) => {
199                assert_eq!(k1, "k");
200                assert_eq!(k2, "j");
201            }
202            other => panic!("unexpected events: {other:?}"),
203        }
204    }
205
206    #[test]
207    fn all_topics_covers_every_variant() {
208        // Confirm every topic that variants advertise is listed in
209        // ALL_TOPICS — guards against drift between TelemetryEvent and
210        // the dashboard topic enumeration.
211        let samples = [
212            TelemetryEvent::ActorStopped { path: "/x".into() }.topic(),
213            TelemetryEvent::DeadLetter(DeadLetterRecord {
214                seq: 0,
215                recipient: "/x".into(),
216                sender: None,
217                message_type: "test".into(),
218                message_preview: "p".into(),
219                timestamp: "now".into(),
220            })
221            .topic(),
222            TelemetryEvent::DDataUpdated { key: "k".into() }.topic(),
223        ];
224        for t in samples {
225            assert!(TelemetryEvent::ALL_TOPICS.contains(&t), "topic {t} missing from ALL_TOPICS");
226        }
227        // No duplicates.
228        let mut sorted = TelemetryEvent::ALL_TOPICS.to_vec();
229        sorted.sort();
230        sorted.dedup();
231        assert_eq!(sorted.len(), TelemetryEvent::ALL_TOPICS.len());
232    }
233}
atomr_telemetry/bus.rs

atomr_telemetry/
bus.rs
