chore(health): move health checking to saluki-core and refactor for supervision

Cargo.toml

@@ -24,7 +24,6 @@ members = [
   "lib/saluki-core",
   "lib/saluki-env",
   "lib/saluki-error",
-  "lib/saluki-health",
   "lib/saluki-io",
   "lib/saluki-metadata",
   "lib/saluki-metrics",
@@ -54,7 +53,6 @@ saluki-context = { path = "lib/saluki-context" }
 saluki-core = { path = "lib/saluki-core" }
 saluki-env = { path = "lib/saluki-env" }
 saluki-error = { path = "lib/saluki-error" }
-saluki-health = { path = "lib/saluki-health" }
 saluki-io = { path = "lib/saluki-io" }
 saluki-metadata = { path = "lib/saluki-metadata" }
 saluki-metrics = { path = "lib/saluki-metrics" }

bin/agent-data-plane/Cargo.toml

@@ -40,7 +40,6 @@ saluki-context = { workspace = true }
 saluki-core = { workspace = true }
 saluki-env = { workspace = true }
 saluki-error = { workspace = true }
-saluki-health = { workspace = true }
 saluki-io = { workspace = true }
 saluki-metadata = { workspace = true }
 saluki-metrics = { workspace = true }

bin/agent-data-plane/src/cli/run.rs

@@ -29,11 +29,11 @@ use saluki_components::{
     },
 };
 use saluki_config::{ConfigurationLoader, GenericConfiguration};
+use saluki_core::health::HealthRegistry;
 use saluki_core::runtime::SupervisorError;
 use saluki_core::topology::TopologyBlueprint;
 use saluki_env::EnvironmentProvider as _;
 use saluki_error::{generic_error, ErrorContext as _, GenericError};
-use saluki_health::HealthRegistry;
 use tokio::{select, time::interval};
 use tracing::{error, info, warn};
 

bin/agent-data-plane/src/env_provider.rs

@@ -1,13 +1,13 @@
 use memory_accounting::ComponentRegistry;
 use saluki_config::GenericConfiguration;
+use saluki_core::health::HealthRegistry;
 use saluki_env::{
     autodiscovery::providers::BoxedAutodiscoveryProvider,
     host::providers::{BoxedHostProvider, FixedHostProvider, RemoteAgentHostProvider},
     workload::providers::{RemoteAgentWorkloadAPIHandler, RemoteAgentWorkloadProvider},
     EnvironmentProvider,
 };
 use saluki_error::GenericError;
-use saluki_health::HealthRegistry;
 use tracing::{debug, warn};
 
 use crate::config::DataPlaneConfiguration;

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

@@ -8,12 +8,14 @@ use saluki_app::{
 };
 use saluki_components::destinations::DogStatsDStatisticsConfiguration;
 use saluki_config::GenericConfiguration;
-use saluki_core::runtime::{
-    InitializationError, ProcessShutdown, RestartStrategy, RuntimeConfiguration, Supervisable, Supervisor,
-    SupervisorFuture,
+use saluki_core::{
+    health::HealthRegistry,
+    runtime::{
+        InitializationError, ProcessShutdown, RestartStrategy, RuntimeConfiguration, Supervisable, Supervisor,
+        SupervisorFuture,
+    },
 };
-use saluki_error::{generic_error, ErrorContext as _, GenericError};
-use saluki_health::HealthRegistry;
+use saluki_error::{ErrorContext as _, GenericError};
 use saluki_io::net::{build_datadog_agent_server_tls_config, get_ipc_cert_file_path, ServerConfig};
 use tracing::info;
 
@@ -41,39 +43,6 @@ fn get_cert_path_from_config(config: &GenericConfiguration) -> Result<PathBuf, G
     ))
 }
 
-/// A worker that runs the health registry.
-pub struct HealthRegistryWorker {
-    health_registry: HealthRegistry,
-}
-
-impl HealthRegistryWorker {
-    /// Creates a new `HealthRegistryWorker`.
-    pub fn new(health_registry: HealthRegistry) -> Self {
-        Self { health_registry }
-    }
-}
-
-#[async_trait]
-impl Supervisable for HealthRegistryWorker {
-    fn name(&self) -> &str {
-        "health-registry"
-    }
-
-    async fn initialize(&self, process_shutdown: ProcessShutdown) -> Result<SupervisorFuture, InitializationError> {
-        // Spawn the health registry runner with the shutdown signal. The runner will exit gracefully when the shutdown
-        // signal is received, and the response receiver will be returned to the registry state so that a subsequent
-        // spawn can succeed if the supervisor restarts this worker.
-        let handle = self.health_registry.clone().spawn(process_shutdown).await?;
-
-        Ok(Box::pin(async move {
-            match handle.await {
-                Ok(()) => Ok(()),
-                Err(e) => Err(generic_error!("Health registry task panicked: {}", e)),
-            }
-        }))
-    }
-}
-
 /// A worker that serves the unprivileged HTTP API.
 pub struct UnprivilegedApiWorker {
     dp_config: DataPlaneConfiguration,
@@ -208,11 +177,11 @@ pub async fn create_control_plane_supervisor(
         .with_dedicated_runtime(RuntimeConfiguration::single_threaded())
         .with_restart_strategy(RestartStrategy::one_to_one());
 
-    // TODO: just make the API handler for `ComponentRegistry` cloneable so we can create/hold on to it in
+    supervisor.add_worker(health_registry.worker());
+
+    // TODO: Just make the API handler for `ComponentRegistry` cloneable so we can create/hold on to it in
     // `UnprivilegedApiWorker` without having to create a scoped one here just to maintain the ownership necessary
     let scoped_registry = component_registry.get_or_create("control-plane");
-
-    supervisor.add_worker(HealthRegistryWorker::new(health_registry.clone()));
     supervisor.add_worker(UnprivilegedApiWorker::new(
         dp_config.clone(),
         health_registry,

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

@@ -1,9 +1,9 @@
 use memory_accounting::ComponentRegistry;
 use saluki_components::destinations::DogStatsDStatisticsConfiguration;
 use saluki_config::GenericConfiguration;
+use saluki_core::health::HealthRegistry;
 use saluki_core::runtime::Supervisor;
 use saluki_error::GenericError;
-use saluki_health::HealthRegistry;
 
 mod control_plane;
 pub use self::control_plane::create_control_plane_supervisor;

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

@@ -3,6 +3,7 @@ use std::{num::NonZeroUsize, time::Duration};
 use async_trait::async_trait;
 use memory_accounting::{ComponentRegistry, MemoryLimiter};
 use saluki_components::{destinations::PrometheusConfiguration, sources::InternalMetricsConfiguration};
+use saluki_core::health::HealthRegistry;
 use saluki_core::{
     runtime::{
         InitializationError, ProcessShutdown, RestartStrategy, RuntimeConfiguration, Supervisable, Supervisor,
@@ -11,7 +12,6 @@ use saluki_core::{
     topology::TopologyBlueprint,
 };
 use saluki_error::{generic_error, GenericError};
-use saluki_health::HealthRegistry;
 use tracing::info;
 
 use crate::config::DataPlaneConfiguration;

lib/saluki-core/Cargo.toml

@@ -21,10 +21,10 @@ metrics-util = { workspace = true, features = ["recency", "registry"] }
 ordered-float = { workspace = true }
 paste = { workspace = true }
 pin-project = { workspace = true }
+saluki-api = { workspace = true }
 saluki-common = { workspace = true }
 saluki-context = { workspace = true }
 saluki-error = { workspace = true }
-saluki-health = { workspace = true }
 saluki-metrics = { workspace = true }
 serde = { workspace = true, features = ["derive"] }
 serde_json = { workspace = true }

lib/saluki-core/src/components/decoders/context.rs

@@ -1,6 +1,6 @@
 use memory_accounting::ComponentRegistry;
-use saluki_health::Health;
 
+use crate::health::Health;
 use crate::{
     components::ComponentContext,
     topology::{EventsDispatcher, PayloadsConsumer, TopologyContext},

lib/saluki-core/src/components/destinations/context.rs

@@ -1,6 +1,6 @@
 use memory_accounting::ComponentRegistry;
-use saluki_health::Health;
 
+use crate::health::Health;
 use crate::{
     components::ComponentContext,
     topology::{EventsConsumer, TopologyContext},

lib/saluki-core/src/components/encoders/context.rs

@@ -1,6 +1,6 @@
 use memory_accounting::ComponentRegistry;
-use saluki_health::Health;
 
+use crate::health::Health;
 use crate::{
     components::ComponentContext,
     topology::{EventsConsumer, PayloadsDispatcher, TopologyContext},

lib/saluki-core/src/components/forwarders/context.rs

@@ -1,6 +1,6 @@
 use memory_accounting::ComponentRegistry;
-use saluki_health::Health;
 
+use crate::health::Health;
 use crate::{
     components::ComponentContext,
     topology::{PayloadsConsumer, TopologyContext},

lib/saluki-core/src/components/relays/context.rs

@@ -1,8 +1,8 @@
 use std::sync::Arc;
 
 use memory_accounting::ComponentRegistry;
-use saluki_health::Health;
 
+use crate::health::Health;
 use crate::{
     components::ComponentContext,
     topology::{shutdown::ComponentShutdownHandle, PayloadsDispatcher, TopologyContext},

lib/saluki-core/src/components/sources/context.rs

@@ -1,8 +1,8 @@
 use std::sync::Arc;
 
 use memory_accounting::ComponentRegistry;
-use saluki_health::Health;
 
+use crate::health::Health;
 use crate::{
     components::ComponentContext,
     topology::{shutdown::ComponentShutdownHandle, EventsDispatcher, TopologyContext},

lib/saluki-core/src/components/transforms/context.rs

@@ -1,6 +1,6 @@
 use memory_accounting::ComponentRegistry;
-use saluki_health::Health;
 
+use crate::health::Health;
 use crate::{
     components::ComponentContext,
     topology::{EventsConsumer, EventsDispatcher, TopologyContext},

lib/saluki-health/src/api.rs → lib/saluki-core/src/health/api.rs

@@ -9,7 +9,7 @@ use saluki_api::{
 use saluki_common::collections::FastHashMap;
 use serde::Serialize;
 
-use crate::RegistryState;
+use super::RegistryState;
 
 #[derive(Serialize)]
 struct SimpleComponentState {
@@ -84,7 +84,7 @@ pub struct HealthAPIHandler {
 }
 
 impl HealthAPIHandler {
-    pub(crate) fn from_state(inner: Arc<Mutex<RegistryState>>) -> Self {
+    pub(super) fn from_state(inner: Arc<Mutex<RegistryState>>) -> Self {
         Self {
             state: HealthRegistryState { inner },
         }