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

saluki_core/diagnostic/
emitter.rs
