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/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/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/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/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/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/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 000000000..228fa98e4 Binary files /dev/null and b/crates/runner-shared/testdata/telemetry_logs_v1.msgpack.zst differ 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 000000000..8197de67d Binary files /dev/null and b/crates/runner-shared/testdata/telemetry_spans_v1.msgpack.zst differ 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/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/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 => { 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(); 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 +}