diff --git a/Cargo.lock b/Cargo.lock index 43c21f16f..7fd546028 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", @@ -735,8 +738,10 @@ dependencies = [ "nix 0.31.3", "object", "open", + "opentelemetry-proto", "predicates", "procfs", + "prost", "rand 0.10.1", "rayon", "regex", @@ -807,6 +812,8 @@ dependencies = [ "compression-core", "flate2", "memchr", + "zstd", + "zstd-safe", ] [[package]] @@ -2724,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" @@ -3129,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 4416ad621..64d51c12d 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 } @@ -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 = [ 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/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); + } +} 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:?}"), + } +}