From 8184d3ccc8f0b8cfd1a5b16e1edc194f8febd1cb Mon Sep 17 00:00:00 2001 From: Luc Varoqui Date: Thu, 15 Oct 2026 17:53:17 +0200 Subject: [PATCH 1/2] feat: collect the OTLP data exported by the benchmarks Add an experimental `--experimental-otel` flag (CODSPEED_EXPERIMENTAL_OTEL). When set, each run part starts an OTLP/HTTP receiver on a free localhost port and passes its URL to the benchmark in CODSPEED_OTEL_ENDPOINT. The receiver accepts protobuf and JSON exports of traces, logs and metrics, plain or gzipped. While the benchmark runs, it only writes each request body to disk as received, so it adds no compression work next to the measured process. Once the process exits, the requests are decompressed and packed into a single `otel.tar.zst` in the profile folder, uploaded with the other results. Compressing all requests as one stream with long-distance matching takes advantage of the content that repeats across export batches. Refs COD-3863 Co-Authored-By: Claude Opus 5.5 --- Cargo.lock | 5 + Cargo.toml | 7 +- src/cli/exec/mod.rs | 1 + src/cli/experimental.rs | 13 + src/cli/run/mod.rs | 2 + src/executor/config.rs | 3 + src/executor/orchestrator.rs | 21 +- src/instruments/mod.rs | 1 + src/instruments/otel_receiver.rs | 549 +++++++++++++++++++++++++++++++ 9 files changed, 599 insertions(+), 3 deletions(-) create mode 100644 src/instruments/otel_receiver.rs diff --git a/Cargo.lock b/Cargo.lock index 43c21f16f..7a7632fea 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -720,6 +720,9 @@ dependencies = [ "gimli", "git2", "gql_client", + "http-body-util", + "hyper", + "hyper-util", "indicatif", "insta", "instrument-hooks-bindings", @@ -807,6 +810,8 @@ dependencies = [ "compression-core", "flate2", "memchr", + "zstd", + "zstd-safe", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 4416ad621..4f5c61e3e 100644 --- a/Cargo.toml +++ b/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 } diff --git a/src/cli/exec/mod.rs b/src/cli/exec/mod.rs index 2302f4839..afe480081 100644 --- a/src/cli/exec/mod.rs +++ b/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, diff --git a/src/cli/experimental.rs b/src/cli/experimental.rs index ec3155682..f4b9bfe1b 100644 --- a/src/cli/experimental.rs +++ b/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 } diff --git a/src/cli/run/mod.rs b/src/cli/run/mod.rs index e6750c1eb..1d9be93a9 100644 --- a/src/cli/run/mod.rs +++ b/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, diff --git a/src/executor/config.rs b/src/executor/config.rs index f3ec640c8..59890a6e6 100644 --- a/src/executor/config.rs +++ b/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, } } } diff --git a/src/executor/orchestrator.rs b/src/executor/orchestrator.rs index ad2c9c5f7..bffbfeb92 100644 --- a/src/executor/orchestrator.rs +++ b/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())); } diff --git a/src/instruments/mod.rs b/src/instruments/mod.rs index c16a00de2..efe2ab986 100644 --- a/src/instruments/mod.rs +++ b/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 { diff --git a/src/instruments/otel_receiver.rs b/src/instruments/otel_receiver.rs new file mode 100644 index 000000000..959e4ab5d --- /dev/null +++ b/src/instruments/otel_receiver.rs @@ -0,0 +1,549 @@ +//! OTLP/HTTP receiver storing the export requests of the benchmark process. +//! +//! While the benchmark runs, each request body is written as received, one file per request, +//! to `/otel/-.[.gz]`. On stop, these files are +//! decompressed and packed into `/otel.tar.zst`, and the directory is removed. +//! Nothing is left behind when the benchmark exports nothing. + +use std::io; +use std::net::{Ipv4Addr, SocketAddr}; +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; + +use async_compression::Level; +use async_compression::tokio::bufread::GzipDecoder; +use async_compression::tokio::write::ZstdEncoder; +use async_compression::zstd::CParameter; +use http_body_util::{BodyExt, Full, Limited}; +use hyper::body::{Bytes, Incoming}; +use hyper::header::{CONTENT_ENCODING, CONTENT_TYPE, HeaderValue}; +use hyper::service::service_fn; +use hyper::{Method, Request, Response, StatusCode}; +use hyper_util::rt::TokioIo; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpListener; +use tokio::sync::oneshot; +use tokio::task::{JoinHandle, JoinSet}; + +use crate::prelude::*; + +/// Base URL of the receiver, as read by the CodSpeed integrations. +pub const OTEL_ENDPOINT_ENV: &str = "CODSPEED_OTEL_ENDPOINT"; + +/// Profile folder subdirectory holding the received requests until they are archived. +const SPOOL_DIR: &str = "otel"; + +/// Profile folder file holding the archived requests. +pub const OTEL_ARCHIVE: &str = "otel.tar.zst"; + +/// Largest window zstd decoders accept without raising their limit. +const ARCHIVE_WINDOW_LOG: u32 = 27; + +const MAX_REQUEST_BODY_BYTES: usize = 64 * 1024 * 1024; + +const PROTOBUF_CONTENT_TYPE: &str = "application/x-protobuf"; +const JSON_CONTENT_TYPE: &str = "application/json"; + +pub struct OtelReceiver { + endpoint: String, + shutdown_tx: oneshot::Sender<()>, + server: JoinHandle<()>, + store: Arc, + archive_path: PathBuf, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct OtelReceiverStats { + pub requests: u64, + /// Size of the request bodies as received. + pub received_bytes: u64, + pub archive_bytes_count: u64, +} + +impl OtelReceiver { + /// Starts listening on a free localhost port, storing the received requests in + /// `profile_folder`. + /// + /// The server stops when the receiver is stopped or dropped. Dropping it leaves the requests + /// unarchived. + pub async fn start(profile_folder: &Path) -> Result { + let listener = TcpListener::bind(SocketAddr::from((Ipv4Addr::LOCALHOST, 0))) + .await + .context("Failed to bind the OTLP receiver")?; + let endpoint = format!("http://{}", listener.local_addr()?); + + let store = Arc::new(RequestStore::new(profile_folder.join(SPOOL_DIR))); + let (shutdown_tx, shutdown_rx) = oneshot::channel(); + let server = tokio::spawn(serve(listener, store.clone(), shutdown_rx)); + + debug!("OTLP receiver listening on {endpoint}"); + Ok(Self { + endpoint, + shutdown_tx, + server, + store, + archive_path: profile_folder.join(OTEL_ARCHIVE), + }) + } + + pub fn endpoint(&self) -> &str { + &self.endpoint + } + + /// Stops the receiver and archives the received requests. A request is stored before it is + /// acknowledged, so everything an exited process has sent is archived. + pub async fn stop(self) -> Result { + drop(self.shutdown_tx); + if let Err(error) = self.server.await { + warn!("The OTLP receiver failed: {error}"); + } + let archive_bytes_count = self + .store + .archive(&self.archive_path) + .await + .context("Failed to archive the OTLP requests")?; + Ok(OtelReceiverStats { + requests: self.store.requests(), + received_bytes: self.store.received_bytes.load(Ordering::Relaxed), + archive_bytes_count, + }) + } +} + +/// Serves until `shutdown_rx` resolves, which also happens when its sender is dropped. +async fn serve( + listener: TcpListener, + store: Arc, + mut shutdown_rx: oneshot::Receiver<()>, +) { + let mut connections = JoinSet::new(); + loop { + let stream = tokio::select! { + _ = &mut shutdown_rx => break, + Some(_) = connections.join_next() => continue, + accepted = listener.accept() => match accepted { + Ok((stream, _)) => stream, + Err(error) => { + debug!("OTLP receiver failed to accept a connection: {error}"); + continue; + } + }, + }; + + let store = store.clone(); + connections.spawn(async move { + let service = service_fn(move |request| { + let store = store.clone(); + async move { Ok::<_, std::convert::Infallible>(handle(request, &store).await) } + }); + if let Err(error) = hyper::server::conn::http1::Builder::new() + .serve_connection(TokioIo::new(stream), service) + .await + { + debug!("OTLP receiver connection error: {error}"); + } + }); + } + + // Exporters keep idle connections open, so the remaining ones are cancelled rather than + // drained. + connections.shutdown().await; +} + +async fn handle(request: Request, store: &RequestStore) -> Response> { + let Some(signal) = Signal::from_path(request.uri().path()) else { + return status_response(StatusCode::NOT_FOUND); + }; + if request.method() != Method::POST { + return status_response(StatusCode::METHOD_NOT_ALLOWED); + } + + let content_type = request + .headers() + .get(CONTENT_TYPE) + .and_then(|value| value.to_str().ok()) + .and_then(Encoding::from_content_type); + let Some(encoding) = content_type else { + return status_response(StatusCode::UNSUPPORTED_MEDIA_TYPE); + }; + let is_gzipped = match request + .headers() + .get(CONTENT_ENCODING) + .map(|value| value.as_bytes()) + { + None | Some(b"identity") => false, + Some(b"gzip") => true, + Some(_) => return status_response(StatusCode::UNSUPPORTED_MEDIA_TYPE), + }; + + let body = match Limited::new(request.into_body(), MAX_REQUEST_BODY_BYTES) + .collect() + .await + { + Ok(body) => body.to_bytes(), + Err(error) => { + debug!("OTLP receiver failed to read a request body: {error}"); + return status_response(StatusCode::PAYLOAD_TOO_LARGE); + } + }; + + if let Err(error) = store.write(signal, encoding, is_gzipped, &body).await { + warn!( + "Failed to store an OTLP {} request: {error:#}", + signal.name() + ); + return status_response(StatusCode::INTERNAL_SERVER_ERROR); + } + + // An empty protobuf body decodes as an `Export*ServiceResponse` with no partial success. + let response_body = match encoding { + Encoding::Protobuf => Bytes::new(), + Encoding::Json => Bytes::from_static(b"{}"), + }; + let mut response = Response::new(Full::new(response_body)); + response.headers_mut().insert( + CONTENT_TYPE, + HeaderValue::from_static(encoding.content_type()), + ); + response +} + +fn status_response(status: StatusCode) -> Response> { + let mut response = Response::new(Full::new(Bytes::new())); + *response.status_mut() = status; + response +} + +#[derive(Debug, Clone, Copy)] +enum Signal { + Traces, + Logs, + Metrics, +} + +impl Signal { + fn from_path(path: &str) -> Option { + match path.trim_end_matches('/') { + "/v1/traces" => Some(Self::Traces), + "/v1/logs" => Some(Self::Logs), + "/v1/metrics" => Some(Self::Metrics), + _ => None, + } + } + + fn name(self) -> &'static str { + match self { + Self::Traces => "traces", + Self::Logs => "logs", + Self::Metrics => "metrics", + } + } +} + +#[derive(Debug, Clone, Copy)] +enum Encoding { + Protobuf, + Json, +} + +impl Encoding { + fn from_content_type(content_type: &str) -> Option { + let mime = content_type.split(';').next()?.trim(); + if mime.eq_ignore_ascii_case(PROTOBUF_CONTENT_TYPE) { + Some(Self::Protobuf) + } else if mime.eq_ignore_ascii_case(JSON_CONTENT_TYPE) { + Some(Self::Json) + } else { + None + } + } + + fn content_type(self) -> &'static str { + match self { + Self::Protobuf => PROTOBUF_CONTENT_TYPE, + Self::Json => JSON_CONTENT_TYPE, + } + } + + fn extension(self) -> &'static str { + match self { + Self::Protobuf => "pb", + Self::Json => "json", + } + } +} + +struct RequestStore { + dir: PathBuf, + next_sequence: AtomicU64, + received_bytes: AtomicU64, +} + +impl RequestStore { + fn new(dir: PathBuf) -> Self { + Self { + dir, + next_sequence: AtomicU64::new(0), + received_bytes: AtomicU64::new(0), + } + } + + async fn write( + &self, + signal: Signal, + encoding: Encoding, + is_gzipped: bool, + body: &[u8], + ) -> Result<()> { + tokio::fs::create_dir_all(&self.dir) + .await + .with_context(|| format!("Failed to create {}", self.dir.display()))?; + + let sequence = self.next_sequence.fetch_add(1, Ordering::Relaxed); + let gzip_suffix = if is_gzipped { ".gz" } else { "" }; + let path = self.dir.join(format!( + "{}-{sequence:06}.{}{gzip_suffix}", + signal.name(), + encoding.extension() + )); + tokio::fs::write(&path, body) + .await + .with_context(|| format!("Failed to write {}", path.display()))?; + + self.received_bytes + .fetch_add(body.len() as u64, Ordering::Relaxed); + Ok(()) + } + + /// A sequence number taken by a failed write is counted as well. + fn requests(&self) -> u64 { + self.next_sequence.load(Ordering::Relaxed) + } + + /// Packs the stored requests, decompressed and without their `.gz` suffix, into a + /// zstd-compressed tar at `archive_path`, then removes them. Returns the archive size, 0 when + /// no request was stored. + async fn archive(&self, archive_path: &Path) -> Result { + let mut entries = match tokio::fs::read_dir(&self.dir).await { + Ok(entries) => entries, + Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(0), + Err(error) => { + return Err(error) + .with_context(|| format!("Failed to read {}", self.dir.display())); + } + }; + let mut paths = Vec::new(); + while let Some(entry) = entries.next_entry().await? { + paths.push(entry.path()); + } + // Sorting by name puts the requests of a signal next to each other, where the + // compressor can match their repeated content. + paths.sort(); + + let file = tokio::fs::File::create(archive_path) + .await + .with_context(|| format!("Failed to create {}", archive_path.display()))?; + let encoder = ZstdEncoder::with_quality_and_params( + file, + Level::Default, + &[ + CParameter::enable_long_distance_matching(true), + CParameter::window_log(ARCHIVE_WINDOW_LOG), + ], + ); + let mut tar = tokio_tar::Builder::new(encoder); + for path in &paths { + let Some(file_name) = path.file_name().and_then(|name| name.to_str()) else { + continue; + }; + let content = tokio::fs::read(path) + .await + .with_context(|| format!("Failed to read {}", path.display()))?; + let (entry_name, content) = match file_name.strip_suffix(".gz") { + Some(entry_name) => { + let mut decoded = Vec::new(); + if let Err(error) = GzipDecoder::new(content.as_slice()) + .read_to_end(&mut decoded) + .await + { + warn!("Skipping the OTLP request {file_name}, not valid gzip: {error}"); + continue; + } + (entry_name, decoded) + } + None => (file_name, content), + }; + + let mut header = tokio_tar::Header::new_gnu(); + header.set_size(content.len() as u64); + header.set_mode(0o644); + tar.append_data(&mut header, entry_name, content.as_slice()) + .await?; + } + let mut encoder = tar.into_inner().await?; + encoder.shutdown().await?; + + tokio::fs::remove_dir_all(&self.dir) + .await + .with_context(|| format!("Failed to remove {}", self.dir.display()))?; + Ok(tokio::fs::metadata(archive_path).await?.len()) + } +} + +#[cfg(test)] +mod tests { + use async_compression::tokio::bufread::ZstdDecoder; + use async_compression::tokio::write::GzipEncoder; + use futures::StreamExt; + + use super::*; + + type Headers<'a> = &'a [(&'a str, &'a str)]; + + async fn post( + receiver: &OtelReceiver, + path: &str, + headers: Headers<'_>, + body: impl Into>, + ) -> reqwest::Response { + let mut request = reqwest::Client::new() + .post(format!("{}{path}", receiver.endpoint())) + .body(body.into()); + for (name, value) in headers { + request = request.header(*name, *value); + } + request.send().await.unwrap() + } + + async fn gzip(data: &[u8]) -> Vec { + let mut encoder = GzipEncoder::new(Vec::new()); + encoder.write_all(data).await.unwrap(); + encoder.shutdown().await.unwrap(); + encoder.into_inner() + } + + /// Returns the name and content of each archive entry, in order. + async fn archived_requests(archive_path: &Path) -> Vec<(String, Vec)> { + let compressed = std::fs::read(archive_path).unwrap(); + let mut archive = tokio_tar::Archive::new(ZstdDecoder::new(compressed.as_slice())); + let mut entries = archive.entries().unwrap(); + let mut requests = Vec::new(); + while let Some(entry) = entries.next().await { + let mut entry = entry.unwrap(); + let name = entry.path().unwrap().to_string_lossy().into_owned(); + let mut content = Vec::new(); + entry.read_to_end(&mut content).await.unwrap(); + requests.push((name, content)); + } + requests + } + + #[tokio::test] + async fn archives_the_decompressed_requests() { + let profile_folder = tempfile::tempdir().unwrap(); + let receiver = OtelReceiver::start(profile_folder.path()).await.unwrap(); + + let response = post( + &receiver, + "/v1/traces", + &[("content-type", PROTOBUF_CONTENT_TYPE)], + b"trace-bytes", + ) + .await; + assert_eq!(response.status(), StatusCode::OK); + assert_eq!( + response.headers()[CONTENT_TYPE.as_str()], + PROTOBUF_CONTENT_TYPE + ); + assert!(response.bytes().await.unwrap().is_empty()); + + let response = post( + &receiver, + "/v1/logs", + &[("content-type", "application/json; charset=utf-8")], + b"{\"resourceLogs\":[]}", + ) + .await; + assert_eq!(response.status(), StatusCode::OK); + assert_eq!(response.text().await.unwrap(), "{}"); + + let gzipped_metrics = gzip(b"metric-bytes").await; + let gzipped_metrics_len = gzipped_metrics.len() as u64; + let response = post( + &receiver, + "/v1/metrics", + &[ + ("content-type", PROTOBUF_CONTENT_TYPE), + ("content-encoding", "gzip"), + ], + gzipped_metrics, + ) + .await; + assert_eq!(response.status(), StatusCode::OK); + + let stats = receiver.stop().await.unwrap(); + let archive_path = profile_folder.path().join(OTEL_ARCHIVE); + assert_eq!( + stats, + OtelReceiverStats { + requests: 3, + received_bytes: 11 + 19 + gzipped_metrics_len, + archive_bytes_count: std::fs::metadata(&archive_path).unwrap().len(), + } + ); + assert_eq!( + archived_requests(&archive_path).await, + vec![ + ("logs-000001.json".into(), b"{\"resourceLogs\":[]}".to_vec()), + ("metrics-000002.pb".into(), b"metric-bytes".to_vec()), + ("traces-000000.pb".into(), b"trace-bytes".to_vec()), + ] + ); + assert!(!profile_folder.path().join(SPOOL_DIR).exists()); + } + + #[tokio::test] + async fn rejects_unsupported_requests_without_storing_them() { + let profile_folder = tempfile::tempdir().unwrap(); + let receiver = OtelReceiver::start(profile_folder.path()).await.unwrap(); + let protobuf = [("content-type", PROTOBUF_CONTENT_TYPE)]; + + let cases: [(&str, Headers, StatusCode); 4] = [ + ("/v1/profiles", &protobuf, StatusCode::NOT_FOUND), + ("/v1/traces", &[], StatusCode::UNSUPPORTED_MEDIA_TYPE), + ( + "/v1/traces", + &[("content-type", "text/plain")], + StatusCode::UNSUPPORTED_MEDIA_TYPE, + ), + ( + "/v1/traces", + &[ + ("content-type", PROTOBUF_CONTENT_TYPE), + ("content-encoding", "zstd"), + ], + StatusCode::UNSUPPORTED_MEDIA_TYPE, + ), + ]; + for (path, headers, expected_status) in cases { + let response = post(&receiver, path, headers, b"body").await; + assert_eq!(response.status(), expected_status, "{path} {headers:?}"); + } + + let response = reqwest::get(format!("{}/v1/traces", receiver.endpoint())) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::METHOD_NOT_ALLOWED); + + assert_eq!( + receiver.stop().await.unwrap(), + OtelReceiverStats { + requests: 0, + received_bytes: 0, + archive_bytes_count: 0, + } + ); + assert_eq!(std::fs::read_dir(profile_folder.path()).unwrap().count(), 0); + } +} From 33db967cdd2c724eb5d9057362809960d940ba5d Mon Sep 17 00:00:00 2001 From: Luc Varoqui Date: Thu, 15 Oct 2026 18:32:59 +0200 Subject: [PATCH 2/2] test: run the divan otel example end to end with OTLP collection Build the divan `otel_example` from a codspeed-rust checkout next to this repository, run it through `codspeed run --experimental-otel` in walltime mode, and check the profile folder holds an `otel.tar.zst` whose trace requests contain the benchmark's spans. The test is ignored by default since CI has no codspeed-rust checkout. Run it with `cargo test --test otel -- --ignored`. Refs COD-3863 Co-Authored-By: Claude Opus 5.5 --- Cargo.lock | 43 ++++ Cargo.toml | 6 + .../instrument-hooks | 2 +- .../instrument-hooks-bindings/src/bindings.rs | 8 + crates/instrument-hooks-bindings/src/lib.rs | 37 ++- .../src/update-bindings.sh | 1 + scripts/run-otel-example.sh | 32 +++ tests/otel.rs | 217 ++++++++++++++++++ 8 files changed, 344 insertions(+), 2 deletions(-) create mode 100755 scripts/run-otel-example.sh create mode 100644 tests/otel.rs diff --git a/Cargo.lock b/Cargo.lock index 7a7632fea..7fd546028 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -738,8 +738,10 @@ dependencies = [ "nix 0.31.3", "object", "open", + "opentelemetry-proto", "predicates", "procfs", + "prost", "rand 0.10.1", "rayon", "regex", @@ -2729,6 +2731,46 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7c87def4c32ab89d880effc9e097653c8da5d6ef28e6b539d313baaacfbafcbe" +[[package]] +name = "opentelemetry" +version = "0.33.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8c1f1a6ac9c875a4941e4c7c085b61e8f4daa326de03b4701768c3491a6ad9ee" +dependencies = [ + "futures-core", + "futures-sink", + "js-sys", + "pin-project-lite", + "thiserror 2.0.18", +] + +[[package]] +name = "opentelemetry-proto" +version = "0.33.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d576f5dddc3299341bcd85a04be17b2564844c0c9563214f02d85f2524d1afae" +dependencies = [ + "opentelemetry", + "opentelemetry_sdk", + "prost", +] + +[[package]] +name = "opentelemetry_sdk" +version = "0.33.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "372a912db2974d1990207006fd0e3aa0ef3fd4ad3be43c65cfaadf2206866061" +dependencies = [ + "futures-channel", + "futures-executor", + "futures-util", + "opentelemetry", + "percent-encoding", + "portable-atomic", + "rand 0.9.4", + "thiserror 2.0.18", +] + [[package]] name = "option-ext" version = "0.2.0" @@ -3134,6 +3176,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d2ea70524a2f82d518bce41317d0fae74151505651af45faf1ffbd6fd33f0568" dependencies = [ "bytes", + "prost-derive", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 4f5c61e3e..64d51c12d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -99,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 = [ diff --git a/crates/instrument-hooks-bindings/instrument-hooks b/crates/instrument-hooks-bindings/instrument-hooks index b9ddb5bc6..6bb632a38 160000 --- a/crates/instrument-hooks-bindings/instrument-hooks +++ b/crates/instrument-hooks-bindings/instrument-hooks @@ -1 +1 @@ -Subproject commit b9ddb5bc654b2e6fa13eb18efcd3a45e7ecda0bb +Subproject commit 6bb632a383634a4b30d46e7e1a3236df1862dda4 diff --git a/crates/instrument-hooks-bindings/src/bindings.rs b/crates/instrument-hooks-bindings/src/bindings.rs index 3adf72f9b..a8ea5b4e0 100644 --- a/crates/instrument-hooks-bindings/src/bindings.rs +++ b/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; diff --git a/crates/instrument-hooks-bindings/src/lib.rs b/crates/instrument-hooks-bindings/src/lib.rs index 8ea620cb8..7e9afbdfd 100644 --- a/crates/instrument-hooks-bindings/src/lib.rs +++ b/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); + } } diff --git a/crates/instrument-hooks-bindings/src/update-bindings.sh b/crates/instrument-hooks-bindings/src/update-bindings.sh index 469ceb03e..26ba6fbbe 100755 --- a/crates/instrument-hooks-bindings/src/update-bindings.sh +++ b/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" diff --git a/scripts/run-otel-example.sh b/scripts/run-otel-example.sh new file mode 100755 index 000000000..f795b70a0 --- /dev/null +++ b/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 `. +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" diff --git a/tests/otel.rs b/tests/otel.rs new file mode 100644 index 000000000..7011df315 --- /dev/null +++ b/tests/otel.rs @@ -0,0 +1,217 @@ +//! End-to-end OTLP collection of the spans and logs of a benchmark built from the codspeed-rust +//! checkout, timestamped on the clock of the benchmark markers. +//! +//! Needs codspeed-rust checked out next to this repository. Run with +//! `cargo test --test otel -- --ignored`. +//! +//! Set `CODSPEED_OTEL_TEST_PROFILE_FOLDER` to a path to keep the profile folder there for +//! inspection instead of in a temporary directory. Its previous content is removed. +#![cfg(target_os = "linux")] + +use std::path::{Path, PathBuf}; + +use assert_cmd::Command; +use async_compression::tokio::bufread::ZstdDecoder; +use futures::StreamExt; +use instrument_hooks_bindings::MONOTONIC_OFFSET_ATTRIBUTE; +use opentelemetry_proto::tonic::collector::logs::v1::ExportLogsServiceRequest; +use opentelemetry_proto::tonic::collector::trace::v1::ExportTraceServiceRequest; +use opentelemetry_proto::tonic::common::v1::any_value::Value; +use opentelemetry_proto::tonic::resource::v1::Resource; +use prost::Message; +use runner_shared::artifacts::{ArtifactExt, ExecutionTimestamps}; +use runner_shared::fifo::MarkerType; +use tokio::io::AsyncReadExt; + +fn codspeed_rust_dir() -> PathBuf { + let dir = Path::new(env!("CARGO_MANIFEST_DIR")).join("../codspeed-rust"); + assert!( + dir.is_dir(), + "codspeed-rust must be checked out at {}", + dir.display() + ); + dir +} + +/// A command running in the codspeed-rust checkout. The toolchain and package variables that +/// `cargo test` sets would otherwise make its cargo build with this repository's toolchain. +fn codspeed_rust_command(program: impl AsRef) -> Command { + let mut cmd = Command::new(program); + cmd.current_dir(codspeed_rust_dir()); + for (name, _) in std::env::vars_os() { + let name_str = name.to_string_lossy(); + let is_set_by_cargo = name_str == "RUSTUP_TOOLCHAIN" + || name_str == "CARGO" + || (name_str.starts_with("CARGO_") && name_str != "CARGO_HOME"); + if is_set_by_cargo { + cmd.env_remove(&name); + } + } + cmd +} + +/// The profile folder to run the benchmark with: `CODSPEED_OTEL_TEST_PROFILE_FOLDER` when set, +/// emptied first and kept after the test, a temporary directory otherwise. +fn profile_folder() -> (PathBuf, Option) { + match std::env::var_os("CODSPEED_OTEL_TEST_PROFILE_FOLDER") { + Some(dir) => { + let dir = PathBuf::from(dir); + if dir.exists() { + std::fs::remove_dir_all(&dir).unwrap(); + } + std::fs::create_dir_all(&dir).unwrap(); + (dir, None) + } + None => { + let temp_dir = tempfile::tempdir().unwrap(); + (temp_dir.path().to_path_buf(), Some(temp_dir)) + } + } +} + +/// Returns the name and content of each entry of the zstd-compressed tar at `path`. +async fn archive_entries(path: &Path) -> Vec<(String, Vec)> { + let compressed = std::fs::read(path).unwrap(); + let mut archive = tokio_tar::Archive::new(ZstdDecoder::new(compressed.as_slice())); + let mut entries = archive.entries().unwrap(); + let mut result = Vec::new(); + while let Some(entry) = entries.next().await { + let mut entry = entry.unwrap(); + let name = entry.path().unwrap().to_string_lossy().into_owned(); + let mut content = Vec::new(); + entry.read_to_end(&mut content).await.unwrap(); + result.push((name, content)); + } + result +} + +#[tokio::test] +#[ignore = "needs codspeed-rust checked out in ../codspeed-rust"] +async fn collects_the_spans_and_logs_of_the_divan_otel_example() { + let cargo_codspeed = "cargo run -q -p cargo-codspeed -- codspeed"; + let bench_args = "-p codspeed-divan-compat --bench otel_example"; + + codspeed_rust_command("bash") + .args([ + "-c", + &format!("{cargo_codspeed} build -m walltime {bench_args} --features otel"), + ]) + .assert() + .success(); + + let (profile_folder, _temp_dir) = profile_folder(); + codspeed_rust_command(env!("CARGO_BIN_EXE_codspeed")) + // Forces the local provider; the GitHub Actions one needs a workflow event. + .env_remove("GITHUB_ACTIONS") + // Isolation needs sudo. + .env("CODSPEED_ISOLATION", "false") + .args([ + "run", + "--skip-upload", + "--mode", + "walltime", + "--experimental-otel", + ]) + .arg("--profile-folder") + .arg(&profile_folder) + .args(["--", &format!("{cargo_codspeed} run {bench_args}")]) + .assert() + .success(); + + assert!(!profile_folder.join("otel").exists()); + let entries = archive_entries(&profile_folder.join("otel.tar.zst")).await; + let traces: Vec = decode_requests(&entries, "traces-"); + let logs: Vec = decode_requests(&entries, "logs-"); + assert!(!traces.is_empty(), "no trace request in {entries:?}"); + assert!(!logs.is_empty(), "no log request in {entries:?}"); + + // (timestamp, offset to the monotonic clock) of each span start and end and log record. + let mut timestamps = Vec::new(); + let mut span_names = Vec::new(); + for resource_spans in traces.iter().flat_map(|request| &request.resource_spans) { + let offset = monotonic_offset(resource_spans.resource.as_ref()); + for span in resource_spans + .scope_spans + .iter() + .flat_map(|scope| &scope.spans) + { + span_names.push(span.name.as_str()); + timestamps.push((span.start_time_unix_nano, offset)); + timestamps.push((span.end_time_unix_nano, offset)); + } + } + let mut log_bodies = Vec::new(); + for resource_logs in logs.iter().flat_map(|request| &request.resource_logs) { + let offset = monotonic_offset(resource_logs.resource.as_ref()); + for record in resource_logs + .scope_logs + .iter() + .flat_map(|scope| &scope.log_records) + { + if let Some(Value::StringValue(body)) = + record.body.as_ref().and_then(|body| body.value.as_ref()) + { + log_bodies.push(body.as_str()); + } + timestamps.push((record.time_unix_nano, offset)); + } + } + assert!( + span_names.contains(&"parse_and_sum"), + "no `parse_and_sum` span in {span_names:?}" + ); + assert!( + log_bodies.contains(&"skipped invalid fields"), + "no `skipped invalid fields` log in {log_bodies:?}" + ); + + let execution_timestamps = ExecutionTimestamps::decode_from_reader( + std::fs::File::open(profile_folder.join(ExecutionTimestamps::file_name(None))).unwrap(), + ) + .unwrap(); + let marker_timestamps = execution_timestamps + .markers + .iter() + .map(|marker| match marker { + MarkerType::SampleStart(timestamp) + | MarkerType::SampleEnd(timestamp) + | MarkerType::RoundStart(timestamp) + | MarkerType::RoundEnd(timestamp) => *timestamp, + }); + let markers = marker_timestamps.clone().min().unwrap()..=marker_timestamps.max().unwrap(); + for (timestamp, offset) in timestamps { + let monotonic = timestamp - offset; + assert!( + markers.contains(&monotonic), + "timestamp {timestamp} is {monotonic} on the monotonic clock, outside of the \ + benchmark markers {markers:?}" + ); + } +} + +/// Decodes the archived requests whose name starts with `prefix`. +fn decode_requests(entries: &[(String, Vec)], prefix: &str) -> Vec { + entries + .iter() + .filter(|(name, _)| name.starts_with(prefix) && name.ends_with(".pb")) + .map(|(_, content)| M::decode(content.as_slice()).unwrap()) + .collect() +} + +/// The offset to subtract from the timestamps of `resource` to get the monotonic clock of the +/// benchmark markers. +fn monotonic_offset(resource: Option<&Resource>) -> u64 { + let attribute = resource + .into_iter() + .flat_map(|resource| &resource.attributes) + .find(|attribute| attribute.key == MONOTONIC_OFFSET_ATTRIBUTE) + .expect("no monotonic clock offset in the resource"); + match attribute + .value + .as_ref() + .and_then(|value| value.value.as_ref()) + { + Some(Value::IntValue(offset)) => u64::try_from(*offset).unwrap(), + value => panic!("unexpected monotonic clock offset {value:?}"), + } +}