feat: collect the OTLP data exported by the benchmarks

Cargo.toml

@@ -33,8 +33,11 @@ reqwest-retry = "0.9"
 serde = { workspace = true }
 serde_json = { workspace = true, features = ["preserve_order"] }
 url = "2.5.8"
+hyper = { version = "1.9", features = ["server", "http1"] }
+hyper-util = { version = "0.1.20", features = ["tokio"] }
+http-body-util = "0.1.3"
 sha256 = "1.6"
-tokio = { version = "1", features = ["macros", "rt", "process"] }
+tokio = { version = "1", features = ["fs", "macros", "net", "rt", "sync", "process"] }
 tokio-tar = { package = "astral-tokio-tar", version = "0.6.2" }
 tokio-util = "0.7.18"
 crc-fast = { version = "1.9.0", default-features = false, features = ["std"] }
@@ -79,6 +82,13 @@ which = "8.0.2"
 crc32fast = "1.5.0"
 samply = { path = "crates/samply-codspeed/samply" }
 bytes = "1"
+opentelemetry-proto = { version = "0.33", default-features = false, features = [
+  "gen-tonic-messages",
+  "trace",
+  "logs",
+  "with-serde",
+] }
+prost = "0.14"
 
 # Memory profiling (memtrack) and the capability handling around it are Linux-only.
 [target.'cfg(target_os = "linux")'.dependencies]

crates/exec-harness/src/lib.rs

@@ -69,7 +69,7 @@ pub fn execute_benchmarks(
 
     match measurement_mode {
         MeasurementMode::Walltime => walltime::perform(commands),
-        MeasurementMode::Memory => analysis::perform(commands),
+        MeasurementMode::Memory | MeasurementMode::Telemetry => analysis::perform(commands),
         MeasurementMode::Simulation => analysis::perform_with_valgrind(commands),
     }
 }

crates/runner-shared/Cargo.toml

@@ -39,3 +39,7 @@ harness = false
 [[bench]]
 name = "memtrack_reader"
 harness = false
+
+[[bench]]
+name = "telemetry"
+harness = false

crates/runner-shared/benches/memtrack_reader.rs

@@ -36,7 +36,9 @@ fn sizes() -> &'static [usize] {
         .ok()
         .and_then(|mode| MeasurementMode::from_str(&mode, true).ok());
     match mode {
-        Some(MeasurementMode::Simulation | MeasurementMode::Memory) => ANALYSIS_MODE_SIZES,
+        Some(
+            MeasurementMode::Simulation | MeasurementMode::Memory | MeasurementMode::Telemetry,
+        ) => ANALYSIS_MODE_SIZES,
         Some(MeasurementMode::Walltime) | None => WALLTIME_MODE_SIZES,
     }
 }

crates/runner-shared/benches/telemetry.rs

@@ -0,0 +1,69 @@
+//! Writing and reading the telemetry files, on the spans and logs of a telemetry-heavy benchmark:
+//! about 10k spans and 10k logs, from nested `tracing` spans with attributes and an event in each.
+
+use std::io::Cursor;
+use std::path::Path;
+
+use divan::Bencher;
+use divan::counter::{BytesCount, ItemsCount};
+use runner_shared::telemetry::{LogRow, SpanRow, TelemetryReader, TelemetryRow, TelemetryWriter};
+
+#[global_allocator]
+static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
+
+fn main() {
+    divan::main();
+}
+
+const SPANS: &[u8] = include_bytes!("../testdata/telemetry_volume_spans.msgpack.zst");
+const LOGS: &[u8] = include_bytes!("../testdata/telemetry_volume_logs.msgpack.zst");
+
+fn read<T: TelemetryRow>(file: &[u8]) -> Vec<T> {
+    TelemetryReader::<T, _>::new(Cursor::new(file))
+        .unwrap()
+        .collect::<anyhow::Result<_>>()
+        .unwrap()
+}
+
+fn bench_read<T: TelemetryRow>(bencher: Bencher, file: &[u8]) {
+    let rows = read::<T>(file).len();
+    bencher
+        .counter(ItemsCount::new(rows))
+        .counter(BytesCount::new(file.len()))
+        .bench_local(|| read::<T>(file).len());
+}
+
+fn bench_write<T: TelemetryRow + Clone>(bencher: Bencher, file: &[u8]) {
+    let rows = read::<T>(file);
+    let dir = tempfile::tempdir_in(env!("CARGO_TARGET_TMPDIR")).unwrap();
+    let path = dir.path().join("telemetry.msgpack.zst");
+    let write = |rows: Vec<T>, path: &Path| {
+        let mut writer = TelemetryWriter::<T>::create(path).unwrap();
+        writer.write(rows).unwrap();
+        writer.close().unwrap();
+    };
+    bencher
+        .counter(ItemsCount::new(rows.len()))
+        .with_inputs(|| rows.clone())
+        .bench_local_values(|rows| write(rows, &path));
+}
+
+#[divan::bench]
+fn read_spans(bencher: Bencher) {
+    bench_read::<SpanRow>(bencher, SPANS);
+}
+
+#[divan::bench]
+fn read_logs(bencher: Bencher) {
+    bench_read::<LogRow>(bencher, LOGS);
+}
+
+#[divan::bench]
+fn write_spans(bencher: Bencher) {
+    bench_write::<SpanRow>(bencher, SPANS);
+}
+
+#[divan::bench]
+fn write_logs(bencher: Bencher) {
+    bench_write::<LogRow>(bencher, LOGS);
+}

crates/runner-shared/src/lib.rs

@@ -6,5 +6,6 @@ pub mod metadata;
 pub mod module_symbols;
 pub mod perf_event;
 pub mod runtime_env;
+pub mod telemetry;
 pub mod unwind_data;
 pub mod walltime_results;

crates/runner-shared/src/measurement_mode.rs

@@ -8,6 +8,7 @@ pub enum MeasurementMode {
     Memory,
     #[value(alias = "instrumentation")]
     Simulation,
+    Telemetry,
 }
 
 impl MeasurementMode {
@@ -19,6 +20,7 @@ impl MeasurementMode {
             Self::Simulation => "instrumentation",
             Self::Walltime => "walltime",
             Self::Memory => "memory",
+            Self::Telemetry => "telemetry",
         }
     }
 }