From 7a6343c99928dadbdf0a0791ebdc0936902cb192 Mon Sep 17 00:00:00 2001 From: Luc Varoqui Date: Tue, 20 Oct 2026 16:07:54 +0200 Subject: [PATCH 1/3] feat(runner-shared): add a versioned file format for benchmark telemetry Add the files holding the OpenTelemetry spans and logs the benchmarks emit, shared by the runner that writes them and the parsers that read them: `otel/spans.msgpack.zst` and `otel/logs.msgpack.zst` in the profile folder. Each file is a zstd stream of positional MessagePack records, one per span or log, like the memtrack artifacts. Rows are flattened for the tables they are loaded into: ids as hex, attribute values as text, and the times of a span or log on the monotonic clock of the benchmark markers, when the benchmark recorded them, in their own fields. zstd runs with a 16 MB window and long-distance matching, since rows repeat their resource attributes far apart. Versions follow `UnwindData`: each record is wrapped in a versioned enum (`SpanRecord`, `LogRecord`), `SpanRow` and `LogRow` alias the current version, and older versions are upgraded through `From`. Version 1 fixtures must keep decoding. `TelemetryReader` decodes a file from any `Read`, such as an archive entry, as its bytes arrive. Unlike a truncated memtrack stream, a file cut in the middle of a row ends with an error. Refs COD-3863 Co-Authored-By: Claude Opus 5.5 --- crates/runner-shared/src/lib.rs | 1 + crates/runner-shared/src/telemetry/mod.rs | 249 ++++++++++++++++++ crates/runner-shared/src/telemetry/tests.rs | 199 ++++++++++++++ .../testdata/telemetry_logs_v1.msgpack.zst | Bin 0 -> 124 bytes .../testdata/telemetry_spans_v1.msgpack.zst | Bin 0 -> 244 bytes 5 files changed, 449 insertions(+) create mode 100644 crates/runner-shared/src/telemetry/mod.rs create mode 100644 crates/runner-shared/src/telemetry/tests.rs create mode 100644 crates/runner-shared/testdata/telemetry_logs_v1.msgpack.zst create mode 100644 crates/runner-shared/testdata/telemetry_spans_v1.msgpack.zst diff --git a/crates/runner-shared/src/lib.rs b/crates/runner-shared/src/lib.rs index fa108cb64..ea972d114 100644 --- a/crates/runner-shared/src/lib.rs +++ b/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; diff --git a/crates/runner-shared/src/telemetry/mod.rs b/crates/runner-shared/src/telemetry/mod.rs new file mode 100644 index 000000000..b315ac6c7 --- /dev/null +++ b/crates/runner-shared/src/telemetry/mod.rs @@ -0,0 +1,249 @@ +//! The OpenTelemetry spans and logs the benchmarks emitted, as files in the profile folder: +//! [`SPANS_FILE`] and [`LOGS_FILE`]. +//! +//! Each file is a zstd-compressed stream of MessagePack records, one per span or log. It is read +//! front to back, so it decodes as its bytes arrive: see [`TelemetryReader`]. +//! +//! [`SpanRow`] and [`LogRow`] alias the current versions. Each row is stored wrapped in its +//! versioned record ([`SpanRecord`], [`LogRecord`]), so the rows of older versions still decode +//! and are upgraded through `From` impls. Fields are positional: a version's struct never changes +//! once released, and the fixtures of each version must keep decoding. +//! +//! # Version history +//! +//! - **V1** — Original format. +//! +//! When adding a version: add a `SpanRowV{N}` or `LogRowV{N}` struct, a `From` impl, a +//! record variant, update the alias and `from_record`, add the version's fixtures, and append an +//! entry above. A reader has to be deployed before the runners writing its version are released. + +use std::fs::File; +use std::io::{BufRead, BufReader, BufWriter, Read, Write}; +use std::marker::PhantomData; +use std::path::Path; + +use anyhow::Context; +use serde::de::DeserializeOwned; +use serde::{Deserialize, Serialize}; + +/// Profile folder path of the spans file. +pub const SPANS_FILE: &str = "otel/spans.msgpack.zst"; +/// Profile folder path of the logs file. +pub const LOGS_FILE: &str = "otel/logs.msgpack.zst"; + +const ZSTD_LEVEL: i32 = 3; +/// Rows repeat most of their content, such as their resource attributes, but the repeats can be +/// far apart: a larger window lets zstd match them. Decoders need this much memory. +const ZSTD_WINDOW_LOG: u32 = 24; + +pub type SpanRow = SpanRowV1; +pub type SpanEventRow = SpanEventV1; +pub type LogRow = LogRowV1; + +/// A span, as stored with its version. +#[derive(Debug, Serialize, Deserialize)] +pub enum SpanRecord { + V1(SpanRowV1), +} + +/// A log, as stored with its version. +#[derive(Debug, Serialize, Deserialize)] +pub enum LogRecord { + V1(LogRowV1), +} + +/// A row type stored in its own telemetry file, as a versioned record. +pub trait TelemetryRow: Sized { + type Record: Serialize + DeserializeOwned + From; + + /// Upgrades a record of any version to the current row. + fn from_record(record: Self::Record) -> anyhow::Result; +} + +impl From for SpanRecord { + fn from(row: SpanRow) -> Self { + Self::V1(row) + } +} + +impl TelemetryRow for SpanRow { + type Record = SpanRecord; + + fn from_record(record: SpanRecord) -> anyhow::Result { + match record { + SpanRecord::V1(row) => Ok(row), + } + } +} + +impl From for LogRecord { + fn from(row: LogRow) -> Self { + Self::V1(row) + } +} + +impl TelemetryRow for LogRow { + type Record = LogRecord; + + fn from_record(record: LogRecord) -> anyhow::Result { + match record { + LogRecord::V1(row) => Ok(row), + } + } +} + +/// Attributes in their original order, which may repeat a key. +pub type Attributes = Vec<(String, String)>; + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct SpanRowV1 { + /// Span start, in nanoseconds since the Unix epoch. + pub timestamp: i64, + /// Span start on the monotonic clock of the benchmark markers, in nanoseconds, `None` when the + /// span didn't carry it. + pub monotonic_start_ns: Option, + /// Span end on the monotonic clock of the benchmark markers, in nanoseconds, `None` when the + /// span didn't carry it. + pub monotonic_end_ns: Option, + /// Lowercase hex. + pub trace_id: String, + pub span_id: String, + /// Empty for a root span. + pub parent_span_id: String, + pub span_name: String, + /// `Unspecified`, `Internal`, `Server`, `Client`, `Producer` or `Consumer`. + pub span_kind: String, + pub service_name: String, + pub scope_name: String, + pub scope_version: String, + pub duration_ns: u64, + /// `Unset`, `Ok` or `Error`. + pub status_code: String, + pub status_message: String, + pub resource_attributes: Attributes, + /// Without the monotonic start and end. + pub span_attributes: Attributes, + pub events: Vec, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct SpanEventV1 { + /// Nanoseconds since the Unix epoch. + pub timestamp: i64, + pub name: String, + pub attributes: Attributes, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct LogRowV1 { + /// The log time, or the time it was observed when it has none, in nanoseconds since the + /// Unix epoch. + pub timestamp: i64, + /// The log time on the monotonic clock of the benchmark markers, in nanoseconds, `None` when + /// the log didn't carry it. + pub monotonic_time_ns: Option, + /// Lowercase hex, empty when the log was emitted outside of a span. + pub trace_id: String, + pub span_id: String, + pub severity_text: String, + pub severity_number: u8, + pub service_name: String, + pub event_name: String, + pub body: String, + pub scope_name: String, + pub scope_version: String, + pub resource_attributes: Attributes, + /// Without the monotonic time. + pub log_attributes: Attributes, +} + +/// Writes rows to a telemetry file, as records of the current version. +pub struct TelemetryWriter { + serializer: rmp_serde::Serializer>>>, + _rows: PhantomData, +} + +impl TelemetryWriter { + pub fn create(path: &Path) -> anyhow::Result { + if let Some(dir) = path.parent() { + std::fs::create_dir_all(dir) + .with_context(|| format!("Failed to create {}", dir.display()))?; + } + let file = + File::create(path).with_context(|| format!("Failed to create {}", path.display()))?; + let mut encoder = zstd::Encoder::new(BufWriter::new(file), ZSTD_LEVEL)?; + encoder.long_distance_matching(true)?; + encoder.window_log(ZSTD_WINDOW_LOG)?; + Ok(Self { + serializer: rmp_serde::Serializer::new(BufWriter::new(encoder)), + _rows: PhantomData, + }) + } + + pub fn write(&mut self, rows: impl IntoIterator) -> anyhow::Result<()> { + for row in rows { + R::Record::from(row).serialize(&mut self.serializer)?; + } + Ok(()) + } + + pub fn close(self) -> anyhow::Result<()> { + let encoder = self + .serializer + .into_inner() + .into_inner() + .map_err(|error| error.into_error())?; + encoder.finish()?.flush()?; + Ok(()) + } +} + +/// Reads the rows of a telemetry file as its bytes are read from `R`, such as an archive entry, +/// upgraded to the current version. +pub struct TelemetryReader { + deserializer: rmp_serde::Deserializer< + rmp_serde::decode::ReadReader>>>, + >, + /// Set after an error, from which the stream can't recover. + failed: bool, + _rows: PhantomData T>, +} + +impl TelemetryReader { + pub fn new(reader: R) -> anyhow::Result { + let mut decoder = zstd::Decoder::new(reader)?; + decoder.window_log_max(ZSTD_WINDOW_LOG)?; + Ok(Self { + // The deserializer reads a few bytes at a time. + deserializer: rmp_serde::Deserializer::new(BufReader::new(decoder)), + failed: false, + _rows: PhantomData, + }) + } +} + +impl Iterator for TelemetryReader { + type Item = anyhow::Result; + + /// Ends at the end of the file, or after the first error, such as a file cut in the middle + /// of a row. + fn next(&mut self) -> Option { + if self.failed { + return None; + } + let row = match self.deserializer.get_mut().fill_buf() { + Ok([]) => return None, + Ok(_) => T::Record::deserialize(&mut self.deserializer) + .context("Failed to decode a row of the telemetry file") + .and_then(T::from_record), + Err(error) => { + Err(anyhow::Error::from(error).context("Failed to read the telemetry file")) + } + }; + self.failed = row.is_err(); + Some(row) + } +} + +#[cfg(test)] +mod tests; diff --git a/crates/runner-shared/src/telemetry/tests.rs b/crates/runner-shared/src/telemetry/tests.rs new file mode 100644 index 000000000..e0ea0cdc8 --- /dev/null +++ b/crates/runner-shared/src/telemetry/tests.rs @@ -0,0 +1,199 @@ +use super::*; + +fn strings(attributes: &[(&str, &str)]) -> Attributes { + attributes + .iter() + .map(|(key, value)| (key.to_string(), value.to_string())) + .collect() +} + +/// The rows of the version 1 fixtures. +fn v1_spans() -> Vec { + let resource_attributes = strings(&[ + ("service.name", "example"), + ("telemetry.sdk.language", "rust"), + ]); + vec![ + SpanRow { + timestamp: 1_700_000_000_000_001_010, + monotonic_start_ns: Some(1_010), + monotonic_end_ns: Some(1_110), + trace_id: "ab".repeat(16), + span_id: "01".repeat(8), + parent_span_id: String::new(), + span_name: "parse".into(), + span_kind: "Internal".into(), + service_name: "example".into(), + scope_name: "codspeed".into(), + scope_version: "1.0.0".into(), + duration_ns: 100, + status_code: "Error".into(), + status_message: "bad input".into(), + resource_attributes: resource_attributes.clone(), + span_attributes: strings(&[("input_len", "7"), ("input_len", "8")]), + events: vec![SpanEventV1 { + timestamp: 1_700_000_000_000_001_050, + name: "parsed".into(), + attributes: strings(&[("valid", "true")]), + }], + }, + SpanRow { + timestamp: 1_700_000_000_000_001_020, + monotonic_start_ns: None, + monotonic_end_ns: None, + trace_id: "ab".repeat(16), + span_id: "02".repeat(8), + parent_span_id: "01".repeat(8), + span_name: "tokenize".into(), + span_kind: "Unspecified".into(), + service_name: "example".into(), + scope_name: "codspeed".into(), + scope_version: "1.0.0".into(), + duration_ns: 10, + status_code: "Unset".into(), + status_message: String::new(), + resource_attributes, + span_attributes: Vec::new(), + events: Vec::new(), + }, + ] +} + +fn v1_logs() -> Vec { + vec![LogRow { + timestamp: 1_700_000_000_000_001_060, + monotonic_time_ns: Some(1_060), + trace_id: "ab".repeat(16), + span_id: "01".repeat(8), + severity_text: "INFO".into(), + severity_number: 9, + service_name: "example".into(), + event_name: "event benches/example.rs:9".into(), + body: "parsed values".into(), + scope_name: "example".into(), + scope_version: String::new(), + resource_attributes: strings(&[("service.name", "example")]), + log_attributes: strings(&[("count", "3")]), + }] +} + +const V1_SPANS_FIXTURE: &str = "testdata/telemetry_spans_v1.msgpack.zst"; +const V1_LOGS_FIXTURE: &str = "testdata/telemetry_logs_v1.msgpack.zst"; + +fn fixture(path: &str) -> std::path::PathBuf { + Path::new(env!("CARGO_MANIFEST_DIR")).join(path) +} + +fn write(path: &Path, batches: &[&[R]]) { + let mut writer = TelemetryWriter::::create(path).unwrap(); + for rows in batches { + writer.write(rows.to_vec()).unwrap(); + } + writer.close().unwrap(); +} + +fn read(file: &[u8]) -> anyhow::Result> { + TelemetryReader::::new(file)?.collect() +} + +fn read_path(path: &Path) -> Vec { + read(&std::fs::read(path).unwrap()).unwrap() +} + +fn encode(record: &impl Serialize) -> Vec { + rmp_serde::to_vec(record).unwrap() +} + +/// A zstd-compressed stream of encoded `records`. +fn file_with(records: &[Vec]) -> Vec { + zstd::encode_all(records.concat().as_slice(), ZSTD_LEVEL).unwrap() +} + +#[test] +fn reads_back_the_written_rows() { + let dir = tempfile::tempdir().unwrap(); + let spans_path = dir.path().join(SPANS_FILE); + let logs_path = dir.path().join(LOGS_FILE); + let spans = v1_spans(); + + write(&spans_path, &[&spans[..1], &[], &spans[1..]]); + write(&logs_path, &[&v1_logs()]); + + assert_eq!(read_path::(&spans_path), spans); + assert_eq!(read_path::(&logs_path), v1_logs()); +} + +#[test] +fn reads_a_file_without_rows() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join(SPANS_FILE); + + write::(&path, &[]); + + assert_eq!(read_path::(&path), []); +} + +#[test] +fn rejects_a_truncated_file() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join(SPANS_FILE); + write(&path, &[&v1_spans()]); + let file = std::fs::read(&path).unwrap(); + + for length in [0, file.len() / 2, file.len() - 1] { + assert!( + read::(&file[..length]).is_err(), + "{length} of {} bytes", + file.len() + ); + } +} + +/// Each version's fixture must keep decoding into the current rows. +#[test] +fn reads_the_version_1_fixtures() { + let read_fixture = |path| std::fs::read(fixture(path)).unwrap(); + + assert_eq!( + read::(&read_fixture(V1_SPANS_FIXTURE)).unwrap(), + v1_spans() + ); + assert_eq!( + read::(&read_fixture(V1_LOGS_FIXTURE)).unwrap(), + v1_logs() + ); +} + +/// Run with `cargo test -p runner-shared -- --ignored` when adding the fixtures of a new +/// version, after renaming the constants and rows above. +#[test] +#[ignore = "writes the fixtures of the current version"] +fn write_the_current_version_fixtures() { + write(&fixture(V1_SPANS_FIXTURE), &[&v1_spans()]); + write(&fixture(V1_LOGS_FIXTURE), &[&v1_logs()]); +} + +/// What a newer runner writes once a version is added. +#[derive(Serialize)] +enum NewerSpanRecord { + V1(Box), + V2(String), +} + +#[test] +fn rejects_a_row_of_a_newer_version() { + let file = file_with(&[ + encode(&NewerSpanRecord::V1(Box::new(v1_spans()[0].clone()))), + encode(&NewerSpanRecord::V2("parse".into())), + encode(&NewerSpanRecord::V1(Box::new(v1_spans()[1].clone()))), + ]); + + let mut reader = TelemetryReader::::new(file.as_slice()).unwrap(); + assert_eq!(reader.next().unwrap().unwrap(), v1_spans()[0]); + let error = reader.next().unwrap().unwrap_err(); + assert_eq!( + error.to_string(), + "Failed to decode a row of the telemetry file" + ); + assert!(reader.next().is_none()); +} diff --git a/crates/runner-shared/testdata/telemetry_logs_v1.msgpack.zst b/crates/runner-shared/testdata/telemetry_logs_v1.msgpack.zst new file mode 100644 index 0000000000000000000000000000000000000000..85c013ad43a54d2f99fd5500a9d61301f7b455dd GIT binary patch literal 124 zcmV-?0E7Q1wJ-euaGe7HBnE+^RxzE=7nhv=HYx-p%>*RbAYo##FfpV_PDW1&r)79y zZE$R5x@C4{Zge1GWo~0=Wpghsa&tO4t#Dy-b7f>8c42IFWpkjBlB{!Oa&~EBWiD=E eZDo;?rDJb(ZginD1ps!El`0a-nPnimiS7=)K{2-g literal 0 HcmV?d00001 diff --git a/crates/runner-shared/testdata/telemetry_spans_v1.msgpack.zst b/crates/runner-shared/testdata/telemetry_spans_v1.msgpack.zst new file mode 100644 index 0000000000000000000000000000000000000000..43c060577fc2af50f7c73bda66a718eb4fcf9dbc GIT binary patch literal 244 zcmVvb#ru*lBr*8Wp1H2 zIFXYC8m45ClBITGY-wbqbaHiN{J_96sB~{@Wo~JDWvf+gb8uy2X=Z6 Date: Tue, 20 Oct 2026 16:09:22 +0200 Subject: [PATCH 2/3] feat: add a telemetry runner mode on a generic analysis executor Add `codspeed run --mode telemetry`, an experimental mode collecting the OpenTelemetry spans and logs the benchmarks export. It is a separate pass, so nothing has to keep the export out of the measured runs, and the benchmarks run once each instead of once per walltime iteration. The mode runs on a new analysis executor: it runs the benchmarks once each without a profiler, answering the integration as an analysis mode, with analysis instruments hooked onto it. An instrument sets up before the process is spawned, gets the integration's start and stop profiler commands, and stops once the process exits, with the benchmark markers. The executor saves those markers as `ExecutionTimestamps.msgpack` in the profile folder, and fails when no benchmark ran, unless `--allow-empty` is set. Uploads carry `executor: "analysis"` and the enabled instruments, so later analysis instruments only add an instrument. The telemetry mode enables the telemetry instrument: - It starts an OTLP/HTTP receiver on a free localhost port, passed to the benchmark in CODSPEED_OTEL_ENDPOINT. The receiver accepts protobuf and JSON exports, plain or gzipped, and writes each trace and log request to disk as received. Metrics are acknowledged and dropped. - Once the process exits, the requests are decoded and written as the runner-shared telemetry files, one row per span or log, formatted like the OpenTelemetry Collector's ClickHouse exporter. A request that fails to decode is skipped with a warning. - The `codspeed.monotonic_start_ns` and `codspeed.monotonic_end_ns` attributes of a span and the `codspeed.monotonic_time_ns` attribute of a log, their times on the monotonic clock of the benchmark markers, move to their own fields of the rows: along with the markers, they tell the benchmark that emitted each span or log. A warning counts the spans and logs without a start or time. `CODSPEED_RUNNER_MODE` is `telemetry` in this mode, which `cargo codspeed run` reads to pick its telemetry build. exec-harness runs it like memory mode. Refs COD-3863 Co-Authored-By: Claude Opus 5.5 --- Cargo.lock | 98 ++++ Cargo.toml | 12 +- crates/exec-harness/src/lib.rs | 2 +- .../runner-shared/benches/memtrack_reader.rs | 4 +- crates/runner-shared/src/measurement_mode.rs | 3 + src/cli/exec/mod.rs | 5 +- src/executor/analysis/executor.rs | 167 ++++++ src/executor/analysis/instruments/mod.rs | 43 ++ .../analysis/instruments/telemetry.rs | 60 ++ src/executor/analysis/mod.rs | 2 + src/executor/interfaces.rs | 4 + src/executor/mod.rs | 12 + src/executor/orchestrator.rs | 14 +- src/instruments/mod.rs | 15 +- src/instruments/otel/files.rs | 236 ++++++++ src/instruments/otel/mod.rs | 12 + src/instruments/otel/receiver.rs | 548 ++++++++++++++++++ src/instruments/otel/rows.rs | 518 +++++++++++++++++ src/local_logger/icons.rs | 2 + ..._logger__icons__tests__icon_rendering.snap | 1 + src/runner_mode.rs | 4 + src/upload/benchmark_display.rs | 3 + src/upload/upload_metadata.rs | 28 +- src/upload/uploader.rs | 4 +- 24 files changed, 1786 insertions(+), 11 deletions(-) create mode 100644 src/executor/analysis/executor.rs create mode 100644 src/executor/analysis/instruments/mod.rs create mode 100644 src/executor/analysis/instruments/telemetry.rs create mode 100644 src/executor/analysis/mod.rs create mode 100644 src/instruments/otel/files.rs create mode 100644 src/instruments/otel/mod.rs create mode 100644 src/instruments/otel/receiver.rs create mode 100644 src/instruments/otel/rows.rs diff --git a/Cargo.lock b/Cargo.lock index 6d4610443..37d905311 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", @@ -833,6 +838,18 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "const-hex" +version = "1.19.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0e59eef12462b0f9b0a3620219be5d639afd79fe39dff0a42c3997061f9298b4" +dependencies = [ + "cfg-if", + "cpufeatures 0.2.17", + "proptest", + "serde_core", +] + [[package]] name = "core-foundation" version = "0.9.4" @@ -859,6 +876,12 @@ version = "0.8.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" +[[package]] +name = "core_detect" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7f8f80099a98041a3d1622845c271458a2d73e688351bf3cb999266764b81d48" + [[package]] name = "cpp_demangle" version = "0.5.1" @@ -2724,6 +2747,49 @@ 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 = [ + "base64", + "const-hex", + "opentelemetry", + "opentelemetry_sdk", + "prost", + "serde", +] + +[[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" @@ -3122,6 +3188,22 @@ dependencies = [ "hex", ] +[[package]] +name = "proptest" +version = "1.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b8530004ccb15eae51c7e40009fbe317f341f804db54dc033eec1c50be28cfa0" +dependencies = [ + "bitflags", + "chacha20", + "core_detect", + "num-traits", + "rand 0.10.1", + "rand_xorshift", + "regex-syntax", + "unarray", +] + [[package]] name = "prost" version = "0.14.3" @@ -3129,6 +3211,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d2ea70524a2f82d518bce41317d0fae74151505651af45faf1ffbd6fd33f0568" dependencies = [ "bytes", + "prost-derive", ] [[package]] @@ -3318,6 +3401,15 @@ version = "0.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69" +[[package]] +name = "rand_xorshift" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "60aa6af80be32871323012e02e6e65f8a7cc7890931ae421d217ad8fe0df2ccf" +dependencies = [ + "rand_core 0.10.1", +] + [[package]] name = "range-collections" version = "0.4.6" @@ -5019,6 +5111,12 @@ dependencies = [ "libc", ] +[[package]] +name = "unarray" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eaea85b334db583fe3274d12b4cd1880032beab409c0d774be044d4480ab9a94" + [[package]] name = "unicode-ident" version = "1.0.24" diff --git a/Cargo.toml b/Cargo.toml index c43fd4052..8868bf976 100644 --- a/Cargo.toml +++ b/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] diff --git a/crates/exec-harness/src/lib.rs b/crates/exec-harness/src/lib.rs index 48410a54b..22c2040e8 100644 --- a/crates/exec-harness/src/lib.rs +++ b/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), } } diff --git a/crates/runner-shared/benches/memtrack_reader.rs b/crates/runner-shared/benches/memtrack_reader.rs index 87a1d0e6e..4bef13344 100644 --- a/crates/runner-shared/benches/memtrack_reader.rs +++ b/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, } } diff --git a/crates/runner-shared/src/measurement_mode.rs b/crates/runner-shared/src/measurement_mode.rs index c720cea53..1a1e9b244 100644 --- a/crates/runner-shared/src/measurement_mode.rs +++ b/crates/runner-shared/src/measurement_mode.rs @@ -8,6 +8,8 @@ pub enum MeasurementMode { Memory, #[value(alias = "instrumentation")] Simulation, + /// Runs the benchmarks without measuring them, to collect the telemetry they emit + Telemetry, } impl MeasurementMode { @@ -19,6 +21,7 @@ impl MeasurementMode { Self::Simulation => "instrumentation", Self::Walltime => "walltime", Self::Memory => "memory", + Self::Telemetry => "telemetry", } } } diff --git a/src/cli/exec/mod.rs b/src/cli/exec/mod.rs index 2302f4839..1f7a32ae1 100644 --- a/src/cli/exec/mod.rs +++ b/src/cli/exec/mod.rs @@ -79,7 +79,10 @@ fn build_orchestrator_config( working_directory: args.shared.working_directory, targets: vec![target], modes, - instruments: Instruments { mongodb: None }, // exec doesn't support MongoDB + instruments: Instruments { + mongodb: None, // exec doesn't support MongoDB + telemetry: false, + }, perf_unwinding_mode: args.shared.profiler_run_args.perf.perf_unwinding_mode, enable_profiler: args.shared.profiler_run_args.resolve_enable_profiler(), walltime_profiler: args.shared.walltime_profiler, diff --git a/src/executor/analysis/executor.rs b/src/executor/analysis/executor.rs new file mode 100644 index 000000000..a84b1eafd --- /dev/null +++ b/src/executor/analysis/executor.rs @@ -0,0 +1,167 @@ +use super::instruments::enabled_instruments; +use crate::executor::helpers::command::CommandBuilder; +use crate::executor::helpers::env::{build_path_env, get_base_injected_env}; +use crate::executor::helpers::get_bench_command::get_bench_command; +use crate::executor::helpers::run_command_with_log_pipe::run_command_with_log_pipe_and_callback; +use crate::executor::helpers::run_with_env::wrap_with_env; +use crate::executor::shared::fifo::RunnerFifo; +use crate::executor::{ExecutionContext, Executor, ExecutorName, ExecutorSupport, ToolStatus}; +use crate::instruments::mongo_tracer::MongoTracer; +use crate::prelude::*; +use crate::runner_mode::RunnerMode; +use crate::system::SystemInfo; +use async_trait::async_trait; +use runner_shared::artifacts::{ArtifactExt, ExecutionTimestamps}; +use runner_shared::fifo::Command as FifoCommand; +use runner_shared::fifo::IntegrationMode; +use std::cell::OnceCell; +use std::collections::HashMap; +use std::fs::canonicalize; +use std::io::Write; +use tempfile::NamedTempFile; + +/// Runs the benchmarks once each in analysis mode, without a profiler, with the +/// [analysis instruments](super::instruments) enabled in the configuration hooked on. +/// +/// The benchmark markers are saved in the profile folder, so that what the instruments collect +/// can be placed on the clock of each benchmark. +pub struct AnalysisExecutor { + /// The runner mode the benchmarks are told they run in. + mode: RunnerMode, + /// Stashed by [`Executor::run`] for [`Executor::teardown`]. + timestamps: OnceCell, +} + +impl AnalysisExecutor { + pub fn new(mode: RunnerMode) -> Self { + Self { + mode, + timestamps: OnceCell::new(), + } + } + + fn bench_cmd( + &self, + execution_context: &ExecutionContext, + instruments_env: HashMap, + ) -> Result<(NamedTempFile, NamedTempFile, CommandBuilder)> { + let config = &execution_context.config; + let mut extra_env = + get_base_injected_env(self.mode.clone(), &execution_context.profile_folder, config); + extra_env.insert("PATH".into(), build_path_env(config.enable_introspection)?); + extra_env.extend(instruments_env); + + // Written to a script to keep the command's formatting without escaping it. + let mut script_file = NamedTempFile::new()?; + script_file.write_all(get_bench_command(config)?.as_bytes())?; + + let mut bench_cmd = CommandBuilder::new("bash"); + bench_cmd.arg(script_file.path()); + let (mut bench_cmd, env_file) = wrap_with_env(bench_cmd, &extra_env)?; + + if let Some(cwd) = &config.working_directory { + bench_cmd.current_dir(canonicalize(cwd)?); + } + + Ok((env_file, script_file, bench_cmd)) + } +} + +#[async_trait(?Send)] +impl Executor for AnalysisExecutor { + fn name(&self) -> ExecutorName { + ExecutorName::Analysis + } + + fn tool_status(&self) -> Option { + None + } + + fn support_level(&self, _system_info: &SystemInfo) -> ExecutorSupport { + ExecutorSupport::FullySupported + } + + async fn run( + &mut self, + execution_context: &ExecutionContext, + _mongo_tracer: &Option, + ) -> Result<()> { + let mut instruments = enabled_instruments(&execution_context.config.instruments); + let mut instruments_env = HashMap::new(); + for instrument in instruments.iter_mut() { + instrument + .start(&execution_context.profile_folder, &mut instruments_env) + .await?; + } + + let (_env_file, _script_file, cmd_builder) = + self.bench_cmd(execution_context, instruments_env)?; + let cmd = cmd_builder.build().into(); + debug!("cmd: {cmd:?}"); + + let mut runner_fifo = RunnerFifo::new()?; + let timestamps = &self.timestamps; + let fifo_instruments = &mut instruments; + let result = run_command_with_log_pipe_and_callback(cmd, async move |mut child| { + let on_cmd = async |cmd: &FifoCommand| { + match cmd { + // Run each benchmark once. + FifoCommand::GetIntegrationMode => { + return Ok(Some(FifoCommand::IntegrationModeResponse( + IntegrationMode::Analysis, + ))); + } + FifoCommand::StartProfiler => { + for instrument in fifo_instruments.iter_mut() { + instrument.on_start_profiler().await?; + } + } + FifoCommand::StopProfiler => { + for instrument in fifo_instruments.iter_mut() { + instrument.on_stop_profiler().await?; + } + } + _ => {} + } + Ok(None) + }; + let (marker_result, _fifo_data, exit_status) = + runner_fifo.handle_fifo_messages(&mut child, on_cmd).await?; + let _ = timestamps.set(marker_result); + Ok(exit_status) + }) + .await; + + // Stopped even when the process failed, to release what they hold. + for instrument in instruments.iter_mut() { + instrument.stop().await?; + } + + let status = result.map_err(|e| anyhow!("failed to execute the benchmark process. {e}"))?; + debug!("cmd exit status: {status:?}"); + + if !status.success() { + bail!("failed to execute the benchmark process: {status}"); + } + + Ok(()) + } + + async fn teardown(&self, execution_context: &ExecutionContext) -> Result<()> { + let Some(timestamps) = self.timestamps.get() else { + bail!("The benchmark process did not report its benchmarks"); + }; + if timestamps.uri_by_ts.is_empty() { + let message = format!("No benchmarks were run in {} mode", self.mode); + if !execution_context.config.allow_empty { + bail!( + "{message}. Were they built for it, e.g. with `cargo codspeed build -m {}`?", + self.mode + ); + } + info!("{message}."); + } + timestamps.save_to(&execution_context.profile_folder)?; + Ok(()) + } +} diff --git a/src/executor/analysis/instruments/mod.rs b/src/executor/analysis/instruments/mod.rs new file mode 100644 index 000000000..e51655fd6 --- /dev/null +++ b/src/executor/analysis/instruments/mod.rs @@ -0,0 +1,43 @@ +//! Instruments hooked onto the [analysis executor](super::executor::AnalysisExecutor). + +use crate::instruments::Instruments; +use crate::prelude::*; +use async_trait::async_trait; +use std::collections::HashMap; +use std::path::Path; + +pub mod telemetry; + +/// Collects data while the benchmarks run in analysis mode. +#[async_trait(?Send)] +pub trait AnalysisInstrument { + /// Called before the benchmark process is spawned, to set up the instrument and the + /// environment it needs in the benchmark process. + async fn start( + &mut self, + profile_folder: &Path, + env: &mut HashMap, + ) -> Result<()>; + + /// Called when the integration starts measuring a benchmark. + async fn on_start_profiler(&mut self) -> Result<()> { + Ok(()) + } + + /// Called when the integration stops measuring a benchmark. + async fn on_stop_profiler(&mut self) -> Result<()> { + Ok(()) + } + + /// Called once the benchmark process has exited, successfully or not. + async fn stop(&mut self) -> Result<()>; +} + +/// The analysis instruments enabled in `instruments`. +pub fn enabled_instruments(instruments: &Instruments) -> Vec> { + let mut enabled: Vec> = Vec::new(); + if instruments.telemetry { + enabled.push(Box::new(telemetry::TelemetryInstrument::default())); + } + enabled +} diff --git a/src/executor/analysis/instruments/telemetry.rs b/src/executor/analysis/instruments/telemetry.rs new file mode 100644 index 000000000..af39aa0a3 --- /dev/null +++ b/src/executor/analysis/instruments/telemetry.rs @@ -0,0 +1,60 @@ +use super::AnalysisInstrument; +use crate::instruments::otel::{OTEL_ENDPOINT_ENV, OtelReceiver, write_telemetry_files}; +use crate::prelude::*; +use async_trait::async_trait; +use std::collections::HashMap; +use std::path::{Path, PathBuf}; + +/// Receives the OpenTelemetry data the benchmarks export, through an OTLP/HTTP receiver whose +/// URL is passed in [`OTEL_ENDPOINT_ENV`], and writes its spans and logs as telemetry files in the +/// profile folder. +#[derive(Default)] +pub struct TelemetryInstrument { + started: Option<(OtelReceiver, PathBuf)>, +} + +#[async_trait(?Send)] +impl AnalysisInstrument for TelemetryInstrument { + async fn start( + &mut self, + profile_folder: &Path, + env: &mut HashMap, + ) -> Result<()> { + let receiver = OtelReceiver::start(profile_folder).await?; + env.insert(OTEL_ENDPOINT_ENV.into(), receiver.endpoint().into()); + self.started = Some((receiver, profile_folder.to_path_buf())); + Ok(()) + } + + async fn stop(&mut self) -> Result<()> { + let Some((receiver, profile_folder)) = self.started.take() else { + return Ok(()); + }; + let requests = receiver.stop().await; + let (request_count, received_bytes) = (requests.count, requests.received_bytes); + let stats = write_telemetry_files(requests, &profile_folder) + .await + .context("Failed to write the OpenTelemetry data")?; + + if request_count == 0 { + warn!( + "The benchmarks exported no OpenTelemetry data. Make sure their integration \ + exports it, e.g. with the `telemetry` feature of `codspeed-divan-compat`." + ); + } else { + info!( + "Collected {} spans and {} logs from {request_count} OpenTelemetry export \ + requests ({received_bytes} bytes)", + stats.spans, stats.logs + ); + } + if stats.without_monotonic_time > 0 { + warn!( + "{} spans and logs carry no time on the clock of the benchmarks, they won't be \ + attributed to their benchmark. Update their integration.", + stats.without_monotonic_time + ); + } + Ok(()) + } +} diff --git a/src/executor/analysis/mod.rs b/src/executor/analysis/mod.rs new file mode 100644 index 000000000..4ae945324 --- /dev/null +++ b/src/executor/analysis/mod.rs @@ -0,0 +1,2 @@ +pub mod executor; +pub mod instruments; diff --git a/src/executor/interfaces.rs b/src/executor/interfaces.rs index 80d3dd6eb..542fe8d79 100644 --- a/src/executor/interfaces.rs +++ b/src/executor/interfaces.rs @@ -8,6 +8,7 @@ pub enum ExecutorName { Valgrind, WallTime, Memory, + Analysis, } impl fmt::Display for ExecutorName { @@ -16,6 +17,7 @@ impl fmt::Display for ExecutorName { ExecutorName::Valgrind => write!(f, "valgrind"), ExecutorName::WallTime => write!(f, "walltime"), ExecutorName::Memory => write!(f, "memory"), + ExecutorName::Analysis => write!(f, "analysis"), } } } @@ -27,6 +29,7 @@ impl ExecutorName { ExecutorName::Valgrind => "CPU Simulation", ExecutorName::WallTime => "Walltime", ExecutorName::Memory => "Memory", + ExecutorName::Analysis => "Analysis", } } @@ -36,6 +39,7 @@ impl ExecutorName { ExecutorName::Valgrind => Icon::ExecutorValgrind, ExecutorName::WallTime => Icon::ExecutorWallTime, ExecutorName::Memory => Icon::ExecutorMemory, + ExecutorName::Analysis => Icon::ExecutorAnalysis, } } } diff --git a/src/executor/mod.rs b/src/executor/mod.rs index 74885c429..7776b5ddb 100644 --- a/src/executor/mod.rs +++ b/src/executor/mod.rs @@ -1,5 +1,6 @@ use std::fmt::Display; +mod analysis; pub mod config; mod execution_context; pub(crate) mod helpers; @@ -22,6 +23,8 @@ pub use execution_context::ExecutionContext; pub use interfaces::ExecutorName; pub use orchestrator::Orchestrator; +use crate::instruments::Instruments; +use analysis::executor::AnalysisExecutor; #[cfg(target_os = "linux")] use memory::executor::MemoryExecutor; use std::path::Path; @@ -37,6 +40,7 @@ impl Display for RunnerMode { RunnerMode::Walltime => write!(f, "walltime"), #[cfg(target_os = "linux")] RunnerMode::Memory => write!(f, "memory"), + RunnerMode::Telemetry => write!(f, "telemetry"), } } } @@ -53,6 +57,14 @@ pub fn get_executor_from_mode( RunnerMode::Walltime => Box::new(WallTimeExecutor::new(walltime_profiler)), #[cfg(target_os = "linux")] RunnerMode::Memory => Box::new(MemoryExecutor), + RunnerMode::Telemetry => Box::new(AnalysisExecutor::new(RunnerMode::Telemetry)), + } +} + +/// Enables the instruments `mode` runs on top of its executor. +pub fn apply_mode_instruments(mode: &RunnerMode, instruments: &mut Instruments) { + if *mode == RunnerMode::Telemetry { + instruments.telemetry = true; } } diff --git a/src/executor/orchestrator.rs b/src/executor/orchestrator.rs index ad2c9c5f7..33af1d436 100644 --- a/src/executor/orchestrator.rs +++ b/src/executor/orchestrator.rs @@ -1,4 +1,6 @@ -use super::{ExecutionContext, ExecutorName, get_executor_from_mode, run_executor}; +use super::{ + ExecutionContext, ExecutorName, apply_mode_instruments, get_executor_from_mode, run_executor, +}; use crate::api_client::CodSpeedAPIClient; use crate::binary_installer::ensure_binary_installed; use crate::binary_pins::{self, PinnedBinary}; @@ -48,6 +50,13 @@ impl Orchestrator { ); } + if config.modes.contains(&RunnerMode::Telemetry) { + warn!( + "The 'telemetry' runner mode is experimental and may change or be removed without \ + notice." + ); + } + Ok(Orchestrator { config, system_info, @@ -141,9 +150,10 @@ 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); + apply_mode_instruments(part.mode, &mut config.instruments); 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)?; diff --git a/src/instruments/mod.rs b/src/instruments/mod.rs index c16a00de2..a18f9f201 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; #[derive(Debug, Clone, PartialEq, Eq)] pub struct MongoDBConfig { @@ -16,11 +17,15 @@ pub struct MongoDBConfig { #[derive(Debug, Clone, PartialEq, Eq)] pub struct Instruments { pub mongodb: Option, + /// Receive the OpenTelemetry data the benchmarks export. Enabled by the telemetry runner + /// mode, not by `--instruments`. + pub telemetry: bool, } #[derive(Deserialize, Serialize, Debug, Clone, PartialEq, Eq, Hash)] pub enum InstrumentName { MongoDB, + Telemetry, } impl Instruments { @@ -35,6 +40,10 @@ impl Instruments { names.push(InstrumentName::MongoDB); } + if self.telemetry { + names.push(InstrumentName::Telemetry); + } + names } } @@ -64,7 +73,10 @@ impl TryFrom<&RunArgs> for Instruments { None }; - Ok(Self { mongodb }) + Ok(Self { + mongodb, + telemetry: false, + }) } } @@ -76,6 +88,7 @@ impl Instruments { mongodb: Some(MongoDBConfig { uri_env_name: Some("MONGODB_URI".into()), }), + telemetry: false, } } } diff --git a/src/instruments/otel/files.rs b/src/instruments/otel/files.rs new file mode 100644 index 000000000..2e4d1ab05 --- /dev/null +++ b/src/instruments/otel/files.rs @@ -0,0 +1,236 @@ +//! The received spans and logs, written as the telemetry files of the profile folder. + +use std::path::Path; + +use opentelemetry_proto::tonic::collector::logs::v1::ExportLogsServiceRequest; +use opentelemetry_proto::tonic::collector::trace::v1::ExportTraceServiceRequest; +use prost::Message; +use runner_shared::telemetry::{LOGS_FILE, SPANS_FILE, TelemetryRow, TelemetryWriter}; +use serde::de::DeserializeOwned; + +use super::ReceivedRequests; +use super::receiver::{Encoding, Signal, StoredRequest}; +use super::rows::{log_rows, span_rows}; +use crate::prelude::*; + +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)] +pub struct TelemetryStats { + pub spans: u64, + pub logs: u64, + /// Spans and logs without their time on the monotonic clock of the benchmark markers. + pub without_monotonic_time: u64, + /// Requests that failed to decode. + pub skipped_requests: u64, +} + +/// Decodes `requests` into the [`SPANS_FILE`] and [`LOGS_FILE`] of `profile_folder`, then removes +/// them. A file is only written when it has rows. +pub async fn write_telemetry_files( + requests: ReceivedRequests, + profile_folder: &Path, +) -> Result { + let profile_folder = profile_folder.to_path_buf(); + tokio::task::spawn_blocking(move || { + let stats = write_files_blocking(&requests, &profile_folder)?; + requests.remove()?; + Ok(stats) + }) + .await? +} + +fn write_files_blocking( + requests: &ReceivedRequests, + profile_folder: &Path, +) -> Result { + let mut spans = LazyWriter::new(profile_folder.join(SPANS_FILE)); + let mut logs = LazyWriter::new(profile_folder.join(LOGS_FILE)); + let mut stats = TelemetryStats::default(); + + for request in requests.list()? { + let written = match request.signal { + Signal::Traces => decode::(&request).map(|request| { + let rows = span_rows(request); + stats.spans += rows.len() as u64; + stats.without_monotonic_time += rows + .iter() + .filter(|row| row.monotonic_start_ns.is_none()) + .count() as u64; + spans.write(rows) + }), + Signal::Logs => decode::(&request).map(|request| { + let rows = log_rows(request); + stats.logs += rows.len() as u64; + stats.without_monotonic_time += rows + .iter() + .filter(|row| row.monotonic_time_ns.is_none()) + .count() as u64; + logs.write(rows) + }), + }; + match written { + Ok(written) => written?, + Err(error) => { + warn!( + "Skipping the OTLP request {}: {error:#}", + request.path.display() + ); + stats.skipped_requests += 1; + } + } + } + + spans.close()?; + logs.close()?; + Ok(stats) +} + +fn decode(request: &StoredRequest) -> Result { + let body = std::fs::read(&request.path)?; + Ok(match request.encoding { + Encoding::Protobuf => M::decode(body.as_slice())?, + Encoding::Json => serde_json::from_slice(&body)?, + }) +} + +/// Creates its file on the first rows. +struct LazyWriter { + path: std::path::PathBuf, + writer: Option>, +} + +impl LazyWriter { + fn new(path: std::path::PathBuf) -> Self { + Self { path, writer: None } + } + + fn write(&mut self, rows: Vec) -> Result<()> { + if rows.is_empty() { + return Ok(()); + } + let writer = match &mut self.writer { + Some(writer) => writer, + None => self.writer.insert(TelemetryWriter::create(&self.path)?), + }; + writer + .write(rows) + .with_context(|| format!("Failed to write {}", self.path.display())) + } + + fn close(self) -> Result<()> { + if let Some(writer) = self.writer { + writer + .close() + .with_context(|| format!("Failed to write {}", self.path.display()))?; + } + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use runner_shared::telemetry::{LogRow, SpanRow, TelemetryReader}; + + use super::super::OtelReceiver; + use super::super::receiver::tests::post; + use super::super::rows::tests::{START, logs_request, trace_request}; + use super::*; + + fn read(path: &Path) -> Vec { + TelemetryReader::::new(std::fs::File::open(path).unwrap()) + .unwrap() + .collect::>() + .unwrap() + } + + /// An OTLP/JSON request, with hex ids and 64-bit integers as strings. + fn json_logs_request() -> String { + let timestamp = START + 70; + serde_json::json!({ + "resourceLogs": [{ + "resource": {"attributes": [ + {"key": "service.name", "value": {"stringValue": "example"}}, + ]}, + "scopeLogs": [{ + "scope": {"name": "example"}, + "logRecords": [{ + "timeUnixNano": timestamp.to_string(), + "severityNumber": 13, + "severityText": "WARN", + "body": {"stringValue": "skipped invalid fields"}, + "attributes": [ + {"key": "skipped", "value": {"intValue": "1"}}, + ], + "traceId": "ab".repeat(16), + "spanId": "01".repeat(8), + }], + }], + }], + }) + .to_string() + } + + #[tokio::test] + async fn writes_the_received_spans_and_logs() { + let profile_folder = tempfile::tempdir().unwrap(); + let receiver = OtelReceiver::start(profile_folder.path()).await.unwrap(); + let protobuf = [("content-type", "application/x-protobuf")]; + let json = [("content-type", "application/json")]; + for (path, headers, body) in [ + ("/v1/traces", &protobuf, trace_request().encode_to_vec()), + ("/v1/logs", &protobuf, logs_request().encode_to_vec()), + ("/v1/logs", &json, json_logs_request().into_bytes()), + ("/v1/traces", &protobuf, b"not protobuf".to_vec()), + ] { + let response = post(&receiver, path, headers, body).await; + assert!(response.status().is_success()); + } + + let stats = write_telemetry_files(receiver.stop().await, profile_folder.path()) + .await + .unwrap(); + + assert_eq!( + stats, + TelemetryStats { + spans: 2, + logs: 2, + without_monotonic_time: 2, + skipped_requests: 1, + } + ); + assert!(!profile_folder.path().join("otel-requests").exists()); + assert_eq!( + read::(&profile_folder.path().join(SPANS_FILE)), + span_rows(trace_request()) + ); + let logs = read::(&profile_folder.path().join(LOGS_FILE)); + assert_eq!(logs[..1], log_rows(logs_request())); + assert_eq!( + ( + logs[1].body.as_str(), + logs[1].trace_id.as_str(), + logs[1].timestamp, + &logs[1].log_attributes, + ), + ( + "skipped invalid fields", + "ab".repeat(16).as_str(), + (START + 70) as i64, + &vec![("skipped".to_string(), "1".to_string())], + ) + ); + } + + #[tokio::test] + async fn writes_nothing_without_requests() { + let profile_folder = tempfile::tempdir().unwrap(); + let receiver = OtelReceiver::start(profile_folder.path()).await.unwrap(); + + let stats = write_telemetry_files(receiver.stop().await, profile_folder.path()) + .await + .unwrap(); + + assert_eq!(stats, TelemetryStats::default()); + assert_eq!(std::fs::read_dir(profile_folder.path()).unwrap().count(), 0); + } +} diff --git a/src/instruments/otel/mod.rs b/src/instruments/otel/mod.rs new file mode 100644 index 000000000..72e0f6c02 --- /dev/null +++ b/src/instruments/otel/mod.rs @@ -0,0 +1,12 @@ +//! Collection of the OpenTelemetry spans and logs the benchmarks export. +//! +//! The [receiver](receiver::OtelReceiver) stores the OTLP/HTTP requests the benchmarks send while +//! they run. Once they exit, the requests are decoded and [written](files) as the +//! [`runner_shared::telemetry`] files. + +mod files; +mod receiver; +mod rows; + +pub use files::write_telemetry_files; +pub use receiver::{OTEL_ENDPOINT_ENV, OtelReceiver, ReceivedRequests}; diff --git a/src/instruments/otel/receiver.rs b/src/instruments/otel/receiver.rs new file mode 100644 index 000000000..039ad8696 --- /dev/null +++ b/src/instruments/otel/receiver.rs @@ -0,0 +1,548 @@ +//! OTLP/HTTP receiver storing the trace and log export requests of the benchmark process. +//! +//! While the benchmark runs, each request body is decompressed and written as received, one file +//! per request, to `/otel-requests/-.`. Metrics +//! requests are acknowledged and dropped. + +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::tokio::bufread::GzipDecoder; +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; +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 decoded. +const SPOOL_DIR: &str = "otel-requests"; + +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, +} + +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. + 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, + }) + } + + pub fn endpoint(&self) -> &str { + &self.endpoint + } + + /// Stops the receiver. A request is stored before it is acknowledged, so the returned + /// requests hold everything an exited process has sent. + pub async fn stop(self) -> ReceivedRequests { + drop(self.shutdown_tx); + if let Err(error) = self.server.await { + warn!("The OTLP receiver failed: {error}"); + } + ReceivedRequests { + dir: self.store.dir.clone(), + count: self.store.requests(), + received_bytes: self.store.received_bytes.load(Ordering::Relaxed), + } + } +} + +/// The trace and log requests stored by a stopped [`OtelReceiver`]. +#[derive(Debug)] +pub struct ReceivedRequests { + dir: PathBuf, + /// A sequence number taken by a failed write is counted as well. + pub count: u64, + /// Size of the request bodies as received, before decompression. + pub received_bytes: u64, +} + +impl ReceivedRequests { + /// The stored requests, in the order they were received. + pub(super) fn list(&self) -> Result> { + let entries = match std::fs::read_dir(&self.dir) { + Ok(entries) => entries, + Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(Vec::new()), + Err(error) => { + return Err(error) + .with_context(|| format!("Failed to read {}", self.dir.display())); + } + }; + let mut requests = Vec::new(); + for entry in entries { + let path = entry?.path(); + match StoredRequest::from_path(&path) { + Some(request) => requests.push(request), + None => warn!("Ignoring the unexpected OTLP spool file {}", path.display()), + } + } + requests.sort_by_key(|request| request.sequence); + Ok(requests) + } + + /// Removes the stored requests. + pub(super) fn remove(self) -> Result<()> { + match std::fs::remove_dir_all(&self.dir) { + Err(error) if error.kind() != io::ErrorKind::NotFound => { + Err(error).with_context(|| format!("Failed to remove {}", self.dir.display())) + } + _ => Ok(()), + } + } +} + +#[derive(Debug, PartialEq, Eq)] +pub(super) struct StoredRequest { + sequence: u64, + pub signal: Signal, + pub encoding: Encoding, + pub path: PathBuf, +} + +impl StoredRequest { + fn file_name(sequence: u64, signal: Signal, encoding: Encoding) -> String { + format!("{sequence:06}-{}.{}", signal.name(), encoding.extension()) + } + + fn from_path(path: &Path) -> Option { + let (stem, extension) = path.file_name()?.to_str()?.split_once('.')?; + let (sequence, signal) = stem.split_once('-')?; + Some(Self { + sequence: sequence.parse().ok()?, + signal: Signal::from_name(signal)?, + encoding: Encoding::from_extension(extension)?, + path: path.to_path_buf(), + }) + } +} + +/// 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(endpoint) = Endpoint::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 Endpoint::Signal(signal) = endpoint { + let received_bytes = body.len() as u64; + let body = if is_gzipped { + let mut decoded = Vec::new(); + if let Err(error) = GzipDecoder::new(body.as_ref()) + .read_to_end(&mut decoded) + .await + { + debug!("OTLP receiver failed to decompress a request body: {error}"); + return status_response(StatusCode::BAD_REQUEST); + } + Bytes::from(decoded) + } else { + body + }; + if let Err(error) = store.write(signal, encoding, &body, received_bytes).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 Endpoint { + Signal(Signal), + /// Accepted so that exporters don't report errors, but not stored. + Metrics, +} + +impl Endpoint { + fn from_path(path: &str) -> Option { + match path.trim_end_matches('/') { + "/v1/traces" => Some(Self::Signal(Signal::Traces)), + "/v1/logs" => Some(Self::Signal(Signal::Logs)), + "/v1/metrics" => Some(Self::Metrics), + _ => None, + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(super) enum Signal { + Traces, + Logs, +} + +impl Signal { + fn name(self) -> &'static str { + match self { + Self::Traces => "traces", + Self::Logs => "logs", + } + } + + fn from_name(name: &str) -> Option { + match name { + "traces" => Some(Self::Traces), + "logs" => Some(Self::Logs), + _ => None, + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(super) 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", + } + } + + fn from_extension(extension: &str) -> Option { + match extension { + "pb" => Some(Self::Protobuf), + "json" => Some(Self::Json), + _ => None, + } + } +} + +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, + body: &[u8], + received_bytes: u64, + ) -> 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 path = self + .dir + .join(StoredRequest::file_name(sequence, signal, encoding)); + tokio::fs::write(&path, body) + .await + .with_context(|| format!("Failed to write {}", path.display()))?; + + self.received_bytes + .fetch_add(received_bytes, Ordering::Relaxed); + Ok(()) + } + + fn requests(&self) -> u64 { + self.next_sequence.load(Ordering::Relaxed) + } +} + +#[cfg(test)] +pub(super) mod tests { + use async_compression::tokio::write::GzipEncoder; + use tokio::io::AsyncWriteExt; + + use super::*; + + type Headers<'a> = &'a [(&'a str, &'a str)]; + + pub(in super::super) 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() + } + + /// The file name and content of each stored request. + fn stored_requests(requests: &ReceivedRequests) -> Vec<(String, Vec)> { + requests + .list() + .unwrap() + .into_iter() + .map(|request| { + let name = request.path.file_name().unwrap().to_string_lossy().into(); + (name, std::fs::read(&request.path).unwrap()) + }) + .collect() + } + + #[tokio::test] + async fn stores_the_decompressed_trace_and_log_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/metrics", + &[("content-type", PROTOBUF_CONTENT_TYPE)], + b"metric-bytes", + ) + .await; + assert_eq!(response.status(), StatusCode::OK); + + let gzipped_logs = gzip(b"{\"resourceLogs\":[]}").await; + let gzipped_logs_len = gzipped_logs.len() as u64; + let response = post( + &receiver, + "/v1/logs", + &[ + ("content-type", "application/json; charset=utf-8"), + ("content-encoding", "gzip"), + ], + gzipped_logs, + ) + .await; + assert_eq!(response.status(), StatusCode::OK); + assert_eq!(response.text().await.unwrap(), "{}"); + + let requests = receiver.stop().await; + assert_eq!(requests.count, 2); + assert_eq!(requests.received_bytes, 11 + gzipped_logs_len); + assert_eq!( + stored_requests(&requests), + vec![ + ("000000-traces.pb".into(), b"trace-bytes".to_vec()), + ("000001-logs.json".into(), b"{\"resourceLogs\":[]}".to_vec()), + ] + ); + + requests.remove().unwrap(); + 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 gzipped_protobuf = [ + ("content-type", PROTOBUF_CONTENT_TYPE), + ("content-encoding", "gzip"), + ]; + + let cases: [(&str, Headers, StatusCode); 5] = [ + ("/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, + ), + ("/v1/traces", &gzipped_protobuf, StatusCode::BAD_REQUEST), + ]; + 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); + + let requests = receiver.stop().await; + assert_eq!((requests.count, requests.received_bytes), (0, 0)); + assert!(requests.list().unwrap().is_empty()); + assert_eq!(std::fs::read_dir(profile_folder.path()).unwrap().count(), 0); + } +} diff --git a/src/instruments/otel/rows.rs b/src/instruments/otel/rows.rs new file mode 100644 index 000000000..82023af78 --- /dev/null +++ b/src/instruments/otel/rows.rs @@ -0,0 +1,518 @@ +//! Flattening of the OTLP export requests into one row per span or log. +//! +//! Values are formatted like the OpenTelemetry Collector's ClickHouse exporter does, so that +//! these rows can be queried alongside its tables. + +use base64::Engine; +use opentelemetry_proto::tonic::collector::logs::v1::ExportLogsServiceRequest; +use opentelemetry_proto::tonic::collector::trace::v1::ExportTraceServiceRequest; +use opentelemetry_proto::tonic::common::v1::{AnyValue, InstrumentationScope, KeyValue, any_value}; +use opentelemetry_proto::tonic::resource::v1::Resource; +use opentelemetry_proto::tonic::trace::v1::span::SpanKind; +use opentelemetry_proto::tonic::trace::v1::status::StatusCode; + +use runner_shared::telemetry::{Attributes, LogRow, SpanEventRow, SpanRow}; + +const SERVICE_NAME_ATTRIBUTE: &str = "service.name"; + +/// Span attribute holding the span start on the monotonic clock of the benchmark markers. +pub const MONOTONIC_START_ATTRIBUTE: &str = "codspeed.monotonic_start_ns"; + +/// Span attribute holding the span end on the monotonic clock of the benchmark markers. +pub const MONOTONIC_END_ATTRIBUTE: &str = "codspeed.monotonic_end_ns"; + +/// Log attribute holding the log time on the monotonic clock of the benchmark markers. +pub const MONOTONIC_TIME_ATTRIBUTE: &str = "codspeed.monotonic_time_ns"; + +pub fn span_rows(request: ExportTraceServiceRequest) -> Vec { + let mut rows = Vec::new(); + for resource_spans in request.resource_spans { + let resource = ResourceContext::new(resource_spans.resource); + for scope_spans in resource_spans.scope_spans { + let (scope_name, scope_version) = scope_name_and_version(scope_spans.scope); + for mut span in scope_spans.spans { + let monotonic_start_ns = + take_monotonic_time(&mut span.attributes, MONOTONIC_START_ATTRIBUTE); + let monotonic_end_ns = + take_monotonic_time(&mut span.attributes, MONOTONIC_END_ATTRIBUTE); + rows.push(SpanRow { + timestamp: unix_nanos(span.start_time_unix_nano), + monotonic_start_ns, + monotonic_end_ns, + trace_id: hex_id(&span.trace_id), + span_id: hex_id(&span.span_id), + parent_span_id: hex_id(&span.parent_span_id), + span_name: span.name, + span_kind: span_kind(span.kind).into(), + service_name: resource.service_name.clone(), + scope_name: scope_name.clone(), + scope_version: scope_version.clone(), + duration_ns: span + .end_time_unix_nano + .saturating_sub(span.start_time_unix_nano), + status_code: status_code(span.status.as_ref().map_or(0, |status| status.code)) + .into(), + status_message: span.status.map(|status| status.message).unwrap_or_default(), + resource_attributes: resource.attributes.clone(), + span_attributes: attributes(span.attributes), + events: span + .events + .into_iter() + .map(|event| SpanEventRow { + timestamp: unix_nanos(event.time_unix_nano), + name: event.name, + attributes: attributes(event.attributes), + }) + .collect(), + }); + } + } + } + rows +} + +pub fn log_rows(request: ExportLogsServiceRequest) -> Vec { + let mut rows = Vec::new(); + for resource_logs in request.resource_logs { + let resource = ResourceContext::new(resource_logs.resource); + for scope_logs in resource_logs.scope_logs { + let (scope_name, scope_version) = scope_name_and_version(scope_logs.scope); + for mut record in scope_logs.log_records { + let timestamp = match record.time_unix_nano { + 0 => record.observed_time_unix_nano, + timestamp => timestamp, + }; + let monotonic_time_ns = + take_monotonic_time(&mut record.attributes, MONOTONIC_TIME_ATTRIBUTE); + rows.push(LogRow { + timestamp: unix_nanos(timestamp), + monotonic_time_ns, + trace_id: hex_id(&record.trace_id), + span_id: hex_id(&record.span_id), + severity_text: record.severity_text, + severity_number: u8::try_from(record.severity_number).unwrap_or_default(), + service_name: resource.service_name.clone(), + event_name: record.event_name, + body: record.body.map(any_value_string).unwrap_or_default(), + scope_name: scope_name.clone(), + scope_version: scope_version.clone(), + resource_attributes: resource.attributes.clone(), + log_attributes: attributes(record.attributes), + }); + } + } + } + rows +} + +/// What the rows of a resource share. +struct ResourceContext { + service_name: String, + attributes: Attributes, +} + +impl ResourceContext { + fn new(resource: Option) -> Self { + let attributes = attributes( + resource + .map(|resource| resource.attributes) + .unwrap_or_default(), + ); + let service_name = attributes + .iter() + .find(|(key, _)| key == SERVICE_NAME_ATTRIBUTE) + .map(|(_, value)| value.clone()) + .unwrap_or_default(); + Self { + service_name, + attributes, + } + } +} + +fn scope_name_and_version(scope: Option) -> (String, String) { + scope + .map(|scope| (scope.name, scope.version)) + .unwrap_or_default() +} + +/// Moves the monotonic time attribute `key` out of `attributes`. An attribute that isn't a +/// non-negative integer stays. +fn take_monotonic_time(attributes: &mut Vec, key: &str) -> Option { + let index = attributes + .iter() + .position(|attribute| attribute.key == key)?; + let time = match attributes[index].value.as_ref()?.value { + Some(any_value::Value::IntValue(time)) => u64::try_from(time).ok()?, + _ => return None, + }; + attributes.remove(index); + Some(time) +} + +fn attributes(attributes: Vec) -> Attributes { + attributes + .into_iter() + .map(|attribute| { + let value = attribute.value.map(any_value_string).unwrap_or_default(); + (attribute.key, value) + }) + .collect() +} + +/// A timestamp past the range of the timestamp columns is clamped to it. +fn unix_nanos(timestamp: u64) -> i64 { + i64::try_from(timestamp).unwrap_or(i64::MAX) +} + +/// Lowercase hex, or empty for an empty or all-zero (invalid) id. +fn hex_id(id: &[u8]) -> String { + if id.iter().all(|byte| *byte == 0) { + return String::new(); + } + id.iter().map(|byte| format!("{byte:02x}")).collect() +} + +fn span_kind(kind: i32) -> &'static str { + match SpanKind::try_from(kind).unwrap_or(SpanKind::Unspecified) { + SpanKind::Unspecified => "Unspecified", + SpanKind::Internal => "Internal", + SpanKind::Server => "Server", + SpanKind::Client => "Client", + SpanKind::Producer => "Producer", + SpanKind::Consumer => "Consumer", + } +} + +fn status_code(code: i32) -> &'static str { + match StatusCode::try_from(code).unwrap_or(StatusCode::Unset) { + StatusCode::Unset => "Unset", + StatusCode::Ok => "Ok", + StatusCode::Error => "Error", + } +} + +/// Strings as is, and other scalars as text. Bytes are base64-encoded, and arrays and maps are +/// JSON. +fn any_value_string(value: AnyValue) -> String { + match value.value { + // An index into a string table that OTLP traces and logs don't carry. + None | Some(any_value::Value::StringValueStrindex(_)) => String::new(), + Some(any_value::Value::StringValue(value)) => value, + Some(any_value::Value::BoolValue(value)) => value.to_string(), + Some(any_value::Value::IntValue(value)) => value.to_string(), + Some(any_value::Value::DoubleValue(value)) => value.to_string(), + Some(any_value::Value::BytesValue(value)) => { + base64::engine::general_purpose::STANDARD.encode(value) + } + Some(value @ (any_value::Value::ArrayValue(_) | any_value::Value::KvlistValue(_))) => { + any_value_json(Some(value)).to_string() + } + } +} + +fn any_value_json(value: Option) -> serde_json::Value { + use serde_json::Value as Json; + match value { + None | Some(any_value::Value::StringValueStrindex(_)) => Json::Null, + Some(any_value::Value::StringValue(value)) => Json::String(value), + Some(any_value::Value::BoolValue(value)) => Json::Bool(value), + Some(any_value::Value::IntValue(value)) => Json::from(value), + Some(any_value::Value::DoubleValue(value)) => Json::from(value), + Some(any_value::Value::BytesValue(value)) => { + Json::String(base64::engine::general_purpose::STANDARD.encode(value)) + } + Some(any_value::Value::ArrayValue(array)) => Json::Array( + array + .values + .into_iter() + .map(|value| any_value_json(value.value)) + .collect(), + ), + Some(any_value::Value::KvlistValue(list)) => Json::Object( + list.values + .into_iter() + .map(|attribute| { + let value = attribute.value.and_then(|value| value.value); + (attribute.key, any_value_json(value)) + }) + .collect(), + ), + } +} + +#[cfg(test)] +pub(super) mod tests { + use opentelemetry_proto::tonic::common::v1::{ArrayValue, KeyValueList}; + use opentelemetry_proto::tonic::logs::v1::{LogRecord, ResourceLogs, ScopeLogs}; + use opentelemetry_proto::tonic::resource::v1::Resource; + use opentelemetry_proto::tonic::trace::v1::span::Event; + use opentelemetry_proto::tonic::trace::v1::{ResourceSpans, ScopeSpans, Span, Status}; + + use super::*; + + /// The time the requests are relative to. + pub const START: u64 = 1_700_000_000_000_001_000; + /// [`START`] on the monotonic clock. + pub const MONOTONIC_START: u64 = 1_000; + + pub fn string_value(value: &str) -> AnyValue { + AnyValue { + value: Some(any_value::Value::StringValue(value.into())), + } + } + + pub fn key_value(key: &str, value: AnyValue) -> KeyValue { + KeyValue { + key: key.into(), + value: Some(value), + ..Default::default() + } + } + + pub fn resource() -> Option { + Some(Resource { + attributes: vec![key_value("service.name", string_value("example"))], + ..Default::default() + }) + } + + fn scope() -> Option { + Some(InstrumentationScope { + name: "codspeed".into(), + version: "1.0.0".into(), + ..Default::default() + }) + } + + pub fn trace_request() -> ExportTraceServiceRequest { + let root = Span { + trace_id: vec![0xab; 16], + span_id: vec![0x01; 8], + name: "parse".into(), + kind: SpanKind::Internal as i32, + start_time_unix_nano: START + 10, + end_time_unix_nano: START + 110, + attributes: vec![ + key_value( + "input_len", + AnyValue { + value: Some(any_value::Value::IntValue(7)), + }, + ), + key_value( + MONOTONIC_START_ATTRIBUTE, + AnyValue { + value: Some(any_value::Value::IntValue(MONOTONIC_START as i64 + 10)), + }, + ), + key_value( + MONOTONIC_END_ATTRIBUTE, + AnyValue { + value: Some(any_value::Value::IntValue(MONOTONIC_START as i64 + 110)), + }, + ), + ], + events: vec![Event { + time_unix_nano: START + 50, + name: "parsed".into(), + attributes: vec![key_value( + "valid", + AnyValue { + value: Some(any_value::Value::BoolValue(true)), + }, + )], + ..Default::default() + }], + status: Some(Status { + message: "bad input".into(), + code: StatusCode::Error as i32, + }), + ..Default::default() + }; + let child = Span { + trace_id: vec![0xab; 16], + span_id: vec![0x02; 8], + parent_span_id: vec![0x01; 8], + name: "tokenize".into(), + start_time_unix_nano: START + 20, + end_time_unix_nano: START + 30, + ..Default::default() + }; + ExportTraceServiceRequest { + resource_spans: vec![ResourceSpans { + resource: resource(), + scope_spans: vec![ScopeSpans { + scope: scope(), + spans: vec![root, child], + ..Default::default() + }], + ..Default::default() + }], + } + } + + pub fn logs_request() -> ExportLogsServiceRequest { + let record = LogRecord { + observed_time_unix_nano: START + 60, + severity_number: 9, + severity_text: "INFO".into(), + event_name: "event benches/example.rs:9".into(), + body: Some(string_value("parsed values")), + attributes: vec![ + key_value( + "count", + AnyValue { + value: Some(any_value::Value::IntValue(3)), + }, + ), + key_value( + MONOTONIC_TIME_ATTRIBUTE, + AnyValue { + value: Some(any_value::Value::IntValue(MONOTONIC_START as i64 + 60)), + }, + ), + ], + trace_id: vec![0xab; 16], + span_id: vec![0x01; 8], + ..Default::default() + }; + ExportLogsServiceRequest { + resource_logs: vec![ResourceLogs { + resource: resource(), + scope_logs: vec![ScopeLogs { + scope: scope(), + log_records: vec![record], + ..Default::default() + }], + ..Default::default() + }], + } + } + + fn strings(attributes: &[(&str, &str)]) -> Attributes { + attributes + .iter() + .map(|(key, value)| (key.to_string(), value.to_string())) + .collect() + } + + fn expected_resource_attributes() -> Attributes { + strings(&[("service.name", "example")]) + } + + #[test] + fn flattens_the_spans() { + let rows = span_rows(trace_request()); + + let timestamp = START as i64; + assert_eq!( + rows, + vec![ + SpanRow { + timestamp: timestamp + 10, + monotonic_start_ns: Some(MONOTONIC_START + 10), + monotonic_end_ns: Some(MONOTONIC_START + 110), + trace_id: "ab".repeat(16), + span_id: "01".repeat(8), + parent_span_id: String::new(), + span_name: "parse".into(), + span_kind: "Internal".into(), + service_name: "example".into(), + scope_name: "codspeed".into(), + scope_version: "1.0.0".into(), + duration_ns: 100, + status_code: "Error".into(), + status_message: "bad input".into(), + resource_attributes: expected_resource_attributes(), + span_attributes: strings(&[("input_len", "7")]), + events: vec![SpanEventRow { + timestamp: timestamp + 50, + name: "parsed".into(), + attributes: strings(&[("valid", "true")]), + }], + }, + SpanRow { + timestamp: timestamp + 20, + monotonic_start_ns: None, + monotonic_end_ns: None, + trace_id: "ab".repeat(16), + span_id: "02".repeat(8), + parent_span_id: "01".repeat(8), + span_name: "tokenize".into(), + span_kind: "Unspecified".into(), + service_name: "example".into(), + scope_name: "codspeed".into(), + scope_version: "1.0.0".into(), + duration_ns: 10, + status_code: "Unset".into(), + status_message: String::new(), + resource_attributes: expected_resource_attributes(), + span_attributes: Vec::new(), + events: Vec::new(), + }, + ] + ); + } + + #[test] + fn flattens_the_logs_timestamped_when_observed_without_a_time() { + let rows = log_rows(logs_request()); + + assert_eq!( + rows, + vec![LogRow { + timestamp: (START + 60) as i64, + monotonic_time_ns: Some(MONOTONIC_START + 60), + trace_id: "ab".repeat(16), + span_id: "01".repeat(8), + severity_text: "INFO".into(), + severity_number: 9, + service_name: "example".into(), + event_name: "event benches/example.rs:9".into(), + body: "parsed values".into(), + scope_name: "codspeed".into(), + scope_version: "1.0.0".into(), + resource_attributes: expected_resource_attributes(), + log_attributes: strings(&[("count", "3")]), + }] + ); + } + + #[test] + fn has_no_service_name_without_a_resource() { + let mut request = trace_request(); + request.resource_spans[0].resource = None; + + let rows = span_rows(request); + + assert_eq!(rows[0].service_name, ""); + assert_eq!(rows[0].resource_attributes, []); + } + + #[test] + fn formats_the_attribute_values_as_text() { + let value = |value| AnyValue { value: Some(value) }; + let cases = [ + (value(any_value::Value::DoubleValue(1.5)), "1.5"), + (value(any_value::Value::DoubleValue(2.0)), "2"), + (value(any_value::Value::BytesValue(b"hi".to_vec())), "aGk="), + ( + value(any_value::Value::ArrayValue(ArrayValue { + values: vec![string_value("a"), value(any_value::Value::IntValue(1))], + })), + r#"["a",1]"#, + ), + ( + value(any_value::Value::KvlistValue(KeyValueList { + values: vec![key_value("key", string_value("value"))], + })), + r#"{"key":"value"}"#, + ), + (AnyValue { value: None }, ""), + ]; + for (value, expected) in cases { + assert_eq!(any_value_string(value.clone()), expected, "{value:?}"); + } + } +} diff --git a/src/local_logger/icons.rs b/src/local_logger/icons.rs index 340f833e2..ee3ac7d46 100644 --- a/src/local_logger/icons.rs +++ b/src/local_logger/icons.rs @@ -19,6 +19,7 @@ pub enum Icon { ExecutorValgrind, ExecutorWallTime, ExecutorMemory, + ExecutorAnalysis, // Box-drawing (used by rolling buffer) BoxTopLeft, @@ -60,6 +61,7 @@ impl fmt::Display for Icon { Icon::ExecutorValgrind => "\u{2699}", // ⚙ Icon::ExecutorWallTime => "\u{23f1}", // ⏱ Icon::ExecutorMemory => "\u{25a4}", // ▤ + Icon::ExecutorAnalysis => "\u{2234}", // ∴ Icon::BoxTopLeft => "\u{256d}", // ╭ Icon::BoxTopRight => "\u{256e}", // ╮ Icon::BoxBottomLeft => "\u{2570}", // ╰ diff --git a/src/local_logger/snapshots/codspeed_runner__local_logger__icons__tests__icon_rendering.snap b/src/local_logger/snapshots/codspeed_runner__local_logger__icons__tests__icon_rendering.snap index b43812b7e..6824cfa3f 100644 --- a/src/local_logger/snapshots/codspeed_runner__local_logger__icons__tests__icon_rendering.snap +++ b/src/local_logger/snapshots/codspeed_runner__local_logger__icons__tests__icon_rendering.snap @@ -13,6 +13,7 @@ ImpactNeutral: ● ExecutorValgrind: ⚙ ExecutorWallTime: ⏱ ExecutorMemory: ▤ +ExecutorAnalysis: ∴ BoxTopLeft: ╭ BoxTopRight: ╮ BoxBottomLeft: ╰ diff --git a/src/runner_mode.rs b/src/runner_mode.rs index 8f7bec62b..2275aac2e 100644 --- a/src/runner_mode.rs +++ b/src/runner_mode.rs @@ -13,6 +13,9 @@ pub enum RunnerMode { Walltime, #[cfg(target_os = "linux")] Memory, + /// Runs the benchmarks without measuring them, to collect the OpenTelemetry data they export + #[value(hide = true)] + Telemetry, } impl From<&RunnerMode> for runner_shared::measurement_mode::MeasurementMode { @@ -23,6 +26,7 @@ impl From<&RunnerMode> for runner_shared::measurement_mode::MeasurementMode { RunnerMode::Walltime => Self::Walltime, #[cfg(target_os = "linux")] RunnerMode::Memory => Self::Memory, + RunnerMode::Telemetry => Self::Telemetry, } } } diff --git a/src/upload/benchmark_display.rs b/src/upload/benchmark_display.rs index 806537830..e6b882285 100644 --- a/src/upload/benchmark_display.rs +++ b/src/upload/benchmark_display.rs @@ -288,6 +288,8 @@ pub fn build_benchmark_table(results: &[FetchLocalRunBenchmarkResult]) -> String ExecutorName::Valgrind => build_simulation_table(executor_results), ExecutorName::WallTime => build_walltime_table(executor_results), ExecutorName::Memory => build_memory_table(executor_results), + // Not in `executor_order`: analysis runs have no performance results. + ExecutorName::Analysis => continue, }; output.push_str(&table); } @@ -318,6 +320,7 @@ pub fn build_detailed_summary(result: &FetchLocalRunBenchmarkResult) -> String { format!("{name}: {time}") } } + ExecutorName::Analysis => name.to_string(), ExecutorName::Memory => { if let Some(mem) = &result.memory { let peak = style(helpers::format_memory(mem.peak_memory as f64, Some(1))).cyan(); diff --git a/src/upload/upload_metadata.rs b/src/upload/upload_metadata.rs index 5d3c8945b..8dfeb1ee5 100644 --- a/src/upload/upload_metadata.rs +++ b/src/upload/upload_metadata.rs @@ -15,12 +15,15 @@ mod tests { use insta::{assert_json_snapshot, assert_snapshot}; - use crate::executor::ExecutorName; - use crate::instruments::InstrumentName; + use serde_json::json; + + use crate::executor::{ExecutorName, apply_mode_instruments, get_executor_from_mode}; + use crate::instruments::{InstrumentName, Instruments}; use crate::run_environment::{ LocalData, RepositoryProvider, RunEnvironment, RunEnvironmentMetadata, RunEvent, RunPart, Sender, }; + use crate::runner_mode::RunnerMode; use crate::system::SystemInfo; use crate::upload::{ LATEST_UPLOAD_METADATA_VERSION, ProfileArchiveMetadata, Runner, UploadMetadata, @@ -89,6 +92,27 @@ mod tests { assert_json_snapshot!(upload_metadata); } + #[test] + fn test_telemetry_runs_upload_as_analysis_with_the_telemetry_instrument() { + let mut instruments = Instruments { + mongodb: None, + telemetry: false, + }; + apply_mode_instruments(&RunnerMode::Telemetry, &mut instruments); + let runner = Runner { + name: "codspeed-runner".into(), + version: "5.5.0".into(), + instruments: instruments.get_active_instrument_names(), + executor: get_executor_from_mode(&RunnerMode::Telemetry, None).name(), + exclude_allocations: false, + system_info: SystemInfo::test(), + }; + + let runner = serde_json::to_value(&runner).unwrap(); + assert_eq!(runner["executor"], "analysis"); + assert_eq!(runner["instruments"], json!(["Telemetry"])); + } + #[test] fn test_get_local_metadata_hash() { let upload_metadata = UploadMetadata { diff --git a/src/upload/uploader.rs b/src/upload/uploader.rs index 16ef3bd0f..e364141a2 100644 --- a/src/upload/uploader.rs +++ b/src/upload/uploader.rs @@ -50,7 +50,7 @@ fn human_bytes_per_second(bytes: u64, elapsed: std::time::Duration) -> String { /// Create a profile archive from the profile folder /// /// For Valgrind, we create a gzip-compressed tar archive of the entire profile folder. -/// For WallTime and Memory, we create an uncompressed tar archive on disk: their +/// For WallTime, Memory and Analysis, we create an uncompressed tar archive on disk: their /// profiles are already compressed, so gzip would barely shrink them. async fn create_profile_archive( profile_folder: &std::path::Path, @@ -68,7 +68,7 @@ async fn create_profile_archive( let data = gzip_encoder.into_inner(); ProfileArchive::new_compressed_in_memory(data).await? } - ExecutorName::Memory | ExecutorName::WallTime => { + ExecutorName::Memory | ExecutorName::WallTime | ExecutorName::Analysis => { debug!("Creating uncompressed tar archive on disk"); let temp_file = tempfile::NamedTempFile::new()?; let temp_path = temp_file.path().to_path_buf(); From 5c8bbe0251162d4abae5d101fbbf564618da632f Mon Sep 17 00:00:00 2001 From: Luc Varoqui Date: Tue, 20 Oct 2026 16:09:31 +0200 Subject: [PATCH 3/3] test: run the divan telemetry example end to end in telemetry mode Build the divan `telemetry` bench of a codspeed-rust checkout next to this repository with `cargo codspeed build -m telemetry`, run it through `codspeed run --mode telemetry`, and read back its telemetry files. Its benchmarks emit `tracing` spans and events, a `tracing` event outside of any span, records of the `log` crate, and nothing at all. Each span and log is timestamped within the run on the wall clock, and falls within the rounds of the benchmark that emitted it by its time on the monotonic clock. Records keep their severity and call site, the ones below `info` are filtered out, and the benchmark without telemetry has no rows. The test is ignored by default since CI has no codspeed-rust checkout. Run it with `cargo test --test otel -- --ignored`. `scripts/run-otel-example.sh` runs the same example and uploads it. Refs COD-3863 Co-Authored-By: Claude Opus 5.5 --- scripts/run-telemetry-example.sh | 32 ++++ tests/telemetry.rs | 291 +++++++++++++++++++++++++++++++ 2 files changed, 323 insertions(+) create mode 100755 scripts/run-telemetry-example.sh create mode 100644 tests/telemetry.rs diff --git a/scripts/run-telemetry-example.sh b/scripts/run-telemetry-example.sh new file mode 100755 index 000000000..f4ad146cd --- /dev/null +++ b/scripts/run-telemetry-example.sh @@ -0,0 +1,32 @@ +#!/usr/bin/env bash +# Build the divan `telemetry` bench of the codspeed-rust checkout next to +# this repository in the telemetry and walltime modes of `cargo codspeed`, run +# it through `codspeed run --mode telemetry,walltime`, and upload the spans and +# logs it collects along with its walltime results. +# +# 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-examples --bench telemetry" + +$CARGO_CODSPEED build -m telemetry $BENCH_ARGS +$CARGO_CODSPEED build -m walltime $BENCH_ARGS + +env -u GITHUB_ACTIONS CODSPEED_ISOLATION=false \ + "$CODSPEED" run --mode telemetry,walltime "$@" -- \ + "$CARGO_CODSPEED run $BENCH_ARGS" diff --git a/tests/telemetry.rs b/tests/telemetry.rs new file mode 100644 index 000000000..78c86c0d4 --- /dev/null +++ b/tests/telemetry.rs @@ -0,0 +1,291 @@ +//! End-to-end collection of the spans and logs of benchmarks built from the codspeed-rust +//! checkout, written as telemetry files that line up with 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::collections::BTreeSet; +use std::ops::Range; +use std::path::{Path, PathBuf}; + +use assert_cmd::Command; +use runner_shared::artifacts::{ArtifactExt, ExecutionTimestamps}; +use runner_shared::fifo::MarkerType; +use runner_shared::telemetry::{ + LOGS_FILE, LogRow, SPANS_FILE, SpanRow, TelemetryReader, TelemetryRow, +}; + +/// The example of codspeed-rust, whose benchmarks emit telemetry in different ways. +const EXAMPLE_PATH: &str = "crates/divan_compat/examples/benches/telemetry.rs"; + +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)) + } + } +} + +#[test] +#[ignore = "needs codspeed-rust checked out in ../codspeed-rust"] +fn collects_the_spans_and_logs_of_the_divan_telemetry_example() { + let cargo_codspeed = "cargo run -q -p cargo-codspeed -- codspeed"; + let bench_args = "-p codspeed-divan-compat-examples --bench telemetry"; + + codspeed_rust_command("bash") + .args([ + "-c", + &format!("{cargo_codspeed} build -m telemetry {bench_args}"), + ]) + .assert() + .success(); + + let (profile_folder, _temp_dir) = profile_folder(); + let run_start = unix_nanos_now(); + 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", "telemetry"]) + .arg("--profile-folder") + .arg(&profile_folder) + // `cargo codspeed run` picks the telemetry mode from `CODSPEED_RUNNER_MODE`. + .args(["--", &format!("{cargo_codspeed} run {bench_args}")]) + .assert() + .success(); + let run_end = unix_nanos_now(); + + assert!(!profile_folder.join("otel-requests").exists()); + let spans: Vec = read(&profile_folder.join(SPANS_FILE)); + let logs: Vec = read(&profile_folder.join(LOGS_FILE)); + let execution_timestamps = ExecutionTimestamps::decode_from_reader( + std::fs::File::open(profile_folder.join(ExecutionTimestamps::file_name(None))).unwrap(), + ) + .unwrap(); + let regions = benchmark_regions(&execution_timestamps); + for benchmark in [ + "parse_and_sum", + "event_without_span", + "log_records", + "no_telemetry", + ] { + assert!( + regions + .iter() + .any(|(region_uri, _)| *region_uri == uri(benchmark)), + "{benchmark} did not run: {regions:?}" + ); + } + // The benchmark whose rounds a time on the monotonic clock falls within. + let benchmark_at = |monotonic_time: Option| { + let monotonic_time = monotonic_time?; + regions + .iter() + .find(|(_, range)| range.contains(&monotonic_time)) + .map(|(uri, _)| uri.clone()) + }; + let span_benchmark = |span: &SpanRow| benchmark_at(span.monotonic_start_ns); + let log_benchmark = |log: &LogRow| benchmark_at(log.monotonic_time_ns); + let log = |body: &str| { + logs.iter().find(|log| log.body == body).unwrap_or_else(|| { + let bodies = logs.iter().map(|log| &log.body).collect::>(); + panic!("no `{body}` log in {bodies:?}") + }) + }; + + // Timestamps are wall clock times. + let timestamps = spans + .iter() + .map(|span| span.timestamp) + .chain(logs.iter().map(|log| log.timestamp)); + for timestamp in timestamps { + assert!( + (run_start..=run_end).contains(×tamp), + "{timestamp} is outside of the run, from {run_start} to {run_end}" + ); + } + + // Every span and log happens within the rounds of a benchmark. + for span in &spans { + let end = span.monotonic_end_ns; + assert!( + span.monotonic_start_ns <= end, + "span {} ends before it starts", + span.span_name + ); + assert!( + span_benchmark(span).is_some() && span_benchmark(span) == benchmark_at(end), + "span {} runs outside of the rounds of a benchmark", + span.span_name + ); + } + for log in &logs { + assert!( + log_benchmark(log).is_some(), + "log `{}` is outside of the rounds of a benchmark", + log.body + ); + } + + // Nested `tracing` spans, with events in them. + let parse_and_sum_spans = spans + .iter() + .filter(|span| span_benchmark(span) == Some(uri("parse_and_sum"))) + .map(|span| span.span_name.as_str()) + .collect::>(); + assert!( + ["parse_and_sum", "parse", "sum"] + .iter() + .all(|name| parse_and_sum_spans.contains(name)), + "missing spans of parse_and_sum in {parse_and_sum_spans:?}" + ); + assert!(!parse_and_sum_spans.contains("sum_inner")); + let skipped = log("skipped invalid fields"); + assert_eq!(log_benchmark(skipped), Some(uri("parse_and_sum"))); + let skipped_span = spans + .iter() + .find(|span| span.span_id == skipped.span_id) + .expect("the `skipped invalid fields` log is in no span"); + assert_eq!(skipped_span.span_name, "parse_and_sum"); + assert_eq!(skipped_span.trace_id, skipped.trace_id); + + // A `tracing` event outside of any span. + let event = log("summed without a span"); + assert_eq!(log_benchmark(event), Some(uri("event_without_span"))); + assert_eq!(event.severity_text, "INFO"); + assert_eq!(event.trace_id, ""); + + // Records of the `log` crate. + let record = log("summing 3 values, logged with the log crate"); + assert_eq!(log_benchmark(record), Some(uri("log_records"))); + assert_eq!(record.severity_text, "INFO"); + assert_eq!(record.severity_number, 9); + let example = std::fs::read_to_string(codspeed_rust_dir().join(EXAMPLE_PATH)).unwrap(); + let line = example + .lines() + .position(|line| line.contains("logged with the log crate")) + .unwrap() + + 1; + let attribute = |key: &str| { + record + .log_attributes + .iter() + .find(|(attribute_key, _)| attribute_key == key) + .map(|(_, value)| value.as_str()) + }; + assert_eq!(attribute("code.filepath"), Some(EXAMPLE_PATH)); + assert_eq!(attribute("code.lineno"), Some(line.to_string().as_str())); + + // Records below `info` are filtered out, whether they come from `tracing` or `log`. + assert!( + logs.iter() + .all(|log| log.body != "filtered out by the default filter") + ); + + // No telemetry at all. + assert!( + spans + .iter() + .all(|span| span_benchmark(span) != Some(uri("no_telemetry"))) + ); + assert!( + logs.iter() + .all(|log| log_benchmark(log) != Some(uri("no_telemetry"))) + ); +} + +fn unix_nanos_now() -> i64 { + let since_epoch = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap(); + i64::try_from(since_epoch.as_nanos()).unwrap() +} + +fn uri(benchmark: &str) -> String { + format!("{EXAMPLE_PATH}::{benchmark}") +} + +fn read(path: &Path) -> Vec { + TelemetryReader::::new(std::fs::File::open(path).unwrap()) + .unwrap() + .collect::>() + .unwrap() +} + +/// Each benchmark with the time its rounds took, from the first round start to the last round +/// end of its sample, on the clock of the markers. +fn benchmark_regions(timestamps: &ExecutionTimestamps) -> Vec<(String, Range)> { + let mut regions = Vec::new(); + let (mut round_start, mut rounds) = (None, None::>); + let mut markers = timestamps.markers.clone(); + markers.sort_by_key(|marker| match *marker { + MarkerType::SampleStart(timestamp) + | MarkerType::SampleEnd(timestamp) + | MarkerType::RoundStart(timestamp) + | MarkerType::RoundEnd(timestamp) => timestamp, + }); + for marker in markers { + match marker { + MarkerType::SampleStart(_) => rounds = None, + MarkerType::RoundStart(start) => round_start = Some(start), + MarkerType::RoundEnd(end) => { + let start = round_start.take().expect("a round ends before it starts"); + let first_start = rounds.as_ref().map_or(start, |rounds| rounds.start); + rounds = Some(first_start..end); + } + MarkerType::SampleEnd(_) => { + regions.push(rounds.take().expect("a sample has no complete round")); + } + } + } + assert_eq!(regions.len(), timestamps.uri_by_ts.len()); + timestamps + .uri_by_ts + .iter() + .map(|(_, uri)| uri.clone()) + .zip(regions) + .collect() +}