Add basic stress testing framework + counter.add test

Cargo.toml

@@ -36,6 +36,7 @@ members = [
     "examples/jaeger-remote-sampler",
     "examples/zipkin",
     "examples/multiple-span-processors",
-    "examples/zpages"
+    "examples/zpages",
+    "stress",
 ]
 exclude = ["examples/external-otlp-grpcio-async-std"]

stress/Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "stress"
+version = "0.1.0"
+edition = "2021"
+publish = false
+
+[[bin]] # Bin to run the metrics stress tests
+name = "metrics"
+path = "src/metrics.rs"
+doc = false
+
+[dependencies]
+ctrlc = "3.2.5"
+lazy_static = "1.4.0"
+num_cpus = "1.15.0"
+rayon = "1.7.0"
+opentelemetry_api = { path = "../opentelemetry-api", features = ["metrics"] }
+opentelemetry_sdk = { path = "../opentelemetry-sdk", features = ["metrics"] }

stress/README.md

@@ -0,0 +1,36 @@
+# OpenTelemetry Stress Tests
+
+## Why would you need stress test
+
+* It helps you to understand performance.
+* You can keep it running for days and nights to verify stability.
+* You can use it to generate lots of load to your backend system.
+* You can use it with other stress tools (e.g. a memory limiter) to verify how
+  your code reacts to certain resource constraints.
+
+## Usage
+
+Open a console, run the following command from the current folder:
+
+```sh
+cargo run --release --bin X
+```
+
+where `X` is the specific stress test you would like to run.
+
+e.g.
+
+```sh
+cargo run --release --bin metrics
+```
+
+```text
+Throughput: 4813019.00 requests/sec
+Throughput: 4805997.50 requests/sec
+Throughput: 4796934.50 requests/sec
+Throughput: 4817836.50 requests/sec
+Throughput: 4812457.00 requests/sec
+Throughput: 4773890.50 requests/sec
+Throughput: 4760047.00 requests/sec
+Throughput: 4851061.50 requests/sec
+```

stress/src/metrics.rs

@@ -0,0 +1,28 @@
+use lazy_static::lazy_static;
+use opentelemetry_api::{
+    metrics::{Counter, MeterProvider as _},
+    Context,
+};
+use opentelemetry_sdk::metrics::{ManualReader, MeterProvider};
+use std::borrow::Cow;
+
+mod throughput;
+
+lazy_static! {
+    static ref CONTEXT: Context = Context::new();
+    static ref PROVIDER: MeterProvider = MeterProvider::builder()
+        .with_reader(ManualReader::builder().build())
+        .build();
+    static ref COUNTER: Counter<u64> = PROVIDER
+        .meter(<&str as Into<Cow<'static, str>>>::into("test"))
+        .u64_counter("hello")
+        .init();
+}
+
+fn main() {
+    throughput::test_throughput(test_counter);
+}
+
+fn test_counter() {
+    COUNTER.add(&CONTEXT, 1, &[]);
+}

stress/src/throughput.rs

@@ -0,0 +1,51 @@
+use rayon::prelude::*;
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use std::sync::Arc;
+use std::time::Instant;
+
+const SLIDING_WINDOW_SIZE: u64 = 2; // In seconds
+
+static STOP: AtomicBool = AtomicBool::new(false);
+
+pub fn test_throughput<F>(func: F)
+where
+    F: Fn() + Sync + Send + 'static,
+{
+    ctrlc::set_handler(move || {
+        STOP.store(true, Ordering::SeqCst);
+    })
+    .expect("Error setting Ctrl-C handler");
+    let mut start_time = Instant::now();
+    let mut end_time = Instant::now();
+    let mut total_count_old: u64 = 0;
+    let total_count = Arc::new(AtomicU64::new(0));
+    let total_count_clone = Arc::clone(&total_count);
+
+    rayon::spawn( move || {
+        (0..num_cpus::get()).into_par_iter().for_each(|_| loop {
+            func();
+            total_count_clone.fetch_add(1, Ordering::SeqCst);
+            if STOP.load(Ordering::SeqCst) {
+                break;
+            }
+        });
+    });
+
+    loop {
+        let elapsed = end_time.duration_since(start_time).as_secs();
+        if elapsed >= SLIDING_WINDOW_SIZE {
+            let total_count_u64 = total_count.load(Ordering::Relaxed);
+            let current_count = total_count_u64 - total_count_old;
+            total_count_old = total_count_u64;
+            let throughput = current_count as f64 / elapsed as f64;
+            println!("Throughput: {:.2} requests/sec", throughput);
+            start_time = Instant::now();
+        }
+
+        if STOP.load(Ordering::SeqCst) {
+            break;
+        }
+
+        end_time = Instant::now();
+    }
+}