Skip to content

Sum AggregationType ingest + CMS per-item estimate(key) #355

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
zzylol merged 5 commits into from
May 29, 2026
Merged

Sum AggregationType ingest + CMS per-item estimate(key) #355

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
d51cab8
CMS per-item estimate(key): reducer estimate path + engine item-key gate
zzylol May 28, 2026
b9da087
Sum AggregationType envelope ingest (SumAgg -> ExactAgg(Sum))
zzylol May 28, 2026
52b5fa9
Wire item_label controller->data-plane so CMS estimate(key) is reachable
zzylol May 28, 2026
6485f81
data-plane: honest SketchStore memory diagnostic + idle-sid eviction
zzylol May 28, 2026
2748dae
Merge pull request #356 from ProjectASAP/feat/sketchstore-idle-eviction
zzylol May 29, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions control_plane/src/emit/backend_push.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -638,6 +638,7 @@ mod tests {
sketch_kind: SketchKind::DDSketch,
sketch_params: SketchParams::DDSketch(DDSketchParams { alpha: 0.01 }),
grouping: vec![],
item_label: None,
spatial_filter: String::new(),
window_secs: 60,
aggregation_input: AggregationInput::SketchEnvelope,
Expand Down
22 changes: 21 additions & 1 deletion control_plane/src/emit/stage_config.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2694,13 +2694,23 @@ fn build_backend_aggregation_json(agg: &BackendAggregation) -> JsonValue {
// synthesizes for non-sketch (Sum-shaped) workloads. The
// `sketch_kind` / `sketch_params` fields carry sentinel values
// in this case and are not emitted on the wire.
let (aggregation_type, parameters) = match &agg.agg_type_override {
let (aggregation_type, mut parameters) = match &agg.agg_type_override {
Some(s) => (s.clone(), json!({})),
None => (
sketch_kind_to_backend_type(&agg.sketch_kind, &agg.sketch_params).to_string(),
sketch_params_to_json(&agg.sketch_params),
),
};
// Carry the per-item dimension (e.g. "endpoint"/"service") into the
// policy parameters so the data-plane ingest can record it on the CMS
// sid and answer per-item estimate(key). Only set for item_label-mode
// frequency sketches; a subset content-match keeps policy resolution
// working for sketches that don't carry it.
if let Some(label) = &agg.item_label {
if let Some(obj) = parameters.as_object_mut() {
obj.insert("item_label".to_string(), JsonValue::String(label.clone()));
}
}
let aggregation_input = match agg.aggregation_input {
AggregationInput::SketchEnvelope => "sketch_envelope",
AggregationInput::Raw => "raw",
Expand Down Expand Up @@ -3097,6 +3107,7 @@ mod tests {
let cfg = BackendStageConfig {
aggregations: vec![
BackendAggregation {
item_label: None,
aggregation_id: "agg0".into(),
metric_name: "http_latency_ms".into(),
sketch_kind: SketchKind::DDSketch,
Expand All @@ -3108,6 +3119,7 @@ mod tests {
agg_type_override: None,
},
BackendAggregation {
item_label: None,
aggregation_id: "agg1".into(),
metric_name: "http_requests_total".into(),
sketch_kind: SketchKind::Hll,
Expand Down Expand Up @@ -3161,6 +3173,7 @@ mod tests {
let cfg = BackendStageConfig {
aggregations: vec![
BackendAggregation {
item_label: None,
aggregation_id: "agg0".into(),
metric_name: "endpoint_count".into(),
sketch_kind: SketchKind::CountSketch,
Expand All @@ -3176,6 +3189,7 @@ mod tests {
agg_type_override: None,
},
BackendAggregation {
item_label: None,
aggregation_id: "agg1".into(),
metric_name: "endpoint_hits".into(),
sketch_kind: SketchKind::Cms,
Expand Down Expand Up @@ -3247,6 +3261,7 @@ mod tests {
};
BackendStageConfig {
aggregations: vec![BackendAggregation {
item_label: None,
aggregation_id: "agg0".into(),
metric_name: "test_metric".into(),
sketch_kind: kind.clone(),
Expand Down Expand Up @@ -3627,6 +3642,7 @@ mod tests {
fn backend_json_emits_grouping_under_labels() {
let cfg = BackendStageConfig {
aggregations: vec![BackendAggregation {
item_label: None,
aggregation_id: "agg0".into(),
metric_name: "http_latency_ms".into(),
sketch_kind: SketchKind::DDSketch,
Expand Down Expand Up @@ -3672,6 +3688,7 @@ mod tests {
fn phase_b_backend_json_aggregation_readout_alias_snapshot() {
let cfg = BackendStageConfig {
aggregations: vec![BackendAggregation {
item_label: None,
aggregation_id: "phase_b_agg0".into(),
metric_name: "phase_b_metric".into(),
sketch_kind: SketchKind::Kll,
Expand Down Expand Up @@ -3719,6 +3736,7 @@ mod tests {
fn phase_eps1_mode1_aggregation_input_is_sketch_envelope() {
let cfg = BackendStageConfig {
aggregations: vec![BackendAggregation {
item_label: None,
aggregation_id: "agg0".into(),
metric_name: "test_metric".into(),
sketch_kind: SketchKind::DDSketch,
Expand All @@ -3743,6 +3761,7 @@ mod tests {
fn phase_eps1_mode2_aggregation_input_is_raw() {
let cfg = BackendStageConfig {
aggregations: vec![BackendAggregation {
item_label: None,
aggregation_id: "agg0".into(),
metric_name: "test_metric".into(),
sketch_kind: SketchKind::DDSketch,
Expand Down Expand Up @@ -5429,6 +5448,7 @@ mod tests {

let cfg = BackendStageConfig {
aggregations: vec![BackendAggregation {
item_label: None,
aggregation_id: "agg0".to_string(),
metric_name: "http_requests_total_latency_ms".to_string(),
sketch_kind: SketchKind::DDSketch,
Expand Down
1 change: 1 addition & 0 deletions control_plane/src/emit/trait_def.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -234,6 +234,7 @@ mod tests {
fn empty_backend_cfg() -> BackendStageConfig {
BackendStageConfig {
aggregations: vec![BackendAggregation {
item_label: None,
aggregation_id: "agg0".to_string(),
metric_name: "test_metric".to_string(),
sketch_kind: SketchKind::DDSketch,
Expand Down
5 changes: 5 additions & 0 deletions control_plane/src/main.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -752,6 +752,10 @@ async fn handle_plan(
// `QueryWorkload` carries both unambiguously,
// and every aggregation under one workload
// shares them — so the patch is uniform.
let item_labels = emit::collect_metric_to_item_label(
&st.workload_registry,
&st.workload_store,
);
for agg in &mut be.aggregations {
if agg.metric_name.is_empty() {
agg.metric_name = workload.metric_name.clone();
Expand All @@ -760,6 +764,7 @@ async fn handle_plan(
agg.window_secs = workload.time_window.as_secs();
}
agg.grouping = workload.group_by_labels.clone();
agg.item_label = item_labels.get(&agg.metric_name).cloned();
}
// Option B unification: every typed cumulative
// emit (handle_plan here, Replanner triggers
Expand Down
10 changes: 10 additions & 0 deletions control_plane/src/physical/colored_dag/emitter.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -571,6 +571,14 @@ pub struct BackendAggregation {
/// strings are the only reliable source of the names today.
#[serde(default)]
pub grouping: Vec<String>,
/// Per-item dimension (the data-point attribute NAME, e.g. "endpoint"
/// or "service") for an item_label-mode frequency sketch. Like
/// `grouping`, the L5 emitter leaves this `None`; `handle_plan` patches
/// it from the workload's `item_label`. Emitted into the aggregation's
/// `parameters["item_label"]` so the data-plane ingest records it on the
/// CMS sid and can answer per-item `estimate(key)` (FrequencyEstimate).
#[serde(default)]
pub item_label: Option<String>,
/// Phase ε.1 — what shape the backend ingests for this
/// aggregation. Mode 1 (sketch at edge) / sketch_envelope is the
/// default (the wire payload is a sketch state already). Mode 2
Expand Down Expand Up @@ -786,6 +794,7 @@ impl Emitter for ThreeStageEmitter {
aggregation_id: aggregation_id.clone(),
});
backend_aggregations.push(BackendAggregation {
item_label: None,
aggregation_id,
metric_name: edge.source_metric.clone().unwrap_or_default(),
sketch_kind: sketch_type.clone(),
Expand Down Expand Up @@ -869,6 +878,7 @@ impl Emitter for ThreeStageEmitter {
let aid = format!("agg{next_agg_index}");
next_agg_index += 1;
backend_aggregations.push(BackendAggregation {
item_label: None,
aggregation_id: aid,
metric_name: edge.source_metric.clone().unwrap_or_default(),
sketch_kind: family.clone(),
Expand Down
11 changes: 11 additions & 0 deletions control_plane/src/replan.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -598,6 +598,14 @@ impl Replanner {
// `ColumnId`s with no label-name resolution today). The
// `QueryWorkload` carries both unambiguously, and every
// aggregation under one workload shares them.
// Per-metric item_label (the high-card dimension a CMS/CountSketch
// hashes): threaded into the policy params so the data-plane ingest
// records it on the sid and can answer per-item estimate(key).
let item_labels = self
.workload_registry
.as_ref()
.map(|reg| crate::emit::collect_metric_to_item_label(reg, &self.workload_store))
.unwrap_or_default();
for agg in &mut be.aggregations {
if agg.metric_name.is_empty() {
agg.metric_name = workload.metric_name.clone();
Expand All @@ -606,6 +614,7 @@ impl Replanner {
agg.window_secs = workload.time_window.as_secs();
}
agg.grouping = workload.group_by_labels.clone();
agg.item_label = item_labels.get(&agg.metric_name).cloned();
}
return Some(be);
}
Expand Down Expand Up @@ -633,6 +642,7 @@ impl Replanner {
let window_secs = workload.time_window.as_secs().max(1);
Some(BackendStageConfig {
aggregations: vec![BackendAggregation {
item_label: None,
aggregation_id: format!("exact-{}-{}", workload.metric_name, role),
metric_name: workload.metric_name.clone(),
// Sentinel sketch_kind / sketch_params — `agg_type_override`
Expand Down Expand Up @@ -1242,6 +1252,7 @@ mod tests {
("http_requests_total".to_string(), AggRole::Sum),
BackendStageConfig {
aggregations: vec![BackendAggregation {
item_label: None,
aggregation_id: "exact-http_requests_total-sum".to_string(),
metric_name: "http_requests_total".to_string(),
sketch_kind: SketchKind::DDSketch,
Expand Down
34 changes: 34 additions & 0 deletions crates/asap_otel_proto/proto/opentelemetry/proto/metrics/v1/metrics.proto
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -213,6 +213,9 @@ message Metric {
CountSketch countsketch = 15;
CountMinSketch countminsketch = 16;
HLLSketch hllsketch = 17;
// Scalar Sum aggregate (AggregationKind = Sum); field 18 mirrors the
// collector pdata SumAgg oneof tag. Decoded into ExactAgg(Sum).
SumAgg sum_agg = 18;
}

// Additional metadata attributes that describe the metric. [Optional].
Expand Down Expand Up @@ -298,6 +301,37 @@ message DDSketch {
double relative_accuracy = 3;
}

// SumAgg represents a first-class scalar Sum aggregate (AggregationKind = Sum),
// carried as a portable {sum,count} envelope in the sketch bytes. Not a sketch;
// shares the modified-OTLP metric data oneof so the backend decodes it via the
// same envelope path (into the ExactAgg(Sum) accumulator).
message SumAgg {
repeated SumAggDataPoint data_points = 1;
AggregationTemporality aggregation_temporality = 2;
}

// SumAggDataPoint carries a scalar Sum aggregate as a {sum,count} envelope.
message SumAggDataPoint {
repeated opentelemetry.proto.common.v1.KeyValue attributes = 9;
fixed64 start_time_unix_nano = 2;
fixed64 time_unix_nano = 3;
// Serialized SumState envelope (sketchlib SketchEnvelope{sum:SumState}).
bytes sketch = 8;
SumAggEncoding encoding = 10;
repeated Exemplar exemplars = 11;
uint32 flags = 15;
uint64 series_id = 16;
}

// SumAggEncoding identifies how the SumAgg payload bytes are encoded.
enum SumAggEncoding {
SUM_AGG_ENCODING_UNSPECIFIED = 0;
SUM_AGG_ENCODING_PROTO = 1;
SUM_AGG_ENCODING_PROTO_DELTA = 2;
SUM_AGG_ENCODING_MSGPACK = 3;
SUM_AGG_ENCODING_MSGPACK_DELTA = 4;
}

// Summary metric data are used to convey quantile summaries,
// a Prometheus (see: https://prometheus.io/docs/concepts/metric_types/#summary)
// and OpenMetrics (see: https://github.com/prometheus/OpenMetrics/blob/4dbf6075567ab43296eed941037c12951faafb92/protos/prometheus.proto#L45)
Expand Down
44 changes: 44 additions & 0 deletions data_plane/src/drivers/ingest/otel.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1289,6 +1289,23 @@ async fn route_modified_otlp_sketches_to_precompute(
&cfg,
&group_by_keys,
);
// Per-item dimension (item_label) the controller threaded
// into the matched policy's parameters — recorded on the sid
// below so the query engine can answer per-item estimate(key)
// (the CMS/CountSketch FrequencyEstimate gate consults it).
let item_label_for_sid: Option<String> = {
let snap = ingest_state.config_snapshot();
snap.get_aggregation_config(policy_fp.as_u64())
.or_else(|| {
snap.get_all_aggregation_configs()
.values()
.find(|c| c.metric == canonical_name)
})
.and_then(|c| c.parameters.get("item_label"))
.and_then(|v| v.as_str())
.filter(|s| !s.is_empty())
.map(|s| s.to_string())
};
ingest_state.sketch_index.register(SketchInstanceMetadata {
sid,
metric_name: canonical_name.clone(),
Expand All @@ -1305,6 +1322,9 @@ async fn route_modified_otlp_sketches_to_precompute(
expires_at_ms: None,
policy_fp,
});
if let Some(label) = &item_label_for_sid {
ingest_state.sketch_index.set_item_label(sid, label);
}
} else if let Some(existing) = ingest_state.sketch_index.instance(sid) {
// P1-4 (a) — one-way capability UPGRADE. The sid
// was first registered from a non-heap frame
Expand Down Expand Up @@ -2433,6 +2453,7 @@ fn otlp_to_record_count(request: &ExportMetricsServiceRequest) -> usize {
Some(Data::Countsketch(c)) => count += c.data_points.len(),
Some(Data::Countminsketch(c)) => count += c.data_points.len(),
Some(Data::Hllsketch(h)) => count += h.data_points.len(),
Some(Data::SumAgg(sa)) => count += sa.data_points.len(),
None => {}
}
}
Expand Down Expand Up @@ -2522,6 +2543,29 @@ fn otlp_to_metric_points_and_sketches(request: &ExportMetricsServiceRequest) ->
});
}
}
Some(Data::SumAgg(sa)) => {
// First-class Sum AggregationType: each data point carries
// a SumState envelope ({sum,count}) in `sketch`. Decode it
// and feed the sum as a MetricPoint into the SAME
// ExactAgg(Sum) path as a plain delta Sum — the backend sums
// the per-window/per-shard partials for the same sid.
for dp in &sa.data_points {
let value = match crate::precompute_engine::operators::sum_accumulator::SumAccumulator::from_sum_bytes(&dp.sketch) {
Ok(acc) => acc.sum,
Err(e) => {
debug!("asap_edge: SumAgg data point decode failed (skipping): {e}");
continue;
}
};
let labels = merge_point_attributes(&base_labels, &dp.attributes);
points.push(MetricPoint {
name: metric.name.clone(),
labels,
timestamp_nanos: dp.time_unix_nano,
value,
});
}
}
Some(Data::Histogram(hist)) => {
for dp in &hist.data_points {
if let Some((attr_name, payload)) =
Expand Down
Loading