diff --git a/Cargo.lock b/Cargo.lock index 12a44ff1..47258aa7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -82,6 +82,22 @@ version = "1.0.104" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "330a5ed07fa54e4702c9d6c4174f74427fc0ef6e214bbd677ae50a5099946470" +[[package]] +name = "asap-executor" +version = "0.1.0" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=58999df294fc9b0326341c1f20be62ecc8c2975e#58999df294fc9b0326341c1f20be62ecc8c2975e" +dependencies = [ + "asap-types", + "asap_sketchlib", + "chrono", + "futures", + "regex", + "serde", + "serde_json", + "thiserror 2.0.20", + "tracing", +] + [[package]] name = "asap-frontend-common" version = "0.1.0" @@ -265,6 +281,15 @@ version = "2.13.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3ded4057c258ba199e2d26386d3af3780957ecaee6c4ef4041c6b4b8b97c0b06" +[[package]] +name = "block-buffer" +version = "0.10.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3078c7629b62d3f0439517fa394996acacc5cbc91c5a20d8c658e77abd503a71" +dependencies = [ + "generic-array", +] + [[package]] name = "bumpalo" version = "3.20.3" @@ -326,7 +351,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06" dependencies = [ "cfg-if", - "cpufeatures", + "cpufeatures 0.3.1", "rand_core 0.10.1", ] @@ -413,6 +438,15 @@ version = "0.8.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" +[[package]] +name = "cpufeatures" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "59ed5838eebb26a2bb2e58f6d5b5316989ae9d08bab10e0e6d103e656d1b0280" +dependencies = [ + "libc", +] + [[package]] name = "cpufeatures" version = "0.3.1" @@ -422,11 +456,23 @@ dependencies = [ "libc", ] +[[package]] +name = "crypto-common" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78c8292055d1c1df0cce5d180393dc8cce0abec0a7102adb6c7b1eef6016d60a" +dependencies = [ + "generic-array", + "typenum", +] + [[package]] name = "data_plane" version = "0.1.0" dependencies = [ "anyhow", + "asap-executor", + "asap-types", "axum", "clap", "control_plane", @@ -435,6 +481,7 @@ dependencies = [ "reqwest", "serde", "serde_json", + "sha2", "snap", "thiserror 1.0.69", "tokio", @@ -449,6 +496,16 @@ version = "0.5.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7cd812cc2bc1d69d4764bd80df88b4317eaef9e773c75226407d9bc0876b211c" +[[package]] +name = "digest" +version = "0.10.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" +dependencies = [ + "block-buffer", + "crypto-common", +] + [[package]] name = "displaydoc" version = "0.2.7" @@ -513,6 +570,21 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "futures" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a31d2a3fbaaeb2af2368bbdd904aa8e812d3c04a1ee10d3171f52d556e5d0a3" +dependencies = [ + "futures-channel", + "futures-core", + "futures-executor", + "futures-io", + "futures-sink", + "futures-task", + "futures-util", +] + [[package]] name = "futures-channel" version = "0.3.34" @@ -520,6 +592,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b1f9e3d69d39e4862ffed03ed071a76f9a13ba1d9109d355b0f0aa6b15e393c4" dependencies = [ "futures-core", + "futures-sink", ] [[package]] @@ -528,6 +601,40 @@ version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e" +[[package]] +name = "futures-executor" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "031b47cf1a3c6cc8bc2fc76cd437f521619387907d469316e7c0bc278f1f5432" +dependencies = [ + "futures-core", + "futures-task", + "futures-util", +] + +[[package]] +name = "futures-io" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53c0fa8157de1303bfffdaa1cc2a673bfffb60102f76b0ef4441659124373fed" + +[[package]] +name = "futures-macro" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9fb9654ba8355388abeb8dcb4fc62f511300867002afc858860463bdd9fe0c44" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.5", +] + +[[package]] +name = "futures-sink" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1944426bf7d03f1d14f708785e4b33efd750b36d48a157b836b3efc15ede8e1d" + [[package]] name = "futures-task" version = "0.3.34" @@ -540,12 +647,27 @@ version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0d50a92467f8ba5dd6e3ee5d4bd04d73ab2e4e1c44474a0674821dfce14b79bc" dependencies = [ + "futures-channel", "futures-core", + "futures-io", + "futures-macro", + "futures-sink", "futures-task", + "memchr", "pin-project-lite", "slab", ] +[[package]] +name = "generic-array" +version = "0.14.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85649ca51fd72272d7821adaf274ad91c288277713d9c18820d8499a7ff69e9a" +dependencies = [ + "typenum", + "version_check", +] + [[package]] name = "getopts" version = "0.2.24" @@ -1562,6 +1684,17 @@ dependencies = [ "serde", ] +[[package]] +name = "sha2" +version = "0.10.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283" +dependencies = [ + "cfg-if", + "cpufeatures 0.2.17", + "digest", +] + [[package]] name = "sharded-slab" version = "0.1.7" @@ -1960,6 +2093,12 @@ dependencies = [ "rand 0.10.2", ] +[[package]] +name = "typenum" +version = "1.20.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" + [[package]] name = "unicode-ident" version = "1.0.24" @@ -2019,6 +2158,12 @@ dependencies = [ "time", ] +[[package]] +name = "version_check" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" + [[package]] name = "vob" version = "3.0.6" diff --git a/data_plane/Cargo.toml b/data_plane/Cargo.toml index 2aaf8975..cf8e346d 100644 --- a/data_plane/Cargo.toml +++ b/data_plane/Cargo.toml @@ -5,6 +5,9 @@ edition.workspace = true [dependencies] control_plane.workspace = true +planner-types.workspace = true +asap-executor.workspace = true +sha2 = "0.10" serde.workspace = true serde_json.workspace = true thiserror.workspace = true diff --git a/data_plane/src/lib.rs b/data_plane/src/lib.rs index fd483635..7180dacf 100644 --- a/data_plane/src/lib.rs +++ b/data_plane/src/lib.rs @@ -6,6 +6,7 @@ //! query is forwarded to Prometheus. pub mod remote_write; +pub mod sds; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; diff --git a/data_plane/src/sds.rs b/data_plane/src/sds.rs new file mode 100644 index 00000000..05ca7478 --- /dev/null +++ b/data_plane/src/sds.rs @@ -0,0 +1,252 @@ +//! Stored summaries (SDS): what the data plane stores for a plan's +//! ingestion-time outputs, and how a query finds them again. +//! +//! The Planner's selected DAG marks each output computed at ingestion time +//! (`MatAtIngestion`) and read at query time: the executor's materialization +//! frontier. The control plane installs the DAG with each `PaneChain` +//! expanded into its panes, so each such output reads one tumbling pane of +//! raw samples: +//! +//! ```text +//! frontier node <- ... <- TimeRange(width) <- TimeShift(offset)* <- Scan +//! ``` +//! +//! At evaluation time `T` it covers `[T - offset - width, T - offset)`. +//! Outputs that differ only in `offset` compute the same thing over different +//! panes, so they share one [`SummaryDefinition`]: the data plane builds each +//! pane of a definition once, and every output reads the pane it needs. + +use std::collections::{BTreeMap, HashMap}; + +use planner_types::ir::physical_export::{PhysicalASAPDAG, PhysicalASAPDAGNode}; +use planner_types::ir::schema::Schema; +use planner_types::ir::{NonASAPOp, Operator}; +use planner_types::physical::Materialization; +use sha2::{Digest, Sha256}; + +/// Identifies what a stored summary computes, independent of node ids and of +/// which pane an output reads. Equal for equal computations across plans. +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct DefinitionId(pub String); + +/// One computation the data plane maintains pane by pane. +#[derive(Debug, Clone, PartialEq)] +pub struct SummaryDefinition { + pub id: DefinitionId, + /// Pane width: each stored summary covers `[end - pane_ms, end)`. + pub pane_ms: i64, + /// The frontier output that builds a pane; any output of the definition + /// computes the same, and this is the first in node order. + pub builder: u64, + /// The raw input of the build: the `TimeRange` node fed one pane's samples. + pub raw_input: u64, + pub output_schema: Schema, +} + +/// How one frontier output reads its definition: at evaluation time `T`, the +/// pane ending at `T - offset_ms`. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct PaneBinding { + pub offset_ms: i64, +} + +/// The stored summaries of one plan. +#[derive(Debug, Clone, PartialEq, Default)] +pub struct Sds { + pub definitions: BTreeMap, + /// Every frontier output, with its definition and pane. + pub outputs: BTreeMap, +} + +#[derive(Debug, thiserror::Error, PartialEq)] +pub enum SdsError { + #[error("node {0} is not in the DAG")] + UnknownNode(u64), + #[error("ingestion-time output {node} does not read one pane of one Scan: {reason}")] + UnsupportedShape { node: u64, reason: String }, + /// The data plane keeps no query-time result across evaluations; it + /// declares `query_time_results_retained: false`, so the Planner should + /// not select such a plan. + #[error("node {0} is kept across evaluations (MatAtQueryKeptAcrossEvaluations), which the data plane does not support")] + KeptAcrossEvaluations(u64), +} + +impl Sds { + /// The definitions behind `frontier`, the DAG outputs computed at + /// ingestion time and read at query time. + pub fn from_dag(dag: &PhysicalASAPDAG, frontier: &[u64]) -> Result { + let nodes: HashMap = + dag.nodes.iter().map(|n| (n.id as u64, n)).collect(); + let mut inputs: HashMap> = HashMap::new(); + for edge in &dag.edges { + inputs + .entry(edge.consumer as u64) + .or_default() + .push(edge.producer as u64); + } + if let Some(node) = dag.nodes.iter().find(|n| { + matches!( + n.materialization, + Materialization::MatAtQueryKeptAcrossEvaluations(_) + ) + }) { + return Err(SdsError::KeptAcrossEvaluations(node.id as u64)); + } + let mut sds = Sds::default(); + let mut hashes = HashMap::new(); + for &node in frontier { + let (raw_input, pane_ms, offset_ms) = pane_of(node, &nodes, &inputs)?; + let id = DefinitionId(hex(&identity(node, &nodes, &inputs, &mut hashes)?)); + let output_schema = nodes[&node].output_schema.clone(); + let definition = + sds.definitions + .entry(id.clone()) + .or_insert_with(|| SummaryDefinition { + id: id.clone(), + pane_ms, + builder: node, + raw_input, + output_schema, + }); + if node < definition.builder { + definition.builder = node; + definition.raw_input = raw_input; + } + sds.outputs.insert(node, (id, PaneBinding { offset_ms })); + } + Ok(sds) + } +} + +/// `(raw input, pane width, offset)` of a frontier output: follow its single +/// inputs down to the `TimeRange`, then sum the `TimeShift`s down to the `Scan`. +fn pane_of( + node: u64, + nodes: &HashMap, + inputs: &HashMap>, +) -> Result<(u64, i64, i64), SdsError> { + let unsupported = |reason: &str| SdsError::UnsupportedShape { + node, + reason: reason.into(), + }; + let only_input = |id: u64| match inputs.get(&id).map(Vec::as_slice) { + Some([input]) => Ok(*input), + _ => Err(unsupported( + "every operator above the Scan must have one input", + )), + }; + let mut current = node; + let (raw_input, pane_ms) = loop { + match relational(nodes, current)? { + Some(NonASAPOp::TimeRange { range, .. }) => { + break (current, range.as_millis() as i64); + } + Some(NonASAPOp::TimeShift { .. }) => { + return Err(unsupported("a TimeShift above the TimeRange")); + } + Some(NonASAPOp::Scan { .. }) => return Err(unsupported("no TimeRange")), + _ => current = only_input(current)?, + } + }; + let mut offset_ms = 0; + current = only_input(raw_input)?; + loop { + match relational(nodes, current)? { + Some(NonASAPOp::TimeShift { shift, .. }) if shift.at.is_none() => { + offset_ms += shift.offset_ms; + current = only_input(current)?; + } + Some(NonASAPOp::Scan { .. }) => return Ok((raw_input, pane_ms, offset_ms)), + _ => { + return Err(unsupported( + "only offset TimeShifts between TimeRange and Scan", + )) + } + } + } +} + +fn relational<'a>( + nodes: &HashMap, + id: u64, +) -> Result>, SdsError> { + match &nodes.get(&id).ok_or(SdsError::UnknownNode(id))?.payload { + Operator::NonASAP(operator) => Ok(Some(operator)), + Operator::ASAP(_) => Ok(None), + } +} + +/// SHA-256 over a node's payload and output schema, then its inputs' +/// identities in edge order. The payload's input ids are zeroed, so the +/// identity does not depend on node numbering. An offset `TimeShift` has its +/// input's identity: the offset says which pane an output reads +/// ([`PaneBinding`]), not what it computes, so the panes of one computation +/// share an identity whether or not their range is shifted. +fn identity( + node: u64, + nodes: &HashMap, + inputs: &HashMap>, + hashes: &mut HashMap, +) -> Result<[u8; 32], SdsError> { + if let Some(hash) = hashes.get(&node) { + return Ok(*hash); + } + let dag_node = nodes.get(&node).ok_or(SdsError::UnknownNode(node))?; + if let Operator::NonASAP(NonASAPOp::TimeShift { shift, .. }) = &dag_node.payload { + if let (None, Some([input])) = (shift.at, inputs.get(&node).map(Vec::as_slice)) { + let hash = identity(*input, nodes, inputs, hashes)?; + hashes.insert(node, hash); + return Ok(hash); + } + } + let payload = dag_node.payload.map_children(|_| 0); + let mut hasher = Sha256::new(); + hasher.update(serde_json::to_vec(&(&payload, &dag_node.output_schema)).expect("serializes")); + for &input in inputs.get(&node).into_iter().flatten() { + hasher.update(identity(input, nodes, inputs, hashes)?); + } + let hash: [u8; 32] = hasher.finalize().into(); + hashes.insert(node, hash); + Ok(hash) +} + +fn hex(bytes: &[u8]) -> String { + bytes.iter().map(|b| format!("{b:02x}")).collect() +} + +/// Built panes, by definition and pane end. A pane is stored once, when it +/// is complete, and never changes. +#[derive(Debug, Default)] +pub struct SummaryStore { + panes: BTreeMap<(DefinitionId, i64), S>, +} + +impl SummaryStore { + pub fn new() -> Self { + Self { + panes: BTreeMap::new(), + } + } + + pub fn insert(&mut self, definition: &DefinitionId, end_ms: i64, state: S) { + self.panes.insert((definition.clone(), end_ms), state); + } + + pub fn get(&self, definition: &DefinitionId, end_ms: i64) -> Option<&S> { + self.panes.get(&(definition.clone(), end_ms)) + } + + /// Drops the panes of `definition` that end at or before `end_ms`. + pub fn evict_through(&mut self, definition: &DefinitionId, end_ms: i64) { + self.panes + .retain(|(id, end), _| id != definition || *end > end_ms); + } + + pub fn len(&self) -> usize { + self.panes.len() + } + + pub fn is_empty(&self) -> bool { + self.panes.is_empty() + } +} diff --git a/data_plane/tests/sds.rs b/data_plane/tests/sds.rs new file mode 100644 index 00000000..a25a2d07 --- /dev/null +++ b/data_plane/tests/sds.rs @@ -0,0 +1,151 @@ +use std::path::Path; + +use asap_executor::physical_planner::frontier_from_timing; +use control_plane::{plan_workload, DeploymentInputs, InstalledPlan}; +use data_plane::sds::{DefinitionId, Sds, SdsError, SummaryStore}; +use planner_types::ir::physical_export::PhysicalASAPDAG; +use planner_types::ir::{NonASAPOp, Operator}; +use planner_types::physical::{Materialization, Medium}; + +/// The selected plan of a #509 example workload in `examples/`, for a +/// deployment that keeps raw data or not. +fn plan(example: &str, raw_data_retained: bool) -> InstalledPlan { + let path = Path::new(env!("CARGO_MANIFEST_DIR")) + .join(format!("../examples/planner-layering-{example}.json")); + let workload = serde_json::from_str(&std::fs::read_to_string(path).unwrap()).unwrap(); + let mut inputs = DeploymentInputs::default(); + inputs.deployment_capabilities.raw_data_retained = raw_data_retained; + plan_workload(&workload, &inputs).unwrap() +} + +fn sds_of(dag: &PhysicalASAPDAG) -> Result { + Sds::from_dag(dag, &frontier_from_timing(dag).unwrap()) +} + +/// Example 4b's six ingestion-time KLL panes are one definition with a +/// 10-minute pane; each output reads the pane `i * 10min` before evaluation. +#[test] +fn example4b_panes_share_one_definition() { + let plan = plan("example4b", false); + let sds = sds_of(&plan.dag.dag).unwrap(); + assert_eq!(sds.definitions.len(), 1); + let definition = sds.definitions.values().next().unwrap(); + assert_eq!(definition.pane_ms, 600_000); + let mut offsets: Vec = sds + .outputs + .values() + .map(|(id, binding)| { + assert_eq!(id, &definition.id); + binding.offset_ms + }) + .collect(); + offsets.sort(); + assert_eq!(offsets, (0..6).map(|i| i * 600_000).collect::>()); + let raw = plan + .dag + .dag + .nodes + .iter() + .find(|n| n.id as u64 == definition.raw_input) + .unwrap(); + assert!(matches!( + raw.payload, + Operator::NonASAP(NonASAPOp::TimeRange { .. }) + )); +} + +/// A plan computed entirely at query time stores nothing. +#[test] +fn query_time_plans_have_no_definitions() { + let plan = plan("example1", true); + assert_eq!(sds_of(&plan.dag.dag).unwrap(), Sds::default()); +} + +/// Identity depends on what is computed, not on how nodes are numbered. +#[test] +fn identity_ignores_node_numbering() { + let dag = plan("example4b", false).dag.dag; + let mut renumbered = dag.clone(); + let renumber = |id: &usize| id + 100; + for node in &mut renumbered.nodes { + node.id = renumber(&node.id); + node.payload = node.payload.map_children(renumber); + } + for edge in &mut renumbered.edges { + edge.producer = renumber(&edge.producer); + edge.consumer = renumber(&edge.consumer); + } + renumbered.roots = renumbered.roots.iter().map(renumber).collect(); + let ids = |sds: Sds| sds.definitions.into_keys().collect::>(); + assert_eq!( + ids(sds_of(&renumbered).unwrap()), + ids(sds_of(&dag).unwrap()) + ); +} + +/// The same computation over another metric is another definition. +#[test] +fn identity_depends_on_the_source() { + let dag = plan("example4b", false).dag.dag; + let other: PhysicalASAPDAG = serde_json::from_str( + &serde_json::to_string(&dag) + .unwrap() + .replace("latency_ms", "other"), + ) + .unwrap(); + assert_ne!( + sds_of(&other).unwrap().definitions.into_keys().next(), + sds_of(&dag).unwrap().definitions.into_keys().next() + ); +} + +/// An ingestion-time output must read a pane of raw samples. +#[test] +fn rejects_outputs_without_a_pane() { + let dag = plan("example4b", false).dag.dag; + let scan = dag + .nodes + .iter() + .find(|n| matches!(n.payload, Operator::NonASAP(NonASAPOp::Scan { .. }))) + .unwrap(); + let scan = scan.id as u64; + assert_eq!( + Sds::from_dag(&dag, &[scan]), + Err(SdsError::UnsupportedShape { + node: scan, + reason: "no TimeRange".into() + }) + ); +} + +/// A plan that keeps a query-time result across evaluations is not +/// installed: the data plane declares it cannot keep one. +#[test] +fn rejects_results_kept_across_evaluations() { + let mut dag = plan("example4b", true).dag.dag; + let node = &mut dag.nodes[0]; + node.materialization = Materialization::MatAtQueryKeptAcrossEvaluations(Medium::Memory); + let id = node.id as u64; + assert_eq!( + Sds::from_dag(&dag, &[]), + Err(SdsError::KeptAcrossEvaluations(id)) + ); +} + +/// Panes are found by definition and end, and evicted once no query needs them. +#[test] +fn store_finds_and_evicts_panes() { + let (a, b) = (DefinitionId("a".into()), DefinitionId("b".into())); + let mut store = SummaryStore::new(); + for end in [60, 120, 180] { + store.insert(&a, end, format!("a{end}")); + } + store.insert(&b, 60, "b60".to_string()); + assert_eq!(store.get(&a, 120).map(String::as_str), Some("a120")); + assert_eq!(store.get(&a, 90), None); + store.evict_through(&a, 120); + assert_eq!(store.get(&a, 120), None); + assert_eq!(store.get(&a, 180).map(String::as_str), Some("a180")); + assert_eq!(store.get(&b, 60).map(String::as_str), Some("b60")); + assert_eq!(store.len(), 2); +} diff --git a/examples/planner-layering-example3b.json b/examples/planner-layering-example3b.json new file mode 100644 index 00000000..c943e415 --- /dev/null +++ b/examples/planner-layering-example3b.json @@ -0,0 +1,44 @@ +{ + "query_workload": { + "language": "promql", + "query_batch": null, + "repeating_queries": [ + { + "query": "quantile_over_time(0.99, latency_ms[5m])", + "demand": { + "fixed_interval": 60000 + }, + "requirements": { + "accuracy": { + "accuracy_metric": "rank", + "max_error": 0.01, + "failure_probability": 0.01, + "accuracy_scope": "whole_result" + }, + "max_latency_ms": 200.0 + }, + "predictability": { + "predictable": { + "known_at": null + } + } + } + ] + }, + "data_workload": { + "arrival": { + "ingesting": { + "rows_per_second": 66666.66666666667 + } + }, + "sources": { + "latency_ms": { + "series_count": 1000000, + "metric_type": null, + "sample_interval": 15000, + "stream_norms": null, + "max_distinct_per_evaluation": null + } + } + } +} diff --git a/examples/planner-layering-example4b.json b/examples/planner-layering-example4b.json new file mode 100644 index 00000000..ab13a77f --- /dev/null +++ b/examples/planner-layering-example4b.json @@ -0,0 +1,44 @@ +{ + "query_workload": { + "language": "promql", + "query_batch": null, + "repeating_queries": [ + { + "query": "quantile_over_time(0.99, latency_ms[60m])", + "demand": { + "fixed_interval": 600000 + }, + "requirements": { + "accuracy": { + "accuracy_metric": "rank", + "max_error": 0.01, + "failure_probability": 0.01, + "accuracy_scope": "whole_result" + }, + "max_latency_ms": 200.0 + }, + "predictability": { + "predictable": { + "known_at": null + } + } + } + ] + }, + "data_workload": { + "arrival": { + "ingesting": { + "rows_per_second": 1000.0 + } + }, + "sources": { + "latency_ms": { + "series_count": 1000, + "metric_type": null, + "sample_interval": 1000, + "stream_norms": null, + "max_distinct_per_evaluation": null + } + } + } +}