vector/internal_events/
pulsar.rs

1#![allow(dead_code)] // TODO requires optional feature compilation
2
3#[cfg(feature = "sources-pulsar")]
4use metrics::Counter;
5use vector_lib::internal_event::{
6    ComponentEventsDropped, CounterName, InternalEvent, UNINTENTIONAL, error_stage, error_type,
7};
8use vector_lib::{NamedInternalEvent, counter};
9
10#[derive(Debug, NamedInternalEvent)]
11pub struct PulsarSendingError {
12    pub count: usize,
13    pub error: vector_lib::Error,
14}
15
16impl InternalEvent for PulsarSendingError {
17    fn emit(self) {
18        let reason = "A Pulsar sink generated an error.";
19        error!(
20            message = reason,
21            error = %self.error,
22            error_type = error_type::REQUEST_FAILED,
23            stage = error_stage::SENDING,
24        );
25        counter!(
26            CounterName::ComponentErrorsTotal,
27            "error_type" => error_type::REQUEST_FAILED,
28            "stage" => error_stage::SENDING,
29        )
30        .increment(1);
31        emit!(ComponentEventsDropped::<UNINTENTIONAL> {
32            count: self.count,
33            reason,
34        });
35    }
36}
37
38#[derive(NamedInternalEvent)]
39pub struct PulsarPropertyExtractionError<F: std::fmt::Display> {
40    pub property_field: F,
41}
42
43impl<F: std::fmt::Display> InternalEvent for PulsarPropertyExtractionError<F> {
44    fn emit(self) {
45        error!(
46            message = "Failed to extract properties. Value should be a map of String -> Bytes.",
47            error_code = "extracting_property",
48            error_type = error_type::PARSER_FAILED,
49            stage = error_stage::PROCESSING,
50            property_field = %self.property_field,
51        );
52        counter!(
53            CounterName::ComponentErrorsTotal,
54            "error_code" => "extracting_property",
55            "error_type" => error_type::PARSER_FAILED,
56            "stage" => error_stage::PROCESSING,
57        )
58        .increment(1);
59    }
60}
61
62#[cfg(feature = "sources-pulsar")]
63pub enum PulsarErrorEventType {
64    Read,
65    Ack,
66    NAck,
67}
68
69#[cfg(feature = "sources-pulsar")]
70pub struct PulsarErrorEventData {
71    pub msg: String,
72    pub error_type: PulsarErrorEventType,
73}
74
75#[cfg(feature = "sources-pulsar")]
76registered_event!(
77    PulsarErrorEvent => {
78        ack_errors: Counter = counter!(
79            CounterName::ComponentErrorsTotal,
80            "error_code" => "acknowledge_message",
81            "error_type" => error_type::ACKNOWLEDGMENT_FAILED,
82            "stage" => error_stage::RECEIVING,
83        ),
84
85        nack_errors: Counter = counter!(
86            CounterName::ComponentErrorsTotal,
87            "error_code" => "negative_acknowledge_message",
88            "error_type" => error_type::ACKNOWLEDGMENT_FAILED,
89            "stage" => error_stage::RECEIVING,
90        ),
91
92        read_errors: Counter = counter!(
93            CounterName::ComponentErrorsTotal,
94            "error_code" => "reading_message",
95            "error_type" => error_type::READER_FAILED,
96            "stage" => error_stage::RECEIVING,
97        ),
98    }
99
100    fn emit(&self,error:PulsarErrorEventData) {
101        match error.error_type{
102            PulsarErrorEventType::Read => {
103                error!(
104                    message = "Failed to read message.",
105                    error = error.msg,
106                    error_code = "reading_message",
107                    error_type = error_type::READER_FAILED,
108                    stage = error_stage::RECEIVING,
109                );
110
111                self.read_errors.increment(1_u64);
112            }
113            PulsarErrorEventType::Ack => {
114                error!(
115                    message = "Failed to acknowledge message.",
116                    error = error.msg,
117                    error_code = "acknowledge_message",
118                    error_type = error_type::ACKNOWLEDGMENT_FAILED,
119                    stage = error_stage::RECEIVING,
120                );
121
122                self.ack_errors.increment(1_u64);
123            }
124            PulsarErrorEventType::NAck => {
125                error!(
126                    message = "Failed to negatively acknowledge message.",
127                    error = error.msg,
128                    error_code = "negative_acknowledge_message",
129                    error_type = error_type::ACKNOWLEDGMENT_FAILED,
130                    stage = error_stage::RECEIVING,
131                );
132
133                self.nack_errors.increment(1_u64);
134            }
135        }
136    }
137);
vector/internal_events/pulsar.rs

vector/internal_events/
pulsar.rs
