From b4fffe67e3cdd29e190accb67c85ebd6656e881c Mon Sep 17 00:00:00 2001 From: Mindaugas Vinkelis Date: Tue, 29 Oct 2024 13:05:12 -0700 Subject: [PATCH] Metrics collect stress test --- stress/Cargo.toml | 9 +- stress/src/metrics_collect.rs | 239 ++++++++++++++++++++++++++++++++++ 2 files changed, 247 insertions(+), 1 deletion(-) create mode 100644 stress/src/metrics_collect.rs diff --git a/stress/Cargo.toml b/stress/Cargo.toml index 0591cde7eb..d0fbe9124c 100644 --- a/stress/Cargo.toml +++ b/stress/Cargo.toml @@ -4,6 +4,11 @@ version = "0.1.0" edition = "2021" publish = false +[[bin]] +name = "metrics_collect" +path = "src/metrics_collect.rs" +doc = false + [[bin]] # Bin to run the metrics stress tests for Counter name = "metrics" path = "src/metrics_counter.rs" @@ -39,6 +44,7 @@ name = "random" path = "src/random.rs" doc = false + [dependencies] ctrlc = "3.2.5" lazy_static = "1.4.0" @@ -51,6 +57,7 @@ tracing = { workspace = true, features = ["std"]} tracing-subscriber = { workspace = true, features = ["registry", "std"] } num-format = "0.4.4" sysinfo = { version = "0.30.12", optional = true } +clap = { version = "4.5.20", features = ["derive"] } [features] -stats = ["sysinfo"] \ No newline at end of file +stats = ["sysinfo"] diff --git a/stress/src/metrics_collect.rs b/stress/src/metrics_collect.rs new file mode 100644 index 0000000000..444c19d00f --- /dev/null +++ b/stress/src/metrics_collect.rs @@ -0,0 +1,239 @@ +use std::{ + cell::RefCell, + ops::DerefMut, + sync::{ + atomic::{AtomicBool, AtomicUsize, Ordering}, + Arc, Barrier, Weak, + }, + time::{Duration, Instant}, +}; + +use lazy_static::lazy_static; +use opentelemetry::{ + metrics::{Histogram, MeterProvider, MetricResult}, + KeyValue, +}; +use opentelemetry_sdk::{ + metrics::{ + data::{ResourceMetrics, Temporality}, + reader::MetricReader, + InstrumentKind, ManualReader, Pipeline, SdkMeterProvider, + }, + Resource, +}; + +use rand::{ + rngs::{self, SmallRng}, + Rng, SeedableRng, +}; + +use clap::{Parser, ValueEnum}; + +#[derive(Debug, Clone, Copy, ValueEnum)] +enum CliTemporality { + Cumulative, + Delta, +} + +/// Simple program to greet a person +#[derive(Parser, Debug)] +#[command( + version, + about = "Measure metrics performance while collecting", + long_about = "The purpose of this test is to see how collecing interferre with measurements.\n\ + Most of the test measure how fast is collecting phase, but more important is\n\ + that it doesn't \"stop-the-world\" while collection phase is running." +)] +struct Cli { + /// Select collection phase temporality + temporality: CliTemporality, +} + +lazy_static! { + pub static ref ATTRIBUTE_VALUES: [&'static str; 10] = [ + "value1", "value2", "value3", "value4", "value5", "value6", "value7", "value8", "value9", + "value10" + ]; +} + +thread_local! { + + /// Store random number generator for each thread + pub static CURRENT_RNG: RefCell = RefCell::new(rngs::SmallRng::from_entropy()); +} + +fn main() { + let cli = Cli::parse(); + let temporality = match cli.temporality { + CliTemporality::Cumulative => Temporality::Cumulative, + CliTemporality::Delta => Temporality::Delta, + }; + let reader = SharedReader::new( + ManualReader::builder() + .with_temporality(temporality) + .build(), + ); + let provider = SdkMeterProvider::builder() + .with_reader(reader.clone()) + .build(); + // use histogram, as it is a bit more complicated during + let histogram = provider.meter("test").u64_histogram("hello").build(); + + calculate_measurements_during_collection(histogram, reader).print_results(); +} + +fn calculate_measurements_during_collection( + histogram: Histogram, + reader: SharedReader, +) -> MeasurementResults { + // we don't need to use every single CPU, better leave other CPU for operating system work, + // so our running threads could be much more stable in performance. + // just for the record, this is has HUGE effect on my machine (laptop intel i7-1355u) + let num_threads = num_cpus::get() / 2; + + let mut res = MeasurementResults { + total_measurements_count: 0, + total_time_collecting: 0, + num_iterations: 0, + }; + let start = Instant::now(); + while start.elapsed() < Duration::from_secs(3) { + res.num_iterations += 1; + let is_collecting = AtomicBool::new(false); + let measurements_while_collecting = AtomicUsize::new(0); + let time_while_collecting = AtomicUsize::new(0); + let barrier = Barrier::new(num_threads + 1); + std::thread::scope(|s| { + // first create bunch of measurements, + // so that collection phase wouldn't be "empty" + let mut handles = Vec::new(); + for _t in 0..num_threads { + handles.push(s.spawn(|| { + for _i in 0..1000 { + CURRENT_RNG.with(|rng| { + histogram.record( + 1, + &random_attribute_set3( + rng.borrow_mut().deref_mut(), + ATTRIBUTE_VALUES.as_ref(), + ), + ); + }); + } + })); + } + for handle in handles { + handle.join().unwrap(); + } + + // simultaneously start collecting and creating more measurements + for _ in 0..num_threads - 1 { + s.spawn(|| { + barrier.wait(); + let now = Instant::now(); + let mut count = 0; + while is_collecting.load(Ordering::Acquire) { + CURRENT_RNG.with(|rng| { + histogram.record( + 1, + &random_attribute_set3( + rng.borrow_mut().deref_mut(), + ATTRIBUTE_VALUES.as_ref(), + ), + ); + }); + count += 1; + } + measurements_while_collecting.fetch_add(count, Ordering::AcqRel); + time_while_collecting + .fetch_add(now.elapsed().as_micros() as usize, Ordering::AcqRel); + }); + } + + let collect_handle = s.spawn(|| { + let mut rm = ResourceMetrics { + resource: Resource::empty(), + scope_metrics: Vec::new(), + }; + is_collecting.store(true, Ordering::Release); + barrier.wait(); + reader.collect(&mut rm).unwrap(); + is_collecting.store(false, Ordering::Release); + }); + barrier.wait(); + collect_handle.join().unwrap(); + }); + res.total_measurements_count += measurements_while_collecting.load(Ordering::Acquire); + res.total_time_collecting += time_while_collecting.load(Ordering::Acquire); + } + res +} + +struct MeasurementResults { + total_measurements_count: usize, + total_time_collecting: usize, + num_iterations: usize, +} + +impl MeasurementResults { + fn print_results(&self) { + println!( + "{:>10.2} measurements/ms", + self.total_measurements_count as f32 / (self.total_time_collecting as f32 / 1000.0f32) + ); + println!( + "{:>10.2} measurements/it", + self.total_measurements_count as f32 / self.num_iterations as f32, + ); + println!( + "{:>10.2} μs/it", + self.total_time_collecting as f32 / self.num_iterations as f32, + ); + } +} + +fn random_attribute_set3(rng: &mut SmallRng, values: &[&'static str]) -> [KeyValue; 3] { + let len = values.len(); + unsafe { + [ + KeyValue::new("attribute1", *values.get_unchecked(rng.gen_range(0..len))), + KeyValue::new("attribute2", *values.get_unchecked(rng.gen_range(0..len))), + KeyValue::new("attribute3", *values.get_unchecked(rng.gen_range(0..len))), + ] + } +} + +// copy/paste from opentelemetry-sdk/benches/metric.rs +#[derive(Clone, Debug)] +pub struct SharedReader(Arc); + +impl SharedReader { + pub fn new(reader: R) -> Self + where + R: MetricReader, + { + Self(Arc::new(reader)) + } +} + +impl MetricReader for SharedReader { + fn register_pipeline(&self, pipeline: Weak) { + self.0.register_pipeline(pipeline) + } + + fn collect(&self, rm: &mut ResourceMetrics) -> MetricResult<()> { + self.0.collect(rm) + } + + fn force_flush(&self) -> MetricResult<()> { + self.0.force_flush() + } + + fn shutdown(&self) -> MetricResult<()> { + self.0.shutdown() + } + + fn temporality(&self, kind: InstrumentKind) -> Temporality { + self.0.temporality(kind) + } +}