enhancement(core): add unified emitter for diagnostics

Author: tobz

bin/agent-data-plane/src/internal/env/workload/api.rs

@@ -14,8 +14,9 @@ use saluki_context::{
     tags::{SharedTagSet, TagSet},
 };
 use saluki_core::{
-    diagnostic::DiagnosticHandle,
+    diagnostic::DiagnosticsEmitter,
     runtime::{state::DataspaceRegistry, InitializationError, Supervisable, SupervisorFuture},
+    support::SubsystemIdentifier,
 };
 use saluki_env::workload::{
     entity::HighestPrecedenceEntityIdRef,
@@ -210,21 +211,23 @@ impl Supervisable for RemoteAgentWorkloadAPIWorker {
     async fn initialize(&self, process_shutdown: ShutdownHandle) -> Result<SupervisorFuture, InitializationError> {
         let workload_route = DynamicRoute::http(EndpointType::Privileged, &self.handler);
 
-        let state = self.handler.state.clone();
-        let tags_handle = DiagnosticHandle::new("workload-tags-dump.json", move || state.tags_dump_json().into_bytes());
-
-        let state = self.handler.state.clone();
-        let eds_handle = DiagnosticHandle::new("workload-external-data-dump.json", move || {
-            state.eds_dump_json().into_bytes()
-        });
+        let tags_state = self.handler.state.clone();
+        let eds_state = self.handler.state.clone();
 
         Ok(Box::pin(async move {
             let dataspace =
                 DataspaceRegistry::try_current().ok_or_else(|| generic_error!("Dataspace not available."))?;
 
             dataspace.assert(workload_route, "workload-api");
-            dataspace.assert(tags_handle, "diag-workload-tags");
-            dataspace.assert(eds_handle, "diag-workload-eds");
+
+            let diagnostics =
+                DiagnosticsEmitter::from_dataspace(SubsystemIdentifier::from_segments(["workload-api"]), dataspace);
+            diagnostics.register_collector("workload-tags-dump.json", move || {
+                tags_state.tags_dump_json().into_bytes()
+            });
+            diagnostics.register_collector("workload-external-data-dump.json", move || {
+                eds_state.eds_dump_json().into_bytes()
+            });
 
             process_shutdown.await;
             Ok(())

bin/agent-data-plane/src/internal/remote_agent.rs

@@ -20,7 +20,7 @@ use saluki_common::sync::shutdown::ShutdownHandle;
 use saluki_common::task::spawn_traced_named;
 use saluki_config::{dynamic::ConfigUpdate, upsert, GenericConfiguration};
 use saluki_core::{
-    diagnostic::DiagnosticHandle,
+    diagnostic::DiagnosticCollector,
     observability::metrics::{get_shared_metrics_state, AggregatedMetricsProcessor, Reflector, TelemetryProcessor},
     runtime::{
         state::{DataspaceRegistry, IdentifierFilter},
@@ -578,7 +578,7 @@ impl FlareProvider for RemoteAgentImpl {
 
                 // Grab and collect all asserted diagnostic handles
                 if let Some(dataspace) = self.dataspace.get() {
-                    let handles = dataspace.current_values::<DiagnosticHandle>(IdentifierFilter::all());
+                    let handles = dataspace.current_values::<DiagnosticCollector>(IdentifierFilter::all());
                     let total_handles = handles.len();
                     let deadline = tokio::time::Instant::now() + DIAGNOSTIC_COLLECT_TIMEOUT;
 

lib/saluki-app/src/accounting.rs

@@ -12,8 +12,9 @@ use saluki_core::accounting::{
     ComponentBounds, ComponentRegistry, ComponentRegistryHandle, MemoryGrant, MemoryLimiter,
 };
 use saluki_core::{
-    diagnostic::DiagnosticHandle,
+    diagnostic::DiagnosticsEmitter,
     runtime::{state::DataspaceRegistry, InitializationError, Supervisable, SupervisorFuture},
+    support::SubsystemIdentifier,
 };
 use saluki_error::{generic_error, ErrorContext as _, GenericError};
 use serde::Deserialize;
@@ -330,17 +331,22 @@ impl Supervisable for ResourceTelemetryWorker {
         let memory_routes = DynamicRoute::http(EndpointType::Unprivileged, self.component_registry.api_handler());
 
         let component_registry = self.component_registry.clone();
-        let flare_handle = DiagnosticHandle::new("memory_status.json", move || {
-            component_registry.memory_snapshot_json().into_bytes()
-        });
 
         Ok(Box::pin(async move {
             let dataspace =
                 DataspaceRegistry::try_current().ok_or_else(|| generic_error!("Dataspace not available."))?;
 
-            // Register our API routes and diagnostic handle before we actually start running.
+            // Register our API routes before we actually start running.
             dataspace.assert(memory_routes, "resource-telemetry-api");
-            dataspace.assert(flare_handle, "diag-memory");
+
+            // Expose our diagnostic artifact via the diagnostics control surface.
+            let diagnostics = DiagnosticsEmitter::from_dataspace(
+                SubsystemIdentifier::from_segments(["resource-telemetry"]),
+                dataspace,
+            );
+            diagnostics.register_collector("memory_status.json", move || {
+                component_registry.memory_snapshot_json().into_bytes()
+            });
 
             select! {
                 _ = process_shutdown => {},

lib/saluki-app/src/config.rs

@@ -11,8 +11,9 @@ use saluki_api::{
 use saluki_common::sync::shutdown::ShutdownHandle;
 use saluki_config::GenericConfiguration;
 use saluki_core::{
-    diagnostic::DiagnosticHandle,
+    diagnostic::DiagnosticsEmitter,
     runtime::{state::DataspaceRegistry, InitializationError, Supervisable, SupervisorFuture},
+    support::SubsystemIdentifier,
 };
 use saluki_error::generic_error;
 use serde_json::Value;
@@ -91,19 +92,21 @@ impl Supervisable for ConfigWorker {
         let config_route = DynamicRoute::http(EndpointType::Privileged, &self.handler);
 
         let config = self.handler.state.config.clone();
-        let flare_handle = DiagnosticHandle::new("runtime_config_dump.yaml", move || {
-            config
-                .as_typed::<serde_json::Value>()
-                .map(|v| serde_json::to_vec_pretty(&v).unwrap_or_default())
-                .unwrap_or_default()
-        });
 
         Ok(Box::pin(async move {
             let dataspace =
                 DataspaceRegistry::try_current().ok_or_else(|| generic_error!("Dataspace not available."))?;
 
             dataspace.assert(config_route, "config-api");
-            dataspace.assert(flare_handle, "diag-config");
+
+            let diagnostics =
+                DiagnosticsEmitter::from_dataspace(SubsystemIdentifier::from_segments(["config-api"]), dataspace);
+            diagnostics.register_collector("runtime_config_dump.yaml", move || {
+                config
+                    .as_typed::<serde_json::Value>()
+                    .map(|v| serde_json::to_vec_pretty(&v).unwrap_or_default())
+                    .unwrap_or_default()
+            });
 
             process_shutdown.await;
             Ok(())

lib/saluki-core/src/diagnostic.rs

@@ -0,0 +1,39 @@
+//! On-demand diagnostic artifact collection.
+
+use std::sync::Arc;
+
+/// A named, on-demand producer of diagnostic artifact bytes.
+///
+/// A collector pairs an artifact name with a synchronous closure that produces the artifact's bytes when
+/// invoked. Collectors are registered through a [`DiagnosticsEmitter`][super::DiagnosticsEmitter] and gathered on
+/// demand by whichever subsystem is responsible for assembling diagnostic artifacts.
+///
+/// The artifact name is suitable, but not guaranteed, to be used as a filename.
+#[derive(Clone)]
+pub struct DiagnosticCollector {
+    artifact_name: String,
+    collect_fn: Arc<dyn Fn() -> Vec<u8> + Send + Sync>,
+}
+
+impl DiagnosticCollector {
+    /// Creates a new collector with the given artifact name and collection closure.
+    ///
+    /// `collect_fn` runs synchronously and must return promptly, as it can delay the collection of artifacts for the
+    /// whole system.
+    pub fn new(artifact_name: impl Into<String>, collect_fn: impl Fn() -> Vec<u8> + Send + Sync + 'static) -> Self {
+        Self {
+            artifact_name: artifact_name.into(),
+            collect_fn: Arc::new(collect_fn),
+        }
+    }
+
+    /// Returns the artifact name.
+    pub fn artifact_name(&self) -> &str {
+        &self.artifact_name
+    }
+
+    /// Collects and returns the artifact bytes.
+    pub fn collect(&self) -> Vec<u8> {
+        (self.collect_fn)()
+    }
+}