Merge remote-tracking branch 'origin/master' into bugfix-22313-amqp-reconnect

Author: aramperes

.github/actions/spelling/allow.txt

@@ -291,6 +291,7 @@ downsides
 downwardapi
 ede
 emoji
+enableable
 esbuild
 etld
 fakeintake

.github/actions/spelling/expect.txt

@@ -175,6 +175,7 @@ efgh
 Elhage
 emerg
 Enableable
+enableable
 endianess
 endler
 eni

Cargo.lock

@@ -163,7 +163,7 @@ rand_distr = { version = "0.5.1", default-features = false }
 semver = { version = "1.0.26", default-features = false, features = ["serde", "std"] }
 serde_json = { version = "1.0.140", default-features = false, features = ["raw_value", "std"] }
 serde = { version = "1.0.219", default-features = false, features = ["alloc", "derive", "rc"] }
-snafu = { version = "0.7.5", default-features = false, features = ["futures", "std"] }
+snafu = { version = "0.8.0", default-features = false, features = ["futures", "std"] }
 tempfile = "3.20.0"
 tokio = { version = "1.45.1", default-features = false, features = ["full"] }
 toml = { version = "0.8.23", default-features = false, features = ["display", "parse"] }

Cargo.toml

@@ -0,0 +1,3 @@
+The `kubernetes_logs` source now includes a new `max_merged_line_bytes` configuration option. This setting enables users to cap the size of log lines after they’ve been combined using `auto_partial_merge`. Previously, the `max_line_bytes` field only restricted line sizes *before* merging, leaving no practical way to limit the length of merged lines—unless you set a size so tiny that it prevented merging altogether by stopping short of the continuation character. This new option gives you better control over merged line sizes.
+
+authors: ganelo

changelog.d/22581_max_merged_line_bytes.feature.md

@@ -0,0 +1,3 @@
+Added `rate_limit_num` and `rate_limit_duration_secs` options to `kafka` sink, to enable rate limiting this sink.
+
+authors: esensar Quad9DNS

changelog.d/23196_kafka_sink_rate_limit.feature.md

@@ -0,0 +1,3 @@
+Adds support for session tokens in AWS authentication options. When using temporary credentials (access key, secret key, and session token), the session token is required. Temporary credentials can be provided by an external system and updated using the `SECRET` backend.
+
+authors: anil-db

changelog.d/aws-auth-session-token.enhancement.md

@@ -387,7 +387,6 @@ impl ChunkedGelfDecoder {
                     warn!(
                         message_id = message_id,
                         timeout_secs = timeout.as_secs_f64(),
-                        internal_log_rate_limit = true,
                         "Message was not fully received within the timeout window. Discarding it."
                     );
                 }
@@ -409,7 +408,6 @@ impl ChunkedGelfDecoder {
             debug!(
                 message_id = message_id,
                 sequence_number = sequence_number,
-                internal_log_rate_limit = true,
                 "Received a duplicate chunk. Ignoring it."
             );
             return Ok(None);

lib/codecs/src/decoding/framing/chunked_gelf.rs

@@ -14,7 +14,7 @@ impl<E: std::fmt::Display> InternalEvent for DnstapParseWarning<E> {
             error = %self.error,
             stage = error_stage::PROCESSING,
             error_type = error_type::PARSER_FAILED,
-            internal_log_rate_limit = true,
+
         );
     }
 }

lib/dnstap-parser/src/internal_events.rs

@@ -1,11 +1,18 @@
-use std::io::{self, BufRead};
+use std::{
+    cmp::min,
+    io::{self, BufRead},
+};
 
 use bstr::Finder;
 use bytes::BytesMut;
-use tracing::warn;
 
 use crate::FilePosition;
 
+pub struct ReadResult {
+    pub successfully_read: Option<usize>,
+    pub discarded_for_size_and_truncated: Vec<BytesMut>,
+}
+
 /// Read up to `max_size` bytes from `reader`, splitting by `delim`
 ///
 /// The function reads up to `max_size` bytes from `reader`, splitting the input
@@ -29,17 +36,18 @@ use crate::FilePosition;
 /// Benchmarks indicate that this function processes in the high single-digit
 /// GiB/s range for buffers of length 1KiB. For buffers any smaller than this
 /// the overhead of setup dominates our benchmarks.
-pub fn read_until_with_max_size<R: BufRead + ?Sized>(
-    reader: &mut R,
-    position: &mut FilePosition,
-    delim: &[u8],
-    buf: &mut BytesMut,
+pub fn read_until_with_max_size<'a, R: BufRead + ?Sized>(
+    reader: &'a mut R,
+    position: &'a mut FilePosition,
+    delim: &'a [u8],
+    buf: &'a mut BytesMut,
     max_size: usize,
-) -> io::Result<Option<usize>> {
+) -> io::Result<ReadResult> {
     let mut total_read = 0;
     let mut discarding = false;
     let delim_finder = Finder::new(delim);
     let delim_len = delim.len();
+    let mut discarded_for_size_and_truncated = Vec::new();
     loop {
         let available: &[u8] = match reader.fill_buf() {
             Ok(n) => n,
@@ -68,16 +76,20 @@ pub fn read_until_with_max_size<R: BufRead + ?Sized>(
         total_read += used;
 
         if !discarding && buf.len() > max_size {
-            warn!(
-                message = "Found line that exceeds max_line_bytes; discarding.",
-                internal_log_rate_limit = true
-            );
+            // keep only the first <1k bytes to make sure we can actually emit a usable error
+            let length_to_keep = min(1000, max_size);
+            let mut truncated: BytesMut = BytesMut::zeroed(length_to_keep);
+            truncated.copy_from_slice(&buf[0..length_to_keep]);
+            discarded_for_size_and_truncated.push(truncated);
             discarding = true;
         }
 
         if done {
             if !discarding {
-                return Ok(Some(total_read));
+                return Ok(ReadResult {
+                    successfully_read: Some(total_read),
+                    discarded_for_size_and_truncated,
+                });
             } else {
                 discarding = false;
                 buf.clear();
@@ -87,7 +99,10 @@ pub fn read_until_with_max_size<R: BufRead + ?Sized>(
             // us to observe an incomplete write. We return None here and let the loop continue
             // next time the method is called. This is safe because the buffer is specific to this
             // FileWatcher.
-            return Ok(None);
+            return Ok(ReadResult {
+                successfully_read: None,
+                discarded_for_size_and_truncated,
+            });
         }
     }
 }
@@ -99,6 +114,8 @@ mod test {
     use bytes::{BufMut, BytesMut};
     use quickcheck::{QuickCheck, TestResult};
 
+    use crate::buffer::ReadResult;
+
     use super::read_until_with_max_size;
 
     fn qc_inner(chunks: Vec<Vec<u8>>, delim: u8, max_size: NonZeroU8) -> TestResult {
@@ -181,7 +198,10 @@ mod test {
             )
             .unwrap()
             {
-                None => {
+                ReadResult {
+                    successfully_read: None,
+                    discarded_for_size_and_truncated: _,
+                } => {
                     // Subject only returns None if this is the last chunk _and_
                     // the chunk did not contain a delimiter _or_ the delimiter
                     // was outside the max_size range _or_ the current chunk is empty.
@@ -190,7 +210,10 @@ mod test {
                         .any(|details| ((details.chunk_index == idx) && details.within_max_size));
                     assert!(chunk.is_empty() || !has_valid_delimiter)
                 }
-                Some(total_read) => {
+                ReadResult {
+                    successfully_read: Some(total_read),
+                    discarded_for_size_and_truncated: _,
+                } => {
                     // Now that the function has returned we confirm that the
                     // returned details match our `first_delim` and also that
                     // the `buffer` is populated correctly.

lib/file-source/src/buffer.rs

@@ -19,7 +19,7 @@ use tracing::{debug, error, info, trace};
 
 use crate::{
     checkpointer::{Checkpointer, CheckpointsView},
-    file_watcher::FileWatcher,
+    file_watcher::{FileWatcher, RawLineResult},
     fingerprinter::{FileFingerprint, Fingerprinter},
     paths_provider::PathsProvider,
     FileSourceInternalEvents, ReadFrom,
@@ -263,7 +263,19 @@ where
 
                 let start = time::Instant::now();
                 let mut bytes_read: usize = 0;
-                while let Ok(Some(line)) = watcher.read_line() {
+                while let Ok(RawLineResult {
+                    raw_line: Some(line),
+                    discarded_for_size_and_truncated,
+                }) = watcher.read_line()
+                {
+                    discarded_for_size_and_truncated.iter().for_each(|buf| {
+                        self.emitter.emit_file_line_too_long(
+                            &buf.clone(),
+                            self.max_line_bytes,
+                            buf.len(),
+                        )
+                    });
+
                     let sz = line.bytes.len();
                     trace!(
                         message = "Read bytes.",