chore(otlp): move agent to axum (#711)

# What?

Moves OTLP Agent to Axum – also adds a cancellation token to be used on
shutdown

# Motivation

Leverage Axum handle connection errors

---------

Co-authored-by: Copilot <[email protected]>

Author: duncanista

bottlecap/src/bin/bottlecap/main.rs

@@ -484,7 +484,7 @@ async fn extension_loop_active(
     let telemetry_listener_cancel_token =
         setup_telemetry_client(&r.extension_id, logs_agent_channel).await?;
 
-    start_otlp_agent(
+    let otlp_shutdown_token = start_otlp_agent(
         config,
         tags_provider.clone(),
         trace_processor.clone(),
@@ -682,6 +682,9 @@ async fn extension_loop_active(
             if let Some(api_runtime_proxy_cancel_token) = api_runtime_proxy_shutdown_signal {
                 api_runtime_proxy_cancel_token.cancel();
             }
+            if let Some(otlp_shutdown_token) = otlp_shutdown_token {
+                otlp_shutdown_token.cancel();
+            }
             dogstatsd_cancel_token.cancel();
             telemetry_listener_cancel_token.cancel();
 
@@ -1066,18 +1069,20 @@ fn start_otlp_agent(
     tags_provider: Arc<TagProvider>,
     trace_processor: Arc<dyn trace_processor::TraceProcessor + Send + Sync>,
     trace_tx: Sender<SendData>,
-) {
+) -> Option<CancellationToken> {
     if !should_enable_otlp_agent(config) {
-        return;
+        return None;
     }
 
     let agent = OtlpAgent::new(config.clone(), tags_provider, trace_processor, trace_tx);
-
+    let shutdown_token = agent.shutdown_token();
     tokio::spawn(async move {
-        if let Err(e) = agent.start().await {
+        if let Err(e) = agent.start() {
             error!("Error starting OTLP agent: {e:?}");
         }
     });
+
+    Some(shutdown_token)
 }
 
 fn start_api_runtime_proxy(

bottlecap/src/http_client.rs renamed to bottlecap/src/http.rs

@@ -1,4 +1,10 @@
 use crate::config;
+use axum::{
+    extract::{FromRequest, Request},
+    http::{self, StatusCode},
+    response::{IntoResponse, Response},
+};
+use bytes::Bytes;
 use core::time::Duration;
 use datadog_fips::reqwest_adapter::create_reqwest_client_builder;
 use std::error::Error;
@@ -49,3 +55,16 @@ fn build_client(config: Arc<config::Config>) -> Result<reqwest::Client, Box<dyn
         Ok(client.build()?)
     }
 }
+
+pub async fn handler_not_found() -> Response {
+    (StatusCode::NOT_FOUND, "Not Found").into_response()
+}
+
+pub async fn extract_request_body(
+    request: Request,
+) -> Result<(http::request::Parts, Bytes), Box<dyn std::error::Error>> {
+    let (parts, body) = request.into_parts();
+    let bytes = Bytes::from_request(Request::from_parts(parts.clone(), body), &()).await?;
+
+    Ok((parts, bytes))
+}

bottlecap/src/lib.rs

@@ -21,7 +21,7 @@ pub mod config;
 pub mod event_bus;
 pub mod events;
 pub mod fips;
-pub mod http_client;
+pub mod http;
 pub mod lifecycle;
 pub mod logger;
 pub mod logs;

bottlecap/src/logs/flusher.rs

@@ -1,5 +1,5 @@
 use crate::config;
-use crate::http_client;
+use crate::http::get_client;
 use crate::logs::aggregator::Aggregator;
 use crate::FLUSH_RETRY_COUNT;
 use reqwest::header::HeaderMap;
@@ -35,7 +35,7 @@ impl Flusher {
         aggregator: Arc<Mutex<Aggregator>>,
         config: Arc<config::Config>,
     ) -> Self {
-        let client = http_client::get_client(config.clone());
+        let client = get_client(config.clone());
         let mut headers = HeaderMap::new();
         headers.insert(
             "DD-API-KEY",