feat(datadog): validate API keys at runtime to track forwarder health (#1841)

## Summary
<!-- Please provide a brief summary about what this PR does.
This should help the reviewers give feedback faster and with higher quality. -->

ADP now supports runtime API key refresh through a configurable `u64` key (which defaults to `60 minutes`).

Things to note: 

- Additional endpoints added during runtime will not be accounted for since it would involve dynamically altering the topology by creating a new forwarder and related components to support the new endpoint. This _also_ mimics Core Agent behavior.
- Non-zero integer values fall back to the default (`60`)
- Fake API key is not treated as invalid
- Forwarders' health now dictated by whether at least **one** API-key is valid. 
- `200` is the valid status code 
- `403` is invalid
- ADP will treat unexpected status codes as errors and not invalid keys. 
- Integration tests will now use the fake API key (32-char string of all 0s)


## Change Type
- [ ] Bug fix
- [x] New feature
- [ ] Non-functional (chore, refactoring, docs)
- [ ] Performance


## How did you test this PR?
<!-- Please how you tested these changes here -->

Unit tests

## References

<!-- Please list any issues closed by this PR. -->

<!--
- Closes: <issue link>
-->
- Closes #1357
<!-- Any other issues or PRs relevant to this PR? Feel free to list them here. -->


Co-authored-by: lucas.tembras <lucas.tembras@datadoghq.com>

Author: lucastemb

docs/agent-data-plane/configuration/dogstatsd.md

@@ -30,7 +30,6 @@ tracking.
 | `dogstatsd_experimental_http.listen_address`                   | Bind address for experimental HTTP DSD listener | [#1682] |
 | `dogstatsd_pipe_name`                                          | Windows named pipe path                         | [#1466] |
 | `dogstatsd_windows_pipe_security_descriptor`                   | Windows named pipe ACL descriptor               | [#1466] |
-| `forwarder_apikey_validation_interval`                         | API key check interval (minutes)                | [#1357] |
 | `serializer_experimental_use_v3_api.compression_level`         | V3 API zstd compression level                   | [#1468] |
 | `serializer_experimental_use_v3_api.series.endpoints`          | Endpoints enabled for V3 series API             | [#1468] |
 | `serializer_experimental_use_v3_api.series.shadow_sample_rate` | V3 series shadow mode sample rate               | [#1468] |
@@ -158,18 +157,19 @@ deployment path, ADP does not implement `heroku_dyno`. See [#1753].
 The following settings are recognized by both ADP and the core agent, but with different behavior or
 default values.
 
-| Config Key                         | Description                               |
-| ---------------------------------- | ----------------------------------------- |
-| `dogstatsd_mapper_cache_size`      | Mapper result LRU cache size              |
-| `dogstatsd_metrics_stats_enable`   | Enable per-metric debug stats             |
-| `forwarder_high_prio_buffer_size`  | High-priority request queue size          |
-| `forwarder_num_workers`            | Concurrent forwarder workers              |
-| `log_level`                        | Log verbosity directives                  |
-| `min_tls_version`                  | Minimum TLS version for HTTPS connections |
-| `multi_region_failover.enabled`    | Enable multi-region failover mode         |
-| `serializer_zstd_compressor_level` | Zstd compression level                    |
-| `skip_ssl_validation`              | Skip TLS cert validation                  |
-| `statsd_forward_host`              | UDP packet forwarding destination host    |
+| Config Key                             | Description                               |
+| -------------------------------------- | ----------------------------------------- |
+| `dogstatsd_mapper_cache_size`          | Mapper result LRU cache size              |
+| `dogstatsd_metrics_stats_enable`       | Enable per-metric debug stats             |
+| `forwarder_apikey_validation_interval` | API key check interval (minutes)          |
+| `forwarder_high_prio_buffer_size`      | High-priority request queue size          |
+| `forwarder_num_workers`                | Concurrent forwarder workers              |
+| `log_level`                            | Log verbosity directives                  |
+| `min_tls_version`                      | Minimum TLS version for HTTPS connections |
+| `multi_region_failover.enabled`        | Enable multi-region failover mode         |
+| `serializer_zstd_compressor_level`     | Zstd compression level                    |
+| `skip_ssl_validation`                  | Skip TLS cert validation                  |
+| `statsd_forward_host`                  | UDP packet forwarding destination host    |
 
 ### `dogstatsd_mapper_cache_size`
 
@@ -191,6 +191,20 @@ mapper, clear `dogstatsd_mapper_profiles` instead when running ADP.
 
 See `dogstatsd_stats_enable`
 
+### `forwarder_apikey_validation_interval`
+
+ADP supports `forwarder_apikey_validation_interval` for Datadog intake forwarding. The value is in minutes,
+defaults to `60`, and falls back to that default for non-positive values.
+
+ADP validates configured API keys when the forwarder starts, periodically at the configured interval, and
+after API-key-related runtime config updates. Validation controls readiness only: confirmed invalid keys can
+make the forwarder not ready, but the forwarder keeps running and transient validation failures do not prove
+that keys are invalid.
+
+Support is partial because ADP only tracks `additional_endpoints` domains present at forwarder startup.
+Runtime updates can rotate keys for those domains, but brand-new domains are not added or validated until
+restart.
+
 ### `forwarder_high_prio_buffer_size`
 
 ADP implements `forwarder_high_prio_buffer_size` and matches the core agent's default of `100`. The

lib/datadog-agent/config-testing/src/config_registry/forwarder.rs

@@ -313,6 +313,17 @@ crate::declare_annotations! {
         test_json: None,
         pipeline_affinity: PipelineAffinity::CrossCutting,
     };
+    /// `forwarder_apikey_validation_interval`-API key check interval (minutes)
+    FORWARDER_APIKEY_VALIDATION_INTERVAL = SalukiAnnotation {
+        schema: &schema::FORWARDER_APIKEY_VALIDATION_INTERVAL,
+        support_level: SupportLevel::Partial,
+        additional_yaml_paths: &[],
+        env_var_override: None,
+        used_by: &[structs::FORWARDER_CONFIGURATION],
+        value_type_override: Some(ValueType::Integer),
+        test_json: None,
+        pipeline_affinity: PipelineAffinity::CrossCutting,
+    };
     /// `forwarder_max_concurrent_requests`-Max concurrent HTTP requests
     FORWARDER_MAX_CONCURRENT_REQUESTS = SalukiAnnotation {
         schema: &schema::FORWARDER_MAX_CONCURRENT_REQUESTS,

lib/datadog-agent/config-testing/src/config_registry/unsupported.rs

@@ -82,17 +82,6 @@ crate::declare_annotations! {
         test_json: None,
         pipeline_affinity: PipelineAffinity::Pipelines(&[Pipeline::DogStatsD]),
     };
-    /// `forwarder_apikey_validation_interval`-API key check interval (minutes)
-    FORWARDER_APIKEY_VALIDATION_INTERVAL = SalukiAnnotation {
-        schema: &schema::FORWARDER_APIKEY_VALIDATION_INTERVAL,
-        support_level: SupportLevel::Incompatible(Severity::Medium),
-        additional_yaml_paths: &[],
-        env_var_override: None,
-        used_by: &[],
-        value_type_override: None,
-        test_json: None,
-        pipeline_affinity: PipelineAffinity::CrossCutting,
-    };
     /// `forwarder_flush_to_disk_mem_ratio`-Mem-to-disk flush threshold
     FORWARDER_FLUSH_TO_DISK_MEM_RATIO = SalukiAnnotation {
         schema: &schema::FORWARDER_FLUSH_TO_DISK_MEM_RATIO,

lib/datadog-agent/config/schema/schema_overlay.yaml

@@ -1143,13 +1143,27 @@ inventory:
         config_registry_filename: encoders.rs
 
   forwarder_apikey_validation_interval:
-    support: none
-    severity: medium
-    planned: true
+    support: partial
     pipelines: [cross_cutting]
     description: "API key check interval (minutes)"
-    documentation: "Core agent supports runtime API key refresh without restart. ADP has partial infrastructure (config-side listeners) but the forwarder does not act on updates."
+    documentation: |
+      ADP supports `forwarder_apikey_validation_interval` for Datadog intake forwarding. The value is in minutes,
+      defaults to `60`, and falls back to that default for non-positive values.
+
+      ADP validates configured API keys when the forwarder starts, periodically at the configured interval, and
+      after API-key-related runtime config updates. Validation controls readiness only: confirmed invalid keys can
+      make the forwarder not ready, but the forwarder keeps running and transient validation failures do not prove
+      that keys are invalid.
+
+      Support is partial because ADP only tracks `additional_endpoints` domains present at forwarder startup.
+      Runtime updates can rotate keys for those domains, but brand-new domains are not added or validated until
+      restart.
     issue: "#1357"
+    test_support:
+      used_by: [ForwarderConfiguration]
+      value_type_override: integer
+      additional_attributes:
+        config_registry_filename: forwarder.rs
 
   forwarder_backoff_base:
     support: full

lib/datadog-agent/config/src/classifier/classifier_data.rs

@@ -291,13 +291,6 @@ pub(crate) static CLASSIFIER_ENTRIES: &[ClassifierEntry] = &[
         pipeline_affinity: PipelineAffinity::Pipelines(&[Pipeline::DogStatsD]),
         default: Some("\"\""),
     },
-    ClassifierEntry {
-        yaml_path: "forwarder_apikey_validation_interval",
-        aliases: &[],
-        support_level: SupportLevel::Incompatible(Severity::Medium),
-        pipeline_affinity: PipelineAffinity::CrossCutting,
-        default: Some("60"),
-    },
     ClassifierEntry {
         yaml_path: "heroku_dyno",
         aliases: &[],

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

@@ -33,6 +33,14 @@ const fn default_forwarder_connection_reset_interval() -> u64 {
     0
 }
 
+const fn default_api_key_validation_interval_mins() -> u64 {
+    60
+}
+
+const fn default_api_key_validation_interval_config_mins() -> i64 {
+    default_api_key_validation_interval_mins() as i64
+}
+
 const MIN_TLS_VERSION_TLS10: &str = "tlsv1.0";
 const MIN_TLS_VERSION_TLS11: &str = "tlsv1.1";
 const MIN_TLS_VERSION_TLS12: &str = "tlsv1.2";
@@ -259,6 +267,18 @@ pub struct ForwarderConfiguration {
     /// outbound intake request. The data plane does not perform local tag validation based on this setting.
     #[serde(default)]
     allow_arbitrary_tags: bool,
+
+    /// API key validation interval, in minutes.
+    ///
+    /// All values that are less than or equal to zero will be ignored, and the default
+    /// value will be used.
+    ///
+    /// Defaults to 60 minutes.
+    #[serde(
+        default = "default_api_key_validation_interval_config_mins",
+        rename = "forwarder_apikey_validation_interval"
+    )]
+    api_key_validation_interval_mins: i64,
 }
 
 impl ForwarderConfiguration {
@@ -267,6 +287,15 @@ impl ForwarderConfiguration {
         let mut forwarder_config = config.as_typed::<Self>()?;
         forwarder_config.parsed_min_tls_version = min_tls_version_from_config_value(&forwarder_config.min_tls_version);
 
+        if forwarder_config.api_key_validation_interval_mins <= 0 {
+            warn!(
+                config_key = "forwarder_apikey_validation_interval",
+                fallback_minutes = default_api_key_validation_interval_mins(),
+                "Configured API key validation interval is invalid; using default."
+            );
+            forwarder_config.api_key_validation_interval_mins = default_api_key_validation_interval_mins() as i64;
+        }
+
         // Handle fixing up the forwarder storage path if it's empty.
         forwarder_config.retry.fix_empty_storage_path(config);
 
@@ -407,6 +436,11 @@ impl ForwarderConfiguration {
     pub const fn allow_arbitrary_tags(&self) -> bool {
         self.allow_arbitrary_tags
     }
+
+    /// Returns the API key validation interval.
+    pub const fn api_key_validation_interval(&self) -> Duration {
+        Duration::from_mins(self.api_key_validation_interval_mins as u64)
+    }
 }
 
 #[cfg(test)]
@@ -583,6 +617,32 @@ mod tests {
         assert_eq!(config.endpoint_concurrency(), 12);
     }
 
+    #[tokio::test]
+    async fn api_key_validation_interval_parsing() {
+        let cases = [
+            ("missing", serde_json::json!({}), Duration::from_mins(60)),
+            (
+                "positive",
+                serde_json::json!({ "forwarder_apikey_validation_interval": 5i64 }),
+                Duration::from_mins(5),
+            ),
+            (
+                "zero",
+                serde_json::json!({ "forwarder_apikey_validation_interval": 0i64 }),
+                Duration::from_mins(60),
+            ),
+            (
+                "negative",
+                serde_json::json!({ "forwarder_apikey_validation_interval": -1i64 }),
+                Duration::from_mins(60),
+            ),
+        ];
+
+        for (case_name, extra_config, expected_interval) in cases {
+            let config = forwarder_config_from(config_with(extra_config), None).await;
+            assert_eq!(config.api_key_validation_interval(), expected_interval, "{case_name}");
+        }
+    }
     #[tokio::test]
     async fn skip_ssl_validation_defaults_to_false() {
         let config = forwarder_config_from(base_config(), None).await;

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

@@ -280,6 +280,11 @@ impl RoutableEndpoint {
         &self.endpoint
     }
 
+    /// Returns the resolved endpoint mutably.
+    pub(crate) const fn endpoint_mut(&mut self) -> &mut ResolvedEndpoint {
+        &mut self.endpoint
+    }
+
     /// Consumes the routable endpoint and returns its parts.
     pub(crate) fn into_parts(self) -> (EndpointRoute, ResolvedEndpoint) {
         (self.route, self.endpoint)

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

@@ -40,6 +40,7 @@ use super::{
     middleware::{for_resolved_endpoint, with_allow_arbitrary_tags, with_version_info},
     telemetry::{ComponentTelemetry, SharedTransactionQueueTelemetry, TransactionQueueTelemetry},
     transaction::{Metadata, Transaction, TransactionBody},
+    validation::ApiKeyValidator,
     METRIC_INTAKE_PATHS,
 };
 
@@ -234,6 +235,16 @@ where
             io_shutdown_rx,
         }
     }
+
+    /// Returns API key validation for the startup endpoint set.
+    pub(crate) fn api_key_validator(&self) -> ApiKeyValidator {
+        ApiKeyValidator::new(
+            self.endpoints.clone(),
+            self.client.clone(),
+            self.live_config.clone(),
+            self.config.api_key_validation_interval(),
+        )
+    }
 }
 
 #[allow(clippy::too_many_arguments)]

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

@@ -9,6 +9,7 @@ pub mod request_builder;
 mod retry;
 pub mod telemetry;
 pub mod transaction;
+pub mod validation;
 
 use saluki_core::data_model::event::trace::{AttributeValue, Trace};
 use stringtheory::MetaString;