- add CLAUDE.md to .gitignore

- disable log agent by default — require explicit DD_LOGS_ENABLED=true
- log the actual error when reqwest client build() fails in start_log_agent
- fail fast in start_log_agent when OPW URL is empty
- apply rustfmt to integration test formatting.

Author: litianningdatadog

.gitignore

@@ -1,3 +1,4 @@
 /target
 /.idea
+/CLAUDE.md
 

CLAUDE.md

@@ -16,6 +16,5 @@ pub const DEFAULT_SITE: &str = "datadoghq.com";
 /// Default flush timeout in seconds.
 pub const DEFAULT_FLUSH_TIMEOUT_SECS: u64 = 5;
 
-/// Default zstd compression level. Valid range: 1 (fastest) to 21 (best ratio).
 /// Negative values enable ultra-fast modes. Level 3 is the zstd library default.
 pub const DEFAULT_COMPRESSION_LEVEL: i32 = 3;

crates/datadog-log-agent/src/constants.rs

@@ -12,9 +12,7 @@
 
 #![allow(clippy::disallowed_methods)] // plain reqwest::Client is fine against local mock server
 
-use datadog_log_agent::{
-    AggregatorService, FlusherMode, LogEntry, LogFlusher, LogFlusherConfig,
-};
+use datadog_log_agent::{AggregatorService, FlusherMode, LogEntry, LogFlusher, LogFlusherConfig};
 use mockito::{Matcher, Server};
 use std::time::Duration;
 
@@ -254,7 +252,11 @@ async fn test_max_entries_fits_in_one_batch() {
     handle.insert_batch(entries).expect("insert");
 
     let batches = handle.get_batches().await.expect("get_batches");
-    assert_eq!(batches.len(), 1, "exactly MAX_BATCH_ENTRIES fits in one batch");
+    assert_eq!(
+        batches.len(),
+        1,
+        "exactly MAX_BATCH_ENTRIES fits in one batch"
+    );
 
     let arr: serde_json::Value = serde_json::from_slice(&batches[0]).expect("valid JSON");
     assert_eq!(arr.as_array().unwrap().len(), MAX);
@@ -442,7 +444,9 @@ async fn test_opw_mode_disables_compression_regardless_of_config() {
 
     let (svc, handle) = AggregatorService::new();
     let _task = tokio::spawn(svc.run());
-    handle.insert_batch(vec![entry("not compressed in OPW")]).expect("insert");
+    handle
+        .insert_batch(vec![entry("not compressed in OPW")])
+        .expect("insert");
 
     // use_compression: true — but OPW mode overrides this to false
     let config = LogFlusherConfig {
@@ -457,7 +461,9 @@ async fn test_opw_mode_disables_compression_regardless_of_config() {
         flush_timeout: Duration::from_secs(5),
     };
 
-    let result = LogFlusher::new(config, build_client(), handle).flush().await;
+    let result = LogFlusher::new(config, build_client(), handle)
+        .flush()
+        .await;
 
     assert!(result);
     mock.assert_async().await;
@@ -485,7 +491,9 @@ async fn test_retry_on_500_succeeds_on_second_attempt() {
 
     let (svc, handle) = AggregatorService::new();
     let _task = tokio::spawn(svc.run());
-    handle.insert_batch(vec![entry("retry me")]).expect("insert");
+    handle
+        .insert_batch(vec![entry("retry me")])
+        .expect("insert");
 
     let result = LogFlusher::new(opw_config(&server.url()), build_client(), handle)
         .flush()
@@ -507,7 +515,9 @@ async fn test_permanent_error_on_403_no_retry() {
 
     let (svc, handle) = AggregatorService::new();
     let _task = tokio::spawn(svc.run());
-    handle.insert_batch(vec![entry("forbidden")]).expect("insert");
+    handle
+        .insert_batch(vec![entry("forbidden")])
+        .expect("insert");
 
     let result = LogFlusher::new(opw_config(&server.url()), build_client(), handle)
         .flush()
@@ -530,7 +540,9 @@ async fn test_exhausted_retries_returns_false() {
 
     let (svc, handle) = AggregatorService::new();
     let _task = tokio::spawn(svc.run());
-    handle.insert_batch(vec![entry("keep failing")]).expect("insert");
+    handle
+        .insert_batch(vec![entry("keep failing")])
+        .expect("insert");
 
     let result = LogFlusher::new(opw_config(&server.url()), build_client(), handle)
         .flush()
@@ -564,7 +576,9 @@ async fn test_additional_endpoints_receive_same_batch() {
 
     let (svc, handle) = AggregatorService::new();
     let _task = tokio::spawn(svc.run());
-    handle.insert_batch(vec![entry("multi-endpoint")]).expect("insert");
+    handle
+        .insert_batch(vec![entry("multi-endpoint")])
+        .expect("insert");
 
     let config = LogFlusherConfig {
         api_key: "test-api-key".to_string(),
@@ -578,7 +592,9 @@ async fn test_additional_endpoints_receive_same_batch() {
         flush_timeout: Duration::from_secs(5),
     };
 
-    let result = LogFlusher::new(config, build_client(), handle).flush().await;
+    let result = LogFlusher::new(config, build_client(), handle)
+        .flush()
+        .await;
 
     assert!(result);
     primary_mock.assert_async().await;
@@ -620,7 +636,9 @@ async fn test_additional_endpoint_failure_does_not_affect_return_value() {
         flush_timeout: Duration::from_secs(5),
     };
 
-    let result = LogFlusher::new(config, build_client(), handle).flush().await;
+    let result = LogFlusher::new(config, build_client(), handle)
+        .flush()
+        .await;
 
     assert!(
         result,

crates/datadog-log-agent/tests/integration_test.rs

@@ -27,8 +27,8 @@ use libdd_trace_utils::{config_utils::read_cloud_env, trace_utils::EnvironmentTy
 
 use datadog_fips::reqwest_adapter::create_reqwest_client_builder;
 use datadog_log_agent::{
-    AggregatorHandle as LogAggregatorHandle, AggregatorService as LogAggregatorService, LogFlusher,
-    LogFlusherConfig,
+    AggregatorHandle as LogAggregatorHandle, AggregatorService as LogAggregatorService,
+    FlusherMode as LogFlusherMode, LogFlusher, LogFlusherConfig,
 };
 use dogstatsd::{
     aggregator::{AggregatorHandle, AggregatorService},
@@ -112,8 +112,8 @@ pub async fn main() {
         .or_else(|_| env::var("HTTPS_PROXY"))
         .ok();
     let dd_logs_enabled = env::var("DD_LOGS_ENABLED")
-        .map(|val| val.to_lowercase() != "false")
-        .unwrap_or(true);
+        .map(|val| val.to_lowercase() == "true")
+        .unwrap_or(false);
     debug!("Starting serverless trace mini agent");
 
     let env_filter = format!("h2=off,hyper=off,rustls=off,{}", log_level);
@@ -360,21 +360,30 @@ fn start_log_agent(dd_api_key: Option<String>, https_proxy: Option<String>) -> O
                     Err(e) => error!("invalid HTTPS proxy for log agent: {e}"),
                 }
             }
-            builder.build().ok()
+            match builder.build() {
+                Ok(c) => Some(c),
+                Err(e) => {
+                    error!("failed to build HTTP client for log agent: {e}");
+                    None
+                }
+            }
         });
 
-    let client = match client {
-        Some(c) => c,
-        None => {
-            error!("failed to build HTTP client for log agent, log flushing disabled");
-            return None;
-        }
-    };
+    let client = client?; // error already logged above
 
     let config = LogFlusherConfig {
         api_key,
         ..LogFlusherConfig::from_env()
     };
+
+    // Fail fast: OPW mode with an empty URL will always produce a network error at flush time.
+    if let LogFlusherMode::ObservabilityPipelinesWorker { url } = &config.mode {
+        if url.is_empty() {
+            error!("OPW mode enabled but DD_OBSERVABILITY_PIPELINES_WORKER_LOGS_URL is empty — log agent disabled");
+            return None;
+        }
+    }
+
     Some(LogFlusher::new(config, client, handle))
 }
 
@@ -412,6 +421,27 @@ mod log_agent_integration_tests {
         handle.shutdown().expect("shutdown");
     }
 
+    /// start_log_agent must reject OPW mode with an empty URL.
+    #[test]
+    fn test_opw_empty_url_is_detected() {
+        let config = LogFlusherConfig {
+            api_key: "key".to_string(),
+            site: "datadoghq.com".to_string(),
+            mode: FlusherMode::ObservabilityPipelinesWorker { url: String::new() },
+            additional_endpoints: Vec::new(),
+            use_compression: false,
+            compression_level: 3,
+            flush_timeout: std::time::Duration::from_secs(5),
+        };
+        assert!(
+            matches!(
+                &config.mode,
+                FlusherMode::ObservabilityPipelinesWorker { url } if url.is_empty()
+            ),
+            "should detect empty OPW URL"
+        );
+    }
+
     /// Verify the LogFlusher struct is constructible from within this crate — compile-time test.
     #[allow(dead_code)]
     fn _assert_log_flusher_constructible() {