chore(config): migrate DatadogEventsConfiguration to typed config (#1981)

## AI Summary

Build `DatadogEventsConfiguration` from the typed shared configuration
model instead of deserializing the generic configuration map. Remove
component-local source defaults and retire the configuration smoke test.

## Change Type
- [x] Non-functional (chore, refactoring, docs)

## How did you test this PR?

- `make build-schema-overlay && make fmt`
- `make check-fmt check-docs`
- `make test`
- `make check-deny check-unused-deps check-licenses check-features
generate-api-docs`
- `cargo check -p saluki-components -p agent-data-plane`

## References

No issue.

Author: webern

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

@@ -412,7 +412,7 @@ async fn create_topology(
     }
 
     if dp_config.events_pipeline_required() {
-        add_baseline_events_pipeline_to_blueprint(&mut blueprint, config).await?;
+        add_baseline_events_pipeline_to_blueprint(&mut blueprint, config_system).await?;
     }
 
     if dp_config.service_checks_pipeline_required() {
@@ -599,11 +599,15 @@ async fn add_baseline_logs_pipeline_to_blueprint(
 }
 
 async fn add_baseline_events_pipeline_to_blueprint(
-    blueprint: &mut TopologyBlueprint, config: &GenericConfiguration,
+    blueprint: &mut TopologyBlueprint, config_system: &ConfigurationSystem,
 ) -> Result<(), GenericError> {
-    let dd_events_config = DatadogEventsConfiguration::from_configuration(config)
-        .map(BufferedIncrementalConfiguration::from_encoder_builder)
-        .error_context("Failed to configure Datadog Events encoder.")?;
+    let saluki = config_system.config();
+    let dd_events_config = DatadogEventsConfiguration::from_configuration(
+        &saluki.shared.metrics_encoding,
+        &saluki.shared.endpoints.compression,
+    )
+    .map(BufferedIncrementalConfiguration::from_encoder_builder)
+    .error_context("Failed to configure Datadog Events encoder.")?;
 
     blueprint
         .add_encoder("dd_events_encode", dd_events_config)?

lib/agent-data-plane-config-system/src/translators/datadog_translator.rs

@@ -22,7 +22,7 @@ use agent_data_plane_config::control::ListenAddress;
 use agent_data_plane_config::domains::dogstatsd::{
     FilterAction, MapperProfile, MetricMapping, MetricTagFilterEntry, OriginTagCardinality,
 };
-use agent_data_plane_config::shared::ForwarderHttpProtocol;
+use agent_data_plane_config::shared::{ForwarderHttpProtocol, ZSTD_DEFAULT_OVERRIDE};
 use agent_data_plane_config::SalukiConfiguration;
 use bytesize::ByteSize;
 use datadog_agent_config::{drive, DatadogConfigWitness, DatadogConfiguration, TranslateError, TranslateErrors};
@@ -952,7 +952,19 @@ impl DatadogConfigWitness for DatadogTranslator<'_> {
     }
 
     fn consume_serializer_zstd_compressor_level(&mut self, value: i64) {
-        self.config.shared.endpoints.compression.zstd_compressor_level = value as i32;
+        // TODO: The core Agent streams a fully resolved config, so its schema default for
+        // `serializer_zstd_compressor_level` arrives here as a concrete value rather than being
+        // absent. When the incoming level matches that Agent default we swap in ADP's intended
+        // level; without this the Agent default would silently override it. We compare against the
+        // schema-generated default, so an operator who deliberately sets exactly the Agent default is
+        // indistinguishable from the default itself and also gets overridden. Removing that ambiguity
+        // needs per-value source tracking (user-set vs Agent default), which is follow-up work.
+        let agent_default = DatadogConfiguration::default().serializer_zstd_compressor_level;
+        self.config.shared.endpoints.compression.zstd_compressor_level = if value == agent_default {
+            ZSTD_DEFAULT_OVERRIDE
+        } else {
+            value as i32
+        };
     }
 
     fn consume_site(&mut self, value: String) {

lib/agent-data-plane-config/src/shared.rs

@@ -118,6 +118,14 @@ pub struct Tls {
     pub sslkeylogfile: String,
 }
 
+/// zstd compression level ADP substitutes when the core Agent supplies its own schema default. ADP
+/// compresses harder than the Agent, whose schema default for `serializer_zstd_compressor_level` is
+/// lower. Because the Agent streams a fully resolved config, that default arrives as a concrete
+/// value; when the incoming level is exactly the Agent default, the translator swaps in this level
+/// so the Agent default does not quietly lower ADP's compression. Any other value is treated as an
+/// explicit choice and left untouched.
+pub const ZSTD_DEFAULT_OVERRIDE: i32 = 3;
+
 /// Payload compression settings applied before transmission.
 #[derive(Clone, Debug, Default, PartialEq, Serialize)]
 pub struct Compression {

lib/saluki-components/src/encoders/datadog/events/mod.rs

@@ -1,11 +1,11 @@
+use agent_data_plane_config::shared::{Compression, MetricsEncoding};
 use async_trait::async_trait;
 use datadog_protos::events as proto;
 use facet::Facet;
 use http::{uri::PathAndQuery, HeaderValue, Method, Uri};
 use protobuf::{rt::WireType, CodedOutputStream};
 use resource_accounting::{MemoryBounds, MemoryBoundsBuilder};
 use saluki_common::iter::ReusableDeduplicator;
-use saluki_config::GenericConfiguration;
 use saluki_context::tags::Tag;
 use saluki_core::{
     components::{encoders::*, ComponentContext},
@@ -19,7 +19,6 @@ use saluki_core::{
 use saluki_error::{ErrorContext as _, GenericError};
 use saluki_io::compression::CompressionScheme;
 use saluki_metrics::MetricsBuilder;
-use serde::Deserialize;
 use tracing::{debug, error, warn};
 
 use crate::common::datadog::{
@@ -28,95 +27,56 @@ use crate::common::datadog::{
     request_builder::{EndpointEncoder, RequestBuilder},
     telemetry::ComponentTelemetry,
     DEFAULT_INTAKE_COMPRESSED_SIZE_LIMIT, DEFAULT_INTAKE_UNCOMPRESSED_SIZE_LIMIT,
-    DEFAULT_SERIALIZER_COMPRESSED_SIZE_LIMIT, DEFAULT_SERIALIZER_UNCOMPRESSED_SIZE_LIMIT,
 };
 
-const DEFAULT_SERIALIZER_COMPRESSOR_KIND: &str = "zstd";
 const MAX_EVENTS_PER_PAYLOAD: usize = 100;
 const EVENTS_FIELD_NUMBER: u32 = 1;
 
 static CONTENT_TYPE_PROTOBUF: HeaderValue = HeaderValue::from_static("application/x-protobuf");
 
-fn default_serializer_compressor_kind() -> String {
-    DEFAULT_SERIALIZER_COMPRESSOR_KIND.to_owned()
-}
-
-const fn default_zstd_compressor_level() -> i32 {
-    3
-}
-
-const fn default_max_payload_size() -> usize {
-    DEFAULT_SERIALIZER_COMPRESSED_SIZE_LIMIT
-}
-
-const fn default_max_uncompressed_payload_size() -> usize {
-    DEFAULT_SERIALIZER_UNCOMPRESSED_SIZE_LIMIT
-}
-
-const fn default_log_payloads() -> bool {
-    false
-}
-
 /// Datadog Events incremental encoder.
 ///
 /// Generates Datadog Events payloads for the Datadog platform.
-#[derive(Deserialize, Facet)]
-#[cfg_attr(test, derive(Debug, PartialEq, serde::Serialize))]
+#[derive(Facet)]
 pub struct DatadogEventsConfiguration {
     /// Maximum compressed size, in bytes, of an events payload.
     ///
     /// This uses the same generic event payload setting as the Datadog Agent. ADP sends events to
     /// `/api/v1/events_batch`, so the effective value is clamped to that endpoint's global intake limit of 3,200,000
     /// bytes. If set to `0`, every non-empty compressed payload exceeds the limit and is dropped during flush.
-    ///
-    /// Defaults to 2,621,440 bytes.
-    #[serde(rename = "serializer_max_payload_size", default = "default_max_payload_size")]
     max_payload_size: usize,
 
     /// Maximum uncompressed size, in bytes, of an events payload.
     ///
     /// This uses the same generic event payload setting as the Datadog Agent. ADP sends events to
     /// `/api/v1/events_batch`, so the effective value is clamped to that endpoint's global intake limit of 62,914,560
     /// bytes. Values smaller than the minimum endpoint framing size prevent the request builder from starting.
-    ///
-    /// Defaults to 4,194,304 bytes.
-    #[serde(
-        rename = "serializer_max_uncompressed_payload_size",
-        default = "default_max_uncompressed_payload_size"
-    )]
     max_uncompressed_payload_size: usize,
 
     /// Compression kind to use for the request payloads.
-    ///
-    /// Defaults to `zstd`.
-    #[serde(
-        rename = "serializer_compressor_kind",
-        default = "default_serializer_compressor_kind"
-    )]
     compressor_kind: String,
 
     /// Compressor level to use when the compressor kind is `zstd`.
-    ///
-    /// Defaults to 3.
-    #[serde(
-        rename = "serializer_zstd_compressor_level",
-        default = "default_zstd_compressor_level"
-    )]
     zstd_compressor_level: i32,
 
     /// Whether to log event payload contents before encoding.
     ///
     /// This logs decoded event objects, not the encoded HTTP body.
-    ///
-    /// Defaults to `false`.
-    #[serde(default = "default_log_payloads")]
     log_payloads: bool,
 }
 
 impl DatadogEventsConfiguration {
-    /// Creates a new `DatadogEventsConfiguration` from the given configuration.
-    pub fn from_configuration(config: &GenericConfiguration) -> Result<Self, GenericError> {
-        Ok(config.as_typed()?)
+    /// Creates a new `DatadogEventsConfiguration` from the shared typed configuration.
+    pub fn from_configuration(
+        metrics_encoding: &MetricsEncoding, compression: &Compression,
+    ) -> Result<Self, GenericError> {
+        Ok(Self {
+            max_payload_size: metrics_encoding.max_payload_size,
+            max_uncompressed_payload_size: metrics_encoding.max_uncompressed_payload_size,
+            compressor_kind: compression.compressor_kind.clone(),
+            zstd_compressor_level: compression.zstd_compressor_level,
+            log_payloads: metrics_encoding.log_payloads,
+        })
     }
 }
 
@@ -328,27 +288,32 @@ fn encode_eventd(eventd: &EventD, tags_deduplicator: &mut ReusableDeduplicator<T
 }
 
 #[cfg(test)]
-mod config_smoke {
-    use datadog_agent_config_testing::config_registry::structs;
-    use datadog_agent_config_testing::run_config_smoke_tests;
-    use serde_json::json;
+mod tests {
+    use agent_data_plane_config::shared::{Compression, MetricsEncoding};
 
     use super::DatadogEventsConfiguration;
-    use crate::config::{DatadogRemapper, KEY_ALIASES};
-
-    #[tokio::test]
-    async fn smoke_test() {
-        run_config_smoke_tests(
-            structs::DATADOG_EVENTS_CONFIGURATION,
-            &[],
-            json!({}),
-            |cfg| {
-                cfg.as_typed::<DatadogEventsConfiguration>()
-                    .expect("DatadogEventsConfiguration should deserialize")
-            },
-            KEY_ALIASES,
-            DatadogRemapper::new,
-        )
-        .await
+
+    #[test]
+    fn from_configuration_reads_typed_shared_values() {
+        let metrics_encoding = MetricsEncoding {
+            max_payload_size: 123,
+            max_uncompressed_payload_size: 456,
+            log_payloads: true,
+            ..Default::default()
+        };
+
+        let compression = Compression {
+            compressor_kind: "gzip".to_owned(),
+            zstd_compressor_level: 7,
+        };
+
+        let config = DatadogEventsConfiguration::from_configuration(&metrics_encoding, &compression)
+            .expect("typed configuration should be accepted");
+
+        assert_eq!(config.max_payload_size, 123);
+        assert_eq!(config.max_uncompressed_payload_size, 456);
+        assert_eq!(config.compressor_kind, "gzip");
+        assert_eq!(config.zstd_compressor_level, 7);
+        assert!(config.log_payloads);
     }
 }