vector/internal_events/
common.rs

1use std::time::Instant;
2
3use vector_lib::NamedInternalEvent;
4pub use vector_lib::internal_event::EventsReceived;
5use vector_lib::internal_event::{
6    ComponentEventsDropped, CounterName, HistogramName, INTENTIONAL, InternalEvent, UNINTENTIONAL,
7    error_stage, error_type,
8};
9use vector_lib::{counter, histogram};
10
11#[derive(Debug, NamedInternalEvent)]
12pub struct KeyOutsideBasePrefixError<'a> {
13    /// Bounded preview of the rejected key — never the full rendered value,
14    /// so attacker-controlled input can't amplify into logs and secrets in
15    /// templated header/query fields don't leak.
16    pub key_preview: &'a str,
17    /// Full byte length of the rejected rendered value.
18    pub key_len: usize,
19    pub message: &'a str,
20}
21
22impl InternalEvent for KeyOutsideBasePrefixError<'_> {
23    fn emit(self) {
24        error!(
25            message = "Rendered key is outside the configured base prefix; dropping event.",
26            key_preview = self.key_preview,
27            key_len = self.key_len,
28            error = self.message,
29            error_type = error_type::CONFINEMENT_FAILED,
30            stage = error_stage::PROCESSING,
31        );
32        counter!(
33            CounterName::ComponentErrorsTotal,
34            "error_type" => error_type::CONFINEMENT_FAILED,
35            "stage" => error_stage::PROCESSING,
36        )
37        .increment(1);
38        emit!(ComponentEventsDropped::<INTENTIONAL> {
39            count: 1,
40            reason: "Rendered key outside base prefix.",
41        });
42    }
43}
44
45#[derive(Debug, NamedInternalEvent)]
46pub struct EndpointBytesReceived<'a> {
47    pub byte_size: usize,
48    pub protocol: &'a str,
49    pub endpoint: &'a str,
50}
51
52impl InternalEvent for EndpointBytesReceived<'_> {
53    fn emit(self) {
54        trace!(
55            message = "Bytes received.",
56            byte_size = %self.byte_size,
57            protocol = %self.protocol,
58            endpoint = %self.endpoint,
59        );
60        counter!(
61            CounterName::ComponentReceivedBytesTotal,
62            "protocol" => self.protocol.to_owned(),
63            "endpoint" => self.endpoint.to_owned(),
64        )
65        .increment(self.byte_size as u64);
66    }
67}
68
69#[derive(Debug, NamedInternalEvent)]
70pub struct EndpointBytesSent<'a> {
71    pub byte_size: usize,
72    pub protocol: &'a str,
73    pub endpoint: &'a str,
74}
75
76impl InternalEvent for EndpointBytesSent<'_> {
77    fn emit(self) {
78        trace!(
79            message = "Bytes sent.",
80            byte_size = %self.byte_size,
81            protocol = %self.protocol,
82            endpoint = %self.endpoint
83        );
84        counter!(
85            CounterName::ComponentSentBytesTotal,
86            "protocol" => self.protocol.to_string(),
87            "endpoint" => self.endpoint.to_string()
88        )
89        .increment(self.byte_size as u64);
90    }
91}
92
93#[derive(Debug, NamedInternalEvent)]
94pub struct SocketOutgoingConnectionError<E> {
95    pub error: E,
96}
97
98impl<E: std::error::Error> InternalEvent for SocketOutgoingConnectionError<E> {
99    fn emit(self) {
100        error!(
101            message = "Unable to connect.",
102            error = %self.error,
103            error_code = "failed_connecting",
104            error_type = error_type::CONNECTION_FAILED,
105            stage = error_stage::SENDING,
106        );
107        counter!(
108            CounterName::ComponentErrorsTotal,
109            "error_code" => "failed_connecting",
110            "error_type" => error_type::CONNECTION_FAILED,
111            "stage" => error_stage::SENDING,
112        )
113        .increment(1);
114    }
115}
116
117const STREAM_CLOSED: &str = "stream_closed";
118
119#[derive(Debug, NamedInternalEvent)]
120pub struct StreamClosedError {
121    pub count: usize,
122}
123
124impl InternalEvent for StreamClosedError {
125    fn emit(self) {
126        error!(
127            message = "Failed to forward event(s), downstream is closed.",
128            error_code = STREAM_CLOSED,
129            error_type = error_type::WRITER_FAILED,
130            stage = error_stage::SENDING,
131        );
132        counter!(
133            CounterName::ComponentErrorsTotal,
134            "error_code" => STREAM_CLOSED,
135            "error_type" => error_type::WRITER_FAILED,
136            "stage" => error_stage::SENDING,
137        )
138        .increment(1);
139        emit!(ComponentEventsDropped::<UNINTENTIONAL> {
140            count: self.count,
141            reason: "Downstream is closed.",
142        });
143    }
144}
145
146#[derive(Debug, NamedInternalEvent)]
147pub struct CollectionCompleted {
148    pub start: Instant,
149    pub end: Instant,
150}
151
152impl InternalEvent for CollectionCompleted {
153    fn emit(self) {
154        debug!(message = "Collection completed.");
155        counter!(CounterName::CollectCompletedTotal).increment(1);
156        histogram!(HistogramName::CollectDurationSeconds).record(self.end - self.start);
157    }
158}
159
160#[derive(Debug, NamedInternalEvent)]
161pub struct SinkRequestBuildError<E> {
162    pub error: E,
163}
164
165impl<E: std::fmt::Display> InternalEvent for SinkRequestBuildError<E> {
166    fn emit(self) {
167        // Providing the name of the sink with the build error is not necessary because the emitted log
168        // message contains the sink name in `component_type` field thanks to `tracing` spans. For example:
169        // "<timestamp> ERROR sink{component_kind="sink" component_id=sink0 component_type=aws_s3 component_name=sink0}: vector::internal_events::common: Failed to build request."
170        error!(
171            message = format!("Failed to build request."),
172            error = %self.error,
173            error_type = error_type::ENCODER_FAILED,
174            stage = error_stage::PROCESSING,
175        );
176        counter!(
177            CounterName::ComponentErrorsTotal,
178            "error_type" => error_type::ENCODER_FAILED,
179            "stage" => error_stage::PROCESSING,
180        )
181        .increment(1);
182    }
183}
vector/internal_events/common.rs

vector/internal_events/
common.rs
