atomr_telemetry/
lib.rs

1//! # atomr-telemetry
2//!
3//! Optional probe surface for observing a running `atomr` node.
4//!
5//! The crate is passive and opt-in. Construct a [`TelemetryExtension`] and
6//! register it on an [`atomr_core::actor::ActorSystem`] via
7//! `atomr_core::actor::Extensions`. Subsystems check for the extension
8//! at runtime (cheap `Arc<T>` lookup) and, when present, emit snapshots +
9//! events into the telemetry [`bus::TelemetryBus`]. When absent, there is
10//! no cost beyond a single `DashMap` lookup.
11//!
12//! See [`crate::exporters`] for the Prometheus / OpenTelemetry exporters
13//! gated behind cargo features.
14
15#![forbid(unsafe_code)]
16#![deny(rust_2018_idioms)]
17
18pub mod actor_registry;
19pub mod bus;
20pub mod cluster;
21pub mod ddata;
22pub mod dead_letters;
23pub mod dto;
24pub mod exporters;
25pub mod persistence;
26pub mod remote;
27pub mod sharding;
28pub mod streams;
29
30use std::sync::Arc;
31
32use atomr_core::actor::ActorSystem;
33use parking_lot::RwLock;
34
35use crate::actor_registry::ActorRegistry;
36use crate::bus::TelemetryBus;
37use crate::dead_letters::DeadLetterFeed;
38
39/// The telemetry extension. Construct once per actor system, register via
40/// the actor system's extensions, and all other probes will pick it up.
41pub struct TelemetryExtension {
42    pub node: String,
43    pub bus: TelemetryBus,
44    pub actors: Arc<ActorRegistry>,
45    pub dead_letters: Arc<DeadLetterFeed>,
46    pub cluster: Arc<cluster::ClusterProbe>,
47    pub sharding: Arc<sharding::ShardingProbe>,
48    pub persistence: Arc<persistence::PersistenceProbe>,
49    pub remote: Arc<remote::RemoteProbe>,
50    pub streams: Arc<streams::StreamsProbe>,
51    pub ddata: Arc<ddata::DDataProbe>,
52    pub(crate) exporters: RwLock<Vec<Arc<dyn exporters::Exporter>>>,
53}
54
55impl TelemetryExtension {
56    /// Build a telemetry extension for the given node name. Channel
57    /// capacity controls how many in-flight `TelemetryEvent`s the broadcast
58    /// bus will buffer per subscriber.
59    pub fn new(node: impl Into<String>, channel_capacity: usize) -> Arc<Self> {
60        let bus = TelemetryBus::new(channel_capacity);
61        Arc::new(Self {
62            node: node.into(),
63            actors: Arc::new(ActorRegistry::new(bus.clone())),
64            dead_letters: Arc::new(DeadLetterFeed::new(bus.clone(), 1024)),
65            cluster: Arc::new(cluster::ClusterProbe::new(bus.clone())),
66            sharding: Arc::new(sharding::ShardingProbe::new(bus.clone())),
67            persistence: Arc::new(persistence::PersistenceProbe::new(bus.clone())),
68            remote: Arc::new(remote::RemoteProbe::new(bus.clone())),
69            streams: Arc::new(streams::StreamsProbe::new(bus.clone())),
70            ddata: Arc::new(ddata::DDataProbe::new(bus.clone())),
71            bus,
72            exporters: RwLock::new(Vec::new()),
73        })
74    }
75
76    /// Install this extension on the given `ActorSystem`. Returns a clone
77    /// of the shared `Arc<TelemetryExtension>`; the caller may keep it to
78    /// feed probes directly from application code.
79    pub fn install(self: Arc<Self>, system: &ActorSystem) -> Arc<Self> {
80        system.extensions().register(TelemetryHandle(self.clone()));
81        system.set_spawn_observer(self.actors.clone());
82        system.set_dead_letter_observer(self.dead_letters.clone());
83        self
84    }
85
86    /// Look up an installed extension on an `ActorSystem`.
87    pub fn from_system(system: &ActorSystem) -> Option<Arc<Self>> {
88        system.extensions().get::<TelemetryHandle>().map(|h| h.0.clone())
89    }
90
91    /// Register an exporter. Exporters receive every event published to
92    /// the bus and may poll probes for snapshots on their own cadence.
93    pub fn add_exporter(&self, exporter: Arc<dyn exporters::Exporter>) {
94        self.bus.attach_exporter(exporter.clone());
95        self.exporters.write().push(exporter);
96    }
97
98    /// Install the FR-11 OTel span exporter and return the
99    /// [`TraceContextInterceptor`]
100    /// to register on actor `Props` via `Props::with_interceptor`. The
101    /// metrics exporter path is independent and untouched.
102    #[cfg(feature = "otel")]
103    pub fn add_span_exporter(
104        &self,
105        exporter: Arc<exporters::otel_tracer::OtelTracerExporter>,
106        probe: exporters::otel_tracer::SpanProbeConfig,
107    ) -> Arc<exporters::otel_tracer::TraceContextInterceptor> {
108        self.bus.attach_span_exporter(exporter, probe)
109    }
110
111    /// Snapshot the full telemetry state of this node (one JSON payload).
112    pub fn snapshot(&self) -> dto::NodeSnapshot {
113        dto::NodeSnapshot {
114            node: self.node.clone(),
115            generated_at: chrono::Utc::now().to_rfc3339(),
116            actors: self.actors.snapshot(),
117            dead_letters: self.dead_letters.recent(100),
118            cluster: self.cluster.snapshot(),
119            sharding: self.sharding.snapshot(),
120            persistence: self.persistence.snapshot(),
121            remote: self.remote.snapshot(),
122            streams: self.streams.snapshot(),
123            ddata: self.ddata.snapshot(),
124        }
125    }
126}
127
128/// Shim so we can register `Arc<TelemetryExtension>` in the typed
129/// `Extensions` bag under a stable handle type.
130pub struct TelemetryHandle(pub Arc<TelemetryExtension>);
131
132/// Re-exports of the FR-11 OTel span model (gated behind the `otel` feature).
133#[cfg(feature = "otel")]
134pub use crate::exporters::otel_tracer::{OtelTracerExporter, SpanProbeConfig, TraceContextInterceptor};
atomr_telemetry/lib.rs

atomr_telemetry/
lib.rs
