forked from private-attribution/ipa
-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge remote-tracking branch 'origin/ipa-metrics-migrate' into perf-w…
…ork-oct2024
- Loading branch information
Showing
20 changed files
with
301 additions
and
279 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 |
|---|---|---|
| @@ -1,48 +1,56 @@ | ||
| use std::{io::stderr, thread}; | ||
| use std::{io, thread, thread::JoinHandle}; | ||
|
|
||
| use metrics_tracing_context::TracingContextLayer; | ||
| use metrics_util::{ | ||
| debugging::{DebuggingRecorder, Snapshotter}, | ||
| layers::Layer, | ||
| use ipa_metrics::{ | ||
| MetricChannelType, MetricsCollectorController, MetricsCurrentThreadContext, MetricsProducer, | ||
| }; | ||
| use tokio::runtime::Builder; | ||
|
|
||
| use crate::telemetry::stats::Metrics; | ||
|
|
||
| /// Collects metrics using `DebuggingRecorder` and dumps them to `stderr` when dropped. | ||
| /// Holds a reference to metrics controller and producer | ||
| pub struct CollectorHandle { | ||
| snapshotter: Snapshotter, | ||
| thread_handle: JoinHandle<()>, | ||
| /// This will be used once we start consuming metrics | ||
| _controller: MetricsCollectorController, | ||
| producer: MetricsProducer, | ||
| } | ||
|
|
||
| /// | ||
| /// Initializes this collector by installing `DebuggingRecorder` to keep track of metrics | ||
| /// emitted from different parts of the app. | ||
| /// | ||
| /// ## Panics | ||
| /// Panics if metric recorder has already been set | ||
| #[must_use] | ||
| pub fn install_collector() -> CollectorHandle { | ||
| let recorder = DebuggingRecorder::new(); | ||
| let snapshotter = recorder.snapshotter(); | ||
|
|
||
| // use span fields as dimensions for metric | ||
| let recorder = TracingContextLayer::all().layer(recorder); | ||
| metrics::set_boxed_recorder(Box::new(recorder)) | ||
| .expect("Metric recorder has been installed already"); | ||
|
|
||
| // register metrics | ||
| crate::telemetry::metrics::register(); | ||
| tracing::info!("Metrics enabled"); | ||
|
|
||
| CollectorHandle { snapshotter } | ||
| /// ## Errors | ||
| /// If it fails to start a new thread | ||
| pub fn install_collector() -> io::Result<CollectorHandle> { | ||
| let (producer, controller, handle) = | ||
| ipa_metrics::install_new_thread(MetricChannelType::Unbounded)?; | ||
| tracing::info!("Metrics engine is enabled"); | ||
|
|
||
| Ok(CollectorHandle { | ||
| thread_handle: handle, | ||
| _controller: controller, | ||
| producer, | ||
| }) | ||
| } | ||
|
|
||
| impl Drop for CollectorHandle { | ||
| fn drop(&mut self) { | ||
| if !thread::panicking() { | ||
| let stats = Metrics::from_snapshot(self.snapshotter.snapshot()); | ||
| stats | ||
| .print(&mut stderr()) | ||
| .expect("Failed to dump metrics to stderr"); | ||
| } | ||
| if !thread::panicking() && !self.thread_handle.is_finished() { | ||
| tracing::warn!("Metrics thread is still running"); | ||
| }; | ||
| } | ||
| } | ||
|
|
||
| impl CollectorHandle { | ||
| pub fn tokio_bind<'a>(&self, target: &'a mut Builder) -> &'a mut Builder { | ||
| let flush_fn = || MetricsCurrentThreadContext::flush(); | ||
|
|
||
| target | ||
| .on_thread_start({ | ||
| let producer = self.producer.clone(); | ||
| move || { | ||
| producer.install(); | ||
| } | ||
| }) | ||
| .on_thread_stop(flush_fn) | ||
| .on_thread_park(flush_fn) | ||
| } | ||
| } |
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
Oops, something went wrong.