From a65578556d6786fd543c559452dc8c5d10619f73 Mon Sep 17 00:00:00 2001 From: Luc Varoqui Date: Tue, 20 Oct 2026 16:07:43 +0200 Subject: [PATCH 1/4] feat: accept the telemetry clock offset of instrument hooks Bump instrument-hooks to protocol version 3. Integrations timestamp their OpenTelemetry spans and logs with the monotonic clock of the benchmark markers, shifted by `instrument_hooks_monotonic_offset` to read as wall clock times. In analysis mode, the hooks now send that offset to the runner in a new `SetTelemetryMonotonicOffset { pid, offset_ns }` command, so that the runner can place the telemetry back on the clock of the markers. The runner accepts protocol version 3 and acknowledges the command in every executor. Executors that don't collect telemetry ignore it. Refs COD-3863 Co-Authored-By: Claude Opus 5.5 --- crates/instrument-hooks-bindings/instrument-hooks | 2 +- crates/instrument-hooks-bindings/src/bindings.rs | 6 ++++++ crates/runner-shared/src/fifo.rs | 7 ++++++- src/executor/shared/fifo.rs | 5 +++++ 4 files changed, 18 insertions(+), 2 deletions(-) diff --git a/crates/instrument-hooks-bindings/instrument-hooks b/crates/instrument-hooks-bindings/instrument-hooks index b9ddb5bc6..f1b12073f 160000 --- a/crates/instrument-hooks-bindings/instrument-hooks +++ b/crates/instrument-hooks-bindings/instrument-hooks @@ -1 +1 @@ -Subproject commit b9ddb5bc654b2e6fa13eb18efcd3a45e7ecda0bb +Subproject commit f1b12073f4f86cd5cd44aa3dffb04f0e6b1ad3ad diff --git a/crates/instrument-hooks-bindings/src/bindings.rs b/crates/instrument-hooks-bindings/src/bindings.rs index 3adf72f9b..a6202a8cf 100644 --- a/crates/instrument-hooks-bindings/src/bindings.rs +++ b/crates/instrument-hooks-bindings/src/bindings.rs @@ -56,6 +56,12 @@ unsafe extern "C" { unsafe extern "C" { pub fn instrument_hooks_current_timestamp() -> u64; } +unsafe extern "C" { + pub fn instrument_hooks_monotonic_offset() -> i64; +} +unsafe extern "C" { + pub fn instrument_hooks_callgrind_add_obj_skip(path: *const ::std::os::raw::c_char) -> u8; +} pub const instrument_hooks_feature_t_FEATURE_DISABLE_CALLGRIND_MARKERS: instrument_hooks_feature_t = 0; pub type instrument_hooks_feature_t = ::std::os::raw::c_uint; diff --git a/crates/runner-shared/src/fifo.rs b/crates/runner-shared/src/fifo.rs index db3094332..05a661a22 100644 --- a/crates/runner-shared/src/fifo.rs +++ b/crates/runner-shared/src/fifo.rs @@ -8,7 +8,7 @@ pub const RUNNER_ACK_FIFO: &str = "/tmp/runner.ack.fifo"; /// of integrations supporting the new protocol version a significant amount of time before /// releasing the runner. pub const MINIMAL_SUPPORTED_PROTOCOL_VERSION: u64 = 1; -pub const CURRENT_PROTOCOL_VERSION: u64 = 2; +pub const CURRENT_PROTOCOL_VERSION: u64 = 3; const _: () = assert!( MINIMAL_SUPPORTED_PROTOCOL_VERSION <= CURRENT_PROTOCOL_VERSION, @@ -63,4 +63,9 @@ pub enum Command { SetVersion(u64), GetIntegrationMode, IntegrationModeResponse(IntegrationMode), + /// The offset the process adds to its monotonic clock to timestamp its OpenTelemetry data. + SetTelemetryMonotonicOffset { + pid: i32, + offset_ns: i64, + }, } diff --git a/src/executor/shared/fifo.rs b/src/executor/shared/fifo.rs index bdff3cd61..80b41a71c 100644 --- a/src/executor/shared/fifo.rs +++ b/src/executor/shared/fifo.rs @@ -232,6 +232,11 @@ impl RunnerFifo { markers.push(*marker); self.send_cmd(FifoCommand::Ack).await?; } + // Sent in every analysis mode, but only used by the executors that + // collect telemetry. + FifoCommand::SetTelemetryMonotonicOffset { .. } => { + self.send_cmd(FifoCommand::Ack).await?; + } FifoCommand::SetVersion(protocol_version) => { match protocol_version.cmp(&runner_shared::fifo::CURRENT_PROTOCOL_VERSION) { Ordering::Less => { From 959ce97a40c7301a19903002ca2ce61fe4e751e5 Mon Sep 17 00:00:00 2001 From: Luc Varoqui Date: Tue, 20 Oct 2026 16:07:54 +0200 Subject: [PATCH 2/4] 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, the benchmark in its own field, and the span or log time relative to the benchmark start when known. 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 | 248 ++++++++++++++++++ crates/runner-shared/src/telemetry/tests.rs | 200 ++++++++++++++ .../testdata/telemetry_logs_v1.msgpack.zst | Bin 0 -> 129 bytes .../testdata/telemetry_spans_v1.msgpack.zst | Bin 0 -> 253 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..a482e73d6 --- /dev/null +++ b/crates/runner-shared/src/telemetry/mod.rs @@ -0,0 +1,248 @@ +//! 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 { + /// Empty when the span has no benchmark attribute. + pub benchmark_uri: String, + /// Span start, in nanoseconds since the Unix epoch. + pub timestamp: i64, + /// Span start minus the benchmark start, `None` when the benchmark start is unknown. + pub benchmark_offset_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 benchmark attribute. + 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 { + /// Empty when the log has no benchmark attribute. + pub benchmark_uri: String, + /// The log time, or the time it was observed when it has none, in nanoseconds since the + /// Unix epoch. + pub timestamp: i64, + /// Log time minus the benchmark start, `None` when the benchmark start is unknown. + pub benchmark_offset_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 benchmark attribute. + 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..bdd34e1d6 --- /dev/null +++ b/crates/runner-shared/src/telemetry/tests.rs @@ -0,0 +1,200 @@ +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 { + benchmark_uri: "benches/example.rs::parse".into(), + timestamp: 1_700_000_000_000_001_010, + benchmark_offset_ns: Some(10), + 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 { + benchmark_uri: String::new(), + timestamp: 1_700_000_000_000_001_020, + benchmark_offset_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 { + benchmark_uri: "benches/example.rs::parse".into(), + timestamp: 1_700_000_000_000_001_060, + benchmark_offset_ns: Some(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: "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..228fa98e4ad840bfdb9dab065a829641f371369b GIT binary patch literal 129 zcmV-{0Dk`{wJ-euaK!@vBnE+^RxzHrVr6b)Xk~LRWq4t2aBO8Ra&tO5aA9(DWzQFv zoc=Z{1SCA!AYo##FfpV_PDW1&r@Cc!Wo~pJIjv+Mc42IFWpkjBlB{!Oa&~EBWiD=E jZDo;?rDJb(ZginD2LN`GMV0s)7j`2VEMx}AqlvBz-dZyq 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..8197de67dd0d47cdcd0ebbbafc1a14c7c62dcb05 GIT binary patch literal 253 zcmVa>YURDc=8#vTMYg*Wczn!DN@&L!WA!7A?W;Z2Xd7>-Z8y{3}cL5Rt9i>1dss2gl{ zoCZNaMMoh4V-FoZbIP{Z^Nd=VRoFDHn*_^KhnFAoly3X)8oCl?k2B5&4sxiq)KNNL0k`ViAz0ui8) zL=71f0TBS%+sOpm!=@4ZH#6kq8wk4Xu?OoE#l%ROY{2A#uiCrTSHy7r2m%K3c%mx< DoA`PW literal 0 HcmV?d00001 From 09e19d3cd0909b3e5f391166621f6cb6c9ed6071 Mon Sep 17 00:00:00 2001 From: Luc Varoqui Date: Tue, 20 Oct 2026 16:09:22 +0200 Subject: [PATCH 3/4] 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. - Each row is placed on the clock of its benchmark: the clock offset the benchmark process sent in `SetTelemetryMonotonicOffset` is subtracted from its timestamp, and the result is measured from the start marker of the benchmark named by its `codspeed.benchmark.uri` attribute. The offset is left out when either is missing. With several benchmark processes, the first offset is used, with a warning when another differs from it by more than 1 ms. `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 | 215 +++++++ src/executor/analysis/instruments/mod.rs | 52 ++ .../analysis/instruments/telemetry.rs | 66 +++ 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/benchmark_clock.rs | 148 +++++ src/instruments/otel/files.rs | 230 ++++++++ src/instruments/otel/mod.rs | 15 + src/instruments/otel/receiver.rs | 548 ++++++++++++++++++ src/instruments/otel/rows.rs | 512 ++++++++++++++++ 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 +- 25 files changed, 1988 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/benchmark_clock.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 43c21f16f..26be8e930 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 4416ad621..33a876452 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..531ac0ae3 --- /dev/null +++ b/src/executor/analysis/executor.rs @@ -0,0 +1,215 @@ +use super::instruments::{AnalysisRun, 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 mut telemetry_monotonic_offset_ns = None; + let fifo_offset_ns = &mut telemetry_monotonic_offset_ns; + 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?; + } + } + FifoCommand::SetTelemetryMonotonicOffset { pid, offset_ns } => { + record_telemetry_monotonic_offset(fifo_offset_ns, *pid, *offset_ns); + return Ok(Some(FifoCommand::Ack)); + } + _ => {} + } + 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. + let run = AnalysisRun { + timestamps: self.timestamps.get(), + telemetry_monotonic_offset_ns, + }; + for instrument in instruments.iter_mut() { + instrument.stop(&run).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(()) + } +} + +/// Two processes computing their offset at different times differ by a few microseconds at +/// most, unless the wall clock was adjusted in between. +const MAX_TELEMETRY_MONOTONIC_OFFSET_DIFFERENCE_NS: i64 = 1_000_000; + +/// Keeps the first offset a benchmark process sent. +fn record_telemetry_monotonic_offset(recorded: &mut Option, pid: i32, offset_ns: i64) { + match *recorded { + None => *recorded = Some(offset_ns), + Some(first_offset_ns) + if first_offset_ns.abs_diff(offset_ns) + > MAX_TELEMETRY_MONOTONIC_OFFSET_DIFFERENCE_NS.unsigned_abs() => + { + warn!( + "Process {pid} timestamps its telemetry {} ns apart from the first benchmark \ + process, its telemetry will be off by as much", + i128::from(offset_ns) - i128::from(first_offset_ns) + ); + } + Some(_) => {} + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn keeps_the_first_telemetry_monotonic_offset() { + let mut recorded = None; + + record_telemetry_monotonic_offset(&mut recorded, 1, 1_000); + record_telemetry_monotonic_offset(&mut recorded, 2, 1_500); + record_telemetry_monotonic_offset(&mut recorded, 3, 1_000 + 2 * 1_000_000); + + assert_eq!(recorded, Some(1_000)); + } +} diff --git a/src/executor/analysis/instruments/mod.rs b/src/executor/analysis/instruments/mod.rs new file mode 100644 index 000000000..186a031a5 --- /dev/null +++ b/src/executor/analysis/instruments/mod.rs @@ -0,0 +1,52 @@ +//! Instruments hooked onto the [analysis executor](super::executor::AnalysisExecutor). + +use crate::instruments::Instruments; +use crate::prelude::*; +use async_trait::async_trait; +use runner_shared::artifacts::ExecutionTimestamps; +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, run: &AnalysisRun<'_>) -> Result<()>; +} + +/// What the benchmark process reported to the runner. +pub struct AnalysisRun<'a> { + /// The benchmark markers. + pub timestamps: Option<&'a ExecutionTimestamps>, + /// What the benchmark process adds to the marker clock to timestamp its OpenTelemetry data. + pub telemetry_monotonic_offset_ns: Option, +} + +/// 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..70c6a5b50 --- /dev/null +++ b/src/executor/analysis/instruments/telemetry.rs @@ -0,0 +1,66 @@ +use super::{AnalysisInstrument, AnalysisRun}; +use crate::instruments::otel::{ + BenchmarkClock, 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, run: &AnalysisRun<'_>) -> 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 clock = match (run.timestamps, run.telemetry_monotonic_offset_ns) { + (Some(timestamps), Some(offset_ns)) => BenchmarkClock::new(timestamps, offset_ns), + (_, None) if request_count > 0 => { + warn!( + "The benchmark process did not send its telemetry clock offset, the telemetry \ + won't be placed on the clock of its benchmark. Update its integration." + ); + BenchmarkClock::default() + } + _ => BenchmarkClock::default(), + }; + let stats = write_telemetry_files(requests, &profile_folder, clock) + .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 `otel` feature of `codspeed-divan-compat`." + ); + } else { + info!( + "Collected {} spans and {} logs from {request_count} OpenTelemetry export \ + requests ({received_bytes} bytes)", + stats.spans, stats.logs + ); + } + 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..10e7567f8 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 with 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/benchmark_clock.rs b/src/instruments/otel/benchmark_clock.rs new file mode 100644 index 000000000..ce3d81db3 --- /dev/null +++ b/src/instruments/otel/benchmark_clock.rs @@ -0,0 +1,148 @@ +use std::collections::HashMap; + +use itertools::Itertools; +use runner_shared::artifacts::ExecutionTimestamps; +use runner_shared::fifo::MarkerType; + +use crate::prelude::*; + +/// The start of each benchmark, to measure the telemetry of the benchmark process from. +#[derive(Debug, Default)] +pub struct BenchmarkClock { + /// On the monotonic clock of the benchmark markers. Sorted, one per run of the benchmark. + starts_by_uri: HashMap>, + /// What the benchmark process adds to the marker clock to timestamp its telemetry. + monotonic_offset_ns: i64, +} + +impl BenchmarkClock { + /// Pairs the `SampleStart` markers with the benchmark uris, both in the order the runner + /// received them. When their counts differ, no benchmark has a start. + pub fn new(timestamps: &ExecutionTimestamps, monotonic_offset_ns: i64) -> Self { + let starts = timestamps + .markers + .iter() + .filter_map(|marker| match marker { + MarkerType::SampleStart(timestamp) => Some(*timestamp), + _ => None, + }) + .collect_vec(); + if starts.len() != timestamps.uri_by_ts.len() { + warn!( + "Found {} benchmark samples for {} benchmarks, the telemetry won't be placed on \ + the clock of its benchmark", + starts.len(), + timestamps.uri_by_ts.len() + ); + return Self::default(); + } + + let mut starts_by_uri = HashMap::>::new(); + for (start, (_, uri)) in starts.into_iter().zip(×tamps.uri_by_ts) { + starts_by_uri.entry(uri.clone()).or_default().push(start); + } + Self { + starts_by_uri, + monotonic_offset_ns, + } + } + + /// Nanoseconds between the start of the benchmark `uri` and `timestamp`, a time of the + /// benchmark process's telemetry. A benchmark that ran several times is measured from its + /// last start before `timestamp`, or its first start if none is. + pub fn offset_ns(&self, uri: &str, timestamp: u64) -> Option { + let starts = self.starts_by_uri.get(uri)?; + let timestamp = + u64::try_from(i128::from(timestamp) - i128::from(self.monotonic_offset_ns)).ok()?; + let start = match starts.partition_point(|start| *start <= timestamp) { + 0 => starts.first()?, + started => &starts[started - 1], + }; + i64::try_from(i128::from(timestamp) - i128::from(*start)).ok() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn timestamps(uris: &[(u64, &str)], markers: &[MarkerType]) -> ExecutionTimestamps { + let uri_by_ts = uris + .iter() + .map(|(timestamp, uri)| (*timestamp, uri.to_string())) + .collect_vec(); + ExecutionTimestamps::new(&uri_by_ts, markers) + } + + #[test] + fn measures_from_the_start_of_the_benchmark() { + let clock = BenchmarkClock::new( + ×tamps( + &[(150, "a"), (350, "b")], + &[ + MarkerType::SampleStart(100), + MarkerType::RoundStart(110), + MarkerType::RoundEnd(120), + MarkerType::SampleEnd(140), + MarkerType::SampleStart(300), + MarkerType::SampleEnd(340), + ], + ), + 0, + ); + + assert_eq!(clock.offset_ns("a", 125), Some(25)); + assert_eq!(clock.offset_ns("b", 300), Some(0)); + // Telemetry emitted before the benchmark started. + assert_eq!(clock.offset_ns("b", 290), Some(-10)); + assert_eq!(clock.offset_ns("c", 125), None); + } + + #[test] + fn measures_a_benchmark_run_twice_from_its_last_start() { + let clock = BenchmarkClock::new( + ×tamps( + &[(150, "a"), (350, "a")], + &[ + MarkerType::SampleStart(100), + MarkerType::SampleEnd(140), + MarkerType::SampleStart(300), + MarkerType::SampleEnd(340), + ], + ), + 0, + ); + + assert_eq!(clock.offset_ns("a", 90), Some(-10)); + assert_eq!(clock.offset_ns("a", 125), Some(25)); + assert_eq!(clock.offset_ns("a", 310), Some(10)); + } + + #[test] + fn moves_the_timestamps_back_to_the_marker_clock() { + let clock = BenchmarkClock::new( + ×tamps( + &[(150, "a")], + &[MarkerType::SampleStart(100), MarkerType::SampleEnd(140)], + ), + 1_000, + ); + + assert_eq!(clock.offset_ns("a", 1_125), Some(25)); + // Before the start of the marker clock. + assert_eq!(clock.offset_ns("a", 999), None); + } + + #[test] + fn has_no_start_when_the_samples_do_not_match_the_benchmarks() { + let clock = BenchmarkClock::new( + ×tamps( + &[(150, "a"), (350, "b")], + &[MarkerType::SampleStart(100), MarkerType::SampleEnd(140)], + ), + 0, + ); + + assert_eq!(clock.offset_ns("a", 125), None); + } +} diff --git a/src/instruments/otel/files.rs b/src/instruments/otel/files.rs new file mode 100644 index 000000000..fd58fdcf4 --- /dev/null +++ b/src/instruments/otel/files.rs @@ -0,0 +1,230 @@ +//! 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::receiver::{Encoding, Signal, StoredRequest}; +use super::rows::{log_rows, span_rows}; +use super::{BenchmarkClock, ReceivedRequests}; +use crate::prelude::*; + +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)] +pub struct TelemetryStats { + pub spans: u64, + pub logs: u64, + /// Requests that failed to decode. + pub skipped_requests: u64, +} + +/// Decodes `requests` into the [`SPANS_FILE`] and [`LOGS_FILE`] of `profile_folder`, placed on +/// `clock`, then removes them. A file is only written when it has rows. +pub async fn write_telemetry_files( + requests: ReceivedRequests, + profile_folder: &Path, + clock: BenchmarkClock, +) -> Result { + let profile_folder = profile_folder.to_path_buf(); + tokio::task::spawn_blocking(move || { + let stats = write_files_blocking(&requests, &profile_folder, &clock)?; + requests.remove()?; + Ok(stats) + }) + .await? +} + +fn write_files_blocking( + requests: &ReceivedRequests, + profile_folder: &Path, + clock: &BenchmarkClock, +) -> 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, clock); + stats.spans += rows.len() as u64; + spans.write(rows) + }), + Signal::Logs => decode::(&request).map(|request| { + let rows = log_rows(request, clock); + stats.logs += rows.len() 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::{ + BENCHMARK_START, MONOTONIC_OFFSET, URI, clock, 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 = MONOTONIC_OFFSET + BENCHMARK_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": "codspeed.benchmark.uri", "value": {"stringValue": URI}}, + {"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(), clock()) + .await + .unwrap(); + + assert_eq!( + stats, + TelemetryStats { + spans: 2, + logs: 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(), &clock()) + ); + let logs = read::(&profile_folder.path().join(LOGS_FILE)); + assert_eq!(logs[..1], log_rows(logs_request(), &clock())); + assert_eq!( + ( + logs[1].body.as_str(), + logs[1].trace_id.as_str(), + logs[1].benchmark_offset_ns, + &logs[1].log_attributes, + ), + ( + "skipped invalid fields", + "ab".repeat(16).as_str(), + Some(70), + &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(), clock()) + .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..7bdd3855f --- /dev/null +++ b/src/instruments/otel/mod.rs @@ -0,0 +1,15 @@ +//! 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, placed on the clock of the benchmark that emitted +//! each span or log. + +mod benchmark_clock; +mod files; +mod receiver; +mod rows; + +pub use benchmark_clock::BenchmarkClock; +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..f021c37d4 --- /dev/null +++ b/src/instruments/otel/rows.rs @@ -0,0 +1,512 @@ +//! 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}; + +use super::BenchmarkClock; + +/// Span and log attribute naming the benchmark that emitted them. +pub const BENCHMARK_URI_ATTRIBUTE: &str = "codspeed.benchmark.uri"; + +const SERVICE_NAME_ATTRIBUTE: &str = "service.name"; + +pub fn span_rows(request: ExportTraceServiceRequest, clock: &BenchmarkClock) -> 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 span in scope_spans.spans { + let (benchmark_uri, span_attributes) = split_benchmark_uri(span.attributes); + rows.push(SpanRow { + benchmark_offset_ns: clock.offset_ns(&benchmark_uri, span.start_time_unix_nano), + benchmark_uri, + timestamp: unix_nanos(span.start_time_unix_nano), + 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, + 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, clock: &BenchmarkClock) -> 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 record in scope_logs.log_records { + let (benchmark_uri, log_attributes) = split_benchmark_uri(record.attributes); + let timestamp = match record.time_unix_nano { + 0 => record.observed_time_unix_nano, + timestamp => timestamp, + }; + rows.push(LogRow { + benchmark_offset_ns: clock.offset_ns(&benchmark_uri, timestamp), + benchmark_uri, + timestamp: unix_nanos(timestamp), + 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, + }); + } + } + } + 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 benchmark attribute out of `attributes`. +fn split_benchmark_uri(attributes: Vec) -> (String, Attributes) { + let mut benchmark_uri = String::new(); + let mut other_attributes = Vec::with_capacity(attributes.len()); + for (key, value) in self::attributes(attributes) { + if key == BENCHMARK_URI_ATTRIBUTE { + benchmark_uri = value; + } else { + other_attributes.push((key, value)); + } + } + (benchmark_uri, other_attributes) +} + +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 runner_shared::artifacts::ExecutionTimestamps; + use runner_shared::fifo::MarkerType; + + use super::*; + + pub const URI: &str = "benches/example.rs::parse"; + /// The integration's timestamps are this far ahead of the marker clock. + pub const MONOTONIC_OFFSET: u64 = 1_700_000_000_000_000_000; + /// On the marker clock. + pub const BENCHMARK_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 clock() -> BenchmarkClock { + BenchmarkClock::new( + &ExecutionTimestamps::new( + &[(BENCHMARK_START + 500, URI.into())], + &[ + MarkerType::SampleStart(BENCHMARK_START), + MarkerType::SampleEnd(BENCHMARK_START + 400), + ], + ), + MONOTONIC_OFFSET as i64, + ) + } + + 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: MONOTONIC_OFFSET + BENCHMARK_START + 10, + end_time_unix_nano: MONOTONIC_OFFSET + BENCHMARK_START + 110, + attributes: vec![ + key_value(BENCHMARK_URI_ATTRIBUTE, string_value(URI)), + key_value( + "input_len", + AnyValue { + value: Some(any_value::Value::IntValue(7)), + }, + ), + ], + events: vec![Event { + time_unix_nano: MONOTONIC_OFFSET + BENCHMARK_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: MONOTONIC_OFFSET + BENCHMARK_START + 20, + end_time_unix_nano: MONOTONIC_OFFSET + BENCHMARK_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: MONOTONIC_OFFSET + BENCHMARK_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(BENCHMARK_URI_ATTRIBUTE, string_value(URI)), + ], + 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(), &clock()); + + let timestamp = (MONOTONIC_OFFSET + BENCHMARK_START) as i64; + assert_eq!( + rows, + vec![ + SpanRow { + benchmark_uri: URI.into(), + timestamp: timestamp + 10, + benchmark_offset_ns: Some(10), + 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 { + benchmark_uri: String::new(), + timestamp: timestamp + 20, + benchmark_offset_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(), &clock()); + + assert_eq!( + rows, + vec![LogRow { + benchmark_uri: URI.into(), + timestamp: (MONOTONIC_OFFSET + BENCHMARK_START + 60) as i64, + benchmark_offset_ns: Some(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, &clock()); + + 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 cf0d27885d908686acd3961d436a8aa6fdf5f883 Mon Sep 17 00:00:00 2001 From: Luc Varoqui Date: Tue, 20 Oct 2026 16:09:31 +0200 Subject: [PATCH 4/4] test: run the divan otel example end to end in telemetry mode Build the divan `otel_example` 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: they hold the benchmark's spans and logs, each tagged with a benchmark that ran and placed within its benchmark's sample. 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-otel-example.sh | 32 +++++++ tests/otel.rs | 171 ++++++++++++++++++++++++++++++++++++ 2 files changed, 203 insertions(+) create mode 100755 scripts/run-otel-example.sh create mode 100644 tests/otel.rs diff --git a/scripts/run-otel-example.sh b/scripts/run-otel-example.sh new file mode 100755 index 000000000..553a3ea40 --- /dev/null +++ b/scripts/run-otel-example.sh @@ -0,0 +1,32 @@ +#!/usr/bin/env bash +# Build the divan `otel_example` of the codspeed-rust checkout next to this +# repository in the telemetry mode of `cargo codspeed`, run it through +# `codspeed run --mode telemetry`, and upload the spans and logs it collects. +# +# The upload goes to the CodSpeed instance of the selected runner profile: pick +# one with `CODSPEED_PROFILE` (see `codspeed profile list`). Arguments are +# passed to `codspeed run`, e.g. `--skip-upload` or `--profile-folder `. +set -euo pipefail + +RUNNER_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +CODSPEED_RUST_DIR="$RUNNER_DIR/../codspeed-rust" + +if [ ! -d "$CODSPEED_RUST_DIR" ]; then + echo "ERROR: codspeed-rust must be checked out at $CODSPEED_RUST_DIR." >&2 + exit 1 +fi + +cargo build --manifest-path "$RUNNER_DIR/Cargo.toml" --bin codspeed +CODSPEED="$RUNNER_DIR/target/debug/codspeed" + +cd "$CODSPEED_RUST_DIR" +CARGO_CODSPEED="cargo run -q -p cargo-codspeed -- codspeed" +BENCH_ARGS="-p codspeed-divan-compat --bench otel_example" + +$CARGO_CODSPEED build -m telemetry $BENCH_ARGS --features otel + +# The local provider: the GitHub Actions one needs a workflow event. Isolation +# needs sudo. +env -u GITHUB_ACTIONS CODSPEED_ISOLATION=false \ + "$CODSPEED" run --mode telemetry "$@" -- \ + "$CARGO_CODSPEED run $BENCH_ARGS" diff --git a/tests/otel.rs b/tests/otel.rs new file mode 100644 index 000000000..e509b4f2e --- /dev/null +++ b/tests/otel.rs @@ -0,0 +1,171 @@ +//! End-to-end collection of the spans and logs of a benchmark built from the codspeed-rust +//! checkout, written as telemetry files on the clock of the benchmark markers. +//! +//! Needs codspeed-rust checked out next to this repository. Run with +//! `cargo test --test otel -- --ignored`. +//! +//! Set `CODSPEED_OTEL_TEST_PROFILE_FOLDER` to a path to keep the profile folder there for +//! inspection instead of in a temporary directory. Its previous content is removed. +#![cfg(target_os = "linux")] + +use std::path::{Path, PathBuf}; + +use assert_cmd::Command; +use runner_shared::artifacts::{ArtifactExt, ExecutionTimestamps}; +use runner_shared::fifo::MarkerType; +use runner_shared::telemetry::{ + LOGS_FILE, LogRow, SPANS_FILE, SpanRow, TelemetryReader, TelemetryRow, +}; + +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_otel_example() { + let cargo_codspeed = "cargo run -q -p cargo-codspeed -- codspeed"; + let bench_args = "-p codspeed-divan-compat --bench otel_example"; + + codspeed_rust_command("bash") + .args([ + "-c", + &format!("{cargo_codspeed} build -m telemetry {bench_args} --features otel"), + ]) + .assert() + .success(); + + let (profile_folder, _temp_dir) = profile_folder(); + codspeed_rust_command(env!("CARGO_BIN_EXE_codspeed")) + // Forces the local provider; the GitHub Actions one needs a workflow event. + .env_remove("GITHUB_ACTIONS") + // Isolation needs sudo. + .env("CODSPEED_ISOLATION", "false") + .args(["run", "--skip-upload", "--mode", "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(); + + 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 span_names = spans.iter().map(|span| &span.span_name).collect::>(); + assert!( + span_names.iter().any(|name| *name == "parse_and_sum"), + "no `parse_and_sum` span in {span_names:?}" + ); + let log_bodies = logs.iter().map(|log| &log.body).collect::>(); + assert!( + log_bodies + .iter() + .any(|body| *body == "skipped invalid fields"), + "no `skipped invalid fields` log in {log_bodies:?}" + ); + + let execution_timestamps = ExecutionTimestamps::decode_from_reader( + std::fs::File::open(profile_folder.join(ExecutionTimestamps::file_name(None))).unwrap(), + ) + .unwrap(); + let benchmark_uris = execution_timestamps + .uri_by_ts + .iter() + .map(|(_, uri)| uri) + .collect::>(); + let longest_sample = longest_sample_ns(&execution_timestamps.markers); + + // Every span and log happens while its benchmark runs. + let span_times = spans.iter().map(|span| { + let end = span + .benchmark_offset_ns + .map(|offset| offset + span.duration_ns as i64); + (&span.benchmark_uri, span.benchmark_offset_ns, end) + }); + let log_times = logs.iter().map(|log| { + ( + &log.benchmark_uri, + log.benchmark_offset_ns, + log.benchmark_offset_ns, + ) + }); + for (row, (uri, start, end)) in span_times.chain(log_times).enumerate() { + assert!(benchmark_uris.contains(&uri), "unknown benchmark {uri:?}"); + let (Some(start), Some(end)) = (start, end) else { + panic!("row {row} has no benchmark offset"); + }; + assert!( + start >= 0 && end <= longest_sample, + "row {row} runs from {start} to {end} ns into its benchmark, outside of the \ + {longest_sample} ns of the longest benchmark sample" + ); + } +} + +fn read(path: &Path) -> Vec { + TelemetryReader::::new(std::fs::File::open(path).unwrap()) + .unwrap() + .collect::>() + .unwrap() +} + +fn longest_sample_ns(markers: &[MarkerType]) -> i64 { + let mut longest = 0; + let mut start = None; + for marker in markers { + match *marker { + MarkerType::SampleStart(timestamp) => start = Some(timestamp), + MarkerType::SampleEnd(end) => { + if let Some(start) = start.take() { + longest = longest.max(end - start); + } + } + MarkerType::RoundStart(_) | MarkerType::RoundEnd(_) => {} + } + } + longest as i64 +}