feat: collect the OTLP data exported by the benchmarks

Cargo.toml

@@ -33,13 +33,16 @@ 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"] }
 base64 = "0.22.1"
-async-compression = { version = "0.4.42", features = ["tokio", "gzip"] }
+async-compression = { version = "0.4.42", features = ["tokio", "gzip", "zstd"] }
 schemars = "1.2.1"
 simplelog = { version = "0.12.2", default-features = false, features = ["termcolor"] }
 tempfile = { workspace = true }
@@ -96,6 +99,12 @@ shell-quote = "0.7.2"
 assert_cmd = "2.2"
 predicates = "3.1.4"
 strum = { version = "0.28.0", features = ["derive"] }
+opentelemetry-proto = { version = "0.33", default-features = false, features = [
+  "gen-tonic-messages",
+  "trace",
+  "logs",
+] }
+prost = "0.14"
 
 [workspace]
 members = [

crates/instrument-hooks-bindings/src/bindings.rs

@@ -4,6 +4,8 @@ pub const MARKER_TYPE_SAMPLE_START: u32 = 0;
 pub const MARKER_TYPE_SAMPLE_END: u32 = 1;
 pub const MARKER_TYPE_BENCHMARK_START: u32 = 2;
 pub const MARKER_TYPE_BENCHMARK_END: u32 = 3;
+pub const INSTRUMENT_HOOKS_MONOTONIC_OFFSET_ATTRIBUTE: &[u8; 35] =
+    b"codspeed.clock.monotonic_offset_ns\0";
 #[repr(C)]
 #[derive(Debug, Copy, Clone)]
 pub struct InstrumentHooks {
@@ -56,6 +58,12 @@ unsafe extern "C" {
 unsafe extern "C" {
     pub fn instrument_hooks_current_timestamp() -> u64;
 }
+unsafe extern "C" {
+    pub fn instrument_hooks_monotonic_offset() -> i64;
+}
+unsafe extern "C" {
+    pub fn instrument_hooks_callgrind_add_obj_skip(path: *const ::std::os::raw::c_char) -> u8;
+}
 pub const instrument_hooks_feature_t_FEATURE_DISABLE_CALLGRIND_MARKERS: instrument_hooks_feature_t =
     0;
 pub type instrument_hooks_feature_t = ::std::os::raw::c_uint;

crates/instrument-hooks-bindings/src/lib.rs

@@ -1,4 +1,4 @@
-#[cfg(use_instrument_hooks)]
+// Also compiled without instrument hooks for its constants; its functions are then never called.
 mod ffi;
 
 #[cfg(use_instrument_hooks)]
@@ -138,6 +138,13 @@ mod linux_impl {
             }
         }
 
+        /// Offset, in nanoseconds, to add to [`Self::current_timestamp`] to read it as a wall
+        /// clock time since the Unix epoch, or 0 where the monotonic clock is unavailable. Cached
+        /// for the process.
+        pub fn monotonic_offset() -> i64 {
+            unsafe { ffi::instrument_hooks_monotonic_offset() }
+        }
+
         pub fn disable_callgrind_markers() {
             unsafe {
                 ffi::instrument_hooks_set_feature(
@@ -194,10 +201,25 @@ mod other_impl {
             0
         }
 
+        pub fn monotonic_offset() -> i64 {
+            0
+        }
+
         pub fn disable_callgrind_markers() {}
     }
 }
 
+/// OpenTelemetry resource attribute holding [`InstrumentHooks::monotonic_offset`] when spans and
+/// logs are timestamped with the shifted monotonic clock.
+pub const MONOTONIC_OFFSET_ATTRIBUTE: &str =
+    match std::ffi::CStr::from_bytes_with_nul(ffi::INSTRUMENT_HOOKS_MONOTONIC_OFFSET_ATTRIBUTE) {
+        Ok(name) => match name.to_str() {
+            Ok(name) => name,
+            Err(_) => panic!("the monotonic offset attribute is not UTF-8"),
+        },
+        Err(_) => panic!("the monotonic offset attribute is not a C string"),
+    };
+
 #[cfg(use_instrument_hooks)]
 pub use linux_impl::InstrumentHooks;
 
@@ -219,4 +241,17 @@ mod tests {
         hooks.add_benchmark_timestamps(start, end);
         assert!(!hooks.is_instrumented() || hooks.stop_benchmark().is_ok());
     }
+
+    #[cfg(all(use_instrument_hooks, any(target_os = "linux", target_os = "macos")))]
+    #[test]
+    fn monotonic_offset_shifts_the_monotonic_clock_to_the_wall_clock() {
+        let offset = InstrumentHooks::monotonic_offset();
+        assert_eq!(offset, InstrumentHooks::monotonic_offset());
+        let shifted_ns = InstrumentHooks::current_timestamp() as i64 + offset;
+        let wall_ns = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap()
+            .as_nanos() as i64;
+        assert!((wall_ns - shifted_ns).abs() < 1_000_000_000);
+    }
 }

crates/instrument-hooks-bindings/src/update-bindings.sh

@@ -9,4 +9,5 @@ bindgen "$SCRIPT_DIR/../instrument-hooks/includes/core.h" \
   --rust-target 1.88 \
   --allowlist-function "instrument_hooks_.*" \
   --allowlist-var "MARKER_TYPE_.*" \
+  --allowlist-var "INSTRUMENT_HOOKS_.*" \
   --allowlist-type "instrument_hooks_feature_t"

scripts/run-otel-example.sh

@@ -0,0 +1,32 @@
+#!/usr/bin/env bash
+# Run the divan `otel_example` of the codspeed-rust checkout next to this
+# repository through `codspeed run --experimental-otel` in walltime mode, and
+# upload the results with the collected OTLP data.
+#
+# The upload goes to the CodSpeed instance of the selected runner profile: pick
+# one with `CODSPEED_PROFILE` (see `codspeed profile list`). Arguments are
+# passed to `codspeed run`, e.g. `--skip-upload` or `--profile-folder <dir>`.
+set -euo pipefail
+
+RUNNER_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+CODSPEED_RUST_DIR="$RUNNER_DIR/../codspeed-rust"
+
+if [ ! -d "$CODSPEED_RUST_DIR" ]; then
+  echo "ERROR: codspeed-rust must be checked out at $CODSPEED_RUST_DIR." >&2
+  exit 1
+fi
+
+cargo build --manifest-path "$RUNNER_DIR/Cargo.toml" --bin codspeed
+CODSPEED="$RUNNER_DIR/target/debug/codspeed"
+
+cd "$CODSPEED_RUST_DIR"
+CARGO_CODSPEED="cargo run -q -p cargo-codspeed -- codspeed"
+BENCH_ARGS="-p codspeed-divan-compat --bench otel_example"
+
+$CARGO_CODSPEED build -m walltime $BENCH_ARGS --features otel
+
+# The local provider: the GitHub Actions one needs a workflow event. Isolation
+# needs sudo.
+env -u GITHUB_ACTIONS CODSPEED_ISOLATION=false \
+  "$CODSPEED" run --mode walltime --experimental-otel "$@" -- \
+  "$CARGO_CODSPEED run $BENCH_ARGS"

src/cli/exec/mod.rs

@@ -94,6 +94,7 @@ fn build_orchestrator_config(
         poll_results_options,
         extra_env: HashMap::new(),
         fair_sched: args.shared.experimental.experimental_fair_sched,
+        collect_otel: args.shared.experimental.experimental_otel,
         cycle_estimation,
         exclude_allocations,
         simulation_track_subprocess: args.shared.simulation_track_subprocess,

src/cli/experimental.rs

@@ -26,6 +26,16 @@ pub struct ExperimentalArgs {
     )]
     pub experimental_disable_pythonmalloc_override: bool,
 
+    /// Receive the OpenTelemetry data exported by the benchmarks over OTLP/HTTP, and upload it
+    /// with the results. The receiver's URL is passed in `CODSPEED_OTEL_ENDPOINT`.
+    #[arg(
+        long,
+        default_value_t = false,
+        help_heading = "Experimental",
+        env = "CODSPEED_EXPERIMENTAL_OTEL"
+    )]
+    pub experimental_otel: bool,
+
     /// Deprecated alias for `--cycle-estimation`, still honored for now.
     #[arg(long, hide = true, env = "CODSPEED_EXPERIMENTAL_CYCLE_ESTIMATION")]
     pub experimental_cycle_estimation: bool,
@@ -69,6 +79,9 @@ impl ExperimentalArgs {
         if self.experimental_disable_pythonmalloc_override {
             flags.push("--experimental-disable-pythonmalloc-override");
         }
+        if self.experimental_otel {
+            flags.push("--experimental-otel");
+        }
         flags
     }
 

src/cli/run/mod.rs

@@ -83,6 +83,7 @@ impl RunArgs {
                 },
                 experimental: ExperimentalArgs {
                     experimental_fair_sched: false,
+                    experimental_otel: false,
                     experimental_memory_capture_stack: None,
                     experimental_cycle_estimation: false,
                     experimental_exclude_allocations: false,
@@ -141,6 +142,7 @@ fn build_orchestrator_config(
         poll_results_options,
         extra_env: HashMap::new(),
         fair_sched: args.shared.experimental.experimental_fair_sched,
+        collect_otel: args.shared.experimental.experimental_otel,
         cycle_estimation,
         exclude_allocations,
         simulation_track_subprocess: args.shared.simulation_track_subprocess,

src/executor/config.rs

@@ -105,6 +105,8 @@ pub struct OrchestratorConfig {
     /// Capture allocation call stacks in memory mode, so allocations can be
     /// attributed to the code that made them.
     pub memory_capture_stack: bool,
+    /// Receive the OTLP data exported by the benchmarks and store it in the profile folder.
+    pub collect_otel: bool,
 }
 
 /// Per-execution configuration passed to executors.
@@ -271,6 +273,7 @@ impl OrchestratorConfig {
             memory_track_physical: true,
             disable_pythonmalloc_override: false,
             memory_capture_stack: true,
+            collect_otel: false,
         }
     }
 }

src/executor/orchestrator.rs

@@ -7,6 +7,7 @@ use crate::cli::run::logger::Logger;
 use crate::executor::config::BenchmarkTarget;
 use crate::executor::config::OrchestratorConfig;
 use crate::executor::helpers::profile_folder::create_profile_folder;
+use crate::instruments::otel_receiver::{OTEL_ENDPOINT_ENV, OtelReceiver};
 use crate::prelude::*;
 use crate::run_environment::{self, RunEnvironment, RunEnvironmentProvider};
 use crate::runner_mode::RunnerMode;
@@ -141,13 +142,23 @@ impl Orchestrator {
         }
 
         for (run_part_index, part) in run_parts.into_iter().enumerate() {
-            let config = self
+            let mut config = self
                 .config
                 .executor_config_for_command(part.command, !part.uses_exec_harness);
             let mut executor = get_executor_from_mode(part.mode, self.config.walltime_profiler);
             let profile_folder =
                 self.resolve_profile_folder(&executor.name(), run_part_index, total_parts)?;
 
+            let otel_receiver = if self.config.collect_otel && !self.config.skip_run {
+                let receiver = OtelReceiver::start(&profile_folder).await?;
+                config
+                    .extra_env
+                    .insert(OTEL_ENDPOINT_ENV.into(), receiver.endpoint().into());
+                Some(receiver)
+            } else {
+                None
+            };
+
             let ctx = ExecutionContext::new(config, profile_folder);
 
             let rolling_buffer_label =
@@ -162,6 +173,14 @@ impl Orchestrator {
             )
             .await?;
 
+            if let Some(receiver) = otel_receiver {
+                let stats = receiver.stop().await?;
+                info!(
+                    "Collected {} OpenTelemetry export requests ({} bytes, {} bytes archived)",
+                    stats.requests, stats.received_bytes, stats.archive_bytes_count
+                );
+            }
+
             all_completed_runs.push((ctx, executor.name()));
         }
 

src/instruments/mod.rs

@@ -7,6 +7,7 @@ use crate::cli::run::RunArgs;
 use crate::prelude::*;
 
 pub mod mongo_tracer;
+pub mod otel_receiver;
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct MongoDBConfig {