fixup: bridge DD_* fallbacks for saluki-only keys

Saluki-only keys set via DD_* env vars were silently
dropped, causing SMP memory regressions in the DSD
quality_gates benchmarks (~20% RSS increase).

Adds bridge_dd_fallbacks() to read saluki-only values
from the Datadog source when the Saluki source has not
set them. Covers aggregate_context_limit, interner
sizes, autoscale, and OTLP context limits.

Author: webern

lib/agent-data-plane-config-system/src/bootstrap.rs

@@ -22,6 +22,7 @@
 use std::path::PathBuf;
 
 use agent_data_plane_config::{BootstrapConfiguration, LocalApiBootstrap, RuntimeAuthority, SalukiOnlyConfiguration};
+use bytesize::ByteSize;
 use datadog_agent_config::{DatadogRemapper, KEY_ALIASES};
 use saluki_config_tools::{ConfigurationLoader, GenericConfiguration};
 use saluki_error::{generic_error, GenericError};
@@ -173,14 +174,11 @@ pub(crate) fn load_local_sources(
         .as_typed()
         .map_err(|e| generic_error!("Failed to parse the Saluki-schema-only configuration: {}", e))?;
 
-    // `dogstatsd_tcp_port` is not in the Datadog core schema, so it cannot go through the overlay
-    // witness. The Saluki source owns this key, but existing deployments set it via `DD_*` env
-    // vars. Bridge the Datadog source value into the Saluki-only struct when not already set.
-    if saluki_only.dogstatsd.tcp_port.is_none() {
-        if let Ok(Some(v)) = datadog_generic.try_get_typed::<u16>("dogstatsd_tcp_port") {
-            saluki_only.dogstatsd.tcp_port = Some(v);
-        }
-    }
+    // DD->Saluki migration bridge: Saluki-schema-only keys are owned by the Saluki source
+    // (SALUKI_* / saluki.yaml), but existing deployments and test configs set them via DD_* env
+    // vars. Bridge each Datadog source value into the Saluki-only struct when the Saluki source
+    // has not already set it. SALUKI_* always wins (it is checked first).
+    bridge_dd_fallbacks(&mut saluki_only, &datadog_generic);
     let saluki_bootstrap = saluki_generic
         .as_typed()
         .map_err(|e| generic_error!("Failed to parse the Saluki bootstrap slice: {}", e))?;
@@ -244,3 +242,47 @@ fn patch_snapshot_pipeline_gates(
 
     Ok(())
 }
+
+/// Bridges Saluki-schema-only keys from the Datadog source (`DD_*`) into the Saluki-only struct.
+///
+/// Saluki-schema-only keys are owned by the Saluki source (`SALUKI_*` / `saluki.yaml`). The clean
+/// end state reads them only from there. However, existing deployments and test infrastructure set
+/// these keys via `DD_*` env vars. This function provides a bounded migration-compatibility layer:
+/// each value is read from the Datadog source only when the Saluki source has not already set it.
+fn bridge_dd_fallbacks(saluki_only: &mut SalukiOnlyConfiguration, dd: &GenericConfiguration) {
+    // ---- dogstatsd ----
+    if saluki_only.dogstatsd.tcp_port.is_none() {
+        if let Ok(Some(v)) = dd.try_get_typed::<u16>("dogstatsd_tcp_port") {
+            saluki_only.dogstatsd.tcp_port = Some(v);
+        }
+    }
+    if saluki_only.dogstatsd.string_interner_size_bytes.is_none() {
+        if let Ok(Some(v)) = dd.try_get_typed::<ByteSize>("dogstatsd_string_interner_size_bytes") {
+            saluki_only.dogstatsd.string_interner_size_bytes = Some(v.0);
+        }
+    }
+    if saluki_only.dogstatsd.autoscale_udp_listeners.is_none() {
+        if let Ok(Some(v)) = dd.try_get_typed::<bool>("dogstatsd_autoscale_udp_listeners") {
+            saluki_only.dogstatsd.autoscale_udp_listeners = Some(v);
+        }
+    }
+
+    // ---- aggregate ----
+    if saluki_only.aggregate.context_limit.is_none() {
+        if let Ok(Some(v)) = dd.try_get_typed::<usize>("aggregate_context_limit") {
+            saluki_only.aggregate.context_limit = Some(v);
+        }
+    }
+
+    // ---- otlp ----
+    if saluki_only.otlp.cached_contexts_limit.is_none() {
+        if let Ok(Some(v)) = dd.try_get_typed::<usize>("otlp_cached_context_limit") {
+            saluki_only.otlp.cached_contexts_limit = Some(v);
+        }
+    }
+    if saluki_only.otlp.traces_string_interner_size.is_none() {
+        if let Ok(Some(v)) = dd.try_get_typed::<ByteSize>("otlp_config.traces.string_interner_size") {
+            saluki_only.otlp.traces_string_interner_size = Some(v.0);
+        }
+    }
+}

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

@@ -200,9 +200,8 @@ fn assemble(
 /// autodiscovery). In that case we establish a lightweight Agent connection for the env-provider
 /// attachments without subscribing to the config stream.
 async fn start_local(
-    bootstrap: BootstrapConfiguration, saluki_only: SalukiOnlyConfiguration,
-    datadog_snapshot: serde_json::Value, standalone_mode: bool, otlp_enabled: bool,
-    registration: RemoteAgentRegistration,
+    bootstrap: BootstrapConfiguration, saluki_only: SalukiOnlyConfiguration, datadog_snapshot: serde_json::Value,
+    standalone_mode: bool, otlp_enabled: bool, registration: RemoteAgentRegistration,
 ) -> Result<StartedConfigurationSystem, GenericError> {
     let datadog: DatadogConfiguration = serde_json::from_value(datadog_snapshot.clone())
         .map_err(|e| generic_error!("Failed to parse the local Datadog snapshot at startup: {}", e))?;