From f8bd707d54dee0e27c976102c6a7c26a17f5be4b Mon Sep 17 00:00:00 2001 From: Drew Newberry Date: Tue, 8 Sep 2026 15:28:36 -0700 Subject: [PATCH 1/3] perf(isolation): add TCP and DNS benchmark harnesses Signed-off-by: Drew Newberry --- crates/openshell-sandbox/Cargo.toml | 5 + .../openshell-sandbox/src/bin/seccomp-perf.rs | 93 +++++ crates/openshell-sandbox/src/lib.rs | 2 + crates/openshell-sandbox/src/perf.rs | 329 +++++++++++++++++ e2e/rust/Cargo.toml | 4 + e2e/rust/tests/internet_network_perf.rs | 170 +++++++++ e2e/rust/tests/live_internet_traffic_perf.rs | 337 ++++++++++++++++++ tasks/rust.toml | 4 + 8 files changed, 944 insertions(+) create mode 100644 crates/openshell-sandbox/src/bin/seccomp-perf.rs create mode 100644 crates/openshell-sandbox/src/perf.rs create mode 100644 e2e/rust/tests/internet_network_perf.rs create mode 100644 e2e/rust/tests/live_internet_traffic_perf.rs diff --git a/crates/openshell-sandbox/Cargo.toml b/crates/openshell-sandbox/Cargo.toml index 7972b34700..1bfb1c2aaa 100644 --- a/crates/openshell-sandbox/Cargo.toml +++ b/crates/openshell-sandbox/Cargo.toml @@ -14,6 +14,11 @@ repository.workspace = true name = "openshell-sandbox" path = "src/main.rs" +[[bin]] +name = "openshell-seccomp-perf" +path = "src/bin/seccomp-perf.rs" +required-features = ["perf-harness"] + [features] perf-harness = [] diff --git a/crates/openshell-sandbox/src/bin/seccomp-perf.rs b/crates/openshell-sandbox/src/bin/seccomp-perf.rs new file mode 100644 index 0000000000..f1354087cb --- /dev/null +++ b/crates/openshell-sandbox/src/bin/seccomp-perf.rs @@ -0,0 +1,93 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Microbenchmark entry point for the production seccomp network broker. + +use std::net::SocketAddr; + +use clap::{Parser, Subcommand}; +use openshell_sandbox::perf::{BenchmarkOptions, Layer, Protocol}; + +#[derive(Debug, Parser)] +#[command( + about = "Measure native and seccomp-filtered socket performance", + long_about = "Measure native and seccomp-filtered socket performance. UDP unconnected means destination-bearing SOCK_DGRAM traffic, not SOCK_RAW. General external UDP and SOCK_RAW are currently denied by the sandbox." +)] +struct Cli { + /// Benchmark layer: native, filtered, or all. + #[arg(long, default_value = "all", value_parser = ["native", "filtered", "all"])] + layer: String, + /// Protocol: tcp-connect, tcp-stream, udp-connected, udp-unconnected, or all. + #[arg( + long, + default_value = "all", + value_parser = ["tcp-connect", "tcp-stream", "udp-connected", "udp-unconnected", "all"] + )] + protocol: String, + #[arg(long, default_value_t = 10_000)] + iterations: u64, + #[arg(long, default_value_t = 1_000)] + warmup: u64, + #[arg(long, default_value_t = 1)] + concurrency: usize, + #[arg(long, default_value_t = 64)] + payload_bytes: usize, + #[command(subcommand)] + command: Option, +} + +#[derive(Debug, Subcommand)] +enum Command { + #[command(hide = true)] + Worker { + #[arg(long)] + protocol: Protocol, + #[arg(long)] + target: SocketAddr, + #[arg(long)] + iterations: u64, + #[arg(long)] + warmup: u64, + #[arg(long)] + concurrency: usize, + #[arg(long)] + payload_bytes: usize, + }, +} + +fn main() -> anyhow::Result<()> { + let cli = Cli::parse(); + if let Some(Command::Worker { + protocol, + target, + iterations, + warmup, + concurrency, + payload_bytes, + }) = cli.command + { + let report = openshell_sandbox::perf::run_worker( + protocol, + target, + iterations, + warmup, + concurrency, + payload_bytes, + )?; + println!("{}", serde_json::to_string(&report)?); + return Ok(()); + } + + let options = BenchmarkOptions { + layers: Layer::selection(&cli.layer)?, + protocols: Protocol::selection(&cli.protocol)?, + iterations: cli.iterations, + warmup: cli.warmup, + concurrency: cli.concurrency, + payload_bytes: cli.payload_bytes, + }; + for report in openshell_sandbox::perf::run(options)? { + println!("{}", serde_json::to_string(&report)?); + } + Ok(()) +} diff --git a/crates/openshell-sandbox/src/lib.rs b/crates/openshell-sandbox/src/lib.rs index 78bc3d2629..4928fe6f40 100644 --- a/crates/openshell-sandbox/src/lib.rs +++ b/crates/openshell-sandbox/src/lib.rs @@ -18,6 +18,8 @@ pub mod main_session; pub mod managed_children; #[cfg(target_os = "linux")] mod network_broker; +#[cfg(all(target_os = "linux", feature = "perf-harness"))] +pub mod perf; #[cfg(unix)] pub mod process; mod pty; diff --git a/crates/openshell-sandbox/src/perf.rs b/crates/openshell-sandbox/src/perf.rs new file mode 100644 index 0000000000..cc3b7da10f --- /dev/null +++ b/crates/openshell-sandbox/src/perf.rs @@ -0,0 +1,329 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Opt-in microbenchmarks for the production seccomp network listener. + +use std::io::{self, Read as _, Write as _}; +use std::net::{IpAddr, Ipv4Addr, SocketAddr, TcpListener, TcpStream}; +use std::process::{Command, Stdio}; +use std::sync::{Arc, Barrier, mpsc}; +use std::thread; +use std::time::Instant; + +use anyhow::{Context as _, bail}; +use clap::ValueEnum; +use openshell_isolation_interface::linux::workload_launcher; +use serde::{Deserialize, Serialize}; +use socket2::{Domain, Socket, Type}; + +use crate::network_broker::NetworkBroker; + +#[derive(Clone, Copy, Debug, Deserialize, Serialize, ValueEnum)] +#[serde(rename_all = "kebab-case")] +pub enum Layer { + Native, + Filtered, +} + +impl Layer { + pub fn selection(value: &str) -> anyhow::Result> { + match value { + "all" => Ok(vec![Self::Native, Self::Filtered]), + "native" => Ok(vec![Self::Native]), + "filtered" => Ok(vec![Self::Filtered]), + _ => bail!("unknown layer {value}"), + } + } +} + +#[derive(Clone, Copy, Debug, Deserialize, Serialize, ValueEnum)] +#[serde(rename_all = "kebab-case")] +pub enum Protocol { + TcpConnect, + TcpStream, +} + +impl Protocol { + pub fn selection(value: &str) -> anyhow::Result> { + match value { + "all" => Ok(vec![Self::TcpConnect, Self::TcpStream]), + _ => Ok(vec![ + Self::from_str(value, true).map_err(|error| anyhow::anyhow!("{error}"))?, + ]), + } + } +} + +#[derive(Debug)] +pub struct BenchmarkOptions { + pub layers: Vec, + pub protocols: Vec, + pub iterations: u64, + pub warmup: u64, + pub concurrency: usize, + pub payload_bytes: usize, +} + +#[derive(Debug, Deserialize, Serialize)] +pub struct BenchmarkReport { + pub layer: Layer, + pub protocol: Protocol, + pub iterations: u64, + pub concurrency: usize, + pub payload_bytes: usize, + pub capability_scope: String, + pub elapsed_ms: f64, + pub operations_per_second: f64, + pub throughput_mbit_per_second: f64, + pub latency_ns_p50: u64, + pub latency_ns_p95: u64, + pub latency_ns_p99: u64, +} + +pub fn run(options: BenchmarkOptions) -> anyhow::Result> { + if options.iterations == 0 || options.concurrency == 0 { + bail!("iterations and concurrency must be greater than zero"); + } + if options.payload_bytes == 0 || options.payload_bytes > 65_507 { + bail!("payload-bytes must be between 1 and 65507"); + } + + let executable = std::env::current_exe().context("resolve benchmark executable")?; + let mut reports = Vec::new(); + let mut filtered_runtime = None; + for layer in options.layers { + if matches!(layer, Layer::Filtered) && filtered_runtime.is_none() { + let (launcher, listener) = workload_launcher::start()?; + let broker = NetworkBroker::start_for_test(listener)?; + broker.confirm_healthy()?; + filtered_runtime = Some((launcher, broker)); + } + for protocol in &options.protocols { + let fixture = Fixture::start(*protocol, filtered_runtime.as_ref().map(|(_, b)| b))?; + let mut command = Command::new(&executable); + command + .arg("worker") + .arg("--protocol") + .arg( + protocol + .to_possible_value() + .expect("protocol value") + .get_name(), + ) + .arg("--target") + .arg(fixture.target.to_string()) + .arg("--iterations") + .arg(options.iterations.to_string()) + .arg("--warmup") + .arg(options.warmup.to_string()) + .arg("--concurrency") + .arg(options.concurrency.to_string()) + .arg("--payload-bytes") + .arg(options.payload_bytes.to_string()) + .stdout(Stdio::piped()) + .stderr(Stdio::inherit()); + let output = match layer { + Layer::Native => command.output(), + Layer::Filtered => filtered_runtime + .as_ref() + .expect("filtered runtime") + .0 + .execute(move || command.output())?, + } + .context("run benchmark worker")?; + if !output.status.success() { + bail!("benchmark worker exited with {}", output.status); + } + let mut report: BenchmarkReport = + serde_json::from_slice(&output.stdout).context("decode worker report")?; + report.layer = layer; + reports.push(report); + } + } + Ok(reports) +} + +struct Fixture { + target: SocketAddr, +} + +impl Fixture { + fn start(protocol: Protocol, _broker: Option<&NetworkBroker>) -> io::Result { + match protocol { + Protocol::TcpConnect => start_tcp_fixture(false), + Protocol::TcpStream => start_tcp_fixture(true), + } + } +} + +fn start_tcp_fixture(echo: bool) -> io::Result { + let socket = Socket::new(Domain::IPV4, Type::STREAM, None)?; + socket.set_reuse_address(true)?; + socket.bind(&SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0).into())?; + socket.listen(4_096)?; + let listener: TcpListener = socket.into(); + let target = listener.local_addr()?; + thread::Builder::new() + .name("seccomp-perf-tcp-fixture".into()) + .spawn(move || { + for stream in listener.incoming() { + let Ok(mut stream) = stream else { break }; + let _ = stream.set_nodelay(true); + if echo { + let _ = thread::Builder::new() + .name("seccomp-perf-tcp-echo".into()) + .spawn(move || { + let mut buffer = vec![0_u8; 65_507]; + while let Ok(length) = stream.read(&mut buffer) { + if length == 0 || stream.write_all(&buffer[..length]).is_err() { + break; + } + } + }); + } + } + })?; + Ok(Fixture { target }) +} + +#[allow(clippy::cast_precision_loss)] +pub fn run_worker( + protocol: Protocol, + target: SocketAddr, + iterations: u64, + warmup: u64, + concurrency: usize, + payload_bytes: usize, +) -> anyhow::Result { + let barrier = Arc::new(Barrier::new(concurrency + 1)); + let (sender, receiver) = mpsc::channel(); + let mut threads = Vec::with_capacity(concurrency); + for _ in 0..concurrency { + let barrier = Arc::clone(&barrier); + let sender = sender.clone(); + threads.push(thread::spawn(move || { + let result = worker_loop( + protocol, + target, + iterations, + warmup, + payload_bytes, + &barrier, + ); + let _ = sender.send(result); + })); + } + drop(sender); + let started = Instant::now(); + barrier.wait(); + let mut samples = Vec::new(); + for result in receiver { + samples.extend(result?); + } + let elapsed = started.elapsed(); + for worker in threads { + worker + .join() + .map_err(|_| anyhow::anyhow!("worker panicked"))?; + } + samples.sort_unstable(); + let operations = iterations.saturating_mul(concurrency as u64); + let seconds = elapsed.as_secs_f64(); + Ok(BenchmarkReport { + layer: Layer::Native, + protocol, + iterations: operations, + concurrency, + payload_bytes, + capability_scope: protocol.capability_scope().to_string(), + elapsed_ms: seconds * 1_000.0, + operations_per_second: operations as f64 / seconds, + throughput_mbit_per_second: operations as f64 * payload_bytes as f64 * 8.0 + / seconds + / 1_000_000.0, + latency_ns_p50: percentile(&samples, 50), + latency_ns_p95: percentile(&samples, 95), + latency_ns_p99: percentile(&samples, 99), + }) +} + +impl Protocol { + const fn capability_scope(self) -> &'static str { + match self { + Self::TcpConnect => "implemented local TCP socket/connect interception", + Self::TcpStream => "implemented established TCP fast path", + } + } +} + +fn worker_loop( + protocol: Protocol, + target: SocketAddr, + iterations: u64, + warmup: u64, + payload_bytes: usize, + barrier: &Barrier, +) -> anyhow::Result> { + let payload = vec![0x5a; payload_bytes]; + let mut tcp = if matches!(protocol, Protocol::TcpStream) { + let stream = TcpStream::connect(target)?; + let _ = stream.set_nodelay(true); + Some(stream) + } else { + None + }; + let mut response = vec![0_u8; payload_bytes]; + for _ in 0..warmup { + one_operation(protocol, target, &payload, &mut response, tcp.as_mut())?; + } + barrier.wait(); + let mut samples = Vec::with_capacity(usize::try_from(iterations).unwrap_or(0)); + for _ in 0..iterations { + let started = Instant::now(); + one_operation(protocol, target, &payload, &mut response, tcp.as_mut())?; + samples.push(u64::try_from(started.elapsed().as_nanos()).unwrap_or(u64::MAX)); + } + Ok(samples) +} + +fn one_operation( + protocol: Protocol, + target: SocketAddr, + payload: &[u8], + response: &mut [u8], + tcp: Option<&mut TcpStream>, +) -> io::Result<()> { + match protocol { + Protocol::TcpConnect => { + let stream = TcpStream::connect(target)?; + let _ = stream.set_nodelay(true); + } + Protocol::TcpStream => { + let stream = tcp.expect("TCP stream initialized"); + stream.write_all(payload)?; + stream.read_exact(response)?; + } + } + Ok(()) +} + +fn percentile(samples: &[u64], percentile: usize) -> u64 { + if samples.is_empty() { + return 0; + } + let index = (samples.len() - 1) * percentile / 100; + samples[index] +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn percentiles_are_stable() { + let samples = (1..=100).collect::>(); + assert_eq!(percentile(&samples, 50), 50); + assert_eq!(percentile(&samples, 95), 95); + assert_eq!(percentile(&samples, 99), 99); + } +} diff --git a/e2e/rust/Cargo.toml b/e2e/rust/Cargo.toml index 8eb3403c61..189e65ad8a 100644 --- a/e2e/rust/Cargo.toml +++ b/e2e/rust/Cargo.toml @@ -243,3 +243,7 @@ module_name_repetitions = "allow" must_use_candidate = "allow" missing_errors_doc = "allow" missing_panics_doc = "allow" +[[test]] +name = "live_internet_traffic_perf" +path = "tests/live_internet_traffic_perf.rs" +required-features = ["e2e-host-gateway"] diff --git a/e2e/rust/tests/internet_network_perf.rs b/e2e/rust/tests/internet_network_perf.rs new file mode 100644 index 0000000000..70c0f1af25 --- /dev/null +++ b/e2e/rust/tests/internet_network_perf.rs @@ -0,0 +1,170 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Opt-in Internet benchmark for the complete sandbox-to-supervisor data path. + +#![cfg(feature = "e2e-host-gateway")] + +use std::io::Write as _; +use std::process::Stdio; + +use openshell_e2e::harness::sandbox::SandboxGuard; +use tempfile::NamedTempFile; + +const BENCHMARK: &str = r#" +import http.client +import json +import os +import socket +import ssl +import statistics +import time + +HOST = "example.com" +ITERATIONS = int(os.environ.get("OPENSHELL_INET_PERF_ITERATIONS", "20")) + +def percentile(values, fraction): + values = sorted(values) + return values[min(len(values) - 1, int((len(values) - 1) * fraction))] + +def measure(name, operation, iterations=ITERATIONS): + operation() + samples = [] + started = time.perf_counter_ns() + for _ in range(iterations): + before = time.perf_counter_ns() + operation() + samples.append((time.perf_counter_ns() - before) / 1_000_000) + elapsed = (time.perf_counter_ns() - started) / 1_000_000_000 + return { + "name": name, + "iterations": iterations, + "mean_ms": statistics.fmean(samples), + "p50_ms": percentile(samples, 0.50), + "p95_ms": percentile(samples, 0.95), + "ops_per_second": iterations / elapsed, + } + +def dns_lookup(): + result = socket.getaddrinfo(HOST, 443, socket.AF_INET, socket.SOCK_STREAM) + if not result: + raise RuntimeError("DNS returned no IPv4 addresses") + +def tcp_connect(): + with socket.create_connection((HOST, 443), timeout=10): + pass + +tls_context = ssl.create_default_context() + +def https_cold(): + connection = http.client.HTTPSConnection(HOST, 443, timeout=10, context=tls_context) + try: + connection.request("HEAD", "/", headers={"Connection": "close"}) + response = connection.getresponse() + response.read() + if response.status != 200: + raise RuntimeError(f"unexpected HTTP status {response.status}") + finally: + connection.close() + +warm_connection = http.client.HTTPSConnection(HOST, 443, timeout=10, context=tls_context) + +def https_reuse(): + warm_connection.request("HEAD", "/", headers={"Connection": "keep-alive"}) + response = warm_connection.getresponse() + response.read() + if response.status != 200: + raise RuntimeError(f"unexpected HTTP status {response.status}") + +try: + metrics = [ + measure("dns_lookup", dns_lookup), + measure("tcp_connect", tcp_connect), + measure("https_cold", https_cold, max(5, ITERATIONS // 2)), + measure("https_reuse", https_reuse), + ] +finally: + warm_connection.close() + +print(json.dumps({"metrics": metrics}, separators=(",", ":"))) +"#; + +fn write_policy() -> Result { + let mut file = NamedTempFile::new().map_err(|error| format!("create policy: {error}"))?; + write!( + file, + r#"version: 1 + +filesystem_policy: + include_workdir: true + read_only: [/usr, /lib, /proc, /dev/urandom, /app, /etc, /var/log] + read_write: [/sandbox, /tmp, /dev/null] + +landlock: + compatibility: best_effort + +network_policies: + internet_performance: + name: internet_performance + endpoints: + - host: example.com + port: 80 + protocol: tcp + - host: example.com + port: 443 + protocol: tcp + binaries: + - path: "/**" +"# + ) + .map_err(|error| format!("write policy: {error}"))?; + file.flush() + .map_err(|error| format!("flush policy: {error}"))?; + Ok(file) +} + +async fn run_host_benchmark() -> Result { + let output = tokio::process::Command::new("python3") + .args(["-c", BENCHMARK]) + .env("OPENSHELL_INET_PERF_ITERATIONS", "20") + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .output() + .await + .map_err(|error| format!("run host benchmark: {error}"))?; + let combined = format!( + "{}{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + if !output.status.success() { + return Err(format!("host benchmark failed: {combined}")); + } + Ok(combined) +} + +#[tokio::test] +#[ignore = "manual Internet performance benchmark"] +async fn benchmark_complete_internet_path() { + let policy = write_policy().expect("write Internet benchmark policy"); + let policy_path = policy.path().to_str().expect("UTF-8 policy path"); + let sandbox = SandboxGuard::create(&["--policy", policy_path]) + .await + .expect("create benchmark sandbox"); + + for round in 1..=3 { + let host = run_host_benchmark().await.expect("host benchmark"); + println!("INTERNET_PERF host round={round} {}", host.trim()); + let mediated = sandbox + .exec(&[ + "sh", + "-c", + "OPENSHELL_INET_PERF_ITERATIONS=20 python3 -c \"$1\"", + "openshell-internet-perf", + BENCHMARK, + ]) + .await + .expect("sandbox benchmark"); + println!("INTERNET_PERF sandbox round={round} {}", mediated.trim()); + } +} diff --git a/e2e/rust/tests/live_internet_traffic_perf.rs b/e2e/rust/tests/live_internet_traffic_perf.rs new file mode 100644 index 0000000000..e8727bbce3 --- /dev/null +++ b/e2e/rust/tests/live_internet_traffic_perf.rs @@ -0,0 +1,337 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Opt-in live-Internet benchmark shaped like a short coding-agent session. + +#![cfg(feature = "e2e-host-gateway")] + +use std::io::Write as _; +use std::process::Stdio; +use std::time::Instant; + +use openshell_e2e::harness::sandbox::SandboxGuard; +use tempfile::NamedTempFile; + +const DEFAULT_ROUNDS: usize = 10; + +const BENCHMARK: &str = r#" +import concurrent.futures +import http.client +import json +import os +import shutil +import socket +import ssl +import statistics +import subprocess +import tempfile +import time +import urllib.parse + +MODE = os.environ.get("OPENSHELL_LIVE_PERF_MODE", "direct") +TLS = ssl.create_default_context() +USER_AGENT = "OpenShell-live-Internet-benchmark/1" + +METADATA_URLS = [ + "https://raw.githubusercontent.com/NVIDIA/OpenShell/main/README.md", + "https://pypi.org/pypi/requests/json", + "https://registry.npmjs.org/typescript/latest", + "https://docs.python.org/3/", +] +DNS_HOSTS = [ + "github.com", + "raw.githubusercontent.com", + "pypi.org", + "registry.npmjs.org", + "crates.io", + "docs.python.org", + "speed.cloudflare.com", +] + +def percentile(values, fraction): + values = sorted(values) + return values[min(len(values) - 1, int((len(values) - 1) * fraction))] + +def request(url, method="GET", max_bytes=None): + parsed = urllib.parse.urlsplit(url) + connection = http.client.HTTPSConnection(parsed.hostname, parsed.port or 443, timeout=20, context=TLS) + path = parsed.path or "/" + if parsed.query: + path += "?" + parsed.query + started = time.perf_counter_ns() + connection.request(method, path, headers={"User-Agent": USER_AGENT, "Connection": "close"}) + response = connection.getresponse() + first = response.read(1) + first_byte_ms = (time.perf_counter_ns() - started) / 1_000_000 + body_bytes = len(first) + while max_bytes is None or body_bytes < max_bytes: + remaining = None if max_bytes is None else max_bytes - body_bytes + chunk = response.read(65536 if remaining is None else min(65536, remaining)) + if not chunk: + break + body_bytes += len(chunk) + total_ms = (time.perf_counter_ns() - started) / 1_000_000 + status = response.status + connection.close() + if status < 200 or status >= 400: + raise RuntimeError(f"{url} returned HTTP {status}") + return {"status": status, "bytes": body_bytes, "first_byte_ms": first_byte_ms, "total_ms": total_ms} + +def metric(name, operation): + started = time.perf_counter_ns() + try: + detail = operation() + return { + "name": name, + "ok": True, + "elapsed_ms": (time.perf_counter_ns() - started) / 1_000_000, + "detail": detail, + } + except Exception as error: + return { + "name": name, + "ok": False, + "elapsed_ms": (time.perf_counter_ns() - started) / 1_000_000, + "error": f"{type(error).__name__}: {error}", + } + +def dns_set(): + samples = [] + for host in DNS_HOSTS: + started = time.perf_counter_ns() + addresses = socket.getaddrinfo(host, 443, socket.AF_UNSPEC, socket.SOCK_STREAM) + samples.append({ + "host": host, + "elapsed_ms": (time.perf_counter_ns() - started) / 1_000_000, + "addresses": len(addresses), + }) + timings = [sample["elapsed_ms"] for sample in samples] + return { + "lookups": samples, + "p50_ms": percentile(timings, 0.50), + "p95_ms": percentile(timings, 0.95), + } + +def metadata_serial(): + return {"requests": [request(url, max_bytes=2_000_000) for url in METADATA_URLS]} + +def metadata_concurrent(): + urls = METADATA_URLS + METADATA_URLS + with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool: + responses = list(pool.map(lambda url: request(url, max_bytes=2_000_000), urls)) + return {"requests": responses} + +def https_reuse(): + connection = http.client.HTTPSConnection("example.com", 443, timeout=20, context=TLS) + samples = [] + try: + for _ in range(20): + started = time.perf_counter_ns() + connection.request("HEAD", "/", headers={"User-Agent": USER_AGENT, "Connection": "keep-alive"}) + response = connection.getresponse() + response.read() + if response.status != 200: + raise RuntimeError(f"example.com returned HTTP {response.status}") + samples.append((time.perf_counter_ns() - started) / 1_000_000) + finally: + connection.close() + return { + "requests": len(samples), + "p50_ms": percentile(samples, 0.50), + "p95_ms": percentile(samples, 0.95), + "mean_ms": statistics.fmean(samples), + } + +def git_clone(): + if shutil.which("git") is None: + return {"skipped": "git is not installed in the workload image"} + with tempfile.TemporaryDirectory(prefix="openshell-live-git-") as directory: + checkout = os.path.join(directory, "sampleproject") + command = [ + "git", "-c", "advice.detachedHead=false", "clone", "--quiet", + "--depth", "1", "--filter=blob:none", + "https://github.com/pypa/sampleproject.git", checkout, + ] + output = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=60, check=False) + if output.returncode != 0: + raise RuntimeError(output.stderr.decode(errors="replace")[-500:]) + files = sum(len(names) for _, _, names in os.walk(checkout)) + size = sum(os.path.getsize(os.path.join(root, name)) for root, _, names in os.walk(checkout) for name in names) + return {"files": files, "checkout_bytes": size} + +def package_download(): + metadata = request("https://pypi.org/pypi/idna/json", max_bytes=2_000_000) + connection = http.client.HTTPSConnection("pypi.org", 443, timeout=20, context=TLS) + connection.request("GET", "/pypi/idna/json", headers={"User-Agent": USER_AGENT}) + response = connection.getresponse() + document = json.loads(response.read()) + connection.close() + wheels = [entry for entry in document["urls"] if entry["packagetype"] == "bdist_wheel"] + if not wheels: + raise RuntimeError("PyPI returned no idna wheel") + artifact = request(wheels[0]["url"], max_bytes=5_000_000) + return {"metadata": metadata, "artifact": artifact, "filename": wheels[0]["filename"]} + +def bulk_download(): + return request("https://speed.cloudflare.com/__down?bytes=5242880", max_bytes=5242880) + +def denied_destination(): + if MODE != "sandbox": + return {"skipped": "policy denial applies only to sandbox mode"} + started = time.perf_counter_ns() + try: + socket.create_connection(("example.net", 443), timeout=5).close() + except OSError as error: + return { + "denied": True, + "latency_ms": (time.perf_counter_ns() - started) / 1_000_000, + "error": str(error), + } + raise RuntimeError("destination omitted from policy was reachable") + +started = time.perf_counter_ns() +metrics = [ + metric("dns_service_set", dns_set), + metric("https_metadata_serial", metadata_serial), + metric("https_metadata_concurrent", metadata_concurrent), + metric("https_reuse", https_reuse), + metric("git_clone", git_clone), + metric("package_download", package_download), + metric("bulk_download_5mib", bulk_download), + metric("policy_denial", denied_destination), +] +document = { + "schema": "openshell.live-internet-perf.v1", + "mode": MODE, + "total_ms": (time.perf_counter_ns() - started) / 1_000_000, + "metrics": metrics, +} +print(json.dumps(document, separators=(",", ":"))) +if any(item["name"] == "policy_denial" and not item["ok"] for item in metrics): + raise SystemExit(1) +"#; + +fn rounds() -> usize { + std::env::var("OPENSHELL_LIVE_PERF_ROUNDS") + .ok() + .and_then(|value| value.parse().ok()) + .filter(|value| *value > 0) + .unwrap_or(DEFAULT_ROUNDS) +} + +fn write_policy() -> Result { + let mut file = NamedTempFile::new().map_err(|error| format!("create policy: {error}"))?; + write!( + file, + r#"version: 1 + +filesystem_policy: + include_workdir: true + read_only: [/usr, /lib, /proc, /dev/urandom, /app, /etc, /var/log] + read_write: [/sandbox, /tmp, /dev/null] + +landlock: + compatibility: best_effort + +network_policies: + live_internet_performance: + name: live_internet_performance + endpoints: + - {{ host: example.com, port: 443, protocol: tcp }} + - {{ host: github.com, port: 443, protocol: tcp }} + - {{ host: raw.githubusercontent.com, port: 443, protocol: tcp }} + - {{ host: pypi.org, port: 443, protocol: tcp }} + - {{ host: files.pythonhosted.org, port: 443, protocol: tcp }} + - {{ host: registry.npmjs.org, port: 443, protocol: tcp }} + - {{ host: crates.io, port: 443, protocol: tcp }} + - {{ host: docs.python.org, port: 443, protocol: tcp }} + - {{ host: speed.cloudflare.com, port: 443, protocol: tcp }} + binaries: + - path: "/**" +"# + ) + .map_err(|error| format!("write policy: {error}"))?; + file.flush() + .map_err(|error| format!("flush policy: {error}"))?; + Ok(file) +} + +async fn run_direct() -> Result { + run_python("direct").await +} + +async fn run_python(mode: &str) -> Result { + let output = tokio::process::Command::new("python3") + .args(["-c", BENCHMARK]) + .env("OPENSHELL_LIVE_PERF_MODE", mode) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .output() + .await + .map_err(|error| format!("run {mode} live Internet benchmark: {error}"))?; + let combined = format!( + "{}{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + if !output.status.success() { + return Err(format!( + "{mode} live Internet benchmark failed (exit {:?}): {combined}", + output.status.code() + )); + } + Ok(combined) +} + +async fn run_sandbox(sandbox: &SandboxGuard) -> Result { + sandbox + .exec(&[ + "sh", + "-c", + "OPENSHELL_LIVE_PERF_MODE=sandbox python3 -c \"$1\"", + "openshell-live-internet-perf", + BENCHMARK, + ]) + .await +} + +#[tokio::test] +#[ignore = "manual live-Internet performance benchmark"] +async fn benchmark_live_internet_agent_traffic() { + let policy = write_policy().expect("write live Internet benchmark policy"); + let policy_path = policy.path().to_str().expect("UTF-8 policy path"); + let create_started = Instant::now(); + let mut sandbox = SandboxGuard::create(&["--policy", policy_path]) + .await + .expect("create live Internet benchmark sandbox"); + println!( + "LIVE_INTERNET_PERF create {{\"elapsed_ms\":{}}}", + create_started.elapsed().as_secs_f64() * 1000.0 + ); + + for round in 1..=rounds() { + if round % 2 == 1 { + let direct = run_direct().await.expect("direct live Internet benchmark"); + println!("LIVE_INTERNET_PERF direct round={round} {}", direct.trim()); + let mediated = run_sandbox(&sandbox) + .await + .expect("sandbox live Internet benchmark"); + println!( + "LIVE_INTERNET_PERF sandbox round={round} {}", + mediated.trim() + ); + } else { + let mediated = run_sandbox(&sandbox) + .await + .expect("sandbox live Internet benchmark"); + println!( + "LIVE_INTERNET_PERF sandbox round={round} {}", + mediated.trim() + ); + let direct = run_direct().await.expect("direct live Internet benchmark"); + println!("LIVE_INTERNET_PERF direct round={round} {}", direct.trim()); + } + } + + sandbox.cleanup().await; +} diff --git a/tasks/rust.toml b/tasks/rust.toml index 50eda4118c..a1ecfc7890 100644 --- a/tasks/rust.toml +++ b/tasks/rust.toml @@ -3,6 +3,10 @@ # Rust check, lint, and format tasks +["perf:seccomp"] +description = "Compare native and sandbox-filtered TCP and DNS performance" +run = "cargo run --release -p openshell-sandbox --features perf-harness --bin openshell-seccomp-perf --" + ["rust:check"] description = "Check all Rust crates for errors" run = "cargo check --workspace" From 0f97afb5f5944f8b4713487fa121cf76b12c149a Mon Sep 17 00:00:00 2001 From: Drew Newberry Date: Tue, 8 Sep 2026 17:05:50 -0700 Subject: [PATCH 2/3] fix(perf): validate measurements and bound worker startup Signed-off-by: Drew Newberry --- architecture/sandbox.md | 17 + .../openshell-sandbox/src/bin/seccomp-perf.rs | 67 ++- crates/openshell-sandbox/src/perf.rs | 318 ++++++++----- e2e/rust/tests/live_internet_traffic_perf.rs | 433 +++++++++++++++++- 4 files changed, 682 insertions(+), 153 deletions(-) diff --git a/architecture/sandbox.md b/architecture/sandbox.md index c0d73da059..4dc6871a90 100644 --- a/architecture/sandbox.md +++ b/architecture/sandbox.md @@ -612,3 +612,20 @@ broker as its sender, not the original calling agent's `SI_USER` identity. Queued signals preserve permitted application siginfo payloads; they cannot forge kernel-generated or `SI_TKILL` codes. Programs requiring original sender identity must account for this mediation boundary. + +## Performance Measurement + +The optional `perf-harness` feature provides a local TCP microbenchmark, not a +full sandbox benchmark. It separates connection establishment from established +stream I/O. Connections and warmup complete before measurement starts; +connection-only results have no payload-throughput value. Worker preparation +fails with an error rather than leaving peers blocked at a barrier. + +The opt-in E2E benchmarks exercise DNS and TCP through both sandbox and +supervisor to Internet destinations. The live-traffic benchmark alternates a +direct **host-process** baseline with sandbox traffic for ten rounds by default; +this baseline is not raw Docker. Its typed reports distinguish successful, +failed, and skipped samples. Only successful samples are performance results: +every comparable scenario needs successful coverage, and every sandbox denial +round must report an explicit permission error. HTTP error responses, truncated +downloads, and failed DNS queries cannot count as fast successful requests. diff --git a/crates/openshell-sandbox/src/bin/seccomp-perf.rs b/crates/openshell-sandbox/src/bin/seccomp-perf.rs index f1354087cb..04cec02cf0 100644 --- a/crates/openshell-sandbox/src/bin/seccomp-perf.rs +++ b/crates/openshell-sandbox/src/bin/seccomp-perf.rs @@ -5,25 +5,21 @@ use std::net::SocketAddr; -use clap::{Parser, Subcommand}; +use clap::{Parser, Subcommand, ValueEnum}; use openshell_sandbox::perf::{BenchmarkOptions, Layer, Protocol}; #[derive(Debug, Parser)] #[command( about = "Measure native and seccomp-filtered socket performance", - long_about = "Measure native and seccomp-filtered socket performance. UDP unconnected means destination-bearing SOCK_DGRAM traffic, not SOCK_RAW. General external UDP and SOCK_RAW are currently denied by the sandbox." + long_about = "Measure native and seccomp-filtered TCP connection establishment and established-stream performance. Full supervisor-mediated DNS and Internet traffic are measured by the separate E2E harness." )] struct Cli { /// Benchmark layer: native, filtered, or all. - #[arg(long, default_value = "all", value_parser = ["native", "filtered", "all"])] - layer: String, - /// Protocol: tcp-connect, tcp-stream, udp-connected, udp-unconnected, or all. - #[arg( - long, - default_value = "all", - value_parser = ["tcp-connect", "tcp-stream", "udp-connected", "udp-unconnected", "all"] - )] - protocol: String, + #[arg(long, default_value = "all", value_enum)] + layer: LayerSelection, + /// TCP scenario to measure. + #[arg(long, default_value = "all", value_enum)] + protocol: ProtocolSelection, #[arg(long, default_value_t = 10_000)] iterations: u64, #[arg(long, default_value_t = 1_000)] @@ -79,8 +75,8 @@ fn main() -> anyhow::Result<()> { } let options = BenchmarkOptions { - layers: Layer::selection(&cli.layer)?, - protocols: Protocol::selection(&cli.protocol)?, + layers: cli.layer.values(), + protocols: cli.protocol.values(), iterations: cli.iterations, warmup: cli.warmup, concurrency: cli.concurrency, @@ -91,3 +87,48 @@ fn main() -> anyhow::Result<()> { } Ok(()) } + +#[derive(Clone, Copy, Debug, ValueEnum)] +enum LayerSelection { + Native, + Filtered, + All, +} + +impl LayerSelection { + fn values(self) -> Vec { + match self { + Self::Native => vec![Layer::Native], + Self::Filtered => vec![Layer::Filtered], + Self::All => vec![Layer::Native, Layer::Filtered], + } + } +} + +#[derive(Clone, Copy, Debug, ValueEnum)] +enum ProtocolSelection { + TcpConnect, + TcpStream, + All, +} + +impl ProtocolSelection { + fn values(self) -> Vec { + match self { + Self::TcpConnect => vec![Protocol::TcpConnect], + Self::TcpStream => vec![Protocol::TcpStream], + Self::All => vec![Protocol::TcpConnect, Protocol::TcpStream], + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + #[test] + fn unsupported_udp_is_rejected_by_cli() { + for value in ["udp-connected", "udp-unconnected"] { + assert!(Cli::try_parse_from(["seccomp-perf", "--protocol", value]).is_err()); + } + } +} diff --git a/crates/openshell-sandbox/src/perf.rs b/crates/openshell-sandbox/src/perf.rs index cc3b7da10f..484a27d640 100644 --- a/crates/openshell-sandbox/src/perf.rs +++ b/crates/openshell-sandbox/src/perf.rs @@ -6,9 +6,9 @@ use std::io::{self, Read as _, Write as _}; use std::net::{IpAddr, Ipv4Addr, SocketAddr, TcpListener, TcpStream}; use std::process::{Command, Stdio}; -use std::sync::{Arc, Barrier, mpsc}; +use std::sync::mpsc; use std::thread; -use std::time::Instant; +use std::time::{Duration, Instant}; use anyhow::{Context as _, bail}; use clap::ValueEnum; @@ -25,17 +25,6 @@ pub enum Layer { Filtered, } -impl Layer { - pub fn selection(value: &str) -> anyhow::Result> { - match value { - "all" => Ok(vec![Self::Native, Self::Filtered]), - "native" => Ok(vec![Self::Native]), - "filtered" => Ok(vec![Self::Filtered]), - _ => bail!("unknown layer {value}"), - } - } -} - #[derive(Clone, Copy, Debug, Deserialize, Serialize, ValueEnum)] #[serde(rename_all = "kebab-case")] pub enum Protocol { @@ -43,17 +32,6 @@ pub enum Protocol { TcpStream, } -impl Protocol { - pub fn selection(value: &str) -> anyhow::Result> { - match value { - "all" => Ok(vec![Self::TcpConnect, Self::TcpStream]), - _ => Ok(vec![ - Self::from_str(value, true).map_err(|error| anyhow::anyhow!("{error}"))?, - ]), - } - } -} - #[derive(Debug)] pub struct BenchmarkOptions { pub layers: Vec, @@ -74,20 +52,19 @@ pub struct BenchmarkReport { pub capability_scope: String, pub elapsed_ms: f64, pub operations_per_second: f64, - pub throughput_mbit_per_second: f64, + /// Payload throughput; absent for connection-only scenarios. + pub throughput_mbit_per_second: Option, pub latency_ns_p50: u64, pub latency_ns_p95: u64, pub latency_ns_p99: u64, } pub fn run(options: BenchmarkOptions) -> anyhow::Result> { - if options.iterations == 0 || options.concurrency == 0 { - bail!("iterations and concurrency must be greater than zero"); - } - if options.payload_bytes == 0 || options.payload_bytes > 65_507 { - bail!("payload-bytes must be between 1 and 65507"); - } - + validate_options( + options.iterations, + options.concurrency, + options.payload_bytes, + )?; let executable = std::env::current_exe().context("resolve benchmark executable")?; let mut reports = Vec::new(); let mut filtered_runtime = None; @@ -104,12 +81,7 @@ pub fn run(options: BenchmarkOptions) -> anyhow::Result> { command .arg("worker") .arg("--protocol") - .arg( - protocol - .to_possible_value() - .expect("protocol value") - .get_name(), - ) + .arg(protocol.cli_name()) .arg("--target") .arg(fixture.target.to_string()) .arg("--iterations") @@ -126,7 +98,7 @@ pub fn run(options: BenchmarkOptions) -> anyhow::Result> { Layer::Native => command.output(), Layer::Filtered => filtered_runtime .as_ref() - .expect("filtered runtime") + .ok_or_else(|| anyhow::anyhow!("filtered runtime was not initialized"))? .0 .execute(move || command.output())?, } @@ -186,6 +158,29 @@ fn start_tcp_fixture(echo: bool) -> io::Result { Ok(Fixture { target }) } +const OPERATION_TIMEOUT: Duration = Duration::from_secs(10); +const START_TIMEOUT: Duration = Duration::from_secs(30); + +fn validate_options( + iterations: u64, + concurrency: usize, + payload_bytes: usize, +) -> anyhow::Result<()> { + if iterations == 0 || concurrency == 0 || concurrency > 256 { + bail!("iterations must be positive and concurrency between 1 and 256"); + } + if iterations + .checked_mul(u64::try_from(concurrency)?) + .is_none_or(|count| count > 8_000_000) + { + bail!("at most 8000000 measured samples are supported"); + } + if payload_bytes == 0 || payload_bytes > 65_507 { + bail!("payload-bytes must be between 1 and 65507"); + } + Ok(()) +} + #[allow(clippy::cast_precision_loss)] pub fn run_worker( protocol: Protocol, @@ -195,39 +190,88 @@ pub fn run_worker( concurrency: usize, payload_bytes: usize, ) -> anyhow::Result { - let barrier = Arc::new(Barrier::new(concurrency + 1)); - let (sender, receiver) = mpsc::channel(); - let mut threads = Vec::with_capacity(concurrency); - for _ in 0..concurrency { - let barrier = Arc::clone(&barrier); - let sender = sender.clone(); - threads.push(thread::spawn(move || { - let result = worker_loop( - protocol, - target, - iterations, - warmup, - payload_bytes, - &barrier, + validate_options(iterations, concurrency, payload_bytes)?; + let (mut samples, elapsed) = thread::scope(|scope| -> anyhow::Result<_> { + let preparation_deadline = Instant::now() + START_TIMEOUT; + let (ready_tx, ready_rx) = mpsc::channel(); + let mut starters = Vec::with_capacity(concurrency); + let mut workers = Vec::with_capacity(concurrency); + for index in 0..concurrency { + let ready_tx = ready_tx.clone(); + let (start_tx, start_rx) = mpsc::channel(); + // Dropping these senders also releases ready workers when another + // worker fails preparation or thread creation. + starters.push(start_tx); + workers.push( + thread::Builder::new() + .name(format!("perf-worker-{index}")) + .spawn_scoped(scope, move || -> anyhow::Result> { + let payload = vec![0x5a; payload_bytes]; + let mut response = vec![0; payload_bytes]; + let prepared = (|| { + let mut connection = WorkerConnection::new(protocol, target)?; + for _ in 0..warmup { + if Instant::now() >= preparation_deadline { + return Err(io::Error::new( + io::ErrorKind::TimedOut, + "benchmark warmup exceeded preparation deadline", + )); + } + connection.operation(&payload, &mut response)?; + } + Ok::<_, io::Error>(connection) + })(); + let mut connection = match prepared { + Ok(connection) => { + ready_tx + .send(Ok(())) + .context("benchmark coordinator closed")?; + connection + } + Err(error) => { + let _ = ready_tx.send(Err(error.to_string())); + return Err(error.into()); + } + }; + start_rx.recv().context("benchmark start cancelled")?; + let mut samples = Vec::with_capacity(usize::try_from(iterations)?); + for _ in 0..iterations { + let started = Instant::now(); + connection.operation(&payload, &mut response)?; + samples.push(u64::try_from(started.elapsed().as_nanos())?); + } + Ok(samples) + })?, ); - let _ = sender.send(result); - })); - } - drop(sender); - let started = Instant::now(); - barrier.wait(); - let mut samples = Vec::new(); - for result in receiver { - samples.extend(result?); - } - let elapsed = started.elapsed(); - for worker in threads { - worker - .join() - .map_err(|_| anyhow::anyhow!("worker panicked"))?; - } + } + drop(ready_tx); + for _ in 0..concurrency { + ready_rx + .recv_timeout(preparation_deadline.saturating_duration_since(Instant::now())) + .context("benchmark preparation did not finish")? + .map_err(anyhow::Error::msg)?; + } + // Every connection and warmup is complete before the measured interval. + let started = Instant::now(); + for starter in starters { + starter + .send(()) + .context("benchmark worker exited before measurement")?; + } + let mut samples = Vec::new(); + for worker in workers { + samples.extend( + worker + .join() + .map_err(|_| anyhow::anyhow!("benchmark worker panicked"))??, + ); + } + Ok((samples, started.elapsed())) + })?; samples.sort_unstable(); - let operations = iterations.saturating_mul(concurrency as u64); + let operations = iterations + .checked_mul(u64::try_from(concurrency)?) + .ok_or_else(|| anyhow::anyhow!("operation count overflow"))?; let seconds = elapsed.as_secs_f64(); Ok(BenchmarkReport { layer: Layer::Native, @@ -238,9 +282,9 @@ pub fn run_worker( capability_scope: protocol.capability_scope().to_string(), elapsed_ms: seconds * 1_000.0, operations_per_second: operations as f64 / seconds, - throughput_mbit_per_second: operations as f64 * payload_bytes as f64 * 8.0 - / seconds - / 1_000_000.0, + // Count payload bytes sent, not an invented payload for connect-only work. + throughput_mbit_per_second: matches!(protocol, Protocol::TcpStream) + .then_some(operations as f64 * payload_bytes as f64 * 8.0 / seconds / 1_000_000.0), latency_ns_p50: percentile(&samples, 50), latency_ns_p95: percentile(&samples, 95), latency_ns_p99: percentile(&samples, 99), @@ -248,6 +292,13 @@ pub fn run_worker( } impl Protocol { + pub const fn cli_name(self) -> &'static str { + match self { + Self::TcpConnect => "tcp-connect", + Self::TcpStream => "tcp-stream", + } + } + const fn capability_scope(self) -> &'static str { match self { Self::TcpConnect => "implemented local TCP socket/connect interception", @@ -256,55 +307,40 @@ impl Protocol { } } -fn worker_loop( - protocol: Protocol, - target: SocketAddr, - iterations: u64, - warmup: u64, - payload_bytes: usize, - barrier: &Barrier, -) -> anyhow::Result> { - let payload = vec![0x5a; payload_bytes]; - let mut tcp = if matches!(protocol, Protocol::TcpStream) { - let stream = TcpStream::connect(target)?; - let _ = stream.set_nodelay(true); - Some(stream) - } else { - None - }; - let mut response = vec![0_u8; payload_bytes]; - for _ in 0..warmup { - one_operation(protocol, target, &payload, &mut response, tcp.as_mut())?; - } - barrier.wait(); - let mut samples = Vec::with_capacity(usize::try_from(iterations).unwrap_or(0)); - for _ in 0..iterations { - let started = Instant::now(); - one_operation(protocol, target, &payload, &mut response, tcp.as_mut())?; - samples.push(u64::try_from(started.elapsed().as_nanos()).unwrap_or(u64::MAX)); - } - Ok(samples) +enum WorkerConnection { + NewConnections(SocketAddr), + Stream(TcpStream), } -fn one_operation( - protocol: Protocol, - target: SocketAddr, - payload: &[u8], - response: &mut [u8], - tcp: Option<&mut TcpStream>, -) -> io::Result<()> { - match protocol { - Protocol::TcpConnect => { - let stream = TcpStream::connect(target)?; - let _ = stream.set_nodelay(true); +impl WorkerConnection { + fn new(protocol: Protocol, target: SocketAddr) -> io::Result { + match protocol { + Protocol::TcpConnect => Ok(Self::NewConnections(target)), + Protocol::TcpStream => Ok(Self::Stream(connect(target)?)), } - Protocol::TcpStream => { - let stream = tcp.expect("TCP stream initialized"); - stream.write_all(payload)?; - stream.read_exact(response)?; + } + + fn operation(&mut self, payload: &[u8], response: &mut [u8]) -> io::Result<()> { + match self { + Self::NewConnections(target) => { + connect(*target)?; + } + Self::Stream(stream) => { + stream.write_all(payload)?; + stream.read_exact(response)?; + } } + Ok(()) } - Ok(()) +} + +fn connect(target: SocketAddr) -> io::Result { + let stream = TcpStream::connect_timeout(&target, OPERATION_TIMEOUT)?; + // This worker deliberately uses blocking std sockets, not Tokio sockets. + stream.set_nodelay(true)?; + stream.set_read_timeout(Some(OPERATION_TIMEOUT))?; + stream.set_write_timeout(Some(OPERATION_TIMEOUT))?; + Ok(stream) } fn percentile(samples: &[u64], percentile: usize) -> u64 { @@ -312,7 +348,7 @@ fn percentile(samples: &[u64], percentile: usize) -> u64 { return 0; } let index = (samples.len() - 1) * percentile / 100; - samples[index] + samples.get(index).copied().unwrap_or(0) } #[cfg(test)] @@ -326,4 +362,50 @@ mod tests { assert_eq!(percentile(&samples, 95), 95); assert_eq!(percentile(&samples, 99), 99); } + + #[test] + fn preparation_failure_returns_instead_of_waiting_for_a_barrier() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let address = listener.local_addr().unwrap(); + drop(listener); + let started = Instant::now(); + assert!(run_worker(Protocol::TcpStream, address, 1, 1, 2, 8).is_err()); + assert!(started.elapsed() < Duration::from_secs(5)); + } + + #[test] + fn connection_only_reports_no_payload_throughput() { + let fixture = Fixture::start(Protocol::TcpConnect, None).unwrap(); + let report = run_worker(Protocol::TcpConnect, fixture.target, 4, 2, 2, 8).unwrap(); + assert_eq!(report.iterations, 8); + assert_eq!(report.throughput_mbit_per_second, None); + } + + #[test] + fn warmup_delay_is_excluded_from_measurement() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let address = listener.local_addr().unwrap(); + let server = thread::spawn(move || { + let (mut stream, _) = listener.accept().unwrap(); + stream.set_nodelay(true).unwrap(); + for round in 0..2 { + let mut byte = [0]; + stream.read_exact(&mut byte).unwrap(); + if round == 0 { + thread::sleep(Duration::from_millis(250)); + } + stream.write_all(&byte).unwrap(); + } + }); + let total = Instant::now(); + let report = run_worker(Protocol::TcpStream, address, 1, 1, 1, 1).unwrap(); + assert!( + total + .elapsed() + .as_secs_f64() + .mul_add(1000.0, -report.elapsed_ms) + >= 200.0 + ); + server.join().unwrap(); + } } diff --git a/e2e/rust/tests/live_internet_traffic_perf.rs b/e2e/rust/tests/live_internet_traffic_perf.rs index e8727bbce3..b338c2ddc8 100644 --- a/e2e/rust/tests/live_internet_traffic_perf.rs +++ b/e2e/rust/tests/live_internet_traffic_perf.rs @@ -5,6 +5,8 @@ #![cfg(feature = "e2e-host-gateway")] +use serde::{Deserialize, Serialize}; +use std::collections::{BTreeMap, BTreeSet}; use std::io::Write as _; use std::process::Stdio; use std::time::Instant; @@ -14,9 +16,388 @@ use tempfile::NamedTempFile; const DEFAULT_ROUNDS: usize = 10; +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)] +#[serde(rename_all = "kebab-case")] +enum BenchmarkMode { + DirectHost, + Sandbox, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +enum MetricName { + DnsServiceSet, + HttpsMetadataSerial, + HttpsMetadataConcurrent, + HttpsReuse, + GitClone, + PackageDownload, + #[serde(rename = "bulk_download_5mib")] + BulkDownload5mib, + PolicyDenial, +} + +const METRICS: [MetricName; 8] = [ + MetricName::DnsServiceSet, + MetricName::HttpsMetadataSerial, + MetricName::HttpsMetadataConcurrent, + MetricName::HttpsReuse, + MetricName::GitClone, + MetricName::PackageDownload, + MetricName::BulkDownload5mib, + MetricName::PolicyDenial, +]; + +#[derive(Debug, Serialize, Deserialize)] +#[serde(tag = "status", rename_all = "snake_case")] +enum Outcome { + Success { elapsed_ms: f64, detail: T }, + Failed { elapsed_ms: f64, error: String }, + Skipped { reason: String }, +} + +impl Outcome { + fn summary(&self) -> Result, &str> { + match self { + Self::Success { elapsed_ms, .. } => Ok(Some(*elapsed_ms)), + Self::Failed { error, .. } => Err(error), + Self::Skipped { .. } => Ok(None), + } + } +} + +#[derive(Debug, Serialize, Deserialize)] +struct HttpMeasurement { + status: u16, + bytes: u64, + first_byte_ms: f64, + total_ms: f64, +} +#[derive(Debug, Serialize, Deserialize)] +struct LookupMeasurement { + host: String, + elapsed_ms: f64, + addresses: usize, +} +#[derive(Debug, Serialize, Deserialize)] +struct DnsMeasurement { + lookups: Vec, + p50_ms: f64, + p95_ms: f64, +} +#[derive(Debug, Serialize, Deserialize)] +struct RequestsMeasurement { + requests: Vec, +} +#[derive(Debug, Serialize, Deserialize)] +struct ReuseMeasurement { + requests: usize, + p50_ms: f64, + p95_ms: f64, + mean_ms: f64, +} +#[derive(Debug, Serialize, Deserialize)] +struct CloneMeasurement { + files: usize, + checkout_bytes: u64, +} +#[derive(Debug, Serialize, Deserialize)] +struct PackageMeasurement { + metadata: HttpMeasurement, + artifact: HttpMeasurement, + filename: String, +} +#[derive(Debug, Serialize, Deserialize)] +struct DenialMeasurement { + denied: bool, + latency_ms: f64, + error: String, +} + +#[derive(Debug, Serialize, Deserialize)] +#[serde(tag = "name", rename_all = "snake_case")] +enum Metric { + DnsServiceSet(Outcome), + HttpsMetadataSerial(Outcome), + HttpsMetadataConcurrent(Outcome), + HttpsReuse(Outcome), + GitClone(Outcome), + PackageDownload(Outcome), + #[serde(rename = "bulk_download_5mib")] + BulkDownload5mib(Outcome), + PolicyDenial(Outcome), +} + +impl Metric { + fn summary(&self) -> (MetricName, Result, &str>) { + match self { + Self::DnsServiceSet(sample) => (MetricName::DnsServiceSet, sample.summary()), + Self::HttpsMetadataSerial(sample) => { + (MetricName::HttpsMetadataSerial, sample.summary()) + } + Self::HttpsMetadataConcurrent(sample) => { + (MetricName::HttpsMetadataConcurrent, sample.summary()) + } + Self::HttpsReuse(sample) => (MetricName::HttpsReuse, sample.summary()), + Self::GitClone(sample) => (MetricName::GitClone, sample.summary()), + Self::PackageDownload(sample) => (MetricName::PackageDownload, sample.summary()), + Self::BulkDownload5mib(sample) => (MetricName::BulkDownload5mib, sample.summary()), + Self::PolicyDenial(Outcome::Success { detail, .. }) if !detail.denied => { + (MetricName::PolicyDenial, Err("destination was not denied")) + } + Self::PolicyDenial(sample) => (MetricName::PolicyDenial, sample.summary()), + } + } +} + +#[derive(Debug, Serialize, Deserialize)] +struct LiveReport { + schema: String, + mode: BenchmarkMode, + total_ms: f64, + metrics: Vec, +} + +#[derive(Default, Debug, Serialize)] +struct SampleCounts { + successful: usize, + failed: usize, + skipped: usize, +} + +#[derive(Default)] +struct Coverage { + counts: BTreeMap<(BenchmarkMode, MetricName), SampleCounts>, +} + +impl Coverage { + fn record(&mut self, report: &LiveReport, expected: BenchmarkMode) -> Result<(), String> { + if report.schema != "openshell.live-internet-perf.v2" + || report.mode != expected + || !report.total_ms.is_finite() + || report.total_ms < 0.0 + { + return Err("unexpected benchmark schema, mode, or duration".to_string()); + } + let names = report + .metrics + .iter() + .map(|metric| metric.summary().0) + .collect::>(); + if report.metrics.len() != METRICS.len() || names != METRICS.into_iter().collect() { + return Err("benchmark scenarios are missing, duplicated, or unknown".to_string()); + } + for metric in &report.metrics { + let (name, sample) = metric.summary(); + let counts = self.counts.entry((expected, name)).or_default(); + match sample { + Ok(Some(duration)) if duration.is_finite() && duration >= 0.0 => { + counts.successful += 1 + } + Ok(Some(_)) => return Err("invalid successful sample duration".to_string()), + Ok(None) => counts.skipped += 1, + Err(_) => counts.failed += 1, + } + } + Ok(()) + } + + fn require_successful_coverage(&self) -> Result<(), String> { + for mode in [BenchmarkMode::DirectHost, BenchmarkMode::Sandbox] { + for name in METRICS { + if mode == BenchmarkMode::DirectHost && name == MetricName::PolicyDenial { + continue; + } + let counts = self.counts.get(&(mode, name)); + if counts.is_none_or(|counts| counts.successful == 0) { + return Err(format!( + "no successful samples for {mode:?}/{name:?}: {counts:?}" + )); + } + if name == MetricName::PolicyDenial + && counts.is_some_and(|counts| counts.failed > 0 || counts.skipped > 0) + { + return Err( + "sandbox denial was not explicitly verified in every round".to_string() + ); + } + } + } + Ok(()) + } +} + +fn record_report( + coverage: &mut Coverage, + output: &str, + expected: BenchmarkMode, + round: usize, +) -> Result<(), String> { + let report: LiveReport = serde_json::from_str(output.trim()) + .map_err(|error| format!("decode benchmark report: {error}"))?; + coverage.record(&report, expected)?; + // Failed durations remain diagnostic only. Consumers must compare Success + // samples, never total_ms (which also includes time spent in failed work). + println!( + "LIVE_INTERNET_PERF {expected:?} round={round} {}", + serde_json::to_string(&report).map_err(|error| error.to_string())? + ); + for ((mode, metric), counts) in &coverage.counts { + println!( + "LIVE_INTERNET_COVERAGE {mode:?}/{metric:?} {}", + serde_json::to_string(counts).map_err(|error| error.to_string())? + ); + } + Ok(()) +} + +#[cfg(test)] +mod report_tests { + use super::*; + + fn success(detail: T) -> Outcome { + Outcome::Success { + elapsed_ms: 1.0, + detail, + } + } + + fn http() -> HttpMeasurement { + HttpMeasurement { + status: 200, + bytes: 5_242_880, + first_byte_ms: 0.5, + total_ms: 1.0, + } + } + + fn report(mode: BenchmarkMode) -> LiveReport { + LiveReport { + schema: "openshell.live-internet-perf.v2".to_string(), + mode, + total_ms: 8.0, + metrics: vec![ + Metric::DnsServiceSet(success(DnsMeasurement { + lookups: vec![], + p50_ms: 1.0, + p95_ms: 1.0, + })), + Metric::HttpsMetadataSerial(success(RequestsMeasurement { + requests: vec![http()], + })), + Metric::HttpsMetadataConcurrent(success(RequestsMeasurement { + requests: vec![http()], + })), + Metric::HttpsReuse(success(ReuseMeasurement { + requests: 20, + p50_ms: 1.0, + p95_ms: 1.0, + mean_ms: 1.0, + })), + Metric::GitClone(success(CloneMeasurement { + files: 3, + checkout_bytes: 20, + })), + Metric::PackageDownload(success(PackageMeasurement { + metadata: http(), + artifact: http(), + filename: "sample.whl".to_string(), + })), + Metric::BulkDownload5mib(success(http())), + Metric::PolicyDenial(if mode == BenchmarkMode::Sandbox { + success(DenialMeasurement { + denied: true, + latency_ms: 1.0, + error: "EPERM".to_string(), + }) + } else { + Outcome::Skipped { + reason: "not applicable".to_string(), + } + }), + ], + } + } + + #[test] + fn typed_report_round_trip_preserves_scenario_names_and_coverage() { + let mut coverage = Coverage::default(); + for mode in [BenchmarkMode::DirectHost, BenchmarkMode::Sandbox] { + let encoded = serde_json::to_string(&report(mode)).unwrap(); + assert!(encoded.contains("bulk_download_5mib")); + let decoded: LiveReport = serde_json::from_str(&encoded).unwrap(); + coverage.record(&decoded, mode).unwrap(); + } + coverage.require_successful_coverage().unwrap(); + } + + #[test] + fn failure_is_not_a_fast_successful_sample() { + let mut coverage = Coverage::default(); + let mut failed = report(BenchmarkMode::Sandbox); + failed.metrics[0] = Metric::DnsServiceSet(Outcome::Failed { + elapsed_ms: 0.01, + error: "unreachable".to_string(), + }); + coverage + .record( + &report(BenchmarkMode::DirectHost), + BenchmarkMode::DirectHost, + ) + .unwrap(); + coverage.record(&failed, BenchmarkMode::Sandbox).unwrap(); + assert!(coverage.require_successful_coverage().is_err()); + let counts = &coverage.counts[&(BenchmarkMode::Sandbox, MetricName::DnsServiceSet)]; + assert_eq!( + (counts.successful, counts.failed, counts.skipped), + (0, 1, 0) + ); + } + + #[test] + fn denial_must_be_explicit_in_every_round() { + let mut coverage = Coverage::default(); + coverage + .record( + &report(BenchmarkMode::DirectHost), + BenchmarkMode::DirectHost, + ) + .unwrap(); + coverage + .record(&report(BenchmarkMode::Sandbox), BenchmarkMode::Sandbox) + .unwrap(); + let mut invalid = report(BenchmarkMode::Sandbox); + invalid.metrics[7] = Metric::PolicyDenial(success(DenialMeasurement { + denied: false, + latency_ms: 1.0, + error: String::new(), + })); + coverage.record(&invalid, BenchmarkMode::Sandbox).unwrap(); + assert!(coverage.require_successful_coverage().is_err()); + } + + #[test] + fn incomplete_or_mislabeled_reports_are_rejected() { + let mut coverage = Coverage::default(); + let mut invalid = report(BenchmarkMode::Sandbox); + assert!( + coverage + .record(&invalid, BenchmarkMode::DirectHost) + .is_err() + ); + invalid.metrics.pop(); + assert!(coverage.record(&invalid, BenchmarkMode::Sandbox).is_err()); + invalid + .metrics + .push(Metric::BulkDownload5mib(success(http()))); + assert!(coverage.record(&invalid, BenchmarkMode::Sandbox).is_err()); + } +} + const BENCHMARK: &str = r#" import concurrent.futures import http.client +import errno import json import os import shutil @@ -28,7 +409,7 @@ import tempfile import time import urllib.parse -MODE = os.environ.get("OPENSHELL_LIVE_PERF_MODE", "direct") +MODE = os.environ.get("OPENSHELL_LIVE_PERF_MODE", "direct-host") TLS = ssl.create_default_context() USER_AGENT = "OpenShell-live-Internet-benchmark/1" @@ -73,24 +454,29 @@ def request(url, method="GET", max_bytes=None): total_ms = (time.perf_counter_ns() - started) / 1_000_000 status = response.status connection.close() - if status < 200 or status >= 400: + if status < 200 or status >= 300: raise RuntimeError(f"{url} returned HTTP {status}") return {"status": status, "bytes": body_bytes, "first_byte_ms": first_byte_ms, "total_ms": total_ms} +class BenchmarkSkip(Exception): + pass + def metric(name, operation): started = time.perf_counter_ns() try: detail = operation() return { "name": name, - "ok": True, + "status": "success", "elapsed_ms": (time.perf_counter_ns() - started) / 1_000_000, "detail": detail, } + except BenchmarkSkip as reason: + return {"name": name, "status": "skipped", "reason": str(reason)} except Exception as error: return { "name": name, - "ok": False, + "status": "failed", "elapsed_ms": (time.perf_counter_ns() - started) / 1_000_000, "error": f"{type(error).__name__}: {error}", } @@ -144,7 +530,7 @@ def https_reuse(): def git_clone(): if shutil.which("git") is None: - return {"skipped": "git is not installed in the workload image"} + raise BenchmarkSkip("git is not installed in the workload image") with tempfile.TemporaryDirectory(prefix="openshell-live-git-") as directory: checkout = os.path.join(directory, "sampleproject") command = [ @@ -173,15 +559,20 @@ def package_download(): return {"metadata": metadata, "artifact": artifact, "filename": wheels[0]["filename"]} def bulk_download(): - return request("https://speed.cloudflare.com/__down?bytes=5242880", max_bytes=5242880) + result = request("https://speed.cloudflare.com/__down?bytes=5242880", max_bytes=5242880) + if result["bytes"] != 5242880: + raise RuntimeError(f"expected 5 MiB, received {result['bytes']} bytes") + return result def denied_destination(): if MODE != "sandbox": - return {"skipped": "policy denial applies only to sandbox mode"} + raise BenchmarkSkip("policy denial applies only to sandbox mode") started = time.perf_counter_ns() try: socket.create_connection(("example.net", 443), timeout=5).close() except OSError as error: + if error.errno not in (errno.EACCES, errno.EPERM): + raise RuntimeError(f"connect failed without an explicit policy denial: {error}") from error return { "denied": True, "latency_ms": (time.perf_counter_ns() - started) / 1_000_000, @@ -201,14 +592,12 @@ metrics = [ metric("policy_denial", denied_destination), ] document = { - "schema": "openshell.live-internet-perf.v1", + "schema": "openshell.live-internet-perf.v2", "mode": MODE, "total_ms": (time.perf_counter_ns() - started) / 1_000_000, "metrics": metrics, } print(json.dumps(document, separators=(",", ":"))) -if any(item["name"] == "policy_denial" and not item["ok"] for item in metrics): - raise SystemExit(1) "#; fn rounds() -> usize { @@ -257,7 +646,7 @@ network_policies: } async fn run_direct() -> Result { - run_python("direct").await + run_python("direct-host").await } async fn run_python(mode: &str) -> Result { @@ -280,7 +669,7 @@ async fn run_python(mode: &str) -> Result { output.status.code() )); } - Ok(combined) + Ok(String::from_utf8_lossy(&output.stdout).into_owned()) } async fn run_sandbox(sandbox: &SandboxGuard) -> Result { @@ -309,29 +698,29 @@ async fn benchmark_live_internet_agent_traffic() { create_started.elapsed().as_secs_f64() * 1000.0 ); + let mut coverage = Coverage::default(); for round in 1..=rounds() { if round % 2 == 1 { let direct = run_direct().await.expect("direct live Internet benchmark"); - println!("LIVE_INTERNET_PERF direct round={round} {}", direct.trim()); + record_report(&mut coverage, &direct, BenchmarkMode::DirectHost, round) + .expect("valid direct-host report"); let mediated = run_sandbox(&sandbox) .await .expect("sandbox live Internet benchmark"); - println!( - "LIVE_INTERNET_PERF sandbox round={round} {}", - mediated.trim() - ); + record_report(&mut coverage, &mediated, BenchmarkMode::Sandbox, round) + .expect("valid sandbox report"); } else { let mediated = run_sandbox(&sandbox) .await .expect("sandbox live Internet benchmark"); - println!( - "LIVE_INTERNET_PERF sandbox round={round} {}", - mediated.trim() - ); + record_report(&mut coverage, &mediated, BenchmarkMode::Sandbox, round) + .expect("valid sandbox report"); let direct = run_direct().await.expect("direct live Internet benchmark"); - println!("LIVE_INTERNET_PERF direct round={round} {}", direct.trim()); + record_report(&mut coverage, &direct, BenchmarkMode::DirectHost, round) + .expect("valid direct-host report"); } } sandbox.cleanup().await; + coverage.require_successful_coverage().expect("benchmark needs successful samples for every scenario; failed/skipped samples are not performance results"); } From bdb0acc96fa0f8fe109ba8dac3e1a0a1696c2a85 Mon Sep 17 00:00:00 2001 From: Drew Newberry Date: Tue, 8 Sep 2026 21:31:43 -0700 Subject: [PATCH 3/3] refactor(perf): remove unused fixture context Signed-off-by: Drew Newberry --- crates/openshell-sandbox/src/perf.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/crates/openshell-sandbox/src/perf.rs b/crates/openshell-sandbox/src/perf.rs index 484a27d640..cc31dde24a 100644 --- a/crates/openshell-sandbox/src/perf.rs +++ b/crates/openshell-sandbox/src/perf.rs @@ -76,7 +76,7 @@ pub fn run(options: BenchmarkOptions) -> anyhow::Result> { filtered_runtime = Some((launcher, broker)); } for protocol in &options.protocols { - let fixture = Fixture::start(*protocol, filtered_runtime.as_ref().map(|(_, b)| b))?; + let fixture = Fixture::start(*protocol)?; let mut command = Command::new(&executable); command .arg("worker") @@ -120,7 +120,7 @@ struct Fixture { } impl Fixture { - fn start(protocol: Protocol, _broker: Option<&NetworkBroker>) -> io::Result { + fn start(protocol: Protocol) -> io::Result { match protocol { Protocol::TcpConnect => start_tcp_fixture(false), Protocol::TcpStream => start_tcp_fixture(true),