Skip to content
Merged
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
88 changes: 42 additions & 46 deletions src/internal_events/aws_sqs.rs
Original file line number Diff line number Diff line change
@@ -1,5 +1,3 @@
#![allow(dead_code)] // TODO requires optional feature compilation

#[cfg(feature = "sources-aws_s3")]
pub use s3::*;
use vector_lib::counter;
Expand Down Expand Up @@ -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)]
Expand All @@ -262,31 +302,6 @@ impl<E: std::fmt::Display> 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")]
Expand All @@ -312,22 +327,3 @@ impl<E: std::fmt::Display> 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);
}
}
15 changes: 11 additions & 4 deletions src/internal_events/http_client_source.rs
Original file line number Diff line number Diff line change
@@ -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!(
Expand Down
16 changes: 14 additions & 2 deletions src/internal_events/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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::*;
Expand Down
157 changes: 83 additions & 74 deletions src/internal_events/parser.rs
Original file line number Diff line number Diff line change
@@ -1,54 +1,12 @@
#![allow(dead_code)] // TODO requires optional feature compilation

use std::borrow::Cow;

use vector_lib::{
NamedInternalEvent, counter,
internal_event::{
ComponentEventsDropped, CounterName, InternalEvent, UNINTENTIONAL, 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()
}
}

#[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;
Expand Down Expand Up @@ -85,38 +43,89 @@ impl<const DROP_EVENT: bool> 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));
}
}
}
4 changes: 2 additions & 2 deletions src/internal_events/prometheus.rs
Original file line number Diff line number Diff line change
@@ -1,5 +1,3 @@
#![allow(dead_code)] // TODO requires optional feature compilation

#[cfg(feature = "sources-prometheus_scrape")]
use std::borrow::Cow;

Expand Down Expand Up @@ -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!(
Expand Down
Loading
Loading