feat(logs): adopt datadog-log-agent shared crate [SVLS-8713]

bottlecap/Cargo.toml

@@ -78,6 +78,7 @@ libdd-trace-obfuscation = { git = "https://github.com/DataDog/libdatadog", rev =
 libdd-trace-stats = { git = "https://github.com/DataDog/libdatadog", rev = "c8121f422d2c8d219f8d421ff3cdb1fcbe9e8b09"  }
 dogstatsd = { git = "https://github.com/DataDog/serverless-components", rev = "28f796bf767fff56caf08153ade5cd80c8e8f705", default-features = false }
 datadog-fips = { git = "https://github.com/DataDog/serverless-components", rev = "28f796bf767fff56caf08153ade5cd80c8e8f705", default-features = false }
+datadog-logs-agent = { git = "https://github.com/DataDog/serverless-components", rev = "8cb5daa7679df7d0f814aad016b592288e91cea5", default-features = false }
 libddwaf = { version = "1.28.1", git = "https://github.com/DataDog/libddwaf-rust", rev = "d1534a158d976bd4f747bf9fcc58e0712d2d17fc", default-features = false, features = ["serde"] }
 
 [dev-dependencies]

bottlecap/src/bin/bottlecap/main.rs

@@ -45,13 +45,7 @@ use bottlecap::{
         listener::Listener as LifecycleListener,
     },
     logger,
-    logs::{
-        agent::LogsAgent,
-        aggregator_service::{
-            AggregatorHandle as LogsAggregatorHandle, AggregatorService as LogsAggregatorService,
-        },
-        flusher::LogsFlusher,
-    },
+    logs::agent::LogsAgent,
     otlp::{agent::Agent as OtlpAgent, should_enable_otlp_agent},
     proxy::{interceptor, should_start_proxy},
     secrets::decrypt,
@@ -79,6 +73,10 @@ use bottlecap::{
     },
 };
 use datadog_fips::reqwest_adapter::create_reqwest_client_builder;
+use datadog_logs_agent::{
+    AggregatorHandle as LogsAggregatorHandle, AggregatorService as LogsAggregatorService,
+    Destination, LogFlusher, LogFlusherConfig, LogsAdditionalEndpoint,
+};
 use decrypt::resolve_secrets;
 use dogstatsd::{
     aggregator::{
@@ -306,7 +304,8 @@ async fn extension_loop_active(
             event_bus_tx.clone(),
             aws_config.is_managed_instance_mode(),
             &shared_client,
-        );
+        )
+        .await;
 
     let (metrics_flushers, metrics_aggregator_handle, dogstatsd_cancel_token) = start_dogstatsd(
         tags_provider.clone(),
@@ -1027,7 +1026,7 @@ fn setup_tag_provider(
     ))
 }
 
-fn start_logs_agent(
+async fn start_logs_agent(
     config: &Arc<Config>,
     api_key_factory: Arc<ApiKeyFactory>,
     tags_provider: &Arc<TagProvider>,
@@ -1036,11 +1035,11 @@ fn start_logs_agent(
     client: &Client,
 ) -> (
     Sender<TelemetryEvent>,
-    LogsFlusher,
+    LogFlusher,
     CancellationToken,
     LogsAggregatorHandle,
 ) {
-    let (aggregator_service, aggregator_handle) = LogsAggregatorService::default();
+    let (aggregator_service, aggregator_handle) = LogsAggregatorService::new();
     // Start service in background
     tokio::spawn(async move {
         aggregator_service.run().await;
@@ -1062,12 +1061,37 @@ fn start_logs_agent(
         drop(agent);
     });
 
-    let flusher = LogsFlusher::new(
-        api_key_factory,
-        aggregator_handle.clone(),
-        config.clone(),
-        client.clone(),
-    );
+    let api_key = api_key_factory.get_api_key().await.unwrap_or_default();
+
+    let mode = if config.observability_pipelines_worker_logs_enabled {
+        Destination::ObservabilityPipelinesWorker {
+            url: config.observability_pipelines_worker_logs_url.clone(),
+        }
+    } else {
+        Destination::Datadog
+    };
+
+    let additional_endpoints: Vec<LogsAdditionalEndpoint> = config
+        .logs_config_additional_endpoints
+        .iter()
+        .map(|ep| LogsAdditionalEndpoint {
+            api_key: ep.api_key.clone(),
+            url: format!("https://{}:{}/api/v2/logs", ep.host, ep.port),
+            is_reliable: ep.is_reliable,
+        })
+        .collect();
+
+    let flusher_config = LogFlusherConfig {
+        api_key,
+        site: config.site.clone(),
+        mode,
+        additional_endpoints,
+        use_compression: config.logs_config_use_compression,
+        compression_level: config.logs_config_compression_level,
+        flush_timeout: std::time::Duration::from_secs(config.flush_timeout),
+    };
+
+    let flusher = LogFlusher::new(flusher_config, client.clone(), aggregator_handle.clone());
     (tx, flusher, cancel_token, aggregator_handle)
 }
 

bottlecap/src/config/env.rs

@@ -711,6 +711,7 @@ impl ConfigSource for EnvConfigSource {
 #[cfg_attr(coverage_nightly, coverage(off))] // Test modules skew coverage metrics
 #[cfg(test)]
 mod tests {
+    #![allow(clippy::result_large_err)]
     use std::time::Duration;
 
     use super::*;

bottlecap/src/config/mod.rs

@@ -811,6 +811,7 @@ pub fn deserialize_optional_duration_from_seconds_ignore_zero<'de, D: Deserializ
 #[cfg_attr(coverage_nightly, coverage(off))] // Test modules skew coverage metrics
 #[cfg(test)]
 pub mod tests {
+    #![allow(clippy::result_large_err)]
     use libdd_trace_obfuscation::replacer::parse_rules_from_string;
 
     use super::*;

bottlecap/src/config/yaml.rs

@@ -742,6 +742,7 @@ impl ConfigSource for YamlConfigSource {
 #[cfg_attr(coverage_nightly, coverage(off))] // Test modules skew coverage metrics
 #[cfg(test)]
 mod tests {
+    #![allow(clippy::result_large_err)]
     use std::path::Path;
     use std::time::Duration;
 

bottlecap/src/flushing/handles.rs

@@ -27,7 +27,7 @@ pub struct MetricsRetryBatch {
 pub struct FlushHandles {
     /// Handles for trace flush operations. Returns failed traces for retry.
     pub trace_flush_handles: Vec<JoinHandle<Vec<SendData>>>,
-    /// Handles for log flush operations. Returns failed request builders for retry.
+    /// Handles for log flush operations. Returns builders for transient failures to retry next invocation.
     pub log_flush_handles: Vec<JoinHandle<Vec<reqwest::RequestBuilder>>>,
     /// Handles for metrics flush operations. Returns batch info for retry.
     pub metric_flush_handles: Vec<JoinHandle<MetricsRetryBatch>>,

bottlecap/src/flushing/service.rs

@@ -8,8 +8,9 @@ use dogstatsd::{
     aggregator::AggregatorHandle as MetricsAggregatorHandle, flusher::Flusher as MetricsFlusher,
 };
 
+use datadog_logs_agent::LogFlusher;
+
 use crate::flushing::handles::{FlushHandles, MetricsRetryBatch};
-use crate::logs::flusher::LogsFlusher;
 use crate::traces::{
     proxy_flusher::Flusher as ProxyFlusher, stats_flusher::StatsFlusher,
     trace_flusher::TraceFlusher,
@@ -23,7 +24,7 @@ use crate::traces::{
 /// - Performing blocking flushes (spawn + await)
 pub struct FlushingService {
     // Flushers
-    logs_flusher: LogsFlusher,
+    logs_flusher: LogFlusher,
     trace_flusher: Arc<TraceFlusher>,
     stats_flusher: Arc<StatsFlusher>,
     proxy_flusher: Arc<ProxyFlusher>,
@@ -40,7 +41,7 @@ impl FlushingService {
     /// Creates a new `FlushingService` with the given flushers.
     #[must_use]
     pub fn new(
-        logs_flusher: LogsFlusher,
+        logs_flusher: LogFlusher,
         trace_flusher: Arc<TraceFlusher>,
         stats_flusher: Arc<StatsFlusher>,
         proxy_flusher: Arc<ProxyFlusher>,
@@ -76,7 +77,7 @@ impl FlushingService {
         let lf = self.logs_flusher.clone();
         self.handles
             .log_flush_handles
-            .push(tokio::spawn(async move { lf.flush(None).await }));
+            .push(tokio::spawn(async move { lf.flush(vec![]).await }));
 
         // Spawn traces flush
         let tf = self.trace_flusher.clone();
@@ -206,7 +207,7 @@ impl FlushingService {
                         match item.try_clone() {
                             Some(item_clone) => {
                                 joinset.spawn(async move {
-                                    lf.flush(Some(item_clone)).await;
+                                    lf.flush(vec![item_clone]).await;
                                 });
                             }
                             None => {
@@ -325,7 +326,7 @@ impl FlushingService {
             .collect();
 
         tokio::join!(
-            self.logs_flusher.flush(None),
+            self.logs_flusher.flush(vec![]),
             futures::future::join_all(metrics_futures),
             self.trace_flusher.flush(None),
             self.stats_flusher.flush(force_stats, None),

bottlecap/src/logs/agent.rs

@@ -6,9 +6,10 @@ use tracing::debug;
 
 use crate::event_bus::Event;
 use crate::extension::telemetry::events::TelemetryEvent;
-use crate::logs::{aggregator_service::AggregatorHandle, processor::LogsProcessor};
+use crate::logs::processor::LogsProcessor;
 use crate::tags;
 use crate::{LAMBDA_RUNTIME_SLUG, config};
+use datadog_logs_agent::AggregatorHandle;
 
 const DRAIN_LOG_INTERVAL: Duration = Duration::from_millis(100);