From b1fcfdae918c171174420d59e691dc04081b8233 Mon Sep 17 00:00:00 2001 From: Jincheng Cao Date: Thu, 22 Oct 2026 14:35:19 -0400 Subject: [PATCH 1/4] chore(internal_events): remove blanket dead_code allows from remaining internal events --- src/internal_events/aws_sqs.rs | 88 ++++++------ src/internal_events/http_client_source.rs | 15 ++- src/internal_events/parser.rs | 157 ++++++++++++---------- src/internal_events/prometheus.rs | 4 +- src/internal_events/unix.rs | 18 ++- 5 files changed, 150 insertions(+), 132 deletions(-) diff --git a/src/internal_events/aws_sqs.rs b/src/internal_events/aws_sqs.rs index 40bb130109527..f8a5ebf424614 100644 --- a/src/internal_events/aws_sqs.rs +++ b/src/internal_events/aws_sqs.rs @@ -1,5 +1,3 @@ -#![allow(dead_code)] // TODO requires optional feature compilation - #[cfg(feature = "sources-aws_s3")] pub use s3::*; use vector_lib::counter; @@ -236,6 +234,48 @@ mod s3 { .increment(1); } } + + #[derive(Debug, NamedInternalEvent)] + pub struct SqsMessageReceiveSucceeded { + pub count: usize, + } + + impl InternalEvent for SqsMessageReceiveSucceeded { + fn emit(self) { + trace!(message = "Received SQS messages.", count = %self.count); + counter!(CounterName::SqsMessageReceiveSucceededTotal).increment(1); + counter!(CounterName::SqsMessageReceivedMessagesTotal).increment(self.count as u64); + } + } + + #[derive(Debug, NamedInternalEvent)] + pub struct SqsMessageProcessingSucceeded<'a> { + pub message_id: &'a str, + } + + impl InternalEvent for SqsMessageProcessingSucceeded<'_> { + fn emit(self) { + trace!(message = "Processed SQS message successfully.", message_id = %self.message_id); + counter!(CounterName::SqsMessageProcessingSucceededTotal).increment(1); + } + } + + #[derive(Debug, NamedInternalEvent)] + pub struct SqsS3EventRecordInvalidEventIgnored<'a> { + pub bucket: &'a str, + pub key: &'a str, + pub kind: &'a str, + pub name: &'a str, + } + + impl InternalEvent for SqsS3EventRecordInvalidEventIgnored<'_> { + fn emit(self) { + warn!(message = "Ignored S3 record in SQS message for an event that was not ObjectCreated.", + bucket = %self.bucket, key = %self.key, kind = %self.kind, name = %self.name); + counter!(CounterName::SqsS3EventRecordIgnoredTotal, "ignore_type" => "invalid_event_kind") + .increment(1); + } + } } #[derive(Debug, NamedInternalEvent)] @@ -262,31 +302,6 @@ impl InternalEvent for SqsMessageReceiveError<'_, E> { } } -#[derive(Debug, NamedInternalEvent)] -pub struct SqsMessageReceiveSucceeded { - pub count: usize, -} - -impl InternalEvent for SqsMessageReceiveSucceeded { - fn emit(self) { - trace!(message = "Received SQS messages.", count = %self.count); - counter!(CounterName::SqsMessageReceiveSucceededTotal).increment(1); - counter!(CounterName::SqsMessageReceivedMessagesTotal).increment(self.count as u64); - } -} - -#[derive(Debug, NamedInternalEvent)] -pub struct SqsMessageProcessingSucceeded<'a> { - pub message_id: &'a str, -} - -impl InternalEvent for SqsMessageProcessingSucceeded<'_> { - fn emit(self) { - trace!(message = "Processed SQS message successfully.", message_id = %self.message_id); - counter!(CounterName::SqsMessageProcessingSucceededTotal).increment(1); - } -} - // AWS SQS source #[cfg(feature = "sources-aws_sqs")] @@ -312,22 +327,3 @@ impl InternalEvent for SqsMessageDeleteError<'_, E> { .increment(1); } } - -// AWS s3 source - -#[derive(Debug, NamedInternalEvent)] -pub struct SqsS3EventRecordInvalidEventIgnored<'a> { - pub bucket: &'a str, - pub key: &'a str, - pub kind: &'a str, - pub name: &'a str, -} - -impl InternalEvent for SqsS3EventRecordInvalidEventIgnored<'_> { - fn emit(self) { - warn!(message = "Ignored S3 record in SQS message for an event that was not ObjectCreated.", - bucket = %self.bucket, key = %self.key, kind = %self.kind, name = %self.name); - counter!(CounterName::SqsS3EventRecordIgnoredTotal, "ignore_type" => "invalid_event_kind") - .increment(1); - } -} diff --git a/src/internal_events/http_client_source.rs b/src/internal_events/http_client_source.rs index 0ba52d722dfee..173f274036251 100644 --- a/src/internal_events/http_client_source.rs +++ b/src/internal_events/http_client_source.rs @@ -1,20 +1,27 @@ -#![allow(dead_code)] // TODO requires optional feature compilation - use vector_lib::{ NamedInternalEvent, counter, internal_event::{CounterName, InternalEvent, error_stage, error_type}, - json_size::JsonSize, }; use super::prelude::http_error_code; +#[cfg(any( + feature = "sources-http_client", + feature = "sources-prometheus_scrape", + feature = "sources-okta" +))] #[derive(Debug, NamedInternalEvent)] pub struct HttpClientEventsReceived { - pub byte_size: JsonSize, + pub byte_size: vector_lib::json_size::JsonSize, pub count: usize, pub url: String, } +#[cfg(any( + feature = "sources-http_client", + feature = "sources-prometheus_scrape", + feature = "sources-okta" +))] impl InternalEvent for HttpClientEventsReceived { fn emit(self) { trace!( diff --git a/src/internal_events/parser.rs b/src/internal_events/parser.rs index f615b1f36ecaa..08a4c7bb18ffa 100644 --- a/src/internal_events/parser.rs +++ b/src/internal_events/parser.rs @@ -1,7 +1,3 @@ -#![allow(dead_code)] // TODO requires optional feature compilation - -use std::borrow::Cow; - use vector_lib::{ NamedInternalEvent, counter, internal_event::{ @@ -9,46 +5,8 @@ use vector_lib::{ }, }; -fn truncate_string_at(s: &str, maxlen: usize) -> Cow<'_, str> { - let ellipsis: &str = "[...]"; - if s.len() >= maxlen { - let mut len = maxlen - ellipsis.len(); - while !s.is_char_boundary(len) { - len -= 1; - } - #[expect( - clippy::string_slice, - reason = "len is adjusted to a char boundary in the loop above" - )] - format!("{}{ellipsis}", &s[..len]).into() - } else { - s.into() - } -} - -#[derive(Debug, NamedInternalEvent)] -pub struct ParserMatchError<'a> { - pub value: &'a [u8], -} - -impl InternalEvent for ParserMatchError<'_> { - fn emit(self) { - error!( - message = "Pattern failed to match.", - error_code = "no_match_found", - error_type = error_type::CONDITION_FAILED, - stage = error_stage::PROCESSING, - field = truncate_string_at(&String::from_utf8_lossy(self.value), 60).as_ref() - ); - counter!( - CounterName::ComponentErrorsTotal, - "error_code" => "no_match_found", - "error_type" => error_type::CONDITION_FAILED, - "stage" => error_stage::PROCESSING, - ) - .increment(1); - } -} +#[cfg(feature = "sources-kubernetes_logs")] +pub use self::kubernetes_logs::*; #[allow(dead_code)] pub const DROP_EVENT: bool = true; @@ -85,38 +43,89 @@ impl InternalEvent for ParserMissingFieldError<'_, DROP_ } } -#[derive(Debug, NamedInternalEvent)] -pub struct ParserConversionError<'a> { - pub name: &'a str, - pub error: crate::types::Error, -} +#[cfg(feature = "sources-kubernetes_logs")] +mod kubernetes_logs { + use std::borrow::Cow; -impl InternalEvent for ParserConversionError<'_> { - fn emit(self) { - error!( - message = "Could not convert types.", - name = %self.name, - error = ?self.error, - error_code = "type_conversion", - error_type = error_type::CONVERSION_FAILED, - stage = error_stage::PROCESSING - ); - counter!( - CounterName::ComponentErrorsTotal, - "error_code" => "type_conversion", - "error_type" => error_type::CONVERSION_FAILED, - "stage" => error_stage::PROCESSING, - "name" => self.name.to_string(), - ) - .increment(1); + use vector_lib::{ + NamedInternalEvent, counter, + internal_event::{CounterName, InternalEvent, error_stage, error_type}, + }; + + fn truncate_string_at(s: &str, maxlen: usize) -> Cow<'_, str> { + let ellipsis: &str = "[...]"; + if s.len() >= maxlen { + let mut len = maxlen - ellipsis.len(); + while !s.is_char_boundary(len) { + len -= 1; + } + #[expect( + clippy::string_slice, + reason = "len is adjusted to a char boundary in the loop above" + )] + format!("{}{ellipsis}", &s[..len]).into() + } else { + s.into() + } } -} -#[cfg(test)] -mod test { - #[test] - fn truncate_utf8() { - let message = "Hello 😁 this is test."; - assert_eq!("Hello [...]", super::truncate_string_at(message, 13)); + #[derive(Debug, NamedInternalEvent)] + pub struct ParserMatchError<'a> { + pub value: &'a [u8], + } + + impl InternalEvent for ParserMatchError<'_> { + fn emit(self) { + error!( + message = "Pattern failed to match.", + error_code = "no_match_found", + error_type = error_type::CONDITION_FAILED, + stage = error_stage::PROCESSING, + field = truncate_string_at(&String::from_utf8_lossy(self.value), 60).as_ref() + ); + counter!( + CounterName::ComponentErrorsTotal, + "error_code" => "no_match_found", + "error_type" => error_type::CONDITION_FAILED, + "stage" => error_stage::PROCESSING, + ) + .increment(1); + } + } + + #[derive(Debug, NamedInternalEvent)] + pub struct ParserConversionError<'a> { + pub name: &'a str, + pub error: crate::types::Error, + } + + impl InternalEvent for ParserConversionError<'_> { + fn emit(self) { + error!( + message = "Could not convert types.", + name = %self.name, + error = ?self.error, + error_code = "type_conversion", + error_type = error_type::CONVERSION_FAILED, + stage = error_stage::PROCESSING + ); + counter!( + CounterName::ComponentErrorsTotal, + "error_code" => "type_conversion", + "error_type" => error_type::CONVERSION_FAILED, + "stage" => error_stage::PROCESSING, + "name" => self.name.to_string(), + ) + .increment(1); + } + } + + #[cfg(test)] + mod test { + #[test] + fn truncate_utf8() { + let message = "Hello 😁 this is test."; + assert_eq!("Hello [...]", super::truncate_string_at(message, 13)); + } } } diff --git a/src/internal_events/prometheus.rs b/src/internal_events/prometheus.rs index e2c8ff4ea9e78..5ccba86236090 100644 --- a/src/internal_events/prometheus.rs +++ b/src/internal_events/prometheus.rs @@ -1,5 +1,3 @@ -#![allow(dead_code)] // TODO requires optional feature compilation - #[cfg(feature = "sources-prometheus_scrape")] use std::borrow::Cow; @@ -44,11 +42,13 @@ impl InternalEvent for PrometheusParseError<'_> { } } +#[cfg(feature = "sources-prometheus_remote_write")] #[derive(Debug, NamedInternalEvent)] pub struct PrometheusRemoteWriteParseError { pub error: prost::DecodeError, } +#[cfg(feature = "sources-prometheus_remote_write")] impl InternalEvent for PrometheusRemoteWriteParseError { fn emit(self) { error!( diff --git a/src/internal_events/unix.rs b/src/internal_events/unix.rs index fc98ff7feacbd..36dac4f71b6f9 100644 --- a/src/internal_events/unix.rs +++ b/src/internal_events/unix.rs @@ -1,7 +1,3 @@ -#![allow(dead_code)] // TODO requires optional feature compilation - -use std::{io::Error, path::Path}; - use vector_lib::internal_event::{ ComponentEventsDropped, CounterName, InternalEvent, UNINTENTIONAL, error_stage, error_type, }; @@ -121,12 +117,22 @@ impl InternalEvent for UnixSendIncompleteError { } } +#[cfg(any( + feature = "sources-utils-net-unix", + feature = "sources-socket", + feature = "sources-dnstap" +))] #[derive(Debug, NamedInternalEvent)] pub struct UnixSocketFileDeleteError<'a> { - pub path: &'a Path, - pub error: Error, + pub path: &'a std::path::Path, + pub error: std::io::Error, } +#[cfg(any( + feature = "sources-utils-net-unix", + feature = "sources-socket", + feature = "sources-dnstap" +))] impl InternalEvent for UnixSocketFileDeleteError<'_> { fn emit(self) { error!( From e7b8f20490038c9634bae3dea603b04b777a1900 Mon Sep 17 00:00:00 2001 From: Jincheng Cao Date: Thu, 22 Oct 2026 16:52:49 -0400 Subject: [PATCH 2/4] chore(internal_events): gate http_client_source on its consuming sources --- src/internal_events/mod.rs | 16 ++++++++++++++-- 1 file changed, 14 insertions(+), 2 deletions(-) diff --git a/src/internal_events/mod.rs b/src/internal_events/mod.rs index c9c89983a7cb0..a3a4f99e9cb56 100644 --- a/src/internal_events/mod.rs +++ b/src/internal_events/mod.rs @@ -78,7 +78,13 @@ mod heartbeat; mod host_metrics; mod http; pub mod http_client; -#[cfg(feature = "sources-utils-http-client")] +#[cfg(any( + feature = "sources-apache_metrics", + feature = "sources-aws_ecs_metrics", + feature = "sources-http_client", + feature = "sources-okta", + feature = "sources-prometheus_scrape", +))] mod http_client_source; #[cfg(feature = "sinks-influxdb")] mod influxdb; @@ -232,7 +238,13 @@ pub(crate) use self::gcp_pubsub::*; pub(crate) use self::grpc::*; #[cfg(feature = "sources-host_metrics")] pub(crate) use self::host_metrics::*; -#[cfg(feature = "sources-utils-http-client")] +#[cfg(any( + feature = "sources-apache_metrics", + feature = "sources-aws_ecs_metrics", + feature = "sources-http_client", + feature = "sources-okta", + feature = "sources-prometheus_scrape", +))] pub(crate) use self::http_client_source::*; #[cfg(feature = "sinks-influxdb")] pub(crate) use self::influxdb::*; From d82be10997eaae6fa6bf85dbe39215cbd497b1be Mon Sep 17 00:00:00 2001 From: Jincheng Cao Date: Fri, 23 Oct 2026 13:42:59 -0400 Subject: [PATCH 3/4] Revert "chore(internal_events): gate http_client_source on its consuming sources" --- src/internal_events/mod.rs | 16 ++-------------- 1 file changed, 2 insertions(+), 14 deletions(-) diff --git a/src/internal_events/mod.rs b/src/internal_events/mod.rs index a3a4f99e9cb56..c9c89983a7cb0 100644 --- a/src/internal_events/mod.rs +++ b/src/internal_events/mod.rs @@ -78,13 +78,7 @@ mod heartbeat; mod host_metrics; mod http; pub mod http_client; -#[cfg(any( - feature = "sources-apache_metrics", - feature = "sources-aws_ecs_metrics", - feature = "sources-http_client", - feature = "sources-okta", - feature = "sources-prometheus_scrape", -))] +#[cfg(feature = "sources-utils-http-client")] mod http_client_source; #[cfg(feature = "sinks-influxdb")] mod influxdb; @@ -238,13 +232,7 @@ pub(crate) use self::gcp_pubsub::*; pub(crate) use self::grpc::*; #[cfg(feature = "sources-host_metrics")] pub(crate) use self::host_metrics::*; -#[cfg(any( - feature = "sources-apache_metrics", - feature = "sources-aws_ecs_metrics", - feature = "sources-http_client", - feature = "sources-okta", - feature = "sources-prometheus_scrape", -))] +#[cfg(feature = "sources-utils-http-client")] pub(crate) use self::http_client_source::*; #[cfg(feature = "sinks-influxdb")] pub(crate) use self::influxdb::*; From b6f8fa5bb75dae9e2dd4849165c8e2707fbaa50f Mon Sep 17 00:00:00 2001 From: Jincheng Cao Date: Fri, 23 Oct 2026 13:44:30 -0400 Subject: [PATCH 4/4] chore(internal_events): remove unused RETAIN_EVENT and redundant DROP_EVENT allow --- src/internal_events/parser.rs | 3 --- 1 file changed, 3 deletions(-) diff --git a/src/internal_events/parser.rs b/src/internal_events/parser.rs index 08a4c7bb18ffa..4e98cc3dd75a4 100644 --- a/src/internal_events/parser.rs +++ b/src/internal_events/parser.rs @@ -8,10 +8,7 @@ use vector_lib::{ #[cfg(feature = "sources-kubernetes_logs")] pub use self::kubernetes_logs::*; -#[allow(dead_code)] pub const DROP_EVENT: bool = true; -#[allow(dead_code)] -pub const RETAIN_EVENT: bool = false; #[derive(Debug, NamedInternalEvent)] pub struct ParserMissingFieldError<'a, const DROP_EVENT: bool> {