feat(logs): adopt datadog-log-agent shared crate

Replace bespoke log aggregator, flusher, and intake types with the
shared datadog-log-agent crate from serverless-components.

Key changes:
- Delete bottlecap/src/logs/{aggregator,aggregator_service,constants,
  flusher}.rs — superseded by the crate
- LogEntry is now the flat datadog_log_agent::LogEntry; lambda context
  (arn, request_id) moves into attributes["lambda"] instead of the old
  nested IntakeLog.message.lambda struct
- LogFlusher (with internal retry) replaces LogsFlusher; flush handle
  type simplifies from Vec<RequestBuilder> to bool
- start_logs_agent made async to resolve the API key upfront and build
  LogFlusherConfig explicitly
- Integration test updated to match the new flat JSON wire format

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: litianningdatadog

bottlecap/Cargo.lock

@@ -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_log_agent::{
+    AggregatorHandle as LogsAggregatorHandle, AggregatorService as LogsAggregatorService,
+    FlusherMode, 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 {
+        FlusherMode::ObservabilityPipelinesWorker {
+            url: config.observability_pipelines_worker_logs_url.clone(),
+        }
+    } else {
+        FlusherMode::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/bin/bottlecap/main.rs

@@ -27,8 +27,8 @@ 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.
-    pub log_flush_handles: Vec<JoinHandle<Vec<reqwest::RequestBuilder>>>,
+    /// Handles for log flush operations. Returns true on success, false on any failure.
+    pub log_flush_handles: Vec<JoinHandle<bool>>,
     /// Handles for metrics flush operations. Returns batch info for retry.
     pub metric_flush_handles: Vec<JoinHandle<MetricsRetryBatch>>,
     /// Handles for proxy flush operations. Returns failed request builders for retry.

bottlecap/src/flushing/handles.rs

@@ -8,8 +8,9 @@ use dogstatsd::{
     aggregator::AggregatorHandle as MetricsAggregatorHandle, flusher::Flusher as MetricsFlusher,
 };
 
+use datadog_log_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().await }));
 
         // Spawn traces flush
         let tf = self.trace_flusher.clone();
@@ -191,32 +192,18 @@ impl FlushingService {
             }
         }
 
-        // Await log handles with retry
+        // Await log handles — retries are handled internally by LogFlusher
         for handle in self.handles.log_flush_handles.drain(..) {
             match handle.await {
-                Ok(retry) => {
-                    if !retry.is_empty() {
-                        debug!(
-                            "FLUSHING_SERVICE | redriving {:?} log payloads",
-                            retry.len()
-                        );
-                    }
-                    for item in retry {
-                        let lf = self.logs_flusher.clone();
-                        match item.try_clone() {
-                            Some(item_clone) => {
-                                joinset.spawn(async move {
-                                    lf.flush(Some(item_clone)).await;
-                                });
-                            }
-                            None => {
-                                error!("FLUSHING_SERVICE | Can't clone redrive log payloads");
-                            }
-                        }
+                Ok(success) => {
+                    if !success {
+                        debug!("FLUSHING_SERVICE | log flush reported a failure");
+                        flush_error = true;
                     }
                 }
                 Err(e) => {
-                    error!("FLUSHING_SERVICE | redrive log error {e:?}");
+                    error!("FLUSHING_SERVICE | log flush task error {e:?}");
+                    flush_error = true;
                 }
             }
         }
@@ -325,7 +312,7 @@ impl FlushingService {
             .collect();
 
         tokio::join!(
-            self.logs_flusher.flush(None),
+            self.logs_flusher.flush(),
             futures::future::join_all(metrics_futures),
             self.trace_flusher.flush(None),
             self.stats_flusher.flush(force_stats, None),

bottlecap/src/flushing/service.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_log_agent::AggregatorHandle;
 
 const DRAIN_LOG_INTERVAL: Duration = Duration::from_millis(100);