Merge branch 'master' into greylilac09/add-incremental-to-absolute

Author: GreyLilac09

.github/actions/spelling/allow.txt

@@ -461,6 +461,7 @@ Softbank
 Sogou
 solarwinds
 Soref
+sortedset
 splunk
 ssh
 staticuser
@@ -524,6 +525,7 @@ XSALSA
 yandex
 Yarvik
 Yifang
+zadd
 zeek
 zookeeper
 Zopo

.github/workflows/semantic.yml

@@ -200,6 +200,7 @@ jobs:
             blackhole sink
             clickhouse sink
             console sink
+            datadog_common sink
             datadog_archives sink
             datadog_events sink
             datadog_logs sink

benches/transform/dedupe.rs

@@ -6,7 +6,7 @@ use criterion::{
     criterion_group, measurement::WallTime, BatchSize, BenchmarkGroup, BenchmarkId, Criterion,
     SamplingMode, Throughput,
 };
-use vector::transforms::dedupe::common::{CacheConfig, FieldMatchConfig};
+use vector::transforms::dedupe::common::{CacheConfig, FieldMatchConfig, TimedCacheConfig};
 use vector::transforms::dedupe::config::DedupeConfig;
 use vector::transforms::dedupe::transform::Dedupe;
 use vector_lib::transform::Transform;
@@ -45,6 +45,7 @@ fn dedupe(c: &mut Criterion) {
             dedupe_config: DedupeConfig {
                 fields: Some(FieldMatchConfig::IgnoreFields(vec!["message".into()])),
                 cache: cache.clone(),
+                time_settings: None,
             },
         },
         // Modification of previous where field "message" is matched.
@@ -54,6 +55,33 @@ fn dedupe(c: &mut Criterion) {
             dedupe_config: DedupeConfig {
                 fields: Some(FieldMatchConfig::MatchFields(vec!["message".into()])),
                 cache: cache.clone(),
+                time_settings: None,
+            },
+        },
+        // Modification of previous where deduplication with max age is used.
+        Param {
+            slug: "field_match_message_timed",
+            input: fixed_stream.clone(),
+            dedupe_config: DedupeConfig {
+                fields: Some(FieldMatchConfig::MatchFields(vec!["message".into()])),
+                cache: cache.clone(),
+                time_settings: Some(TimedCacheConfig {
+                    max_age_ms: Duration::from_secs(5),
+                    refresh_on_drop: false,
+                }),
+            },
+        },
+        // Modification of previous where refresh on drop is enabled.
+        Param {
+            slug: "field_match_message_timed_refresh_on_drop",
+            input: fixed_stream.clone(),
+            dedupe_config: DedupeConfig {
+                fields: Some(FieldMatchConfig::MatchFields(vec!["message".into()])),
+                cache: cache.clone(),
+                time_settings: Some(TimedCacheConfig {
+                    max_age_ms: Duration::from_secs(5),
+                    refresh_on_drop: true,
+                }),
             },
         },
         // Measurement where ignore fields do not exist in the event.
@@ -69,6 +97,7 @@ fn dedupe(c: &mut Criterion) {
                     "cdeab".into(),
                     "bcdea".into(),
                 ])),
+                time_settings: None,
             },
         },
         // Modification of previous where match fields do not exist in the
@@ -85,6 +114,7 @@ fn dedupe(c: &mut Criterion) {
                     "cdeab".into(),
                     "bcdea".into(),
                 ])),
+                time_settings: None,
             },
         },
     ] {

changelog.d/23448_dnstap_tcp_source_throughput_issue.fix.md

@@ -0,0 +1,3 @@
+Fixed an issue with `dnstap` `tcp` source, where it could have a drastically reduced throughput after number of connections exceeded number of available cores.
+
+authors: esensar Quad9DNS

changelog.d/23480_dedupe_transform_time_settings.feature.md

@@ -0,0 +1,3 @@
+Adds `time_settings` configuration for `dedupe` transform, to enable configuring `max_age` for items in the deduplication cache, helping make a distinction between real duplicates and duplication that is expected in the incoming data over a longer period of time.
+
+authors: esensar Quad9DNS

changelog.d/23513_config_prevent_panic.fix.md

@@ -0,0 +1,3 @@
+Prevent panic on non-UTF8 environment variables.
+
+authors: kurochan

changelog.d/23514_datadog_prevent_panic.fix.md

@@ -0,0 +1,3 @@
+Prevents a panic when DD_API_KEY contains newline characters.
+
+authors: kurochan

changelog.d/redis_zadd_in_sink.feature.md

@@ -0,0 +1,3 @@
+Adds support for the Redis ZADD command in the Redis sink.
+
+authors: 5Dev24

lib/vector-buffers/src/buffer_usage_data.rs

@@ -242,8 +242,10 @@ impl BufferUsage {
     /// The `buffer_id` should be a unique name -- ideally the `component_id` of the sink using this buffer -- but is
     /// not used for anything other than reporting, and so has no _requirement_ to be unique.
     pub fn install(self, buffer_id: &str) {
+        let buffer_id = buffer_id.to_string();
         let span = self.span;
         let stages = self.stages;
+        let task_name = format!("buffer usage reporter ({buffer_id})");
 
         let task = async move {
             let mut interval = interval(Duration::from_secs(2));
@@ -264,6 +266,7 @@ impl BufferUsage {
                     let received = stage.received.consume();
                     if received.has_updates() {
                         emit(BufferEventsReceived {
+                            buffer_id: buffer_id.clone(),
                             idx: stage.idx,
                             count: received.event_count,
                             byte_size: received.event_byte_size,
@@ -273,6 +276,7 @@ impl BufferUsage {
                     let sent = stage.sent.consume();
                     if sent.has_updates() {
                         emit(BufferEventsSent {
+                            buffer_id: buffer_id.clone(),
                             idx: stage.idx,
                             count: sent.event_count,
                             byte_size: sent.event_byte_size,
@@ -282,6 +286,7 @@ impl BufferUsage {
                     let dropped = stage.dropped.consume();
                     if dropped.has_updates() {
                         emit(BufferEventsDropped {
+                            buffer_id: buffer_id.clone(),
                             idx: stage.idx,
                             intentional: false,
                             reason: "corrupted_events",
@@ -293,6 +298,7 @@ impl BufferUsage {
                     let dropped_intentional = stage.dropped_intentional.consume();
                     if dropped_intentional.has_updates() {
                         emit(BufferEventsDropped {
+                            buffer_id: buffer_id.clone(),
                             idx: stage.idx,
                             intentional: true,
                             reason: "drop_newest",
@@ -304,7 +310,6 @@ impl BufferUsage {
             }
         };
 
-        let task_name = format!("buffer usage reporter ({buffer_id})");
         spawn_named(task.instrument(span.or_current()), task_name.as_str());
     }
 }