vector_common/internal_event/
mod.rs

1mod bytes_received;
2mod bytes_sent;
3pub mod cached_event;
4pub mod component_events_dropped;
5mod events_received;
6mod events_sent;
7mod optional_tag;
8mod prelude;
9pub mod service;
10
11use std::ops::{Add, AddAssign};
12
13pub use bytes_received::BytesReceived;
14pub use bytes_sent::BytesSent;
15#[allow(clippy::module_name_repetitions)]
16pub use cached_event::{RegisterTaggedInternalEvent, RegisteredEventCache};
17pub use component_events_dropped::{ComponentEventsDropped, INTENTIONAL, UNINTENTIONAL};
18pub use events_received::{EventsReceived, EventsReceivedHandle};
19pub use events_sent::{DEFAULT_OUTPUT, EventsSent, TaggedEventsSent};
20pub use metrics::SharedString;
21pub use optional_tag::OptionalTag;
22pub use prelude::{error_stage, error_type};
23pub use service::{CallError, PollReadyError};
24
25use crate::json_size::JsonSize;
26
27pub trait InternalEvent: Sized {
28    fn emit(self);
29
30    // Optional for backwards compat until all events implement this
31    fn name(&self) -> Option<&'static str> {
32        None
33    }
34}
35
36#[allow(clippy::module_name_repetitions)]
37pub trait RegisterInternalEvent: Sized {
38    type Handle: InternalEventHandle;
39
40    fn register(self) -> Self::Handle;
41
42    fn name(&self) -> Option<&'static str> {
43        None
44    }
45}
46
47#[allow(clippy::module_name_repetitions)]
48pub trait InternalEventHandle: Sized {
49    type Data: Sized;
50    fn emit(&self, data: Self::Data);
51}
52
53// Sets the name of an event if it doesn't have one
54pub struct DefaultName<E> {
55    pub name: &'static str,
56    pub event: E,
57}
58
59impl<E: InternalEvent> InternalEvent for DefaultName<E> {
60    fn emit(self) {
61        self.event.emit();
62    }
63
64    fn name(&self) -> Option<&'static str> {
65        Some(self.event.name().unwrap_or(self.name))
66    }
67}
68
69impl<E: RegisterInternalEvent> RegisterInternalEvent for DefaultName<E> {
70    type Handle = E::Handle;
71
72    fn register(self) -> Self::Handle {
73        self.event.register()
74    }
75
76    fn name(&self) -> Option<&'static str> {
77        Some(self.event.name().unwrap_or(self.name))
78    }
79}
80
81#[cfg(any(test, feature = "test"))]
82pub fn emit(event: impl InternalEvent) {
83    if let Some(name) = event.name() {
84        super::event_test_util::record_internal_event(name);
85    }
86    event.emit();
87}
88
89#[cfg(not(any(test, feature = "test")))]
90pub fn emit(event: impl InternalEvent) {
91    event.emit();
92}
93
94#[cfg(any(test, feature = "test"))]
95pub fn register<E: RegisterInternalEvent>(event: E) -> E::Handle {
96    if let Some(name) = event.name() {
97        super::event_test_util::record_internal_event(name);
98    }
99    event.register()
100}
101
102#[cfg(not(any(test, feature = "test")))]
103pub fn register<E: RegisterInternalEvent>(event: E) -> E::Handle {
104    event.register()
105}
106
107pub type Registered<T> = <T as RegisterInternalEvent>::Handle;
108
109// Wrapper types used to hold data emitted by registered events
110
111#[derive(Clone, Copy)]
112pub struct ByteSize(pub usize);
113
114#[derive(Clone, Copy)]
115pub struct Count(pub usize);
116
117/// Holds the tuple `(count_of_events, estimated_json_size_of_events)`.
118#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
119pub struct CountByteSize(pub usize, pub JsonSize);
120
121impl AddAssign for CountByteSize {
122    fn add_assign(&mut self, rhs: Self) {
123        self.0 += rhs.0;
124        self.1 += rhs.1;
125    }
126}
127
128impl Add<CountByteSize> for CountByteSize {
129    type Output = CountByteSize;
130
131    fn add(self, rhs: CountByteSize) -> Self::Output {
132        CountByteSize(self.0 + rhs.0, self.1 + rhs.1)
133    }
134}
135
136// Wrapper types used to hold parameters for registering events
137
138pub struct Output(pub Option<SharedString>);
139
140pub struct Protocol(pub SharedString);
141
142impl Protocol {
143    pub const HTTP: Protocol = Protocol(SharedString::const_str("http"));
144    pub const HTTPS: Protocol = Protocol(SharedString::const_str("https"));
145    pub const NONE: Protocol = Protocol(SharedString::const_str("none"));
146    pub const TCP: Protocol = Protocol(SharedString::const_str("tcp"));
147    pub const UDP: Protocol = Protocol(SharedString::const_str("udp"));
148    pub const UNIX: Protocol = Protocol(SharedString::const_str("unix"));
149    pub const INTERNAL: Protocol = Protocol(SharedString::const_str("internal"));
150    pub const STATIC: Protocol = Protocol(SharedString::const_str("static"));
151}
152
153impl From<&'static str> for Protocol {
154    fn from(s: &'static str) -> Self {
155        Self(SharedString::const_str(s))
156    }
157}
158
159impl From<Protocol> for SharedString {
160    fn from(value: Protocol) -> Self {
161        value.0
162    }
163}
164
165/// Macro to take care of some of the repetitive boilerplate in implementing a registered event. See
166/// the other events in this module for examples of how to use this.
167///
168/// ## Usage
169///
170/// ```ignore
171/// registered_event!(
172///     Event {
173///         event_field: &'static str,
174///     } => {
175///         handle_field: Counter = counter!("name", "tag" => self.event_field),
176///     }
177///     fn emit(&self, data: DataType) {
178///         self.handle_field.increment(data.0);
179///     }
180/// );
181///
182/// let handle = register!(Event { event_field: "message" });
183///
184/// handle.emit(DataType(123));
185/// ```
186///
187/// In this example, the first set of fields describes the data required to register the event. This
188/// is what would be used by the `register!` macro. For example, `register!(Event { event_field:
189/// "something" })`. The second set of fields describes the data required to store the registered
190/// handle, namely the `Counter`s and `Gauge`s that record the handle from `metrics` as well as any
191/// associated data for emitting traces or debug messages, followed by an initialization assignment
192/// value. The `emit` function is the code required to update the metrics and generate any log
193/// messages.
194#[macro_export]
195macro_rules! registered_event {
196    // A registered event struct with no fields (zero-sized type).
197    ($event:ident => $($tail:tt)*) => {
198        #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
199        pub struct $event;
200
201        $crate::registered_event!(=> $event $($tail)*);
202    };
203
204    // A normal registered event struct.
205    ($event:ident { $( $field:ident: $type:ty, )* } => $($tail:tt)*) => {
206        #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
207        pub struct $event {
208            $( pub $field: $type, )*
209        }
210
211        $crate::registered_event!(=> $event $($tail)*);
212    };
213
214    // Sub-matcher to implement the common bits in the above two cases.
215    (
216        => $event:ident {
217            $( $field:ident: $type:ty = $value:expr, )*
218        }
219
220        fn emit(&$slf:ident, $data_name:ident: $data:ident)
221            $emit_body:block
222
223        $(fn register($fixed_name:ident: $fixed_tags:ty, $tags_name:ident: $tags:ty)
224            $register_body:block)?
225    ) => {
226        paste::paste!{
227            #[derive(Clone)]
228            pub struct [<$event Handle>] {
229                $( $field: $type, )*
230            }
231
232            impl $crate::internal_event::RegisterInternalEvent for $event {
233                type Handle = [<$event Handle>];
234
235                fn name(&self) -> Option<&'static str> {
236                    Some(stringify!($event))
237                }
238
239                fn register($slf) -> Self::Handle {
240                    Self::Handle {
241                        $( $field: $value, )*
242                    }
243                }
244            }
245
246            impl $crate::internal_event::InternalEventHandle for [<$event Handle>] {
247                type Data = $data;
248
249                fn emit(&$slf, $data_name: $data)
250                    $emit_body
251            }
252
253            $(impl $crate::internal_event::cached_event::RegisterTaggedInternalEvent for $event {
254                type Tags = $tags;
255                type Fixed = $fixed_tags;
256
257                fn register(
258                    $fixed_name: $fixed_tags,
259                    $tags_name: $tags,
260                ) -> <Self as $crate::internal_event::RegisterInternalEvent>::Handle {
261                    $register_body
262                }
263            })?
264
265        }
266    };
267}
vector_common/internal_event/mod.rs

vector_common/internal_event/
mod.rs
