diff --git a/src/cmd/src/bin/query_regression_runner/otlp.rs b/src/cmd/src/bin/query_regression_runner/otlp.rs index ceccad0ffc..5be02b0efe 100644 --- a/src/cmd/src/bin/query_regression_runner/otlp.rs +++ b/src/cmd/src/bin/query_regression_runner/otlp.rs @@ -15,7 +15,7 @@ use std::collections::HashMap; use std::fs; use std::path::Path; -use std::process::{Command, Stdio}; +use std::process::{Child, Command, ExitStatus, Stdio}; use std::time::{Duration, Instant}; use regex::Regex; @@ -177,11 +177,13 @@ async fn run_otelgen_load( fs::create_dir_all(&log_dir)?; let stdout_path = log_dir.join("stdout.log"); let stderr_path = log_dir.join("stderr.log"); - let mut child = Command::new(otelgen_bin) - .args(&command[1..]) - .stdout(Stdio::from(fs::File::create(&stdout_path)?)) - .stderr(Stdio::from(fs::File::create(&stderr_path)?)) - .spawn()?; + let mut child = ChildGuard::new( + Command::new(otelgen_bin) + .args(&command[1..]) + .stdout(Stdio::from(fs::File::create(&stdout_path)?)) + .stderr(Stdio::from(fs::File::create(&stderr_path)?)) + .spawn()?, + ); let started = Instant::now(); if load.warmup_seconds > 0 { let _ = wait_for_child(&mut child, Duration::from_secs(load.warmup_seconds)).await?; @@ -196,17 +198,13 @@ async fn run_otelgen_load( .max(60); if !wait_for_child(&mut child, Duration::from_secs(remaining)).await? { timed_out = true; - child.kill()?; - let _ = child.wait()?; + let _ = child.kill_and_wait()?; } } let final_snapshot = fetch_otlp_metrics(client, http_port, &clock).await?; let returncode = match child.try_wait()? { Some(status) => status.code(), - None => { - child.kill()?; - child.wait()?.code() - } + None => child.kill_and_wait()?.code(), }; let elapsed_seconds = started.elapsed().as_secs_f64(); Ok(json!({ @@ -221,7 +219,45 @@ async fn run_otelgen_load( })) } -async fn wait_for_child(child: &mut std::process::Child, timeout: Duration) -> Result { +struct ChildGuard { + child: Child, + reaped: bool, +} + +impl ChildGuard { + fn new(child: Child) -> Self { + Self { + child, + reaped: false, + } + } + + fn try_wait(&mut self) -> std::io::Result> { + let status = self.child.try_wait()?; + if status.is_some() { + self.reaped = true; + } + Ok(status) + } + + fn kill_and_wait(&mut self) -> std::io::Result { + self.child.kill()?; + let status = self.child.wait()?; + self.reaped = true; + Ok(status) + } +} + +impl Drop for ChildGuard { + fn drop(&mut self) { + if !self.reaped { + let _ = self.child.kill(); + let _ = self.child.wait(); + } + } +} + +async fn wait_for_child(child: &mut ChildGuard, timeout: Duration) -> Result { let deadline = Instant::now() + timeout; loop { if child.try_wait()?.is_some() { @@ -485,6 +521,70 @@ mod tests { .unwrap() } + #[cfg(unix)] + #[tokio::test] + async fn metrics_failure_after_spawn_reaps_otelgen() { + use std::io::{Read, Write}; + use std::net::TcpListener; + use std::os::unix::fs::PermissionsExt; + use std::thread; + + let temp_dir = tempfile::tempdir().unwrap(); + let ready_path = temp_dir.path().join("ready"); + let pid_path = temp_dir.path().join("pid"); + let otelgen_path = temp_dir.path().join("otelgen"); + fs::write( + &otelgen_path, + format!( + "#!/bin/sh\necho $$ > {}\ntouch {}\nwhile :; do :; done\n", + pid_path.display(), + ready_path.display(), + ), + ) + .unwrap(); + let mut permissions = fs::metadata(&otelgen_path).unwrap().permissions(); + permissions.set_mode(0o755); + fs::set_permissions(&otelgen_path, permissions).unwrap(); + + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let port = listener.local_addr().unwrap().port(); + let server = thread::spawn(move || { + for status in ["200 OK", "500 Internal Server Error"] { + let (mut stream, _) = listener.accept().unwrap(); + let mut request = [0; 1024]; + let _ = stream.read(&mut request); + if status == "500 Internal Server Error" { + while !ready_path.exists() { + thread::yield_now(); + } + } + write!( + stream, + "HTTP/1.1 {status}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n" + ) + .unwrap(); + } + }); + let client = Client::builder().build().unwrap(); + let mut load = otlp_load_for_test(); + load.warmup_seconds = 0; + + assert!( + run_otelgen_load(&otelgen_path, port, temp_dir.path(), &load, &client) + .await + .is_err() + ); + server.join().unwrap(); + let pid = fs::read_to_string(pid_path).unwrap(); + assert!( + !std::process::Command::new("sh") + .args(["-c", &format!("kill -0 {pid}")]) + .status() + .unwrap() + .success() + ); + } + #[test] fn parses_labeled_otlp_metrics_and_rejects_non_finite_samples() { let metrics = parse_prometheus_metrics( diff --git a/src/cmd/src/bin/query_regression_runner/remote/mod.rs b/src/cmd/src/bin/query_regression_runner/remote/mod.rs index 946b00b61d..82314c473d 100644 --- a/src/cmd/src/bin/query_regression_runner/remote/mod.rs +++ b/src/cmd/src/bin/query_regression_runner/remote/mod.rs @@ -15,11 +15,12 @@ use std::fs; use std::path::{Path, PathBuf}; +use object_store::config::ObjectStoreConfig; use serde_json::{Value, json}; use self::read_bench::run_read_bench; use self::storage::{enforce_storage_thresholds, run_storage_inspection}; -use crate::query_regression_runner::model::DestinationConfig; +use crate::query_regression_runner::model::{DestinationConfig, ReadBenchConfig}; use crate::query_regression_runner::plan::normalized_remote_write; use crate::query_regression_runner::{ FinalizeRemoteArgs, PrepareRemoteArgs, RenderRemoteConfigArgs, Result, @@ -60,14 +61,15 @@ pub(super) async fn run_finalize_remote(args: FinalizeRemoteArgs) -> Result<()> args.candidate_destination.as_deref(), ), ] { - let (data_home, destination) = match (data_home, destination) { - (Some(data_home), None) => (data_home.to_path_buf(), None), + let (data_home, destination, destination_config) = match (data_home, destination) { + (Some(data_home), None) => (data_home.to_path_buf(), None, None), (None, Some(path)) => { let destination: DestinationConfig = toml::from_str(&fs::read_to_string(path)?)?; ( - PathBuf::from(destination.data_home), + PathBuf::from(&destination.data_home), Some(path.to_path_buf()), + Some(destination), ) } (Some(_), Some(_)) => { @@ -83,6 +85,11 @@ pub(super) async fn run_finalize_remote(args: FinalizeRemoteArgs) -> Result<()> .into()); } }; + validate_read_bench_destination( + remote.read_bench.as_ref(), + destination_config.as_ref(), + name, + )?; let target = targets .iter_mut() .find(|target| target.get("name").and_then(Value::as_str) == Some(name)) @@ -160,6 +167,26 @@ pub(super) async fn run_finalize_remote(args: FinalizeRemoteArgs) -> Result<()> Ok(()) } +fn validate_read_bench_destination( + read_bench: Option<&ReadBenchConfig>, + destination: Option<&DestinationConfig>, + name: &str, +) -> Result<()> { + let Some(destination) = destination else { + return Ok(()); + }; + if read_bench.is_some_and(|config| config.enabled) + && !matches!(&destination.object_store, ObjectStoreConfig::File(_)) + { + return Err(format!( + "{name}: read_bench requires a File destination, but --{name}-destination uses {}; disable read_bench to keep footer inspection enabled", + destination.object_store.provider_name(), + ) + .into()); + } + Ok(()) +} + fn is_storage_threshold_entry(threshold: &Value) -> bool { let Some(name) = threshold.get("threshold").and_then(Value::as_str) else { return false; @@ -181,3 +208,65 @@ fn is_storage_threshold_entry(threshold: &Value) -> bool { | "forbid_encodings" )) } + +#[cfg(test)] +mod tests { + use object_store::config::S3Config; + + use super::*; + + fn read_bench(enabled: bool) -> ReadBenchConfig { + ReadBenchConfig { + enabled, + parquetbench: true, + scanbench: false, + iterations: 1, + projection: vec![], + parquet_reader: "default".to_string(), + scan_scanner: "default".to_string(), + parallelism: 1, + max_files: None, + } + } + + fn destination(object_store: ObjectStoreConfig) -> DestinationConfig { + DestinationConfig { + data_home: "/tmp/data".to_string(), + object_store, + } + } + + #[test] + fn read_bench_requires_file_destination() { + let enabled = read_bench(true); + assert!( + validate_read_bench_destination( + Some(&enabled), + Some(&destination(ObjectStoreConfig::default())), + "base", + ) + .is_ok() + ); + + let error = validate_read_bench_destination( + Some(&enabled), + Some(&destination(ObjectStoreConfig::S3(S3Config::default()))), + "candidate", + ) + .unwrap_err(); + assert_eq!( + error.to_string(), + "candidate: read_bench requires a File destination, but --candidate-destination uses S3; disable read_bench to keep footer inspection enabled", + ); + + let disabled = read_bench(false); + assert!( + validate_read_bench_destination( + Some(&disabled), + Some(&destination(ObjectStoreConfig::S3(S3Config::default()))), + "candidate", + ) + .is_ok() + ); + } +}