datadog_agent_runtime/control_plane/
config_updates.rs

1//! Supervised worker that keeps the configuration current.
2
3use std::future::Future;
4
5use async_trait::async_trait;
6use saluki_common::sync::shutdown::ShutdownHandle;
7use saluki_core::runtime::{InitializationError, Supervisable, SupervisorFuture};
8use saluki_error::{ErrorContext as _, GenericError};
9use tokio::select;
10
11/// Applies the configuration updates that a remote agent's configuration system receives from the Datadog Agent.
12pub trait ConfigUpdatesRunner: Send + Sync + 'static {
13    /// Applies each update from the Datadog Agent until the configuration stream closes.
14    ///
15    /// Each call **MUST** continue from where the previous call stopped, so that the caller can run this again after a
16    /// call fails or panics.
17    ///
18    /// # Errors
19    ///
20    /// Returns an error when the configuration stream closes. If the stream can't reopen, each later call **SHOULD**
21    /// fail in the same way at once.
22    fn run(&self) -> impl Future<Output = Result<(), GenericError>> + Send + 'static;
23}
24
25/// A worker that applies configuration updates from the Datadog Agent.
26///
27/// The worker runs until it is shut down, or until the Agent configuration stream closes, in which case it fails. Each
28/// restart continues with the same stream, so a restart recovers from a panic while an update is applied. A restart
29/// does not recover from a closed stream: the stream does not reopen, so the restarted worker fails again at once.
30pub struct ConfigUpdatesWorker<U>
31where
32    U: ConfigUpdatesRunner,
33{
34    updates: U,
35}
36
37impl<U> ConfigUpdatesWorker<U>
38where
39    U: ConfigUpdatesRunner,
40{
41    /// Creates a new [`ConfigUpdatesWorker`] that applies the given updates.
42    pub fn new(updates: U) -> Self {
43        Self { updates }
44    }
45}
46
47#[async_trait]
48impl<U> Supervisable for ConfigUpdatesWorker<U>
49where
50    U: ConfigUpdatesRunner,
51{
52    fn name(&self) -> &str {
53        "config-updates"
54    }
55
56    async fn initialize(&self, process_shutdown: ShutdownHandle) -> Result<SupervisorFuture, InitializationError> {
57        let updates = self.updates.run();
58
59        Ok(Box::pin(async move {
60            select! {
61                _ = process_shutdown => Ok(()),
62                result = updates => result.error_context("Stopped applying configuration updates from the Datadog Agent."),
63            }
64        }))
65    }
66}
datadog_agent_runtime/control_plane/config_updates.rs

datadog_agent_runtime/control_plane/
config_updates.rs
