feat(datadog): Request secrets update from DDAgent

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

@@ -167,12 +167,14 @@ pub async fn handle_run_command(
     let dsd_stats_config = DogStatsDStatisticsConfiguration::new();
 
     // Create the blueprint for our primary topology.
+
     let (blueprint, control_surfaces) = create_topology(
         &config,
         &dp_config,
         &env_provider,
         &component_registry,
         dsd_stats_config.clone(),
+        &ra_bootstrap,
     )
     .await?;
     let (dsd_capture_api_handler, dsd_replay_api_handler) = match control_surfaces.dogstatsd {
@@ -441,6 +443,7 @@ struct DogStatsDControlSurface {
 async fn create_topology(
     config: &GenericConfiguration, dp_config: &DataPlaneConfiguration, env_provider: &ADPEnvironmentProvider,
     component_registry: &ComponentRegistry, dsd_stats_config: DogStatsDStatisticsConfiguration,
+    ra_bootstrap: &Option<RemoteAgentBootstrap>,
 ) -> Result<(TopologyBlueprint, TopologyControlSurfaces), GenericError> {
     let mut blueprint = TopologyBlueprint::new("primary", component_registry);
     let mut control_surfaces = TopologyControlSurfaces::default();
@@ -465,8 +468,13 @@ async fn create_topology(
         || dp_config.service_checks_pipeline_required()
         || dp_config.traces_pipeline_required()
     {
-        let dd_forwarder_config =
+        let mut dd_forwarder_config =
             DatadogConfiguration::from_configuration(config).error_context("Failed to configure Datadog forwarder.")?;
+
+        if let Some(ra_bootstrap) = ra_bootstrap {
+            dd_forwarder_config = dd_forwarder_config
+                .with_config_update_retry_predicate(ra_bootstrap.create_config_update_retry_predicate());
+        }
         blueprint.add_forwarder("dd_out", dd_forwarder_config)?;
     }
 

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

@@ -1,5 +1,5 @@
 use std::collections::HashMap;
-use std::{collections::hash_map::Entry, time::Duration};
+use std::{collections::hash_map::Entry, sync::Arc, time::Duration};
 
 use async_trait::async_trait;
 use chrono::{DateTime, Utc};
@@ -20,6 +20,7 @@ use saluki_core::observability::metrics::{
     get_shared_metrics_state, AggregatedMetricsProcessor, Reflector, TelemetryProcessor,
 };
 use saluki_error::{generic_error, GenericError};
+use saluki_io::net::util::retry::HttpRetryPredicate;
 use saluki_io::net::GrpcTargetAddress;
 use serde_json::{Map, Value};
 use tokio::{
@@ -146,6 +147,31 @@ impl RemoteAgentBootstrap {
 
         receiver
     }
+
+    /// Creates a predicate that requests Agent configuration updates when a 403 response is retried.
+    pub fn create_config_update_retry_predicate(&self) -> HttpRetryPredicate {
+        let client = self.client.clone();
+        let session_id = self.session_id.clone();
+
+        Arc::new(move |_| {
+            let Some(current_session_id) = session_id.get() else {
+                debug!("Cannot request Datadog Agent config updates because no remote agent session ID is available.");
+                return true;
+            };
+
+            let client = client.clone();
+            spawn_traced_named("adp-request-config-updates", async move {
+                match client.request_config_updates(&current_session_id).await {
+                    Ok(_) => debug!(session_id = %current_session_id, "Requested Datadog Agent config updates."),
+                    Err(e) => {
+                        warn!(session_id = %current_session_id, error = %e, "Failed to request Datadog Agent config updates.")
+                    }
+                }
+            });
+
+            true
+        })
+    }
 }
 
 struct RemoteAgentState {

lib/datadog-agent-commons/src/ipc/client/mod.rs

@@ -16,9 +16,9 @@ use saluki_io::net::client::http::HttpsCapableConnectorBuilder;
 use tonic::{
     service::interceptor::InterceptedService,
     transport::{Channel, Endpoint},
-    Code, Request, Response,
+    Code, Request, Response, Status,
 };
-use tracing::warn;
+use tracing::{debug, warn};
 
 use crate::ipc::{config::RemoteAgentClientConfiguration, session::SessionId, tls::build_ipc_client_ipc_tls_config};
 
@@ -222,6 +222,38 @@ impl RemoteAgentClient {
 
         StreamingResponse::from_response_future(async move { client.stream_config_events(request).await })
     }
+
+    /// Requests that the Agent refresh its configuration and publish updates.
+    ///
+    /// A successful RPC response only means the request was delivered to the Agent. Configuration recovery still depends
+    /// on the config stream publishing updated values.
+    ///
+    /// # Errors
+    ///
+    /// If there is an error sending the request to the Agent API, an error will be returned. Older Agents that do not
+    /// implement the RPC are treated as successful no-ops.
+    pub async fn request_config_updates(&self, session_id: &SessionId) -> Result<Response<()>, GenericError> {
+        let mut client = self.secure_client.clone();
+        let mut request = Request::new(());
+
+        request
+            .metadata_mut()
+            .insert("session_id", session_id.to_grpc_header_value());
+
+        match client.request_config_updates(request).await {
+            Ok(response) => Ok(response.map(|_| ())),
+            Err(status) => handle_request_config_updates_error(status).map_err(Into::into),
+        }
+    }
+}
+
+fn handle_request_config_updates_error(status: Status) -> Result<Response<()>, Status> {
+    if status.code() == Code::Unimplemented {
+        debug!("Datadog Agent does not implement RequestConfigUpdates. Continuing without config refresh requests.");
+        Ok(Response::new(()))
+    } else {
+        Err(status)
+    }
 }
 
 async fn try_query_agent_api(
@@ -244,3 +276,26 @@ async fn try_query_agent_api(
         },
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use tonic::Code;
+
+    use super::*;
+
+    #[test]
+    fn request_config_updates_treats_unimplemented_as_noop() {
+        let response = handle_request_config_updates_error(Status::new(Code::Unimplemented, "unknown method"))
+            .expect("unimplemented should be treated as a no-op");
+
+        assert_eq!(response.into_inner(), ());
+    }
+
+    #[test]
+    fn request_config_updates_preserves_other_errors() {
+        let status = handle_request_config_updates_error(Status::new(Code::Unavailable, "agent unavailable"))
+            .expect_err("non-unimplemented errors should be preserved");
+
+        assert_eq!(status.code(), Code::Unavailable);
+    }
+}

lib/protos/datadog/proto/datadog-agent/datadog/api/v1/api.proto

@@ -65,4 +65,7 @@ service AgentSecure {
 
     // Streams config events to the remote agent.
     rpc StreamConfigEvents(datadog.model.v1.ConfigStreamRequest) returns (stream datadog.model.v1.ConfigEvent);
+
+    // Requests that the Agent refresh its configuration and publish updates.
+    rpc RequestConfigUpdates(google.protobuf.Empty) returns (google.protobuf.Empty);
 }

lib/saluki-components/src/common/datadog/config.rs

@@ -10,7 +10,7 @@ use tracing::warn;
 use super::{
     endpoints::{EndpointConfiguration, EndpointRoute, RoutableEndpoint},
     proxy::ProxyConfiguration,
-    retry::RetryConfiguration,
+    retry::{retryable_forbidden_predicate_for_config, RetryConfiguration},
 };
 
 const fn default_endpoint_concurrency() -> usize {
@@ -331,6 +331,14 @@ impl ForwarderConfiguration {
         &self.retry
     }
 
+    /// Configures the retry predicate used to request Core Agent configuration updates for retryable 403 responses.
+    pub(crate) fn with_config_update_retry_predicate(
+        &mut self, config: GenericConfiguration, predicate: saluki_io::net::util::retry::HttpRetryPredicate,
+    ) {
+        self.retry
+            .with_retry_predicate(retryable_forbidden_predicate_for_config(config, predicate));
+    }
+
     /// Returns a reference to the proxy configuration.
     pub const fn proxy(&self) -> &Option<ProxyConfiguration> {
         &self.proxy