diff --git a/.github/workflows/overhead-inspection.yml b/.github/workflows/overhead-inspection.yml new file mode 100644 index 00000000..34037989 --- /dev/null +++ b/.github/workflows/overhead-inspection.yml @@ -0,0 +1,30 @@ +name: Runtime overhead contracts +on: + pull_request: + paths: + - 'data_plane/**' + - 'crates/asap_types/**' + - 'Cargo.toml' + - 'Cargo.lock' + - '.github/workflows/overhead-inspection.yml' + workflow_dispatch: +jobs: + contracts: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: dtolnay/rust-toolchain@stable + - uses: Swatinem/rust-cache@v2 + - run: sudo apt-get update && sudo apt-get install -y protobuf-compiler + - run: cargo test --locked -p data_plane --lib runtime_config::tests + - run: cargo test --locked -p data_plane --example overhead_inspect + - run: >- + cargo run --locked -p data_plane --example overhead_inspect -- + --workers 1,2 --concurrency 1,2 --requests 20 --warmup 2 --repeats 1 + --log-level off --disable-console-log --disable-file-log + --output target/overhead-smoke + - uses: actions/upload-artifact@v4 + if: always() + with: + name: overhead-smoke + path: target/overhead-smoke diff --git a/Cargo.lock b/Cargo.lock index 1e6eed93..28b16752 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1202,6 +1202,7 @@ dependencies = [ "hex", "http-body-util", "lazy_static", + "libc", "memmap2", "moka", "prometheus", diff --git a/data_plane/Cargo.toml b/data_plane/Cargo.toml index 289105b9..6f074e19 100644 --- a/data_plane/Cargo.toml +++ b/data_plane/Cargo.toml @@ -87,6 +87,7 @@ fs2 = "0.4" # none of them. [dev-dependencies] +libc = "0.2" asap-aware-mapping.workspace = true tempfile = "3.20.0" criterion = { version = "0.5", features = ["html_reports"] } diff --git a/data_plane/examples/overhead/fixture.rs b/data_plane/examples/overhead/fixture.rs new file mode 100644 index 00000000..db501079 --- /dev/null +++ b/data_plane/examples/overhead/fixture.rs @@ -0,0 +1,152 @@ +use anyhow::{ensure, Result}; +use asap_sketchlib::{DdSketch, MessagePackCodec}; +use data_plane::drivers::query::{ + adapters::AdapterConfig, servers::http::validate_and_build_runtime_plan, +}; +use data_plane::query_engines::{ + routing::query_engine_routing::QueryEngine, QueryForwardingPolicy, QueryResult, +}; +use data_plane::storage_engines::{ + sketch_db::index::*, + types::{ActivePhysicalPlanHandle, BackendStorageRouting}, +}; +use data_plane::{ASAPQueryEngine, HttpServer, HttpServerConfig}; +use std::{ + collections::{BTreeMap, BTreeSet}, + sync::Arc, +}; + +#[path = "../../tests/support/physical_fixture.rs"] +#[allow(dead_code)] +mod physical_fixture; + +pub const QUERY: &str = "quantile_over_time(0.5, overhead_values[1s])"; +pub const TIME: u64 = 600_000; +pub struct Fixture { + pub raw: Vec, + pub sketch: DdSketch, + pub expected: f64, + pub exact_expected: f64, + pub engine: Arc, + pub server: HttpServer, +} +pub fn exact(values: &[f64]) -> f64 { + let mut sorted = values.to_vec(); + let middle = sorted.len() / 2; + let (_, upper, _) = sorted.select_nth_unstable_by(middle, f64::total_cmp); + let upper = *upper; + if sorted.len().is_multiple_of(2) { + (sorted[..middle] + .iter() + .copied() + .max_by(f64::total_cmp) + .unwrap() + + upper) + / 2.0 + } else { + upper + } +} +impl Fixture { + pub fn new(samples: usize) -> Result { + ensure!(samples > 0, "samples must be positive"); + // Deliberately unsorted, reproducible input; exact includes scratch allocation + // and selection, while prebuilt sketch construction is outside measurement. + let raw: Vec<_> = (0..samples) + .map(|i| 1.0 + ((i * 7919) % 10007) as f64) + .collect(); + let mut sketch = DdSketch::new(0.01); + for value in &raw { + sketch.update(*value); + } + let expected = sketch.quantile(0.5).unwrap(); + let exact_expected = exact(&raw); + ensure!( + (expected - exact(&raw)).abs() <= exact(&raw) * 0.02, + "sketch accuracy gate failed" + ); + let config: asap_types::PrecomputeMaterialization = serde_json::from_value( + serde_json::json!({ + "aggregation_type":"DDSketch", "aggregation_sub_type":"", "metric":"overhead_values", + "window_size":1,"slide_interval":1,"window_type":"tumbling","num_aggregates_to_retain":10, + "parameters":{"alpha":0.01},"pane_origin_ms":0,"partitioning":"per_entity", + "window_layout":{"kind":"pane","pane_secs":1},"grouping_labels":{"labels":[]}, + "aggregated_labels":{"labels":[]},"rollup_labels":{"labels":[]}, + "spatial_filter":"","spatial_filter_normalized":"","original_yaml":"" + }), + )?; + let plan = physical_fixture::artifact_from_materializations(vec![config.clone()]); + let store = Arc::new(SketchStore::new()); + store + .install_precompute_plan( + Arc::new(plan.summary_catalog.clone()), + &plan.precompute_plan, + ) + .map_err(anyhow::Error::msg)?; + let skconfig = SketchConfig::DDSketch { + relative_accuracy: 0.01, + }; + store.register(SummarySeriesMetadata { + storage_handle: 1, + metric_name: "overhead_values".into(), + group_by_keys: BTreeSet::new(), + capability: Some(Capability::QuantileApprox(Some(SketchAlgorithm::DDSketch))), + accuracy: Some(AccuracyBound::from_config(&skconfig)), + agg_kind: AggKind::Sketch { + algorithm: SketchAlgorithm::DDSketch, + config: skconfig, + spatial_filter_canonical: String::new(), + }, + first_seen_unix_ms: 0, + retired_at_ms: None, + expires_at_ms: None, + policy_fp: config.policy_fingerprint(), + }); + ensure!( + store.append_sample( + 1, + BTreeMap::new(), + (TIME - 1000, TIME), + SketchSampleState { + bytes: sketch.to_msgpack().map_err(|e| anyhow::anyhow!("{e}"))?, + encoding: SketchEncoding::MsgpackFull, + } + ), + "sample admission failed" + ); + let active = ActivePhysicalPlanHandle::new( + validate_and_build_runtime_plan(plan, Arc::new(BackendStorageRouting::empty())) + .map_err(anyhow::Error::msg)?, + ); + let engine = Arc::new( + ASAPQueryEngine::new(1000) + .with_sketch_index(store.clone()) + .with_active_physical_plan(active) + .with_query_forwarding_policy(QueryForwardingPolicy::Disabled), + ); + let server = HttpServer::new( + HttpServerConfig { + port: 0, + handle_http_requests: true, + adapter_config: AdapterConfig::prometheus_promql(String::new(), false) + .with_query_forwarding_policy(QueryForwardingPolicy::Disabled), + }, + engine.clone(), + store, + ); + Ok(Self { + raw, + sketch, + expected, + exact_expected, + engine, + server, + }) + } + pub async fn backend(&self) -> Result { + match self.engine.execute_at(QUERY, TIME).await? { + QueryResult::Vector(v) if v.values.len() == 1 => Ok(v.values[0].value), + other => anyhow::bail!("unexpected result {other:?}"), + } + } +} diff --git a/data_plane/examples/overhead_inspect.rs b/data_plane/examples/overhead_inspect.rs new file mode 100644 index 00000000..37894356 --- /dev/null +++ b/data_plane/examples/overhead_inspect.rs @@ -0,0 +1,390 @@ +//! Three-layer overhead inspection; each matrix cell runs in a fresh process. +#[path = "overhead/fixture.rs"] +mod fixture; +use anyhow::{ensure, Context, Result}; +use clap::Parser; +use data_plane::runtime_config::{process_snapshot, LogConfig, RuntimeConfig}; +use serde_json::{json, Value}; +use std::{ + path::PathBuf, + sync::Arc, + time::{Duration, Instant}, +}; +use tokio::task::JoinSet; + +#[derive(Parser, Debug)] +struct Args { + #[command(flatten)] + runtime: RuntimeConfig, + #[command(flatten)] + logging: LogConfig, + #[arg(long, default_value = "1,2,4,8,16", value_delimiter = ',')] + workers: Vec, + #[arg(long, default_value = "1,2,4,8,16", value_delimiter = ',')] + concurrency: Vec, + #[arg( + long, + default_value = "sketch,raw_exact,backend,http", + value_delimiter = ',' + )] + layers: Vec, + #[arg(long, default_value_t = 10000)] + samples: usize, + #[arg(long, default_value_t = 1000)] + requests: usize, + #[arg(long, default_value_t = 100)] + warmup: usize, + #[arg(long, default_value_t = 3)] + repeats: usize, + /// Optional open-loop arrival rate. At the concurrency cap requests are dropped and counted. + #[arg(long)] + rate: Option, + #[arg(long, default_value = "target/overhead")] + output: PathBuf, + #[arg(long, hide = true)] + cell: bool, +} +fn cpu(clock: libc::clockid_t) -> f64 { + let mut time = libc::timespec { + tv_sec: 0, + tv_nsec: 0, + }; + // CLOCK_PROCESS/THREAD_CPUTIME_ID read only initialized stack memory. + assert_eq!(unsafe { libc::clock_gettime(clock, &mut time) }, 0); + time.tv_sec as f64 + time.tv_nsec as f64 / 1e9 +} +fn percentile(xs: &[f64], q: f64) -> Option { + if xs.is_empty() { + None + } else { + Some( + xs[((xs.len() as f64 * q).ceil() as usize) + .saturating_sub(1) + .min(xs.len() - 1)], + ) + } +} +fn main() -> Result<()> { + let args = Args::parse(); + ensure!( + args.requests > 0 + && args.samples > 0 + && args.repeats > 0 + && !args.workers.is_empty() + && !args.concurrency.is_empty(), + "empty experiment" + ); + ensure!( + args.workers.iter().chain(&args.concurrency).all(|n| *n > 0), + "workers and concurrency must be positive" + ); + ensure!( + args.rate.is_none_or(|r| r.is_finite() && r > 0.0), + "rate must be finite and positive" + ); + ensure!( + !args.layers.is_empty() + && args + .layers + .iter() + .all(|l| matches!(l.as_str(), "sketch" | "raw_exact" | "backend" | "http")), + "unknown layer" + ); + std::fs::create_dir_all(&args.output)?; + if args.cell { + return cell(&args); + } + let mut reports = Vec::new(); + for worker in &args.workers { + for concurrency in &args.concurrency { + for layer in &args.layers { + for repeat in 0..args.repeats { + let dir = args + .output + .join(format!("{layer}-w{worker}-c{concurrency}-r{repeat}")); + let mut cmd = std::process::Command::new(std::env::current_exe()?); + cmd.args([ + "--cell", + "--runtime-workers", + &worker.to_string(), + "--runtime-max-blocking-threads", + &args.runtime.runtime_max_blocking_threads.to_string(), + "--concurrency", + &concurrency.to_string(), + "--layers", + layer, + "--samples", + &args.samples.to_string(), + "--requests", + &args.requests.to_string(), + "--warmup", + &args.warmup.to_string(), + "--log-level", + &args.logging.log_level, + "--output", + ]) + .arg(&dir); + if args.logging.disable_console_log { + cmd.arg("--disable-console-log"); + } + if args.logging.disable_file_log { + cmd.arg("--disable-file-log"); + } + if let Some(rate) = args.rate { + cmd.args(["--rate", &rate.to_string()]); + } + ensure!(cmd.status()?.success(), "cell failed: {}", dir.display()); + reports.push(serde_json::from_slice::(&std::fs::read( + dir.join("report.json"), + )?)?); + std::fs::write( + args.output.join("matrix.json"), + serde_json::to_vec_pretty(&reports)?, + )?; + } + } + } + } + let mut csv = String::from("layer,workers,concurrency,completed,dropped,error_rate,completed_per_second,p50_seconds,p95_seconds,p99_seconds,server_and_aux_cpu_seconds_per_completed\n"); + for report in &reports { + let fields = [ + "layer", + "workers", + "concurrency", + "completed", + "dropped", + "error_rate", + "completed_per_second", + "p50_seconds", + "p95_seconds", + "p99_seconds", + "server_and_aux_cpu_seconds_per_completed", + ]; + csv.push_str( + &fields + .iter() + .map(|key| match &report[*key] { + Value::String(s) => s.clone(), + value => value.to_string(), + }) + .collect::>() + .join(","), + ); + csv.push('\n'); + } + std::fs::write(args.output.join("matrix.csv"), csv)?; + Ok(()) +} + +async fn operation( + layer: &str, + fixture: Arc, + handle: tokio::runtime::Handle, + client: reqwest::Client, + url: String, +) -> Result<()> { + let expected = if layer == "raw_exact" { + fixture.exact_expected + } else { + fixture.expected + }; + let value = if layer == "http" { + let response = client + .get(url) + .query(&[("query", fixture::QUERY), ("time", "600")]) + .send() + .await? + .error_for_status()?; + let source = response + .headers() + .get("x-asap-data-source") + .and_then(|h| h.to_str().ok()) + .unwrap_or("") + .to_owned(); + ensure!(!source.contains("fallback"), "external fallback: {source}"); + let v: Value = response.json().await?; + ensure!( + v["status"] == "success" + && v["data"]["result"].as_array().is_some_and(|r| r.len() == 1), + "bad HTTP result: {v}" + ); + v["data"]["result"][0]["value"][1] + .as_str() + .context("missing value")? + .parse::()? + } else { + let layer = layer.to_owned(); + handle + .spawn(async move { + match layer.as_str() { + "sketch" => Ok(std::hint::black_box( + fixture.sketch.quantile(std::hint::black_box(0.5)).unwrap(), + )), + "raw_exact" => Ok(std::hint::black_box(fixture::exact(std::hint::black_box( + &fixture.raw, + )))), + _ => fixture.backend().await, + } + }) + .await?? + }; + ensure!( + value.is_finite() && (value - expected).abs() <= expected.abs() * 1e-10, + "result mismatch {value} != {expected}" + ); + Ok(()) +} + +fn cell(args: &Args) -> Result<()> { + ensure!( + args.layers.len() == 1 && args.concurrency.len() == 1, + "one layer/concurrency per child" + ); + let _guard = args.logging.init(&args.output)?; + let runtime = args.runtime.build()?; + let fixture = Arc::new(fixture::Fixture::new(args.samples)?); + let port = runtime + .block_on(fixture.server.start_test_server()) + .map_err(|e| anyhow::anyhow!("{e}"))?; + // Driver futures run on the calling OS thread, separate from server workers. + let driver = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build()?; + let handle = runtime.handle().clone(); + let layer = &args.layers[0]; + let concurrency = args.concurrency[0]; + let client = reqwest::Client::builder() + .timeout(Duration::from_secs(30)) + .build()?; + let url = format!("http://127.0.0.1:{port}/api/v1/query"); + let before = process_snapshot(); + let revision = std::process::Command::new("git") + .args(["rev-parse", "HEAD"]) + .output() + .ok() + .filter(|o| o.status.success()) + .map(|o| String::from_utf8_lossy(&o.stdout).trim().to_owned()); + let result=driver.block_on(async { + for _ in 0..args.warmup { operation(layer,fixture.clone(),handle.clone(),client.clone(),url.clone()).await?; } + // Direct synchronous kernel timing excludes task dispatch. Concurrent layer + // samples below deliberately include dispatch, reported separately. + let kernel = if matches!(layer.as_str(),"sketch"|"raw_exact") { + let start=Instant::now(); + for _ in 0..args.requests { + if layer=="sketch" {std::hint::black_box(fixture.sketch.quantile(std::hint::black_box(0.5)));} + else {std::hint::black_box(fixture::exact(std::hint::black_box(&fixture.raw)));} + } + Some(start.elapsed().as_secs_f64()/args.requests as f64) + } else {None}; + let start_cpu=cpu(libc::CLOCK_PROCESS_CPUTIME_ID); + let start_driver_cpu=cpu(libc::CLOCK_THREAD_CPUTIME_ID); + let start=Instant::now(); + let mut tasks=JoinSet::new(); + let mut latencies=Vec::new(); let mut service_latencies=Vec::new();let mut errors=Vec::new();let mut dropped=0; + for n in 0..args.requests { + let scheduled=args.rate.map(|r|start+Duration::from_secs_f64(n as f64/r)); + if let Some(at)=scheduled {tokio::time::sleep_until(at.into()).await;} + while let Some(result)=tasks.try_join_next() {record(result?,&mut latencies,&mut service_latencies,&mut errors);} + if tasks.len()>=concurrency { + if args.rate.is_some() {dropped+=1;continue;} + if let Some(result)=tasks.join_next().await {record(result?,&mut latencies,&mut service_latencies,&mut errors);} + } + let fixture=fixture.clone();let handle=handle.clone();let client=client.clone();let url=url.clone();let layer=layer.clone(); + tasks.spawn(async move { + let sent=Instant::now(); + let result=operation(&layer,fixture,handle,client,url).await; + (scheduled.unwrap_or(sent).elapsed().as_secs_f64(),sent.elapsed().as_secs_f64(),result.err().map(|e|e.to_string())) + }); + } + while let Some(result)=tasks.join_next().await {record(result?,&mut latencies,&mut service_latencies,&mut errors);} + let elapsed=start.elapsed().as_secs_f64(); + let total_cpu=cpu(libc::CLOCK_PROCESS_CPUTIME_ID)-start_cpu; + let driver_cpu=cpu(libc::CLOCK_THREAD_CPUTIME_ID)-start_driver_cpu; + latencies.sort_by(f64::total_cmp);service_latencies.sort_by(f64::total_cmp); + let completed=latencies.len(); + Ok::<_,anyhow::Error>(json!({"layer":layer,"warmup_requests":args.warmup,"sketch_alpha":0.01,"quantile":0.5,"series":1,"workers":runtime.metrics().num_workers(),"max_blocking_threads":args.runtime.runtime_max_blocking_threads, + "concurrency":concurrency,"samples":args.samples,"offered_requests":args.requests,"completed":completed,"dropped":dropped,"admitted":args.requests-dropped, + "errors":errors,"error_rate":errors.len() as f64/args.requests as f64,"offered_rate":args.rate, + "actual_offered_per_second":args.requests as f64/elapsed,"actual_admitted_per_second":(args.requests-dropped) as f64/elapsed,"completed_per_second":completed as f64/elapsed, + "elapsed_seconds":elapsed,"p50_seconds":percentile(&latencies,0.5),"p95_seconds":percentile(&latencies,0.95),"p99_seconds":percentile(&latencies,0.99), + "latency_seconds":latencies,"service_latency_seconds":service_latencies,"direct_kernel_seconds_per_query":kernel, + "process_cpu_seconds":total_cpu,"driver_cpu_seconds":driver_cpu,"server_and_aux_cpu_seconds_per_completed":if completed>0 {Some((total_cpu-driver_cpu).max(0.0)/completed as f64)}else{None}, + "logging":args.logging,"effective_log_filter":args.logging.filter(),"before":before,"after":process_snapshot(), + "git_revision":revision,"command":std::env::args().collect::>(),"release_build":!cfg!(debug_assertions),"passed":errors.is_empty() && dropped==0})) + })?; + std::fs::write( + args.output.join("report.json"), + serde_json::to_vec_pretty(&result)?, + )?; + ensure!( + result["errors"].as_array().unwrap().is_empty(), + "query failures; see report.json" + ); + Ok(()) +} +fn record( + sample: (f64, f64, Option), + latencies: &mut Vec, + service: &mut Vec, + errors: &mut Vec, +) { + let (latency, work, error) = sample; + if let Some(error) = error { + errors.push(error); + } else { + latencies.push(latency); + service.push(work); + } +} + +#[cfg(test)] +mod tests { + use super::*; + /// Real installed-plan execution and production HTTP serialization agree with + /// the direct sketch on both single-worker and multithreaded runtimes. + #[test] + fn production_paths_match_kernel() { + for workers in [1, 2] { + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(workers) + .enable_all() + .build() + .unwrap(); + runtime.block_on(async { + let fixture = Arc::new(fixture::Fixture::new(1000).unwrap()); + let port = fixture.server.start_test_server().await.unwrap(); + for layer in ["sketch", "raw_exact", "backend", "http"] { + operation( + layer, + fixture.clone(), + runtime.handle().clone(), + reqwest::Client::new(), + format!("http://127.0.0.1:{port}/api/v1/query"), + ) + .await + .unwrap(); + } + }); + } + } + /// Exact reference follows linear-interpolated median for odd/even unsorted data. + #[test] + fn exact_reference() { + assert_eq!(fixture::exact(&[4., 1., 3., 2.]), 2.5); + assert_eq!(fixture::exact(&[9., 1., 3.]), 3.); + } + /// Tail percentiles use nearest rank, including tiny sample sets. + #[test] + fn ranks() { + assert_eq!(percentile(&[], 0.99), None); + assert_eq!(percentile(&[1., 2., 3.], 0.99), Some(3.)); + } + /// Failed requests are retained as failures, never successful latency samples. + #[test] + fn failure_accounting() { + let (mut l, mut s, mut e) = (vec![], vec![], vec![]); + record((1., 1., Some("failed".into())), &mut l, &mut s, &mut e); + assert!(l.is_empty()); + assert_eq!(e.len(), 1); + } +} diff --git a/data_plane/src/lib.rs b/data_plane/src/lib.rs index 116429c2..cd13b048 100644 --- a/data_plane/src/lib.rs +++ b/data_plane/src/lib.rs @@ -63,3 +63,5 @@ pub type Result = std::result::Result Result<()> { Ok(()) } -#[tokio::main] -async fn main() -> Result<()> { +fn main() -> Result<()> { let args = Args::parse(); + let runtime = args.runtime.build()?; + runtime.block_on(run(args)) +} +async fn run(args: Args) -> Result<()> { validate_profile(&args)?; // Create output directory @@ -519,7 +524,19 @@ async fn main() -> Result<()> { // Initialize logging similar to Python's create_loggers function // Keep the guard alive for the entire lifetime of the application - let _log_guard = setup_logging(&args.output_dir, &args.log_level)?; + let _log_guard = args.logging.init(std::path::Path::new(&args.output_dir))?; + // Persist independently of the log filter, including when all logs are off. + let configuration = serde_json::json!({ + "runtime_workers": tokio::runtime::Handle::current().metrics().num_workers(), + "max_blocking_threads": args.runtime.runtime_max_blocking_threads, + "precompute_worker_tasks": args.precompute_num_workers, + "logging": args.logging, "effective_log_filter": args.logging.filter(), + "process_at_startup": data_plane::runtime_config::process_snapshot() + }); + fs::write( + std::path::Path::new(&args.output_dir).join("runtime-config.json"), + serde_json::to_vec_pretty(&configuration)?, + )?; info!("Starting Query Engine Rust"); info!("Output directory: {}", args.output_dir); @@ -1192,6 +1209,10 @@ async fn main() -> Result<()> { } }) }); + fs::write( + std::path::Path::new(&args.output_dir).join("runtime-ready.json"), + serde_json::to_vec_pretty(&data_plane::runtime_config::process_snapshot())?, + )?; // Wait for shutdown signal tokio::select! { result = server.run() => { @@ -1324,47 +1345,6 @@ async fn spawn_memory_diagnostics( } } -fn setup_logging( - output_dir: &str, - log_level: &str, -) -> Result { - use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt, EnvFilter}; - - // Create env filter that respects RUST_LOG, with fallback to command line arg - let env_filter = EnvFilter::try_from_default_env() - .or_else(|_| EnvFilter::try_new(log_level)) - .unwrap_or_else(|_| EnvFilter::new("info")); - - // Create file appender for logging to file - let file_appender = tracing_appender::rolling::never(output_dir, "query_engine.log"); - let (non_blocking_file, guard) = tracing_appender::non_blocking(file_appender); - - // Create console layer for stdout - let console_layer = tracing_subscriber::fmt::layer() - .with_file(true) - .with_line_number(true) - .with_target(true) - .with_writer(std::io::stdout); - - // Create file layer for file output - let file_layer = tracing_subscriber::fmt::layer() - .with_file(true) - .with_line_number(true) - .with_target(true) - .with_ansi(false) // Disable ANSI color codes in log file - .with_writer(non_blocking_file); - - tracing_subscriber::registry() - .with(env_filter) - .with(console_layer) - .with(file_layer) - .init(); - - info!("Logging initialized (respects RUST_LOG environment variable)"); - info!("Logs will be written to: {}/query_engine.log", output_dir); - Ok(guard) -} - #[cfg(test)] mod tests { use super::{validate_profile, Args}; diff --git a/data_plane/src/runtime_config.rs b/data_plane/src/runtime_config.rs new file mode 100644 index 00000000..c6ab7b88 --- /dev/null +++ b/data_plane/src/runtime_config.rs @@ -0,0 +1,161 @@ +//! Runtime controls shared by the server and the overhead inspection harness. +use clap::Args; +use serde::Serialize; +use std::{num::NonZeroUsize, path::Path}; + +#[derive(Args, Clone, Debug, Serialize)] +pub struct RuntimeConfig { + /// Tokio workers; defaults to TOKIO_WORKER_THREADS, then available parallelism. + #[arg(long)] + pub runtime_workers: Option, + /// Upper bound on Tokio's separate, lazily created blocking pool. + #[arg(long, default_value = "512")] + pub runtime_max_blocking_threads: NonZeroUsize, +} +impl RuntimeConfig { + pub fn workers(&self) -> anyhow::Result { + if let Some(n) = self.runtime_workers { + return Ok(n.get()); + } + if let Ok(n) = std::env::var("TOKIO_WORKER_THREADS") { + return Ok(n.parse::()?.get()); + } + Ok(std::thread::available_parallelism()?.get()) + } + pub fn build(&self) -> anyhow::Result { + Ok(tokio::runtime::Builder::new_multi_thread() + .worker_threads(self.workers()?) + .max_blocking_threads(self.runtime_max_blocking_threads.get()) + .thread_name("asap-runtime") + .enable_all() + .build()?) + } +} + +#[derive(Args, Clone, Debug, Serialize)] +pub struct LogConfig { + /// RUST_LOG takes precedence over this filter. + #[arg(long, default_value = "info")] + pub log_level: String, + #[arg(long)] + pub disable_console_log: bool, + #[arg(long)] + pub disable_file_log: bool, +} +impl LogConfig { + pub fn filter(&self) -> String { + std::env::var("RUST_LOG").unwrap_or_else(|_| self.log_level.clone()) + } + pub fn init( + &self, + directory: &Path, + ) -> anyhow::Result> { + use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt, EnvFilter}; + let filter = EnvFilter::try_new(self.filter())?; + let console = (!self.disable_console_log).then(|| { + tracing_subscriber::fmt::layer() + .with_file(true) + .with_line_number(true) + .with_target(true) + .with_writer(std::io::stdout) + }); + let (file, guard) = if self.disable_file_log { + (None, None) + } else { + std::fs::create_dir_all(directory)?; + let (writer, guard) = tracing_appender::non_blocking(tracing_appender::rolling::never( + directory, + "query_engine.log", + )); + ( + Some( + tracing_subscriber::fmt::layer() + .with_file(true) + .with_line_number(true) + .with_target(true) + .with_ansi(false) + .with_writer(writer), + ), + Some(guard), + ) + }; + tracing_subscriber::registry() + .with(filter) + .with(console) + .with(file) + .try_init()?; + Ok(guard) + } +} + +/// Unknown OS-specific measurements remain null, never zero. +pub fn process_snapshot() -> serde_json::Value { + let read = |p: &str| std::fs::read_to_string(p).ok(); + let status = read("/proc/self/status"); + let status_bytes = |name: &str| { + status.as_deref().and_then(|s| { + s.lines().find_map(|line| { + line.strip_prefix(name)? + .split_whitespace() + .next()? + .parse::() + .ok()? + .checked_mul(1024) + }) + }) + }; + let threads: Vec<_> = std::fs::read_dir("/proc/self/task").into_iter().flatten() + .filter_map(Result::ok).map(|e| serde_json::json!({"tid":e.file_name().to_string_lossy(), "name":std::fs::read_to_string(e.path().join("comm")).ok()})).collect(); + let cgroup = read("/proc/self/cgroup"); + let relative = cgroup + .as_deref() + .and_then(|s| s.lines().find_map(|line| line.strip_prefix("0::"))); + let group = relative.map(|s| Path::new("/sys/fs/cgroup").join(s.trim_start_matches('/'))); + let limits = group.map(|p| { + let ancestors: Vec<_> = p.ancestors().take_while(|a| a.starts_with("/sys/fs/cgroup")) + .map(|a| serde_json::json!({ + "path": a, + "cpu_max": std::fs::read_to_string(a.join("cpu.max")).ok(), + "cpuset_effective": std::fs::read_to_string(a.join("cpuset.cpus.effective")).ok(), + "memory_max": std::fs::read_to_string(a.join("memory.max")).ok() + })).collect(); + serde_json::json!({"path": p, "visible_ancestors_including_self": ancestors}) + }); + serde_json::json!({"cpu_model":read("/proc/cpuinfo").and_then(|s|s.lines().find_map(|l|l.strip_prefix("model name").map(|v|v.trim_start_matches([' ', ':', '\t']).to_owned()))), + "kernel":read("/proc/sys/kernel/osrelease"), "pid":std::process::id(), "available_parallelism":std::thread::available_parallelism().ok().map(|n|n.get()), + "rss_bytes":status_bytes("VmRSS:"), "peak_rss_bytes":status_bytes("VmHWM:"), "status":status, "threads":threads, "cgroup_membership":cgroup, "cgroup_v2":limits}) +} + +#[cfg(test)] +mod tests { + use super::*; + use clap::Parser; + #[derive(Parser)] + struct Cli { + #[command(flatten)] + runtime: RuntimeConfig, + #[command(flatten)] + logs: LogConfig, + } + /// Zero-sized pools fail at argument parsing rather than panicking at startup. + #[test] + fn rejects_zero_pools() { + for flag in ["--runtime-workers", "--runtime-max-blocking-threads"] { + assert!(Cli::try_parse_from(["test", flag, "0"]).is_err()); + } + } + /// Explicit worker configuration reaches the real Tokio runtime. + #[test] + fn configures_workers() { + let cli = Cli::parse_from([ + "test", + "--runtime-workers", + "2", + "--runtime-max-blocking-threads", + "3", + ]); + let runtime = cli.runtime.build().unwrap(); + assert_eq!(runtime.metrics().num_workers(), 2); + assert_eq!(cli.runtime.runtime_max_blocking_threads.get(), 3); + } +} diff --git a/docs/developer_docs/performance/overhead-inspection.md b/docs/developer_docs/performance/overhead-inspection.md new file mode 100644 index 00000000..cccae5df --- /dev/null +++ b/docs/developer_docs/performance/overhead-inspection.md @@ -0,0 +1,125 @@ +# Inspecting runtime overhead (#758) + +Audience: developers measuring execution overhead. The baseline is direct sketch +readout and direct exact computation over the same raw observations. No standalone +sketch server is involved. This experiment does not measure planner quality or +replace the workload benefit gates in #759. + +## Runtime model and controls + +The backend has one multithreaded Tokio runtime. HTTP query handlers, ingestion, +precompute workers, periodic flushing and asynchronous maintenance share its +workers. `--precompute-num-workers` controls ingest worker **tasks**, not OS +threads. Increasing it does not allocate a separate pool. Synchronous CPU work +inside an async task occupies its current runtime worker until it yields. + +`--runtime-workers N` fixes runtime workers. Without the flag, +`TOKIO_WORKER_THREADS` takes precedence over detected available parallelism. +`--runtime-max-blocking-threads N` caps the separate, lazy blocking pool (default +512); it is not a count of always-running threads or a cap on all process threads. +The file logger creates an additional writer thread. Libraries may create their +own threads; `/proc/self/task` inventories capture observed OS threads. + +`--log-level` accepts a tracing filter, including `error` and `off`. `RUST_LOG` +overrides the flag. `--disable-console-log` and `--disable-file-log` independently +remove those sinks; disabling the file sink also removes its writer thread. +Ordinary deployments keep info logging by default. Invalid filters fail startup. + +The server writes `runtime-config.json` even with logging disabled: resolved +workers, blocking pool ceiling, configured precompute tasks, effective filter, +thread inventory, available parallelism, Linux affinity/status and cgroup-v2 +limits. `runtime-ready.json` captures threads after background task setup. These +are snapshots, not a claim that no thread will subsequently be created. Unknown +OS counters are null. Visible cgroup ancestors are recorded because their limits also apply. Limits +outside a container namespace remain unknown; record the deployment's CPU quota +and pinning alongside these artifacts. + +## Reproducible three-layer experiment + +Build once; do not include compilation in timing: + +```sh +cargo build --release --locked -p data_plane --example overhead_inspect +RUST_LOG=off target/release/examples/overhead_inspect \ + --workers 1,2,4,8,16 --concurrency 1,2,4,8,16 \ + --requests 10000 --warmup 1000 --repeats 5 \ + --disable-console-log --disable-file-log --output target/overhead-off +``` + +Run on an otherwise idle host with fixed CPU affinity/quota. Record CPU model, +OS, Git revision and command. Each matrix cell is a fresh process; repeats retain +individual JSON reports and `matrix.csv` rather than merging unlike configurations. Keep raw reports +and compare run-to-run variation before drawing conclusions. Use request counts +large enough to amortize startup and CPU clock resolution. + +The deterministic fixture has one series, one complete one-second pane and a +median query at a fixed timestamp. All layers use the same unsorted positive raw +values and alpha=0.01 DDSketch. Ingestion and sketch construction occur before +measurement. The sketch's result must meet the fixture's 2% median-error gate; +backend and HTTP results must match direct sketch readout. External forwarding +is disabled. The installed plan rejects fallback. This initial workload isolates +warm read overhead; it does not establish results for other sketch families, +series counts, merges across panes, range queries or concurrent ingestion. + +| Layer | Timed work | +| --- | --- | +| `sketch` | Direct DDSketch median on the prebuilt sketch | +| `raw_exact` | Scratch copy and exact median selection over unsorted raw values | +| `backend` | Real ASAPQueryEngine, installed plan, SketchStore lookup/readout and result construction | +| `http` | Real HTTP server over loopback, parsing, routing, execution and response decoding | + +For sketch and raw exact, `direct_kernel_seconds_per_query` measures a separate +synchronous loop with `black_box`, outside task dispatch. Concurrent samples for +those layers include submission to the runtime; do not call their entire latency +"sketch cost". Backend samples also include submission. HTTP uses reused client +connections. All responses pass correctness checks; failed queries are counted +and fail the experiment instead of becoming successful latency samples. + +The driver uses a separate current-thread runtime. CPU reports include whole +process CPU and driver-thread CPU; their difference includes server workers and +auxiliary threads, not just query instructions. The per-completed-query ratio +includes CPU spent on failed requests. RSS/peak RSS include fixture, client, +server and report buffers; peak RSS is process-lifetime, not just the timed phase. +Use an external process profiler for allocation/lock attribution. Do not subtract +layer latencies and label the remainder "trait overhead". + +## Arrival rate, saturation and logging + +Without `--rate`, clients use bounded closed-loop concurrency. With `--rate 1000`, +arrivals follow a fixed schedule independent of response completion. At the +in-flight cap, arrivals are dropped and counted. Latency includes delay from the +scheduled arrival, while `service_latency_seconds` starts at task execution. +Report requested rate, actual throughput, drops and errors together with +p50/p95/p99. Successful-request percentiles do not describe dropped arrivals. +Any drops make `passed=false`, even though the scan continues to collect other +cells. The driver/timer has finite capacity: include scheduling delay in the +interpretation and increase rate gradually; an unsustained requested rate is not +proof of backend saturation. + +Repeat the same matrix with `RUST_LOG=error`, `RUST_LOG=info` and `RUST_LOG=off`, +controlling sinks explicitly. A filter comparison is meaningful only if that +workload actually emits events at those levels. Keep production logging defaults +unchanged. The performance fixture intentionally does not run ingestion or +maintenance background loops; use the production server's runtime artifacts when +profiling those mixed workloads. + +## Profiling and acceptance + +On a host with Linux perf available, profile one cell at a time, for example: + +```sh +CARGO_PROFILE_RELEASE_DEBUG=1 cargo build --release -p data_plane --example overhead_inspect +perf record -g --call-graph dwarf -- target/release/examples/overhead_inspect \ + --cell --runtime-workers 4 --concurrency 8 --layers backend \ + --requests 100000 --warmup 1000 --log-level off --disable-file-log \ + --disable-console-log --output target/overhead-profile +perf report +``` + +Collect CPU stacks and, where appropriate, allocation and off-CPU/lock profiles +before changing abstractions. The deliverable is a reproducible scaling curve +and an evidence-backed explanation of hot paths, not a predetermined speedup. +Unit tests check accounting and runtime controls; a small real four-layer smoke +run checks that the harness reaches the production execution paths. Performance +thresholds belong on a controlled host after variance is established, not noisy +shared CI. No claim about an eight-thread crossover is assumed.