codecs/
internal_events.rs

1//! Internal events for codecs.
2
3use metrics::counter;
4use tracing::error;
5use vector_common::internal_event::{
6    ComponentEventsDropped, InternalEvent, UNINTENTIONAL, emit, error_stage, error_type,
7};
8use vector_common_macros::NamedInternalEvent;
9
10#[derive(Debug, NamedInternalEvent)]
11/// Emitted when a decoder framing error occurs.
12pub struct DecoderFramingError<E> {
13    /// The framing error that occurred.
14    pub error: E,
15}
16
17impl<E: std::fmt::Display> InternalEvent for DecoderFramingError<E> {
18    fn emit(self) {
19        error!(
20            message = "Failed framing bytes.",
21            error = %self.error,
22            error_code = "decoder_frame",
23            error_type = error_type::PARSER_FAILED,
24            stage = error_stage::PROCESSING,
25        );
26        counter!(
27            "component_errors_total",
28            "error_code" => "decoder_frame",
29            "error_type" => error_type::PARSER_FAILED,
30            "stage" => error_stage::PROCESSING,
31        )
32        .increment(1);
33    }
34}
35
36#[derive(Debug, NamedInternalEvent)]
37/// Emitted when a decoder fails to deserialize a frame.
38pub struct DecoderDeserializeError<'a> {
39    /// The deserialize error that occurred.
40    pub error: &'a vector_common::Error,
41}
42
43impl InternalEvent for DecoderDeserializeError<'_> {
44    fn emit(self) {
45        error!(
46            message = "Failed deserializing frame.",
47            error = %self.error,
48            error_code = "decoder_deserialize",
49            error_type = error_type::PARSER_FAILED,
50            stage = error_stage::PROCESSING,
51        );
52        counter!(
53            "component_errors_total",
54            "error_code" => "decoder_deserialize",
55            "error_type" => error_type::PARSER_FAILED,
56            "stage" => error_stage::PROCESSING,
57        )
58        .increment(1);
59    }
60}
61
62#[derive(Debug, NamedInternalEvent)]
63/// Emitted when an encoder framing error occurs.
64pub struct EncoderFramingError<'a> {
65    /// The framing error that occurred.
66    pub error: &'a crate::encoding::BoxedFramingError,
67}
68
69impl InternalEvent for EncoderFramingError<'_> {
70    fn emit(self) {
71        let reason = "Failed framing bytes.";
72        error!(
73            message = reason,
74            error = %self.error,
75            error_code = "encoder_frame",
76            error_type = error_type::ENCODER_FAILED,
77            stage = error_stage::SENDING,
78        );
79        counter!(
80            "component_errors_total",
81            "error_code" => "encoder_frame",
82            "error_type" => error_type::ENCODER_FAILED,
83            "stage" => error_stage::SENDING,
84        )
85        .increment(1);
86        emit(ComponentEventsDropped::<UNINTENTIONAL> { count: 1, reason });
87    }
88}
89
90#[derive(Debug, NamedInternalEvent)]
91/// Emitted when an encoder fails to serialize a frame.
92pub struct EncoderSerializeError<'a> {
93    /// The serialization error that occurred.
94    pub error: &'a vector_common::Error,
95}
96
97impl InternalEvent for EncoderSerializeError<'_> {
98    fn emit(self) {
99        const SERIALIZE_REASON: &str = "Failed serializing frame.";
100        error!(
101            message = SERIALIZE_REASON,
102            error = %self.error,
103            error_code = "encoder_serialize",
104            error_type = error_type::ENCODER_FAILED,
105            stage = error_stage::SENDING,
106        );
107        counter!(
108            "component_errors_total",
109            "error_code" => "encoder_serialize",
110            "error_type" => error_type::ENCODER_FAILED,
111            "stage" => error_stage::SENDING,
112        )
113        .increment(1);
114        emit(ComponentEventsDropped::<UNINTENTIONAL> {
115            count: 1,
116            reason: SERIALIZE_REASON,
117        });
118    }
119}
120
121#[derive(Debug, NamedInternalEvent)]
122/// Emitted when writing encoded bytes fails.
123pub struct EncoderWriteError<'a, E> {
124    /// The write error that occurred.
125    pub error: &'a E,
126    /// The number of events dropped by the failed write.
127    pub count: usize,
128}
129
130impl<E: std::fmt::Display> InternalEvent for EncoderWriteError<'_, E> {
131    fn emit(self) {
132        let reason = "Failed writing bytes.";
133        error!(
134            message = reason,
135            error = %self.error,
136            error_type = error_type::IO_FAILED,
137            stage = error_stage::SENDING,
138        );
139        counter!(
140            "component_errors_total",
141            "error_type" => error_type::ENCODER_FAILED,
142            "stage" => error_stage::SENDING,
143        )
144        .increment(1);
145        if self.count > 0 {
146            emit(ComponentEventsDropped::<UNINTENTIONAL> {
147                count: self.count,
148                reason,
149            });
150        }
151    }
152}
153
154#[cfg(feature = "arrow")]
155#[derive(Debug, NamedInternalEvent)]
156/// Emitted when encoding violates a schema constraint.
157pub struct EncoderNullConstraintError<'a> {
158    /// The schema constraint error that occurred.
159    pub error: &'a vector_common::Error,
160}
161
162#[cfg(feature = "arrow")]
163impl InternalEvent for EncoderNullConstraintError<'_> {
164    fn emit(self) {
165        const CONSTRAINT_REASON: &str = "Schema constraint violation.";
166        error!(
167            message = CONSTRAINT_REASON,
168            error = %self.error,
169            error_code = "encoding_null_constraint",
170            error_type = error_type::ENCODER_FAILED,
171            stage = error_stage::SENDING,
172        );
173        counter!(
174            "component_errors_total",
175            "error_code" => "encoding_null_constraint",
176            "error_type" => error_type::ENCODER_FAILED,
177            "stage" => error_stage::SENDING,
178        )
179        .increment(1);
180        emit(ComponentEventsDropped::<UNINTENTIONAL> {
181            count: 1,
182            reason: CONSTRAINT_REASON,
183        });
184    }
185}
codecs/internal_events.rs

codecs/
internal_events.rs
