forked from vectordotdev/vector
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
chore(observability): emit
component_sent
events by source
and `s…
…ervice` (vectordotdev#17549) Closes vectordotdev#17580 Closes vectordotdev#17581 This is still in draft until I can get the following done. - [ ] ~~There are way more clones that I am happy with here, especially since this is in a hot path. These need reducing.~~ The remaining clones that I would like to remove are in the `get_tags` functions. This didn't seem trivial, and given the fairly positive regression numbers, I think it should be ok to defer for now. - [x] Function documentation. - [ ] Currently source schemas aren't being attached to the event at runtime, so the service meaning can't be retrieved. That won't work until this has been done. This will be a separate PR - vectordotdev#17692 - [x] I've only tested this with the kafka sink so far. I think it should work with all Stream sinks without needing any further modification - but further testing is needed. - [x] Tests. A bunch of tests need writing. - [x] The Vector source tests are failing I think because we now have `EventsSent` and `TaggedEventsSent` which both emit `component_sent_event` events and the test framework doesn't like this. This needs fixed. - [ ] We will need to review every sink to ensure they work with this. All the stream based sinks should, but the others are highly likely to need some work. --------- Signed-off-by: Stephen Wakely <fungus.humungus@gmail.com>
- Loading branch information
1 parent
6a6b42b
commit dcf7f9a
Showing
77 changed files
with
1,387 additions
and
501 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,69 @@ | ||
use std::{ | ||
collections::BTreeMap, | ||
sync::{Arc, RwLock}, | ||
}; | ||
|
||
use derivative::Derivative; | ||
|
||
use super::{InternalEventHandle, RegisterInternalEvent}; | ||
|
||
/// Metrics (eg. `component_sent_event_bytes_total`) may need to emit tags based on | ||
/// values contained within the events. These tags can't be determined in advance. | ||
/// | ||
/// Metrics need to be registered and the handle needs to be held onto in order to | ||
/// prevent them from expiring and being dropped (this would result in the counter | ||
/// resetting to zero). | ||
/// `CachedEvent` is used to maintain a store of these registered metrics. When a | ||
/// new event is emitted for a previously unseen set of tags an event is registered | ||
/// and stored in the cache. | ||
#[derive(Derivative)] | ||
#[derivative(Clone(bound = ""), Default(bound = ""))] | ||
pub struct RegisteredEventCache<Event: RegisterTaggedInternalEvent> { | ||
cache: Arc< | ||
RwLock< | ||
BTreeMap< | ||
<Event as RegisterTaggedInternalEvent>::Tags, | ||
<Event as RegisterInternalEvent>::Handle, | ||
>, | ||
>, | ||
>, | ||
} | ||
|
||
/// This trait must be implemented by events that emit dynamic tags. `register` must | ||
/// be implemented to register an event based on the set of tags passed. | ||
pub trait RegisterTaggedInternalEvent: RegisterInternalEvent { | ||
/// The type that will contain the data necessary to extract the tags | ||
/// that will be used when registering the event. | ||
type Tags; | ||
|
||
fn register(tags: Self::Tags) -> <Self as RegisterInternalEvent>::Handle; | ||
} | ||
|
||
impl<Event, EventHandle, Data, Tags> RegisteredEventCache<Event> | ||
where | ||
Data: Sized, | ||
EventHandle: InternalEventHandle<Data = Data>, | ||
Tags: Ord + Clone, | ||
Event: RegisterInternalEvent<Handle = EventHandle> + RegisterTaggedInternalEvent<Tags = Tags>, | ||
{ | ||
/// Emits the event with the given tags. | ||
/// It will register the event and store in the cache if this has not already | ||
/// been done. | ||
/// | ||
/// # Panics | ||
/// | ||
/// This will panic if the lock is poisoned. | ||
pub fn emit(&self, tags: &Tags, value: Data) { | ||
let read = self.cache.read().unwrap(); | ||
if let Some(event) = read.get(tags) { | ||
event.emit(value); | ||
} else { | ||
let event = <Event as RegisterTaggedInternalEvent>::register(tags.clone()); | ||
event.emit(value); | ||
|
||
// Ensure the read lock is dropped so we can write. | ||
drop(read); | ||
self.cache.write().unwrap().insert(tags.clone(), event); | ||
} | ||
} | ||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,14 @@ | ||
/// The user can configure whether a tag should be emitted. If they configure it to | ||
/// be emitted, but the value doesn't exist - we should emit the tag but with a value | ||
/// of `-`. | ||
#[derive(Clone, Debug, PartialEq, PartialOrd, Eq, Ord, Hash)] | ||
pub enum OptionalTag<T> { | ||
Ignored, | ||
Specified(Option<T>), | ||
} | ||
|
||
impl<T> From<Option<T>> for OptionalTag<T> { | ||
fn from(value: Option<T>) -> Self { | ||
Self::Specified(value) | ||
} | ||
} |
Oops, something went wrong.