Dual Shipping Logs Support (#718)

Adds support for dual shipping metrics to endpoints configured using the
`logs_config` YAML or `DD_LOGS_CONFIG_ADDITIONAL_ENDPOINTS` env var
config.

Implemented a `LogsFlusher` as a wrapper to all the `Flusher` instances
to manages flushing to all configured endpoints.

Moved retry logic to `LogsFlusher`, as the retry request contains the
endpoint details and does not have to be tied to a particular flusher.

---------

Co-authored-by: jordan gonzález <30836115+duncanista@users.noreply.github.com>

Author: shreyamalpani

bottlecap/src/bin/bottlecap/main.rs

@@ -25,7 +25,7 @@ use bottlecap::{
         listener::Listener as LifecycleListener,
     },
     logger,
-    logs::{agent::LogsAgent, flusher::Flusher as LogsFlusher},
+    logs::{agent::LogsAgent, flusher::LogsFlusher},
     otlp::{agent::Agent as OtlpAgent, should_enable_otlp_agent},
     proxy::{interceptor, should_start_proxy},
     secrets::decrypt,

bottlecap/src/config/env.rs

@@ -12,6 +12,9 @@ use crate::{
         deserialize_optional_bool_from_anything, deserialize_string_or_int,
         flush_strategy::FlushStrategy,
         log_level::LogLevel,
+        logs_additional_endpoints::{
+            deserialize_logs_additional_endpoints, LogsAdditionalEndpoint,
+        },
         processing_rule::{deserialize_processing_rules, ProcessingRule},
         service_mapping::deserialize_service_mapping,
         trace_propagation_style::{deserialize_trace_propagation_style, TracePropagationStyle},
@@ -125,6 +128,12 @@ pub struct EnvConfig {
     /// to 9 (maximum compression but higher resource usage). Only takes effect if
     /// `use_compression` is set to `true`.
     pub logs_config_compression_level: Option<i32>,
+    /// @env `DD_LOGS_CONFIG_ADDITIONAL_ENDPOINTS`
+    ///
+    /// Additional endpoints to send logs to.
+    /// <https://docs.datadoghq.com/agent/configuration/dual-shipping/?tab=helm#environment-variable-configuration-6>
+    #[serde(deserialize_with = "deserialize_logs_additional_endpoints")]
+    pub logs_config_additional_endpoints: Vec<LogsAdditionalEndpoint>,
 
     // APM
     //
@@ -322,6 +331,7 @@ fn merge_config(config: &mut Config, env_config: &EnvConfig) {
     merge_option!(config, env_config, logs_config_processing_rules);
     merge_option_to_value!(config, env_config, logs_config_use_compression);
     merge_option_to_value!(config, env_config, logs_config_compression_level);
+    merge_vec!(config, env_config, logs_config_additional_endpoints);
 
     // APM
     merge_hashmap!(config, env_config, service_mapping);
@@ -505,6 +515,10 @@ mod tests {
             );
             jail.set_env("DD_LOGS_CONFIG_USE_COMPRESSION", "false");
             jail.set_env("DD_LOGS_CONFIG_COMPRESSION_LEVEL", "3");
+            jail.set_env(
+                "DD_LOGS_CONFIG_ADDITIONAL_ENDPOINTS",
+                "[{\"api_key\": \"apikey2\", \"Host\": \"agent-http-intake.logs.datadoghq.com\", \"Port\": 443, \"is_reliable\": true}]",
+            );
 
             // APM
             jail.set_env("DD_SERVICE_MAPPING", "old-service:new-service");
@@ -640,6 +654,12 @@ mod tests {
                 }]),
                 logs_config_use_compression: false,
                 logs_config_compression_level: 3,
+                logs_config_additional_endpoints: vec![LogsAdditionalEndpoint {
+                    api_key: "apikey2".to_string(),
+                    host: "agent-http-intake.logs.datadoghq.com".to_string(),
+                    port: 443,
+                    is_reliable: true,
+                }],
                 service_mapping: HashMap::from([(
                     "old-service".to_string(),
                     "new-service".to_string(),

bottlecap/src/config/logs_additional_endpoints.rs

@@ -0,0 +1,67 @@
+use serde::{Deserialize, Deserializer};
+use serde_json::Value;
+use tracing::error;
+
+#[derive(Debug, PartialEq, Clone, Deserialize)]
+pub struct LogsAdditionalEndpoint {
+    pub api_key: String,
+    #[serde(rename = "Host")]
+    pub host: String,
+    #[serde(rename = "Port")]
+    pub port: u32,
+    pub is_reliable: bool,
+}
+
+#[allow(clippy::module_name_repetitions)]
+pub fn deserialize_logs_additional_endpoints<'de, D>(
+    deserializer: D,
+) -> Result<Vec<LogsAdditionalEndpoint>, D::Error>
+where
+    D: Deserializer<'de>,
+{
+    let value = Value::deserialize(deserializer)?;
+
+    match value {
+        Value::String(s) if !s.is_empty() => {
+            // For JSON format (string) in DD_ADDITIONAL_ENDPOINTS
+            Ok(serde_json::from_str(&s).unwrap_or_else(|err| {
+                error!("Failed to deserialize DD_LOGS_CONFIG_ADDITIONAL_ENDPOINTS: {err}");
+                vec![]
+            }))
+        }
+        _ => Ok(Vec::new()),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serde_json::json;
+
+    #[test]
+    fn test_deserialize_logs_additional_endpoints_valid() {
+        let input = json!("[{\"api_key\": \"apiKey2\", \"Host\": \"agent-http-intake.logs.datadoghq.com\", \"Port\": 443, \"is_reliable\": true}]");
+
+        let result = deserialize_logs_additional_endpoints(input).unwrap();
+        let mut expected = Vec::new();
+        expected.push(LogsAdditionalEndpoint {
+            api_key: "apiKey2".to_string(),
+            host: "agent-http-intake.logs.datadoghq.com".to_string(),
+            port: 443,
+            is_reliable: true,
+        });
+
+        assert_eq!(result, expected);
+    }
+
+    #[test]
+    fn test_deserialize_logs_additional_endpoints_invalid() {
+        // input missing "Port" field
+        let input = json!("[{\"api_key\": \"apiKey2\", \"Host\": \"agent-http-intake.logs.datadoghq.com\", \"is_reliable\": true}]");
+
+        let result = deserialize_logs_additional_endpoints(input).unwrap();
+        let expected = Vec::new(); // expect empty list due to invalid input
+
+        assert_eq!(result, expected);
+    }
+}

bottlecap/src/config/mod.rs

@@ -4,13 +4,15 @@ pub mod aws;
 pub mod env;
 pub mod flush_strategy;
 pub mod log_level;
+pub mod logs_additional_endpoints;
 pub mod processing_rule;
 pub mod service_mapping;
 pub mod trace_propagation_style;
 pub mod yaml;
 
 use datadog_trace_obfuscation::replacer::ReplaceRule;
 use datadog_trace_utils::config_utils::{trace_intake_url, trace_intake_url_prefixed};
+
 use serde::{Deserialize, Deserializer};
 use serde_aux::prelude::deserialize_bool_from_anything;
 use serde_json::Value;
@@ -24,6 +26,7 @@ use crate::config::{
     env::EnvConfigSource,
     flush_strategy::FlushStrategy,
     log_level::LogLevel,
+    logs_additional_endpoints::LogsAdditionalEndpoint,
     processing_rule::{deserialize_processing_rules, ProcessingRule},
     trace_propagation_style::TracePropagationStyle,
     yaml::YamlConfigSource,
@@ -260,6 +263,7 @@ pub struct Config {
     pub logs_config_processing_rules: Option<Vec<ProcessingRule>>,
     pub logs_config_use_compression: bool,
     pub logs_config_compression_level: i32,
+    pub logs_config_additional_endpoints: Vec<LogsAdditionalEndpoint>,
 
     // APM
     //
@@ -353,6 +357,7 @@ impl Default for Config {
             logs_config_processing_rules: None,
             logs_config_use_compression: true,
             logs_config_compression_level: 6,
+            logs_config_additional_endpoints: Vec::new(),
 
             // APM
             service_mapping: HashMap::new(),

bottlecap/src/config/yaml.rs

@@ -8,6 +8,7 @@ use crate::{
         deserialize_string_or_int,
         flush_strategy::FlushStrategy,
         log_level::LogLevel,
+        logs_additional_endpoints::LogsAdditionalEndpoint,
         service_mapping::deserialize_service_mapping,
         trace_propagation_style::{deserialize_trace_propagation_style, TracePropagationStyle},
         Config, ConfigError, ConfigSource, ProcessingRule,
@@ -117,6 +118,7 @@ pub struct LogsConfig {
     #[serde(deserialize_with = "deserialize_optional_bool_from_anything")]
     pub use_compression: Option<bool>,
     pub compression_level: Option<i32>,
+    pub additional_endpoints: Vec<LogsAdditionalEndpoint>,
 }
 
 /// APM Config
@@ -392,6 +394,12 @@ fn merge_config(config: &mut Config, yaml_config: &YamlConfig) {
         yaml_config.logs_config,
         compression_level
     );
+    merge_vec!(
+        config,
+        logs_config_additional_endpoints,
+        yaml_config.logs_config,
+        additional_endpoints
+    );
 
     // APM
     merge_hashmap!(config, yaml_config, service_mapping);
@@ -633,8 +641,8 @@ http_protocol: "http1"
 # Endpoints
 additional_endpoints:
   "https://app.datadoghq.com":
-    - apikey2
-    - apikey3
+    - "apikey2"
+    - "apikey3"
   "https://app.datadoghq.eu":
     - apikey4
 
@@ -655,6 +663,11 @@ logs_config:
       pattern: "test-pattern"
   use_compression: false
   compression_level: 3
+  additional_endpoints:
+    - api_key: "apikey2"
+      Host: "agent-http-intake.logs.datadoghq.com"
+      Port: 443
+      is_reliable: true
 
 # APM
 apm_config:
@@ -770,6 +783,12 @@ extension_version: "compatibility"
                 }]),
                 logs_config_use_compression: false,
                 logs_config_compression_level: 3,
+                logs_config_additional_endpoints: vec![LogsAdditionalEndpoint {
+                    api_key: "apikey2".to_string(),
+                    host: "agent-http-intake.logs.datadoghq.com".to_string(),
+                    port: 443,
+                    is_reliable: true,
+                }],
                 service_mapping: HashMap::from([(
                     "old-service".to_string(),
                     "new-service".to_string(),