rqe-optimizer: size DDSketch memory from the metric's value range

docs/rqe_sketch_deployment_v1.md

@@ -57,6 +57,8 @@ For each metric, the caller provides `MetricFacts`:
 - `scrape_interval`: every series yields one sample per scrape.
 - `card(X)`: distinct value combinations of each label set `X` in use,
   including `labels` itself (the raw series count).
+- `value_range` (optional): `(lo, hi)`, the smallest and largest positive
+  sample value, with `0 < lo <= hi`. It sizes DDSketch (see below).
 
 The arrival rate is derived as follows (not provided by the user):
 
@@ -283,7 +285,12 @@ For deployment `D` serving RAQE `i`:
 - `a_D = x / y`: open windows; each sample is inserted into `a_D` instances of its group.
 - `S_i`, `T_i`: RAQE `i`'s lookback and interval.
 - `n_i = S_i / x`: windows merged per query.
-- `m`: memory per instance, measured.
+- `m`: memory per instance, measured. For DDSketch on a metric with a
+  `value_range`, `m = (floor(ln(hi / lo) / ln((1 + alpha) / (1 - alpha))) + 1) × 8`
+  bytes instead: one bucket count per `gamma`-power in the range, capped by
+  the values one instance sees (`λ · x / card(G)` for a window, `λ · L / card(G)`
+  for a query's merge accumulator), as sketch-bench's `dd_footprint` counts them. The range is the metric's, so `m` is an upper
+  bound for a group whose own values span less.
 - `c_ins`: CPU per insert, measured.
 - `c_mrg`: CPU per pairwise merge, measured.
 - `c_qry`: CPU per query of one instance, measured.
@@ -330,9 +337,11 @@ not scale.
 
 Quantile sketches have no keys, but are not strictly fixed either: KLL grows
 slowly with the number of values inserted (about `k × log(n / k)`), and
-DDSketch with the range of values. The model treats them as Fixed at the size
+DDSketch with the range of values. The model treats KLL as Fixed at the size
 the export measured (1,000,000 values per instance), so it misstates instances
-that see far fewer or far more values.
+that see far fewer or far more values. DDSketch is sized from the metric's
+`value_range` when given (capped by values per instance), else from the
+measured size.
 
 The model uses one formula, `card(G) × m`, which is right for both cells in
 use:

rqe-optimizer/examples/small_problem.rs

@@ -93,9 +93,10 @@ fn load_cost_table() -> AtomicCostTable {
 }
 
 /// Both metrics: 5 services × 10 endpoints × 600 pods = 30,000 series
-/// scraped every 15 s, so `λ` = 2,000 samples/sec each.
+/// scraped every 15 s, so `λ` = 2,000 samples/sec each. Durations range from
+/// 1 ms to 60 s, which sizes DDSketch.
 fn facts() -> WorkloadFacts {
-    let http_metric_facts = || MetricFacts {
+    let http_metric_facts = |value_range| MetricFacts {
         labels: label_set(&["service", "endpoint", "pod"]),
         scrape_interval_ms: 15_000,
         cardinality: [
@@ -104,10 +105,11 @@ fn facts() -> WorkloadFacts {
             (label_set(&["service", "endpoint", "pod"]), 30_000),
         ]
         .into(),
+        value_range,
     };
     [
-        (REQUESTS.to_string(), http_metric_facts()),
-        (DURATION.to_string(), http_metric_facts()),
+        (REQUESTS.to_string(), http_metric_facts(None)),
+        (DURATION.to_string(), http_metric_facts(Some((0.001, 60.0)))),
     ]
     .into()
 }

rqe-optimizer/src/analytical_cost_model.rs

@@ -15,7 +15,7 @@
 // ponytail: a family with one fixed-size instance for all groups (HydraKLL,
 // sketch-bench#142) needs a per-family shape so it isn't multiplied by card(G).
 
-use crate::{secs, Capability, Deployment, Mapping, Raqe, WorkloadFacts};
+use crate::{secs, Capability, Deployment, Mapping, Millis, Raqe, WorkloadFacts};
 use std::collections::{BTreeMap, BTreeSet};
 
 pub(crate) const BYTES_PER_GIB: f64 = (1u64 << 30) as f64;
@@ -30,6 +30,9 @@ pub const OUTPUT_BYTES_PER_TOPK_ENTRY: f64 = 16.0;
 // ponytail: fixed k. A large k needs `k` on the RAQE and a benchmark that
 // varies the heap, whose memory the export leaves out.
 pub const TOPK_ENTRIES: f64 = 32.0;
+/// Bytes per DDSketch bucket: one `u64` count, as sketch-bench's
+/// `dd_footprint` counts them.
+pub const DD_BYTES_PER_BUCKET: f64 = 8.0;
 
 /// One phase's resource use.
 #[derive(Debug, Clone, Copy, Default, PartialEq)]
@@ -80,6 +83,37 @@ fn group_count(deployment: &Deployment, facts: &WorkloadFacts) -> f64 {
     facts[&deployment.metric].cardinality[&deployment.grouping_labels] as f64
 }
 
+/// `m`: memory of one instance holding `span_ms` of one group's samples: a
+/// window `x`, or the lookback `L` for a query's merge accumulator.
+/// DDSketch on a metric with a known value range holds one store (values are
+/// positive) of [`dd_bucket_count`] buckets, capped by the values it holds,
+/// `λ · span / card(G)`, as `dd_footprint` caps them by count. The range is the metric's, so this is
+/// an upper bound for a group whose own values span less. Every other row,
+/// and DDSketch without a range, keeps the measured size.
+pub(crate) fn instance_memory_bytes(
+    deployment: &Deployment,
+    facts: &WorkloadFacts,
+    span_ms: Millis,
+) -> f64 {
+    let config = &deployment.config;
+    let metric = &facts[&deployment.metric];
+    let alpha = config.sketch_config["params"]["alpha"].as_f64();
+    match (config.sketch.as_str(), metric.value_range, alpha) {
+        ("dd", Some((lo, hi)), Some(alpha)) => {
+            let values =
+                metric.arrival_rate_per_sec() * secs(span_ms) / group_count(deployment, facts);
+            dd_bucket_count(lo, hi, alpha).min(values.ceil().max(1.0)) * DD_BYTES_PER_BUCKET
+        }
+        _ => config.mem_bytes_per_instance,
+    }
+}
+
+/// Buckets DDSketch needs for values in `[lo, hi]` at relative accuracy
+/// `alpha`: `⌊ln(hi/lo) / ln((1+α)/(1−α))⌋ + 1`, as `dd_footprint` counts them.
+pub(crate) fn dd_bucket_count(lo: f64, hi: f64, alpha: f64) -> f64 {
+    ((hi / lo).ln() / ((1.0 + alpha) / (1.0 - alpha)).ln()).floor() + 1.0
+}
+
 /// `x / y`: windows each sample is inserted into.
 fn open_window_count(deployment: &Deployment) -> f64 {
     deployment
@@ -102,7 +136,7 @@ pub(crate) fn ingest(deployment: &Deployment, facts: &WorkloadFacts) -> PhaseCos
             * open_windows
             * deployment.config.insert_cpu_secs,
         memory_bytes: group_count(deployment, facts)
-            * deployment.config.mem_bytes_per_instance
+            * instance_memory_bytes(deployment, facts, deployment.window_ms)
             * open_windows,
     }
 }
@@ -116,14 +150,18 @@ pub(crate) fn merge(raqe: &Raqe, deployment: &Deployment, facts: &WorkloadFacts)
     PhaseCost {
         cpu_secs_per_sec: groups * merges_per_group * deployment.config.merge_cpu_secs
             / secs(raqe.interval_ms),
-        memory_bytes: groups * merge_memory_per_group(raqe, deployment),
+        memory_bytes: groups * merge_memory_per_group(raqe, deployment, facts),
     }
 }
 
-/// One accumulator if the query merges at all.
-pub(crate) fn merge_memory_per_group(raqe: &Raqe, deployment: &Deployment) -> f64 {
+/// One accumulator if the query merges at all, holding the whole lookback.
+pub(crate) fn merge_memory_per_group(
+    raqe: &Raqe,
+    deployment: &Deployment,
+    facts: &WorkloadFacts,
+) -> f64 {
     if merged_window_count(raqe, deployment) > 1.0 {
-        deployment.config.mem_bytes_per_instance
+        instance_memory_bytes(deployment, facts, raqe.lookback_ms)
     } else {
         0.0
     }
@@ -147,7 +185,9 @@ pub(crate) fn storage_bytes(raqe: &Raqe, deployment: &Deployment, facts: &Worklo
     let closed_windows = deployment
         .closed_instance_count(raqe.lookback_ms)
         .expect("only eligible pairs are costed") as f64;
-    group_count(deployment, facts) * deployment.config.mem_bytes_per_instance * closed_windows
+    group_count(deployment, facts)
+        * instance_memory_bytes(deployment, facts, deployment.window_ms)
+        * closed_windows
 }
 
 /// Serial CPU time of one query, in ms: `card(G) · (c_qry + (L/x − 1) · c_mrg)`.
@@ -210,6 +250,57 @@ mod tests {
     use super::*;
     use crate::test_support::{deployment, facts, raqe};
 
+    #[test]
+    fn dd_bucket_count_is_log_range_over_log_gamma() {
+        // alpha = 1/3 gives gamma = 2: buckets [1, 2), [2, 4), ... up to 1000.
+        assert_eq!(dd_bucket_count(1.0, 1000.0, 1.0 / 3.0), 10.0);
+        assert_eq!(dd_bucket_count(5.0, 5.0, 0.01), 1.0);
+        // 1 ms to 60 s at 1%: ln(60000) / ln(1.01 / 0.99) = 550.1.
+        assert_eq!(dd_bucket_count(0.001, 60.0, 0.01), 551.0);
+    }
+
+    #[test]
+    fn dd_memory_comes_from_the_value_range_when_known() {
+        let mut dd = deployment(1_000.0, 0.0, 0.0, 0.0, 60_000, 60_000);
+        dd.config.sketch = "dd".into();
+        dd.config.sketch_config = serde_json::json!({"params": {"alpha": 1.0 / 3.0}});
+        let mut ranged = facts(1, 1);
+        ranged
+            .get_mut(crate::test_support::METRIC)
+            .unwrap()
+            .value_range = Some((1.0, 1000.0));
+
+        let window = |d: &Deployment| d.window_ms;
+        assert_eq!(instance_memory_bytes(&dd, &ranged, window(&dd)), 80.0); // 10 buckets × 8 B
+        assert_eq!(ingest(&dd, &ranged).memory_bytes, 80.0);
+
+        // 1 sample/sec into a 5 s window: 5 values, fewer than the 10 buckets.
+        let short = Deployment {
+            window_ms: 5_000,
+            slide_ms: 5_000,
+            ..dd.clone()
+        };
+        assert_eq!(ingest(&short, &ranged).memory_bytes, 40.0);
+        // A 15 s query folds 3 windows: its accumulator holds 15 values, so
+        // the 10-bucket range is the cap again, not the window's 5.
+        assert_eq!(
+            merge(&raqe(15_000, 15_000), &short, &ranged).memory_bytes,
+            80.0
+        );
+        assert_eq!(
+            merge(&raqe(10_000, 10_000), &short, &ranged).memory_bytes,
+            80.0
+        );
+
+        // No range, or not DDSketch: the measured size.
+        assert_eq!(
+            instance_memory_bytes(&dd, &facts(1, 1), window(&dd)),
+            1_000.0
+        );
+        let cms = deployment(1_000.0, 0.0, 0.0, 0.0, 60_000, 60_000);
+        assert_eq!(instance_memory_bytes(&cms, &ranged, window(&cms)), 1_000.0);
+    }
+
     #[test]
     fn scores_each_phase() {
         // 4 groups and 7 samples/sec. Window 20 s sliding by 10 s: 2 open

rqe-optimizer/src/candidates.rs

@@ -1,6 +1,6 @@
 //! Candidate generation and eligibility (§3).
 
-use crate::analytical_cost_model::merge_memory_per_group;
+use crate::analytical_cost_model::{instance_memory_bytes, merge_memory_per_group};
 use crate::{AtomicCostEntry, Capability, Deployment, LabelSet, Millis, Raqe, WorkloadFacts};
 use std::collections::{BTreeMap, BTreeSet};
 
@@ -101,6 +101,7 @@ pub fn build_all_candidates(
 ) -> Vec<Deployment> {
     prune_dominated_candidates(
         raqes,
+        facts,
         build_all_candidates_unpruned(raqes, costs, facts, allow_undeployable_families),
     )
 }
@@ -148,7 +149,11 @@ pub fn build_all_candidates_unpruned(
 /// multipliers. So it compares ingest CPU and memory per group, then latency
 /// (merge and query CPU), merge memory and stored memory for each RAQE the
 /// dominated candidate can serve.
-pub fn prune_dominated_candidates(raqes: &[Raqe], candidates: Vec<Deployment>) -> Vec<Deployment> {
+pub fn prune_dominated_candidates(
+    raqes: &[Raqe],
+    facts: &WorkloadFacts,
+    candidates: Vec<Deployment>,
+) -> Vec<Deployment> {
     let eligibility: Vec<Vec<bool>> = candidates
         .iter()
         .map(|candidate| {
@@ -173,6 +178,7 @@ pub fn prune_dominated_candidates(raqes: &[Raqe], candidates: Vec<Deployment>) -
                     candidate,
                     &eligibility[candidate_index],
                     raqes,
+                    facts,
                 ) {
                     Dominance::StrictlyBetter => true,
                     // The earlier of two identical candidates wins, so a tie
@@ -200,6 +206,7 @@ fn dominance(
     original: &Deployment,
     original_coverage: &[bool],
     raqes: &[Raqe],
+    facts: &WorkloadFacts,
 ) -> Dominance {
     let same_group = replacement.capability == original.capability
         && replacement.metric == original.metric
@@ -216,7 +223,10 @@ fn dominance(
     // Each pair is (replacement's cost, original's cost).
     let mut costs = vec![
         (ingest_cpu(replacement), ingest_cpu(original)),
-        (ingest_memory(replacement), ingest_memory(original)),
+        (
+            ingest_memory(replacement, facts),
+            ingest_memory(original, facts),
+        ),
     ];
     for (raqe, _) in raqes
         .iter()
@@ -228,12 +238,12 @@ fn dominance(
             query_latency(original, raqe),
         ));
         costs.push((
-            merge_memory_per_group(raqe, replacement),
-            merge_memory_per_group(raqe, original),
+            merge_memory_per_group(raqe, replacement, facts),
+            merge_memory_per_group(raqe, original, facts),
         ));
         costs.push((
-            stored_memory(replacement, raqe),
-            stored_memory(original, raqe),
+            stored_memory(replacement, raqe, facts),
+            stored_memory(original, raqe, facts),
         ));
     }
 
@@ -269,8 +279,8 @@ fn ingest_cpu(deployment: &Deployment) -> f64 {
 
 /// Open-window bytes per group. `card(G)` is a common multiplier within a
 /// group, so it is left out.
-fn ingest_memory(deployment: &Deployment) -> f64 {
-    open_window_count(deployment) * deployment.config.mem_bytes_per_instance
+fn ingest_memory(deployment: &Deployment, facts: &WorkloadFacts) -> f64 {
+    open_window_count(deployment) * instance_memory_bytes(deployment, facts, deployment.window_ms)
 }
 
 fn query_latency(deployment: &Deployment, raqe: &Raqe) -> f64 {
@@ -282,11 +292,11 @@ fn query_latency(deployment: &Deployment, raqe: &Raqe) -> f64 {
 }
 
 /// Closed-window bytes per group when serving `raqe`.
-fn stored_memory(deployment: &Deployment, raqe: &Raqe) -> f64 {
+fn stored_memory(deployment: &Deployment, raqe: &Raqe, facts: &WorkloadFacts) -> f64 {
     deployment
         .closed_instance_count(raqe.lookback_ms)
         .expect("candidate coverage only contains exactly tiled RAQEs") as f64
-        * deployment.config.mem_bytes_per_instance
+        * instance_memory_bytes(deployment, facts, deployment.window_ms)
 }
 
 pub fn eligible_deployments_for(r: &Raqe, deployments: &[Deployment]) -> Vec<usize> {
@@ -469,7 +479,7 @@ mod tests {
             ..coarse.clone()
         };
 
-        let retained = prune_dominated_candidates(&[r], vec![fine, coarse.clone()]);
+        let retained = prune_dominated_candidates(&[r], &facts(1, 1), vec![fine, coarse.clone()]);
 
         assert_eq!(retained, vec![coarse]);
     }
@@ -493,8 +503,11 @@ mod tests {
             ..large_window.clone()
         };
 
-        let retained =
-            prune_dominated_candidates(&[r], vec![small_window.clone(), large_window.clone()]);
+        let retained = prune_dominated_candidates(
+            &[r],
+            &facts(1, 1),
+            vec![small_window.clone(), large_window.clone()],
+        );
 
         assert_eq!(retained, vec![large_window]);
     }
@@ -528,7 +541,11 @@ mod tests {
             ..whole_window.clone()
         };
 
-        let retained = prune_dominated_candidates(&[r], vec![whole_window.clone(), panes.clone()]);
+        let retained = prune_dominated_candidates(
+            &[r],
+            &facts(1, 1),
+            vec![whole_window.clone(), panes.clone()],
+        );
 
         assert_eq!(retained, vec![whole_window, panes]);
     }
@@ -563,7 +580,8 @@ mod tests {
             ..direct.clone()
         };
 
-        let retained = prune_dominated_candidates(&[r], vec![direct.clone(), halves.clone()]);
+        let retained =
+            prune_dominated_candidates(&[r], &facts(1, 1), vec![direct.clone(), halves.clone()]);
 
         assert_eq!(retained, vec![direct, halves]);
     }