fix(test): clean up OTLP loads and reject unsupported read benches

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
(cherry picked from commit ef2c3861faade406971efb06edc96d5735d4cdfb)

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<bool> {
+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<Option<ExitStatus>> {
+        let status = self.child.try_wait()?;
+        if status.is_some() {
+            self.reaped = true;
+        }
+        Ok(status)
+    }
+
+    fn kill_and_wait(&mut self) -> std::io::Result<ExitStatus> {
+        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<bool> {
     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(

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()
+        );
+    }
+}