saluki_core/diagnostic/
emitter.rs

1//! Subsystem-scoped diagnostics control surface.
2
3// TODO: probably rework process name construction to use `SubsystemIdentifier` under the hood, and expose that, in
4// addition to process ID, as a task-local we can access so that we can make `DiagnosticEmitter::from_current` require
5// no parameters at all while still doing the right thing
6
7use snafu::{OptionExt as _, Snafu};
8use stringtheory::MetaString;
9
10use super::{DiagnosticCollector, DiagnosticEvent};
11use crate::{
12    runtime::state::{DataspaceRegistry, Identifier, IdentifierFilter, Subscription},
13    support::SubsystemIdentifier,
14};
15
16/// Errors that can occur when creating a [`DiagnosticsEmitter`].
17#[derive(Debug, Snafu)]
18#[snafu(context(suffix(false)))]
19pub enum DiagnosticsEmitterError {
20    /// No dataspace was available in the current context.
21    ///
22    /// An emitter can only be created within a running supervision tree, where an ambient dataspace is established.
23    /// This indicates that creation was attempted outside of one.
24    #[snafu(display("no dataspace available in the current context (not running inside a supervision tree)"))]
25    NoDataspace,
26}
27
28/// A subsystem-scoped control surface for exposing diagnostics.
29///
30/// A `DiagnosticsEmitter` is created for a single subsystem, identified by a [`SubsystemIdentifier`], and is attached
31/// to a specific dataspace. It hides the boilerplate of interacting with the dataspace directly, while still using it
32/// under the hood so that other subsystems can subscribe to what is exposed in a decoupled, eventually consistent way.
33///
34/// It exposes two capabilities:
35///
36/// - **Collectors**: named, on-demand producers of artifact bytes (see [`register_collector`]), stored as persistent
37///   dataspace assertions and automatically withdrawn when the owning process exits.
38/// - **Events**: abstract, point-in-time [`DiagnosticEvent`]s (see [`emit`]), delivered as transient dataspace
39///   messages to any subscribers present at the time of emission.
40///
41/// [`register_collector`]: Self::register_collector
42/// [`emit`]: Self::emit
43///
44/// # Example
45///
46/// <!-- vale off -->
47/// ```
48/// use saluki_core::diagnostic::{DiagnosticDetails, DiagnosticEvent, DiagnosticsEmitter};
49/// use saluki_core::runtime::state::DataspaceRegistry;
50/// use saluki_core::support::SubsystemIdentifier;
51///
52/// let dataspace = DataspaceRegistry::new();
53/// let emitter = DiagnosticsEmitter::from_dataspace(
54///     SubsystemIdentifier::from_segments(["my-subsystem"]),
55///     dataspace,
56/// );
57///
58/// // Expose an artifact that is produced on demand:
59/// emitter.register_collector("state.json", || b"{}");
60///
61/// // Emit a point-in-time event:
62/// emitter.emit(DiagnosticEvent::new("credentials rejected", DiagnosticDetails::InvalidApiKey));
63/// ```
64/// <!-- vale on -->
65#[derive(Clone)]
66pub struct DiagnosticsEmitter {
67    base_id: MetaString,
68    dataspace: DataspaceRegistry,
69}
70
71impl DiagnosticsEmitter {
72    /// Creates an emitter for the given subsystem, attaching to the current dataspace.
73    ///
74    /// # Errors
75    ///
76    /// If no dataspace is available, an error is returned.
77    pub fn from_current(id: SubsystemIdentifier) -> Result<Self, DiagnosticsEmitterError> {
78        let dataspace = DataspaceRegistry::try_current().context(NoDataspace)?;
79        Ok(Self::from_dataspace(id, dataspace))
80    }
81
82    /// Creates an emitter for the given subsystem from an already-held dataspace handle.
83    pub fn from_dataspace(id: SubsystemIdentifier, dataspace: DataspaceRegistry) -> Self {
84        let base_id = id.to_string();
85        Self {
86            base_id: base_id.into(),
87            dataspace,
88        }
89    }
90
91    /// Registers a collector for a given artifact.
92    ///
93    /// The collector is exposed until it is explicitly removed via [`unregister_collector`], or until the owning
94    /// process exits, whichever comes first. Registering a collector with an artifact name that is already registered
95    /// by this subsystem replaces the previous one.
96    ///
97    /// Care should be taken when registering a collector:
98    ///
99    /// - the given artifact name _should_ be unique within the overall system, and should be generally suitable as a
100    ///   file name when possible (artifact names are sanitized/normalized where necessary, but may lose useful
101    ///   information in the process)
102    /// - the collection function (`collect_fn`) will be run synchronously and should return promptly, as it can delay
103    ///   the collection of artifacts for the whole system
104    ///
105    /// [`unregister_collector`]: Self::unregister_collector
106    pub fn register_collector<F, T>(&self, artifact_name: impl Into<String>, collect_fn: F)
107    where
108        F: Fn() -> T + Send + Sync + 'static,
109        T: Into<Vec<u8>>,
110    {
111        let collector = DiagnosticCollector::new(artifact_name, collect_fn);
112        let id = self.build_collector_identifier(collector.artifact_name());
113        self.dataspace.assert(collector, id);
114    }
115
116    /// Removes a previously registered collector by name
117    ///
118    /// Does nothing if no collector with that name is currently registered by this subsystem.
119    pub fn unregister_collector(&self, artifact_name: impl AsRef<str>) {
120        let id = self.build_collector_identifier(artifact_name.as_ref());
121        self.dataspace.retract::<DiagnosticCollector>(id);
122    }
123
124    /// Emits a diagnostic event.
125    ///
126    /// Diagnostics events are transient and only delivered to active listeners.
127    pub fn emit(&self, event: DiagnosticEvent) {
128        self.dataspace.send(event, self.base_id.clone());
129    }
130
131    fn build_collector_identifier(&self, artifact_name: &str) -> Identifier {
132        Identifier::named(format!("{}-{}", self.base_id, artifact_name))
133    }
134}
135
136/// Subscribes to diagnostic events matching the given filter, using the current dataspace.
137///
138/// This is the counterpart to [`DiagnosticsEmitter::emit`] for consumers that want to observe events without holding a
139/// [`DataspaceRegistry`] directly.
140///
141/// # Errors
142///
143/// If no dataspace is available, an error is returned.
144pub fn subscribe_events(filter: IdentifierFilter) -> Result<Subscription<DiagnosticEvent>, DiagnosticsEmitterError> {
145    let dataspace = DataspaceRegistry::try_current().context(NoDataspace)?;
146    Ok(dataspace.subscribe::<DiagnosticEvent>(filter))
147}
148
149#[cfg(test)]
150mod tests {
151    use tokio_test::{assert_pending, assert_ready, assert_ready_eq, task::spawn as test_spawn};
152
153    use super::*;
154    use crate::{
155        diagnostic::DiagnosticDetails,
156        runtime::{
157            state::{DataspaceUpdate, CURRENT_DATASPACE},
158            ProcessId,
159        },
160    };
161
162    fn emitter(dataspace: DataspaceRegistry) -> DiagnosticsEmitter {
163        DiagnosticsEmitter::from_dataspace(SubsystemIdentifier::from_segments(["sub"]), dataspace)
164    }
165
166    #[test]
167    fn from_current_without_dataspace_errors() {
168        let result = DiagnosticsEmitter::from_current(SubsystemIdentifier::from_segments(["sub"]));
169        assert!(matches!(result, Err(DiagnosticsEmitterError::NoDataspace)));
170    }
171
172    #[test]
173    fn from_current_inside_dataspace_succeeds() {
174        let registry = DataspaceRegistry::new();
175        CURRENT_DATASPACE.sync_scope(registry, || {
176            assert!(DiagnosticsEmitter::from_current(SubsystemIdentifier::from_segments(["sub"])).is_ok());
177        });
178    }
179
180    #[test]
181    fn register_collector_is_discoverable() {
182        let registry = DataspaceRegistry::new();
183        emitter(registry.clone()).register_collector("state.json", || b"hello");
184
185        let collectors = registry.current_values::<DiagnosticCollector>(IdentifierFilter::all());
186        assert_eq!(collectors.len(), 1);
187        assert_eq!(collectors[0].artifact_name(), "state.json");
188        assert_eq!(collectors[0].collect(), b"hello".to_vec());
189    }
190
191    #[test]
192    fn multiple_collectors_coexist() {
193        let registry = DataspaceRegistry::new();
194        let emitter = emitter(registry.clone());
195        emitter.register_collector("tags.json", || vec![1]);
196        emitter.register_collector("eds.json", || vec![2]);
197
198        let mut names: Vec<String> = registry
199            .current_values::<DiagnosticCollector>(IdentifierFilter::all())
200            .iter()
201            .map(|c| c.artifact_name().to_string())
202            .collect();
203        names.sort();
204        assert_eq!(names, vec!["eds.json".to_string(), "tags.json".to_string()]);
205    }
206
207    #[test]
208    fn reregister_same_artifact_updates() {
209        let registry = DataspaceRegistry::new();
210        let emitter = emitter(registry.clone());
211        emitter.register_collector("state.json", || b"v1");
212        emitter.register_collector("state.json", || b"v2");
213
214        let collectors = registry.current_values::<DiagnosticCollector>(IdentifierFilter::all());
215        assert_eq!(collectors.len(), 1);
216        assert_eq!(collectors[0].collect(), b"v2".to_vec());
217    }
218
219    #[test]
220    fn unregister_collector_retracts() {
221        let registry = DataspaceRegistry::new();
222        let emitter = emitter(registry.clone());
223
224        let mut sub = registry.subscribe::<DiagnosticCollector>(IdentifierFilter::all());
225        emitter.register_collector("state.json", || vec![0]);
226        emitter.unregister_collector("state.json");
227
228        // First update: the assertion, under the derived `<subsystem>-<artifact>` identifier.
229        let mut recv = test_spawn(sub.recv());
230        match assert_ready!(recv.poll()) {
231            Some(DataspaceUpdate::Asserted(id, collector)) => {
232                assert_eq!(id, Identifier::named("sub-state.json"));
233                assert_eq!(collector.artifact_name(), "state.json");
234            }
235            _ => panic!("expected an assertion first"),
236        }
237        drop(recv);
238
239        // Second update: the retraction.
240        let mut recv = test_spawn(sub.recv());
241        match assert_ready!(recv.poll()) {
242            Some(DataspaceUpdate::Retracted(id)) => assert_eq!(id, Identifier::named("sub-state.json")),
243            _ => panic!("expected a retraction second"),
244        }
245    }
246
247    #[test]
248    fn register_collector_is_tagged_to_current_process() {
249        let registry = DataspaceRegistry::new();
250        let emitter = emitter(registry.clone());
251
252        // The collector is asserted under the current process, so it is withdrawn when that process exits.
253        let pid = ProcessId::current();
254        emitter.register_collector("state.json", || vec![0]);
255        assert_eq!(
256            registry
257                .current_values::<DiagnosticCollector>(IdentifierFilter::all())
258                .len(),
259            1
260        );
261
262        // Simulating that process exiting withdraws the collector automatically.
263        registry.retract_all_for_process(pid);
264        assert!(registry
265            .current_values::<DiagnosticCollector>(IdentifierFilter::all())
266            .is_empty());
267    }
268
269    #[test]
270    fn emit_delivers_event_to_subscriber() {
271        let registry = DataspaceRegistry::new();
272        let emitter = emitter(registry.clone());
273
274        let mut sub = registry.subscribe::<DiagnosticEvent>(IdentifierFilter::all());
275        emitter.emit(DiagnosticEvent::new(
276            "credentials rejected",
277            DiagnosticDetails::InvalidApiKey,
278        ));
279
280        let mut recv = test_spawn(sub.recv());
281        assert_ready_eq!(
282            recv.poll(),
283            Some(DataspaceUpdate::Message(
284                Identifier::named("sub"),
285                DiagnosticEvent::new("credentials rejected", DiagnosticDetails::InvalidApiKey)
286            ))
287        );
288    }
289
290    #[test]
291    fn subscribe_events_receives_emitted_event() {
292        let registry = DataspaceRegistry::new();
293        let emitter = emitter(registry.clone());
294
295        CURRENT_DATASPACE.sync_scope(registry, || {
296            let mut sub = subscribe_events(IdentifierFilter::all()).expect("dataspace should be available");
297            emitter.emit(DiagnosticEvent::new("boom", DiagnosticDetails::InvalidApiKey));
298
299            let mut recv = test_spawn(sub.recv());
300            assert_ready_eq!(
301                recv.poll(),
302                Some(DataspaceUpdate::Message(
303                    Identifier::named("sub"),
304                    DiagnosticEvent::new("boom", DiagnosticDetails::InvalidApiKey)
305                ))
306            );
307        });
308    }
309
310    #[test]
311    fn emit_is_transient() {
312        let registry = DataspaceRegistry::new();
313        let emitter = emitter(registry.clone());
314
315        // Emit before anyone is subscribed.
316        emitter.emit(DiagnosticEvent::new("boom", DiagnosticDetails::InvalidApiKey));
317
318        // Events are never stored.
319        assert!(registry
320            .current_values::<DiagnosticEvent>(IdentifierFilter::all())
321            .is_empty());
322
323        // A subscriber that appears afterwards does not receive the already-sent event.
324        let mut sub = registry.subscribe::<DiagnosticEvent>(IdentifierFilter::all());
325        let mut recv = test_spawn(sub.recv());
326        assert_pending!(recv.poll());
327    }
328}
saluki_core/diagnostic/emitter.rs

saluki_core/diagnostic/
emitter.rs
