release: v0.31.2 (#1806)

Fix tight error loop when a generated block exceeds the throttle's
maximum capacity. Oversized blocks are now skipped instead of retried
indefinitely, and the log level is downgraded from ERROR to DEBUG.

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: GeorgeHahn

CHANGELOG.md

@@ -6,6 +6,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+## [0.31.2]
+## Changed
+- Work around tight error loop when a generated block exceeds the throttle's
+  maximum capacity. Oversized blocks are now skipped instead of retried
+  indefinitely, and the log is downgraded from ERROR to DEBUG.
+## Removed
+- Reverted HTTP blackhole histogram distribution tracking (`bytes_received_distr`,
+  `decoded_bytes_received_distr`) that was reapplied after 0.31.1.
+
 ## [0.31.1]
 ## Removed
 - Reverted HTTP blackhole histogram distribution tracking (`bytes_received_distr`,
@@ -23,8 +32,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Observer config now supports `enable_smaps` and `enable_smaps_rollup` options
   to toggle `/proc/{pid}/smaps` and `/proc/{pid}/smaps_rollup` collection.
 - Lading now supports histogram approximations in its capture files.
-- HTTP blackhole now tracks distribution of bytes received, both decoded and
-  compressed.
 - New "Static Chunks" generator that divides static files by lines into blocks
   (as opposed to static which turns each file into a block).
 - Fingerprint mechanism now calculates Shannon entropy.

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "lading"
-version = "0.31.1"
+version = "0.31.2"
 authors = [
   "Brian L. Troutwine <brian.troutwine@datadoghq.com>",
   "George Hahn <george.hahn@datadoghq.com>",

lading/Cargo.toml

@@ -4,17 +4,15 @@
 //!
 //! `bytes_received`: Total bytes received
 //! `total_bytes_received`: Aggregated bytes received across all blackhole types
-//! `bytes_received_distr`: Distribution of compressed bytes per request (with `path` label)
 //! `decoded_bytes_received`: Total decoded bytes received
-//! `decoded_bytes_received_distr`: Distribution of decompressed bytes per request (with `path` label)
 //! `requests_received`: Total requests received
 //!
 
 use bytes::Bytes;
 use http::{HeaderMap, header::InvalidHeaderValue, status::InvalidStatusCode};
 use http_body_util::{BodyExt, combinators::BoxBody};
 use hyper::{Request, Response, StatusCode, header};
-use metrics::{counter, histogram};
+use metrics::counter;
 use serde::{Deserialize, Serialize};
 use std::{net::SocketAddr, time::Duration};
 use tracing::error;
@@ -155,25 +153,20 @@ async fn srv(
 ) -> Result<hyper::Response<BoxBody<Bytes, hyper::Error>>, hyper::Error> {
     counter!("requests_received", &metric_labels).increment(1);
 
-    let path = req.uri().path().to_string();
-
+    // Split into parts
     let (parts, body) = req.into_parts();
 
+    // Convert the `Body` into `Bytes`
     let body: Bytes = body.boxed().collect().await?.to_bytes();
 
     let body_len = body.len() as u64;
     counter!("bytes_received", &metric_labels).increment(body_len);
     counter!("total_bytes_received").increment(body_len);
 
-    let mut labels_with_path = metric_labels.clone();
-    labels_with_path.push(("path".to_string(), path));
-    histogram!("bytes_received_distr", &labels_with_path).record(body.len() as f64);
-
     match crate::codec::decode(parts.headers.get(hyper::header::CONTENT_ENCODING), body) {
         Err(response) => Ok(*response),
         Ok(body) => {
             counter!("decoded_bytes_received", &metric_labels).increment(body.len() as u64);
-            histogram!("decoded_bytes_received_distr", &labels_with_path).record(body.len() as f64);
 
             tokio::time::sleep(response_delay).await;
 

lading/src/blackhole/http.rs

@@ -30,7 +30,7 @@ use tokio::{
     io::{AsyncWriteExt, BufWriter},
     task::{JoinError, JoinHandle},
 };
-use tracing::{error, info};
+use tracing::{debug, error, info};
 
 use lading_payload::block;
 
@@ -385,7 +385,8 @@ impl Child {
                                     &self.labels).await?;
                         }
                         Err(err) => {
-                            error!("Discarding block due to throttle error: {err}");
+                            debug!("Discarding block due to throttle error: {err}");
+                            self.block_cache.advance(&mut handle);
                         }
                     }
                 }

lading/src/generator/file_gen/logrotate.rs

@@ -31,7 +31,7 @@ use tokio::{
     io::{AsyncWriteExt, BufWriter},
     task::{JoinError, JoinSet},
 };
-use tracing::{error, info};
+use tracing::{debug, error, info};
 
 use lading_payload::{self, block};
 
@@ -358,7 +358,8 @@ impl Child {
                             }
                         }
                         Err(err) => {
-                            error!("Discarding block due to throttle error: {err}");
+                            debug!("Discarding block due to throttle error: {err}");
+                            self.block_cache.advance(&mut handle);
                         }
                     }
                 }

lading/src/generator/file_gen/traditional.rs

@@ -24,7 +24,7 @@ use once_cell::sync::OnceCell;
 use rand::{SeedableRng, prelude::StdRng};
 use serde::{Deserialize, Serialize};
 use tokio::sync::Semaphore;
-use tracing::{error, info};
+use tracing::{debug, error, info};
 
 use lading_payload::block;
 
@@ -284,7 +284,8 @@ impl Http {
 
                         }
                         Err(err) => {
-                            error!("Discarding block due to throttle error: {err}");
+                            debug!("Discarding block due to throttle error: {err}");
+                            self.block_cache.advance(&mut handle);
                         }
                     }
                 },

lading/src/generator/http.rs

@@ -38,7 +38,7 @@ use tokio::{
     sync::{Semaphore, SemaphorePermit},
     time::timeout,
 };
-use tracing::{error, info};
+use tracing::{debug, error, info};
 
 use crate::generator::splunk_hec::acknowledgements::Channel;
 use lading_payload::block;
@@ -318,7 +318,8 @@ impl SplunkHec {
                             tokio::spawn(send_hec_request(permit, block_length, labels, channel, client, request, request_shutdown.clone(), uri_clone));
                         }
                         Err(err) => {
-                            error!("Discarding block due to throttle error: {err}");
+                            debug!("Discarding block due to throttle error: {err}");
+                            self.block_cache.advance(&mut handle);
                         }
                     }
                 }

lading/src/generator/splunk_hec.rs

@@ -26,7 +26,7 @@ use tokio::{
     task::{JoinError, JoinSet},
     time::Duration,
 };
-use tracing::{error, info, trace};
+use tracing::{debug, info, trace};
 
 use lading_payload::block;
 
@@ -271,7 +271,8 @@ impl TcpWorker {
                             }
                         }
                         Err(err) => {
-                            error!("Discarding block due to throttle error: {err}");
+                            debug!("Discarding block due to throttle error: {err}");
+                            self.block_cache.advance(&mut handle);
                         }
                     }
                 }

lading/src/generator/tcp.rs

@@ -38,7 +38,7 @@ use std::{
     time::Duration,
 };
 use tokio::sync::Semaphore;
-use tracing::{error, info, warn};
+use tracing::{debug, error, info, warn};
 
 const MAX_RETRY_MILLIS: u16 = 6_400;
 
@@ -380,7 +380,8 @@ impl TraceAgent {
                             });
                         }
                         Err(err) => {
-                            error!("Discarding block due to throttle error: {err}");
+                            debug!("Discarding block due to throttle error: {err}");
+                            self.block_cache.advance(&mut handle);
                         }
                     }
                 },