saluki_core/health/
worker.rs

1use async_trait::async_trait;
2use saluki_api::{DynamicRoute, EndpointType};
3use saluki_common::sync::shutdown::ShutdownHandle;
4use saluki_error::generic_error;
5
6use super::HealthRegistry;
7use crate::{
8    diagnostic::DiagnosticsEmitter,
9    runtime::{state::DataspaceRegistry, InitializationError, Supervisable, SupervisorFuture},
10    support::SubsystemIdentifier,
11};
12
13/// A worker that runs the health registry.
14///
15/// This is the only way to run the health registry's liveness probing event loop. The worker
16/// implements [`Supervisable`], so it should be added to a [`Supervisor`][crate::runtime::Supervisor]
17/// to be managed as part of a supervision tree.
18pub struct HealthRegistryWorker {
19    health_registry: HealthRegistry,
20}
21
22impl HealthRegistryWorker {
23    pub(super) fn new(health_registry: HealthRegistry) -> Self {
24        Self { health_registry }
25    }
26}
27
28#[async_trait]
29impl Supervisable for HealthRegistryWorker {
30    fn name(&self) -> &str {
31        "health-registry"
32    }
33
34    async fn initialize(&self, process_shutdown: ShutdownHandle) -> Result<SupervisorFuture, InitializationError> {
35        let runner = self.health_registry.clone().into_runner()?;
36
37        let health_routes = DynamicRoute::http(EndpointType::Unprivileged, self.health_registry.api_handler());
38
39        let health_registry = self.health_registry.clone();
40
41        Ok(Box::pin(async move {
42            let dataspace =
43                DataspaceRegistry::try_current().ok_or_else(|| generic_error!("Dataspace not available."))?;
44
45            // Register our API routes before we actually start running.
46            dataspace.assert(health_routes, "health-registry-api");
47
48            // Expose our diagnostic artifact via the diagnostics control surface.
49            let diagnostics =
50                DiagnosticsEmitter::from_dataspace(SubsystemIdentifier::from_segments(["health-registry"]), dataspace);
51            diagnostics.register_collector("health.json", move || health_registry.snapshot_json());
52
53            // We pass the shutdown handle into the runner here, instead of our usual `select! { shutdown => ...,
54            // main_loop_future => ... }` pattern because we try to ensure that we give back the liveness receiver
55            // before the runner completes.
56            //
57            // TODO: We should actually use something like a proper mutex guard so that returning the receiver happens
58            // automatically when the runner future goes out of scope and is dropped, since right now we wouldn't be
59            // able to ensure the current behavior (returning the receiver before the runner completes) happens in the
60            // face of an exceptional error.
61            runner.run(process_shutdown).await;
62
63            Ok(())
64        }))
65    }
66}
saluki_core/health/worker.rs

saluki_core/health/
worker.rs
