[APR-278] Add config refresher to update API key

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

@@ -7,6 +7,7 @@
 #![deny(missing_docs)]
 use std::{
     future::pending,
+    sync::Arc,
     time::{Duration, Instant},
 };
 
@@ -19,7 +20,7 @@ use saluki_components::{
         AggregateConfiguration, ChainedConfiguration, HostEnrichmentConfiguration, OriginEnrichmentConfiguration,
     },
 };
-use saluki_config::{ConfigurationLoader, GenericConfiguration};
+use saluki_config::{ConfigurationLoader, GenericConfiguration, RefreshableConfiguration, RefresherConfiguration};
 use saluki_core::topology::TopologyBlueprint;
 use saluki_error::{ErrorContext as _, GenericError};
 use saluki_health::HealthRegistry;
@@ -176,8 +177,13 @@ fn create_topology(
         .with_transform_builder(host_enrichment_config)
         .with_transform_builder(origin_enrichment_config);
     let internal_metrics_remap_config = AgentTelemetryRemapperConfiguration::new();
-    let dd_metrics_config = DatadogMetricsConfiguration::from_configuration(configuration)
+
+    let refresher_configuration = RefresherConfiguration::from_configuration(configuration)?;
+    let refreshable_configuration: Arc<RefreshableConfiguration> = refresher_configuration.build()?;
+
+    let mut dd_metrics_config = DatadogMetricsConfiguration::from_configuration(configuration)
         .error_context("Failed to configure Datadog Metrics destination.")?;
+    dd_metrics_config.add_refreshable_configuration(refreshable_configuration.clone());
     let events_service_checks_config = DatadogEventsServiceChecksConfiguration::from_configuration(configuration)
         .error_context("Failed to configure Datadog Events/Service Checks destination.")?;
 

lib/saluki-components/src/destinations/datadog_metrics/mod.rs

@@ -9,7 +9,7 @@ use http_body::Body;
 use http_body_util::BodyExt as _;
 use hyper::body::Incoming;
 use memory_accounting::{MemoryBounds, MemoryBoundsBuilder};
-use saluki_config::GenericConfiguration;
+use saluki_config::{GenericConfiguration, RefreshableConfiguration};
 use saluki_core::{
     components::{destinations::*, ComponentContext},
     pooling::{FixedSizeObjectPool, ObjectPool},
@@ -141,6 +141,9 @@ pub struct DatadogMetricsConfiguration {
     /// Defaults to empty.
     #[serde(default, rename = "additional_endpoints")]
     additional_endpoints: AdditionalEndpoints,
+
+    #[serde(skip)]
+    config_refresher: Arc<RefreshableConfiguration>,
 }
 
 fn default_request_timeout_secs() -> u64 {
@@ -172,6 +175,11 @@ impl DatadogMetricsConfiguration {
     pub fn from_configuration(config: &GenericConfiguration) -> Result<Self, GenericError> {
         Ok(config.as_typed()?)
     }
+
+    /// Add option to retrieve configuration values from a `RefreshableConfiguration`.
+    pub fn add_refreshable_configuration(&mut self, refresher: Arc<RefreshableConfiguration>) {
+        self.config_refresher = refresher;
+    }
 }
 
 #[async_trait]
@@ -211,11 +219,14 @@ impl DestinationBuilder for DatadogMetricsConfiguration {
         let series_request_builder = RequestBuilder::new(MetricsEndpoint::Series, rb_buffer_pool.clone()).await?;
         let sketches_request_builder = RequestBuilder::new(MetricsEndpoint::Sketches, rb_buffer_pool).await?;
 
+        let refresher = self.config_refresher.clone();
+
         Ok(Box::new(DatadogMetrics {
             service,
             series_request_builder,
             sketches_request_builder,
             endpoints,
+            refresher,
         }))
     }
 }
@@ -259,6 +270,7 @@ where
     series_request_builder: RequestBuilder<O>,
     sketches_request_builder: RequestBuilder<O>,
     endpoints: Vec<ResolvedEndpoint>,
+    refresher: Arc<RefreshableConfiguration>,
 }
 
 #[allow(unused)]
@@ -276,6 +288,7 @@ where
             mut sketches_request_builder,
             service,
             endpoints,
+            refresher,
         } = *self;
 
         let mut health = context.take_health_handle();
@@ -291,6 +304,7 @@ where
             telemetry.clone(),
             context.component_context(),
             endpoints,
+            refresher,
         ));
 
         health.mark_ready();
@@ -430,6 +444,7 @@ where
 async fn run_io_loop<S, B>(
     mut requests_rx: mpsc::Receiver<(usize, Request<B>)>, io_shutdown_tx: oneshot::Sender<()>, service: S,
     telemetry: ComponentTelemetry, component_context: ComponentContext, resolved_endpoints: Vec<ResolvedEndpoint>,
+    refresher: Arc<RefreshableConfiguration>,
 ) where
     S: Service<Request<B>, Response = hyper::Response<Incoming>> + Clone + Send + 'static,
     S::Future: Send,
@@ -455,6 +470,7 @@ async fn run_io_loop<S, B>(
             telemetry.clone(),
             component_context.clone(),
             resolved_endpoint,
+            refresher.clone(),
         ));
 
         endpoint_txs.push((endpoint_url, endpoint_tx));
@@ -489,6 +505,7 @@ async fn run_io_loop<S, B>(
 async fn run_endpoint_io_loop<S, B>(
     mut requests_rx: mpsc::Receiver<(usize, Request<B>)>, task_barrier: Arc<Barrier>, service: S,
     telemetry: ComponentTelemetry, context: ComponentContext, endpoint: ResolvedEndpoint,
+    refresher: Arc<RefreshableConfiguration>,
 ) where
     S: Service<Request<B>, Response = hyper::Response<Incoming>> + Send + 'static,
     S::Future: Send,
@@ -504,7 +521,7 @@ async fn run_endpoint_io_loop<S, B>(
     // of the URI, adding the API key as a header, and so on.
     let mut service = ServiceBuilder::new()
         // Set the request's URI to the endpoint's URI, and add the API key as a header.
-        .map_request(for_resolved_endpoint::<B>(endpoint))
+        .map_request(for_resolved_endpoint::<B>(endpoint, refresher.clone()))
         // Set the User-Agent and DD-Agent-Version headers indicating the version of the data plane sending the request.
         .map_request(with_version_info::<B>())
         // Add telemetry about the requests.
@@ -552,14 +569,21 @@ async fn run_endpoint_io_loop<S, B>(
     task_barrier.wait().await;
 }
 
-fn for_resolved_endpoint<B>(endpoint: ResolvedEndpoint) -> impl Fn(Request<B>) -> Request<B> + Clone {
+#[allow(unused)]
+fn for_resolved_endpoint<B>(
+    endpoint: ResolvedEndpoint, refresher: Arc<RefreshableConfiguration>,
+) -> impl Fn(Request<B>) -> Request<B> + Clone {
     let new_uri_authority = Authority::try_from(endpoint.endpoint().authority())
         .expect("should not fail to construct new endpoint authority");
     let new_uri_scheme =
         Scheme::try_from(endpoint.endpoint().scheme()).expect("should not fail to construct new endpoint scheme");
-    let api_key_value =
-        HeaderValue::from_str(endpoint.api_key()).expect("should not fail to construct API key header value");
-
+    let mut api_key: String;
+    if let Ok(refresher_api_key) = refresher.get_typed::<String>("api_key") {
+        api_key = refresher_api_key.clone();
+    } else {
+        api_key = endpoint.api_key().to_string();
+    }
+    let api_key_value = HeaderValue::from_str(&api_key).expect("should not fail to construct API key header value");
     move |mut request| {
         // Build an updated URI by taking the endpoint URL and slapping the request's URI path on the end of it.
         let new_uri = Uri::builder()

lib/saluki-components/src/destinations/datadog_metrics/request_builder.rs

@@ -105,6 +105,7 @@ impl MetricsEndpoint {
     }
 }
 
+#[allow(unused)]
 pub struct RequestBuilder<O>
 where
     O: ObjectPool<Item = BytesBuffer> + 'static,
@@ -128,7 +129,6 @@ where
     pub async fn new(endpoint: MetricsEndpoint, buffer_pool: O) -> Result<Self, RequestBuilderError> {
         let chunked_buffer_pool = ChunkedBytesBufferObjectPool::new(buffer_pool);
         let compressor = create_compressor(&chunked_buffer_pool).await;
-
         Ok(Self {
             endpoint,
             endpoint_uri: endpoint.endpoint_uri(),

lib/saluki-config/Cargo.toml

@@ -6,8 +6,12 @@ license = { workspace = true }
 repository = { workspace = true }
 
 [dependencies]
+arc-swap = { workspace = true }
 figment = { workspace = true, features = ["env", "json", "yaml"] }
+http = { workspace = true }
+reqwest = { workspace = true, features = ["default-tls", "json"] }
 saluki-error = { workspace = true }
+saluki-io = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
 snafu = { workspace = true }

lib/saluki-config/src/lib.rs

@@ -12,8 +12,10 @@ use snafu::{ResultExt as _, Snafu};
 use tracing::debug;
 
 mod provider;
+mod refresher;
 mod secrets;
 use self::provider::ResolvedProvider;
+pub use self::refresher::{RefreshableConfiguration, RefresherConfiguration};
 
 /// A configuration error.
 #[derive(Debug, Snafu)]