From f96f829e48e645627767ff0c4d88144c87694fc4 Mon Sep 17 00:00:00 2001 From: zz_y Date: Wed, 2 Sep 2026 21:58:43 -0600 Subject: [PATCH 1/3] feat(transport): add transmission plan and frame identity --- control_plane/src/backend_client.rs | 2 + control_plane/src/emit/backend_push.rs | 11 + control_plane/src/main.rs | 1 + control_plane/src/opamp/mod.rs | 10 + control_plane/src/physical/compiler.rs | 304 +++++++++++++++++- data_plane/src/drivers/ingest/otel.rs | 179 ++++++++++- data_plane/src/drivers/query/servers/http.rs | 11 + data_plane/src/main.rs | 13 + .../src/precompute_engine/ingest_handler.rs | 6 + .../types/hot_reload_config.rs | 24 ++ .../control-plane/physical-compiler.md | 13 + .../control-plane/plan-publication.md | 6 +- 12 files changed, 557 insertions(+), 23 deletions(-) diff --git a/control_plane/src/backend_client.rs b/control_plane/src/backend_client.rs index c0d05909b..7fd82482e 100644 --- a/control_plane/src/backend_client.rs +++ b/control_plane/src/backend_client.rs @@ -366,6 +366,7 @@ impl BackendClient { pub async fn post_physical_plan_typed( &self, precompute_plan: &crate::physical::compiler::PrecomputePlan, + transmission_plan: &crate::physical::compiler::TransmissionPlan, backend_plan: Vec, query_plan: &crate::query_plan::QueryPlan, storage_routing: Option, @@ -376,6 +377,7 @@ impl BackendClient { .post(&url) .json(&serde_json::json!({ "precompute_plan": precompute_plan, + "transmission_plan": transmission_plan, "backend_plan": backend_plan, "query_plan": query_plan, "storage_routing": storage_routing, diff --git a/control_plane/src/emit/backend_push.rs b/control_plane/src/emit/backend_push.rs index e3f9985ce..0d8a08ce9 100644 --- a/control_plane/src/emit/backend_push.rs +++ b/control_plane/src/emit/backend_push.rs @@ -252,11 +252,22 @@ async fn push_documents_coupled( .unwrap_or_default(), entries: Default::default(), }; + let transmission_plan = match crate::physical::compiler::TransmissionPlan::build( + precompute_plan.envelope.clone(), + precompute_plan, + ) { + Ok(plan) => plan, + Err(error) => { + warn!(%error, "failed to build compatibility TransmissionPlan"); + return (false, false, 0); + } + }; for attempt in 1..=RETRY_MAX_ATTEMPTS { match client .post_physical_plan_typed( precompute_plan, + &transmission_plan, plan_bytes.clone(), &query_plan, Some(routing.clone()), diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs index 602826745..f3ba480c5 100644 --- a/control_plane/src/main.rs +++ b/control_plane/src/main.rs @@ -650,6 +650,7 @@ async fn handle_compile_and_publish_physical_plan( if let Err(error) = backend .post_physical_plan_typed( &bundle.precompute_plan, + &bundle.transmission_plan, bundle.backend_plan.encode_to_vec(), &bundle.query_plan, None, diff --git a/control_plane/src/opamp/mod.rs b/control_plane/src/opamp/mod.rs index 1c8a41329..80603a285 100644 --- a/control_plane/src/opamp/mod.rs +++ b/control_plane/src/opamp/mod.rs @@ -1035,6 +1035,16 @@ mod tests { output_representation: "summary_state".into(), }, }], + transmission_rules: vec![crate::physical::compiler::TransmissionRule { + materialization: asap_types::PolicyFingerprint(1), + producer_id: collector_id.into(), + schema_id: "asap-query-backend.v1:summary-state:v1:1".into(), + mode: crate::physical::compiler::TransmissionMode::Full, + encoding: crate::physical::compiler::StateEncoding::SketchlibProtobufV1, + emit_every_ms: 60_000, + full_checkpoint_every_ms: None, + destination_ref: "asapquery-backend".into(), + }], } } diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index ca2c2c2bb..d8b52b6b3 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -145,7 +145,7 @@ pub struct DeploymentEnvironment { pub backend_compat: String, } -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub struct PlanEnvelope { pub plan_id: u64, pub plan_version: u64, @@ -188,6 +188,7 @@ pub struct CollectorPlan { pub collector_id: String, pub envelope: PlanEnvelope, pub materializations: Vec, + pub transmission_rules: Vec, } /// Backend-side materialization projection consumed by the streaming @@ -229,7 +230,7 @@ pub struct IngestContract { pub require_registered_producer: bool, } -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)] #[serde(rename_all = "snake_case")] pub enum StateEncoding { SketchlibProtobufV1, @@ -525,10 +526,233 @@ pub struct PhysicalPlan { pub envelope: PlanEnvelope, pub collector_plans: Vec, pub precompute_plan: PrecomputePlan, + pub transmission_plan: TransmissionPlan, pub backend_plan: BackendPlan, pub query_plan: QueryPlan, } +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)] +#[serde(rename_all = "snake_case")] +pub enum TransmissionMode { + Full, + Delta, +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum SequenceScope { + MaterializationWindowProducerEpoch, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct FrameIdentityContract { + pub identity_version: u32, + pub sequence_scope: SequenceScope, + pub require_checkpoint_for_full: bool, + pub require_base_checkpoint_for_delta: bool, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)] +#[serde(deny_unknown_fields)] +pub struct TransmissionRule { + pub materialization: asap_types::PolicyFingerprint, + pub producer_id: String, + pub schema_id: String, + pub mode: TransmissionMode, + pub encoding: StateEncoding, + pub emit_every_ms: u64, + pub full_checkpoint_every_ms: Option, + pub destination_ref: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct TransmissionPlan { + pub envelope: PlanEnvelope, + pub frame_identity: FrameIdentityContract, + pub rules: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum SummaryFrameKind { + Full, + Delta, +} + +/// Identity attached to every summary record. Window bounds come from the +/// data point; the remaining fields are carried as reserved `asap.frame.*` +/// attributes until the modified-OTLP schema gains a dedicated message. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct SummaryFrameIdentity { + pub identity_version: u32, + pub plan_id: u64, + pub plan_version: u64, + pub backend_compat: String, + pub materialization: asap_types::PolicyFingerprint, + pub schema_id: String, + pub producer_id: String, + pub producer_epoch: String, + pub window_start_unix_nano: u64, + pub window_end_unix_nano: u64, + pub sequence: u64, + pub kind: SummaryFrameKind, + pub encoding: StateEncoding, + pub checkpoint_id: Option, + pub base_checkpoint_id: Option, +} + +#[derive(Debug, Error, PartialEq, Eq)] +pub enum TransmissionPlanError { + #[error("TransmissionPlan envelope differs from PrecomputePlan")] + EnvelopeMismatch, + #[error("transmission rules do not exactly match precompute producer bindings")] + ProducerSetMismatch, + #[error("invalid transmission rule for producer {0}")] + InvalidRule(String), + #[error("frame identity is invalid: {0}")] + InvalidFrame(String), + #[error("frame has no matching transmission rule")] + UnknownFrame, +} + +impl TransmissionPlan { + pub fn build( + envelope: PlanEnvelope, + precompute: &PrecomputePlan, + ) -> Result { + if envelope != precompute.envelope { + return Err(TransmissionPlanError::EnvelopeMismatch); + } + let schemas: BTreeMap<_, _> = precompute + .schemas + .iter() + .map(|schema| (schema.materialization, schema)) + .collect(); + let rules = precompute + .producers + .iter() + .map(|producer| { + let schema = schemas + .get(&producer.materialization) + .expect("validated PrecomputePlan schema binding"); + let materialization = precompute + .materializations + .iter() + .find(|m| m.policy_fingerprint() == producer.materialization) + .expect("validated PrecomputePlan materialization binding"); + TransmissionRule { + materialization: producer.materialization, + producer_id: producer.producer_id.clone(), + schema_id: producer.schema_id.clone(), + mode: TransmissionMode::Full, + encoding: schema.encodings[0].clone(), + emit_every_ms: materialization.window_size.saturating_mul(1_000), + full_checkpoint_every_ms: None, + destination_ref: "asapquery-backend".into(), + } + }) + .collect(); + let plan = Self { + envelope, + frame_identity: FrameIdentityContract { + identity_version: 1, + sequence_scope: SequenceScope::MaterializationWindowProducerEpoch, + require_checkpoint_for_full: true, + require_base_checkpoint_for_delta: true, + }, + rules, + }; + plan.validate(precompute)?; + Ok(plan) + } + + pub fn validate(&self, precompute: &PrecomputePlan) -> Result<(), TransmissionPlanError> { + if self.envelope != precompute.envelope { + return Err(TransmissionPlanError::EnvelopeMismatch); + } + let expected: BTreeSet<_> = precompute + .producers + .iter() + .map(|producer| { + ( + producer.materialization, + producer.producer_id.as_str(), + producer.schema_id.as_str(), + ) + }) + .collect(); + let actual: BTreeSet<_> = self + .rules + .iter() + .map(|rule| { + ( + rule.materialization, + rule.producer_id.as_str(), + rule.schema_id.as_str(), + ) + }) + .collect(); + if expected != actual || actual.len() != self.rules.len() { + return Err(TransmissionPlanError::ProducerSetMismatch); + } + for rule in &self.rules { + if rule.emit_every_ms == 0 + || rule.destination_ref.is_empty() + || (rule.mode == TransmissionMode::Delta + && rule + .full_checkpoint_every_ms + .map_or(true, |value| value == 0)) + { + return Err(TransmissionPlanError::InvalidRule(rule.producer_id.clone())); + } + } + Ok(()) + } + + pub fn validate_frame( + &self, + frame: &SummaryFrameIdentity, + ) -> Result<(), TransmissionPlanError> { + if frame.identity_version != self.frame_identity.identity_version + || frame.plan_id != self.envelope.plan_id + || frame.plan_version != self.envelope.plan_version + || frame.backend_compat != self.envelope.backend_compat + || frame.producer_epoch.is_empty() + || frame.sequence == 0 + || frame.window_start_unix_nano >= frame.window_end_unix_nano + || (frame.kind == SummaryFrameKind::Full + && self.frame_identity.require_checkpoint_for_full + && frame.checkpoint_id.is_none()) + || (frame.kind == SummaryFrameKind::Delta + && self.frame_identity.require_base_checkpoint_for_delta + && frame.base_checkpoint_id.is_none()) + { + return Err(TransmissionPlanError::InvalidFrame( + "identity/lifecycle/window/checkpoint fields do not satisfy the active contract" + .into(), + )); + } + let mode = match frame.kind { + SummaryFrameKind::Full => TransmissionMode::Full, + SummaryFrameKind::Delta => TransmissionMode::Delta, + }; + if self.rules.iter().any(|rule| { + rule.materialization == frame.materialization + && rule.producer_id == frame.producer_id + && rule.schema_id == frame.schema_id + && rule.mode == mode + && rule.encoding == frame.encoding + }) { + Ok(()) + } else { + Err(TransmissionPlanError::UnknownFrame) + } + } +} + #[derive(Debug, Error)] pub enum CompileError { #[error("planner revision mismatch: request={request}, compiler={compiler}")] @@ -717,15 +941,7 @@ impl PhysicalCompiler { output_representation: OutputRepresentation::SummaryState, }); } - let collector_plans = environment - .collector_ids - .into_iter() - .map(|collector_id| CollectorPlan { - collector_id, - envelope: envelope.clone(), - materializations: collector_materializations.clone(), - }) - .collect::>(); + let producer_ids = environment.collector_ids.clone(); // Several queries/readouts may intentionally share one maintained // summary. PrecomputePlan is keyed by physical identity, not query ID. let mut materializations_by_fingerprint = BTreeMap::new(); @@ -737,10 +953,6 @@ impl PhysicalCompiler { .or_insert(materialization); } let materializations = materializations_by_fingerprint.into_values().collect(); - let producer_ids = collector_plans - .iter() - .map(|plan| plan.collector_id.clone()) - .collect::>(); let precompute_plan = PrecomputePlan::build( envelope.clone(), materializations, @@ -751,6 +963,25 @@ impl PhysicalCompiler { query_id: "precompute-plan".into(), reason: error.to_string(), })?; + let transmission_plan = TransmissionPlan::build(envelope.clone(), &precompute_plan) + .map_err(|error| CompileError::Query { + query_id: "transmission-plan".into(), + reason: error.to_string(), + })?; + let collector_plans = producer_ids + .into_iter() + .map(|collector_id| CollectorPlan { + transmission_rules: transmission_plan + .rules + .iter() + .filter(|rule| rule.producer_id == collector_id) + .cloned() + .collect(), + collector_id, + envelope: envelope.clone(), + materializations: collector_materializations.clone(), + }) + .collect::>(); let materialization_fingerprints: BTreeSet<_> = backend_plan.materializations.keys().copied().collect(); let mut query_entries = BTreeMap::new(); @@ -850,6 +1081,7 @@ impl PhysicalCompiler { envelope, collector_plans, precompute_plan, + transmission_plan, backend_plan, query_plan, }) @@ -934,6 +1166,14 @@ fn state_schema_id(fingerprint: asap_types::PolicyFingerprint) -> String { fn state_encodings(family: &SummaryFamilyType) -> Vec { match family { SummaryFamilyType::ExactAggregate(..) => vec![StateEncoding::ExactAccumulatorV1], + SummaryFamilyType::Sketch(kind, _) + if matches!( + kind.algorithm(), + SketchAlgorithm::CmsWithHeap | SketchAlgorithm::CountSketchWithHeap + ) => + { + vec![StateEncoding::SketchCoreMsgpackV1] + } SummaryFamilyType::Sketch(..) => vec![ StateEncoding::SketchlibProtobufV1, StateEncoding::SketchCoreMsgpackV1, @@ -1319,6 +1559,39 @@ mod tests { assert_eq!(bundle.precompute_plan.producers.len(), 2); assert_eq!(bundle.precompute_plan.ingest.endpoint_path, "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/v1/metrics"); bundle.precompute_plan.validate().expect("runtime contract"); + bundle + .transmission_plan + .validate(&bundle.precompute_plan) + .expect("transmission contract"); + assert_eq!(bundle.transmission_plan.rules.len(), 2); + let rule = &bundle.transmission_plan.rules[0]; + let frame = SummaryFrameIdentity { + identity_version: 1, + plan_id: bundle.envelope.plan_id, + plan_version: bundle.envelope.plan_version, + backend_compat: bundle.envelope.backend_compat.clone(), + materialization: rule.materialization, + schema_id: rule.schema_id.clone(), + producer_id: rule.producer_id.clone(), + producer_epoch: "boot-1".into(), + window_start_unix_nano: 1, + window_end_unix_nano: 2, + sequence: 1, + kind: SummaryFrameKind::Full, + encoding: rule.encoding.clone(), + checkpoint_id: Some("checkpoint-1".into()), + base_checkpoint_id: None, + }; + bundle + .transmission_plan + .validate_frame(&frame) + .expect("matching frame identity"); + let mut wrong_version = frame; + wrong_version.plan_version += 1; + assert!(matches!( + bundle.transmission_plan.validate_frame(&wrong_version), + Err(TransmissionPlanError::InvalidFrame(_)) + )); assert_eq!(bundle.backend_plan.materializations.len(), 1); assert_eq!( bundle @@ -1368,6 +1641,7 @@ mod tests { assert_eq!(plan.envelope, bundle.envelope); assert_eq!(plan.materializations[0].metric, "m"); assert_eq!(plan.materializations[0].window_secs, 60); + assert_eq!(plan.transmission_rules.len(), 1); assert_eq!( plan.materializations[0].abstract_window_framework, SummaryWindowFramework::Tumbling diff --git a/data_plane/src/drivers/ingest/otel.rs b/data_plane/src/drivers/ingest/otel.rs index 1c4f1fd02..3283f2460 100644 --- a/data_plane/src/drivers/ingest/otel.rs +++ b/data_plane/src/drivers/ingest/otel.rs @@ -30,8 +30,8 @@ use crate::precompute_engine::IngestState; use crate::query_engines::routing::FreshnessProbeCache; use crate::storage_engines::types::AggregateCore; use asap_otel_proto::tonic::collector::metrics::v1::{ - metrics_service_server::MetricsService, ExportMetricsServiceRequest, - ExportMetricsServiceResponse, + metrics_service_server::MetricsService, ExportMetricsPartialSuccess, + ExportMetricsServiceRequest, ExportMetricsServiceResponse, }; use asap_otel_proto::tonic::common::v1::any_value::Value as AnyValueVariant; use asap_otel_proto::tonic::metrics::v1::number_data_point::Value as NumberValue; @@ -216,7 +216,10 @@ impl MetricsService for MetricsServiceImpl { } debug!("OTLP sending response via gRPC"); Ok(Response::new(ExportMetricsServiceResponse { - partial_success: None, + partial_success: (outcome.rejected_frames > 0).then(|| ExportMetricsPartialSuccess { + rejected_data_points: outcome.rejected_frames as i64, + error_message: "summary frame identity rejected by active TransmissionPlan".into(), + }), // Sid bindings the sender should cache. Each entry maps an // `attributes_fingerprint` to the canonical sid the backend's // `SeriesIdResolver` minted (or returned from its cache). The @@ -328,7 +331,7 @@ async fn handle_otlp_http( // are the universal recovery primitive (see proto comment on // `ExportMetricsServiceResponse.unknown_series_ids`). Ok(Json(serde_json::json!({ - "rejected": 0, + "rejected": outcome.rejected_frames, "unknown_series_ids": outcome.unknown_series_ids, "series_assignments_count": outcome.series_assignments.len(), }))) @@ -857,6 +860,7 @@ async fn route_otlp_to_precompute( pub(crate) struct IngestOutcome { pub unknown_series_ids: Vec, pub series_assignments: Vec, + pub rejected_frames: u64, } async fn route_modified_otlp_sketches_to_precompute( @@ -867,6 +871,9 @@ async fn route_modified_otlp_sketches_to_precompute( let ingest_received_at = Instant::now(); let snap = ingest_state.config_snapshot(); + let active_physical_plan = ingest_state + .physical_plan_snapshot() + .filter(|plan| plan.backend_plan.plan_id != 0); let agg_configs = snap.get_all_aggregation_configs(); // Schema retirement #5 — agg_id-keyed registry retired; sid-level // reconcile is the only path going forward. See the raw-OTLP @@ -881,6 +888,7 @@ async fn route_modified_otlp_sketches_to_precompute( let mut routed = 0usize; let mut decoded_failed = 0usize; let mut unconfigured = 0usize; + let mut rejected_frames = 0u64; // CQ-2 — the legacy routing-side WorkerMessage push (the DEPRECATED // dual-write that clones the accumulator into the worker under a // bucket-sid, in tandem with the Phase-5 SketchStore append above) is @@ -1059,7 +1067,34 @@ async fn route_modified_otlp_sketches_to_precompute( None => metric.name.clone(), }; - for dp in dps { + for mut dp in dps { + let frame_identity = if let Some(active) = active_physical_plan.as_ref() { + match take_summary_frame_identity( + &mut dp.attrs, + dp.start_time_unix_nano, + dp.time_unix_nano, + ) + .and_then(|frame| { + if state_encoding_for_wire(dp.encoding) != Some(frame.encoding.clone()) + { + return Err("wire encoding differs from frame identity".into()); + } + active + .transmission_plan + .validate_frame(&frame) + .map_err(|error| error.to_string())?; + Ok(frame) + }) { + Ok(frame) => Some(frame), + Err(error) => { + rejected_frames = rejected_frames.saturating_add(1); + warn!(metric = %metric.name, %error, "rejected summary frame"); + continue; + } + } + } else { + None + }; let series_key = format_series_key(&canonical_name, &dp.attrs); let ts_ms = (dp.time_unix_nano / 1_000_000) as i64; @@ -1211,6 +1246,32 @@ async fn route_modified_otlp_sketches_to_precompute( continue; }; + if let Some(frame) = frame_identity.as_ref() { + let observed_policy = ingest_state + .sketch_index + .instance(sid) + .map(|metadata| metadata.policy_fp) + .unwrap_or_else(|| { + derive_sketch_policy_fp( + ingest_state, + &canonical_name, + sketch_kind_handle_for(&dp), + &dp.container_config, + &dp.attrs.keys().cloned().collect(), + ) + }); + if observed_policy != frame.materialization { + rejected_frames = rejected_frames.saturating_add(1); + warn!( + sid, + declared = frame.materialization.0, + observed = observed_policy.0, + "rejected summary frame with mismatched materialization" + ); + continue; + } + } + // Phase 5 — register a `SketchInstanceMetadata` on // first sight of `sid` and append this DP's sketch // state to the per-sid columnar storage. The instance @@ -1712,6 +1773,7 @@ async fn route_modified_otlp_sketches_to_precompute( IngestOutcome { unknown_series_ids: unknown_sids, series_assignments: new_assignments, + rejected_frames, } } @@ -2019,6 +2081,86 @@ struct ModifiedOtlpSketchDp { container_config: crate::storage_engines::sketch_db::index::SketchConfig, } +fn take_summary_frame_identity( + attrs: &mut HashMap, + window_start_unix_nano: u64, + window_end_unix_nano: u64, +) -> Result { + use control_plane::physical::compiler::{ + StateEncoding, SummaryFrameIdentity, SummaryFrameKind, + }; + + fn required(attrs: &mut HashMap, key: &str) -> Result { + attrs + .remove(key) + .filter(|value| !value.is_empty()) + .ok_or_else(|| format!("missing {key}")) + } + fn number(attrs: &mut HashMap, key: &str) -> Result { + required(attrs, key)? + .parse() + .map_err(|_| format!("invalid {key}")) + } + + let identity_version = u32::try_from(number(attrs, "asap.frame.identity_version")?) + .map_err(|_| "invalid asap.frame.identity_version".to_string())?; + let plan_id = number(attrs, "asap.frame.plan_id")?; + let plan_version = number(attrs, "asap.frame.plan_version")?; + let backend_compat = required(attrs, "asap.frame.backend_compat")?; + let materialization = + asap_types::PolicyFingerprint(number(attrs, "asap.frame.materialization")?); + let schema_id = required(attrs, "asap.frame.schema_id")?; + let producer_id = required(attrs, "asap.frame.producer_id")?; + let producer_epoch = required(attrs, "asap.frame.producer_epoch")?; + let sequence = number(attrs, "asap.frame.sequence")?; + let kind = match required(attrs, "asap.frame.kind")?.as_str() { + "full" => SummaryFrameKind::Full, + "delta" => SummaryFrameKind::Delta, + _ => return Err("invalid asap.frame.kind".into()), + }; + let encoding = match required(attrs, "asap.frame.encoding")?.as_str() { + "sketchlib_protobuf_v1" => StateEncoding::SketchlibProtobufV1, + "sketch_core_msgpack_v1" => StateEncoding::SketchCoreMsgpackV1, + "exact_accumulator_v1" => StateEncoding::ExactAccumulatorV1, + _ => return Err("invalid asap.frame.encoding".into()), + }; + let checkpoint_id = attrs + .remove("asap.frame.checkpoint_id") + .filter(|value| !value.is_empty()); + let base_checkpoint_id = attrs + .remove("asap.frame.base_checkpoint_id") + .filter(|value| !value.is_empty()); + + Ok(SummaryFrameIdentity { + identity_version, + plan_id, + plan_version, + backend_compat, + materialization, + schema_id, + producer_id, + producer_epoch, + window_start_unix_nano, + window_end_unix_nano, + sequence, + kind, + encoding, + checkpoint_id, + base_checkpoint_id, + }) +} + +fn state_encoding_for_wire( + encoding: i32, +) -> Option { + use control_plane::physical::compiler::StateEncoding; + match encoding { + ENCODING_PROTO | ENCODING_PROTO_DELTA => Some(StateEncoding::SketchlibProtobufV1), + ENCODING_MSGPACK | ENCODING_MSGPACK_DELTA => Some(StateEncoding::SketchCoreMsgpackV1), + _ => None, + } +} + /// Decode the typed `sketch` bytes from a modified-OTLP /// `*SketchDataPoint` into a concrete `AggregateCore`. /// @@ -4251,4 +4393,31 @@ mod sid_bucketing_tests { ); } } + + #[test] + fn summary_frame_identity_is_parsed_and_removed_from_series_labels() { + let mut attrs = HashMap::from([ + ("service".into(), "api".into()), + ("asap.frame.identity_version".into(), "1".into()), + ("asap.frame.plan_id".into(), "42".into()), + ("asap.frame.plan_version".into(), "3".into()), + ( + "asap.frame.backend_compat".into(), + "asap-query-backend.v1".into(), + ), + ("asap.frame.materialization".into(), "99".into()), + ("asap.frame.schema_id".into(), "schema-99".into()), + ("asap.frame.producer_id".into(), "edge-a".into()), + ("asap.frame.producer_epoch".into(), "boot-7".into()), + ("asap.frame.sequence".into(), "8".into()), + ("asap.frame.kind".into(), "full".into()), + ("asap.frame.encoding".into(), "sketchlib_protobuf_v1".into()), + ("asap.frame.checkpoint_id".into(), "cp-8".into()), + ]); + let frame = take_summary_frame_identity(&mut attrs, 100, 200).expect("valid identity"); + assert_eq!(frame.plan_id, 42); + assert_eq!(frame.plan_version, 3); + assert_eq!(frame.materialization, asap_types::PolicyFingerprint(99)); + assert_eq!(attrs, HashMap::from([("service".into(), "api".into())])); + } } diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs index 8030f65f7..25b0d5264 100644 --- a/data_plane/src/drivers/query/servers/http.rs +++ b/data_plane/src/drivers/query/servers/http.rs @@ -5297,6 +5297,7 @@ async fn handle_post_backend_plan( #[derive(serde::Deserialize)] struct PhysicalPlanInstallRequest { precompute_plan: control_plane::physical::compiler::PrecomputePlan, + transmission_plan: control_plane::physical::compiler::TransmissionPlan, backend_plan: Vec, query_plan: control_plane::query_plan::QueryPlan, storage_routing: Option, @@ -5341,6 +5342,15 @@ async fn handle_post_physical_plan( ) .into_response(), }; + if let Err(error) = request.transmission_plan.validate(&request.precompute_plan) { + return ( + StatusCode::UNPROCESSABLE_ENTITY, + axum::Json(serde_json::json!({ + "status": "error", "error": format!("TransmissionPlan validation error: {error}") + })), + ) + .into_response(); + } let new_config = crate::storage_engines::types::StreamingConfig::new(runtime_materializations); let new_plan = match control_plane::backend_plan::BackendPlan::decode(&request.backend_plan) { Ok(plan) => plan, @@ -5449,6 +5459,7 @@ async fn handle_post_physical_plan( let plan_id = new_plan.plan_id; let active = crate::storage_engines::types::ActivePhysicalPlan { precompute_plan: request.precompute_plan, + transmission_plan: request.transmission_plan, runtime_config: Arc::new(new_config), backend_plan: Arc::new(new_plan), query_plan: Arc::new(request.query_plan), diff --git a/data_plane/src/main.rs b/data_plane/src/main.rs index 24937765d..58e925ed9 100644 --- a/data_plane/src/main.rs +++ b/data_plane/src/main.rs @@ -451,9 +451,21 @@ async fn main() -> Result<()> { .cloned() .collect(), }; + let initial_transmission_plan = control_plane::physical::compiler::TransmissionPlan { + envelope: initial_precompute_plan.envelope.clone(), + frame_identity: control_plane::physical::compiler::FrameIdentityContract { + identity_version: 1, + sequence_scope: + control_plane::physical::compiler::SequenceScope::MaterializationWindowProducerEpoch, + require_checkpoint_for_full: true, + require_base_checkpoint_for_delta: true, + }, + rules: Vec::new(), + }; let active_physical_plan = data_plane::storage_engines::types::HotReloadActivePhysicalPlan::new( data_plane::storage_engines::types::ActivePhysicalPlan { precompute_plan: initial_precompute_plan, + transmission_plan: initial_transmission_plan, runtime_config: streaming_config.clone(), backend_plan: Arc::new(initial_backend_plan), query_plan: Arc::new(control_plane::query_plan::QueryPlan::empty()), @@ -819,6 +831,7 @@ async fn main() -> Result<()> { let current = active_physical_plan.snapshot(); active_physical_plan.swap(data_plane::storage_engines::types::ActivePhysicalPlan { precompute_plan: current.precompute_plan.clone(), + transmission_plan: current.transmission_plan.clone(), runtime_config: current.runtime_config.clone(), backend_plan: current.backend_plan.clone(), query_plan: current.query_plan.clone(), diff --git a/data_plane/src/precompute_engine/ingest_handler.rs b/data_plane/src/precompute_engine/ingest_handler.rs index 15810ecba..e5eacc1d0 100644 --- a/data_plane/src/precompute_engine/ingest_handler.rs +++ b/data_plane/src/precompute_engine/ingest_handler.rs @@ -180,6 +180,12 @@ impl IngestState { self.hot_reload_config.snapshot() } + pub fn physical_plan_snapshot( + &self, + ) -> Option> { + self.hot_reload_config.physical_plan_snapshot() + } + /// RES-1 — record that a per-series snapshot base for `window_start` /// was just (re)inserted, then opportunistically sweep stale entries. /// diff --git a/data_plane/src/storage_engines/types/hot_reload_config.rs b/data_plane/src/storage_engines/types/hot_reload_config.rs index 8e368b8ff..2b0151528 100644 --- a/data_plane/src/storage_engines/types/hot_reload_config.rs +++ b/data_plane/src/storage_engines/types/hot_reload_config.rs @@ -87,6 +87,7 @@ use crate::storage_engines::types::StreamingConfig; #[derive(Debug, Clone)] pub struct ActivePhysicalPlan { pub precompute_plan: control_plane::physical::compiler::PrecomputePlan, + pub transmission_plan: control_plane::physical::compiler::TransmissionPlan, pub runtime_config: Arc, pub backend_plan: Arc, pub query_plan: Arc, @@ -570,6 +571,10 @@ impl HotReloadStreamingConfig { .unwrap_or_else(|| self.inner.load_full()) } + pub fn physical_plan_snapshot(&self) -> Option> { + self.active.as_ref().map(|active| active.snapshot()) + } + /// Atomically replace the current config. The previous `Arc` is /// dropped when the last reader holding it goes out of scope. /// Returns the `Arc` that was just replaced, for callers that @@ -632,6 +637,25 @@ mod tests { producers: Vec::new(), materializations: Vec::new(), }, + transmission_plan: control_plane::physical::compiler::TransmissionPlan { + envelope: control_plane::physical::compiler::PlanEnvelope { + plan_id, + plan_version, + generated_at_unix_ms: activation_unix_ms, + activation_unix_ms, + expiry_unix_ms, + backend_compat: "asap-query-backend.v1".into(), + planner_revision: control_plane::physical::compiler::PLANNER_REVISION.into(), + capability_snapshot_id: "test".into(), + }, + frame_identity: control_plane::physical::compiler::FrameIdentityContract { + identity_version: 1, + sequence_scope: control_plane::physical::compiler::SequenceScope::MaterializationWindowProducerEpoch, + require_checkpoint_for_full: true, + require_base_checkpoint_for_delta: true, + }, + rules: Vec::new(), + }, runtime_config: Arc::new(StreamingConfig::new(HashMap::new())), backend_plan: Arc::new(control_plane::backend_plan::BackendPlan { plan_id, diff --git a/docs/developer_docs/control-plane/physical-compiler.md b/docs/developer_docs/control-plane/physical-compiler.md index 948297261..c79e70e47 100644 --- a/docs/developer_docs/control-plane/physical-compiler.md +++ b/docs/developer_docs/control-plane/physical-compiler.md @@ -130,6 +130,7 @@ pub struct PhysicalPlan { pub envelope: PlanEnvelope, pub collector_plans: Vec, // complete per-target projections pub precompute_plan: PrecomputePlan, // backend streaming materializations + pub transmission_plan: TransmissionPlan, // producer/frame wire contract pub backend_plan: BackendPlan, pub query_plan: QueryPlan, // node-ID physical serving DAG } @@ -143,6 +144,7 @@ Supporting public types: | `PlanEnvelope` | Shared deterministic `plan_id`, generation time, capability snapshot, and Planner revision. | | `CollectorPlan` | Serializable execution projection consumed by ASAPCollector. | | `PrecomputePlan` | Authoritative materialization, ingest, state-schema, and producer contract consumed directly by the backend runtime. | +| `TransmissionPlan` | Exact per-producer mode, encoding, cadence, destination, checkpoint policy, and frame identity contract. | | `BackendPlan` | Versioned public data-plane materialization/routing contract defined in this repository. | | `QueryPlan` | Canonical-query keyed executable DAG with exact materialization bindings and fallback policy. | @@ -158,6 +160,7 @@ Output definitions: | `envelope` | Shared plan/version/activation/compatibility identity. | | `collector_plans` | One plan per targeted collector, following ASAPCollector's public CollectorPlan schema. | | `precompute_plan` | Materializations plus `/v1/metrics` ingest semantics, typed state schemas/encodings, and allowed producers; it is installed directly and contains no query-string jobs. | +| `transmission_plan` | One rule per producer/materialization/schema binding plus the mandatory frame identity and sequencing scope. | | `backend_plan` | Matching data-plane materialization and routing contract. | | `query_plan` | Canonical query identity, explicit fallback policy, node-ID DAG, and exact per-node materialization bindings. | @@ -272,6 +275,16 @@ route references a declared materialization. A Backend-only precompute has a BackendPlan producer but no Collector materialization; a raw pass-through has matching raw transmission and ingest/archive declarations. +### Summary frame identity + +Until modified OTLP has dedicated identity fields, Collector attaches reserved +data-point attributes under `asap.frame.*`: identity version, plan ID/version, +backend compatibility, materialization and schema IDs, producer ID/epoch, +sequence, full/delta kind, encoding, and checkpoint/base IDs. Window start/end +remain the typed data-point timestamps. The backend removes reserved attributes +before building the series label key and rejects any frame that does not match +the active TransmissionPlan. + The compiler error must identify an unsupported capability, invalid placement, window incompatibility, identity conflict, or invalid selected guarantee. It must not silently substitute another logical summary. diff --git a/docs/developer_docs/control-plane/plan-publication.md b/docs/developer_docs/control-plane/plan-publication.md index 712ab4456..036209930 100644 --- a/docs/developer_docs/control-plane/plan-publication.md +++ b/docs/developer_docs/control-plane/plan-publication.md @@ -39,7 +39,7 @@ validate request and compile one bundle preflight every Collector capability | v -POST one atomic PrecomputePlan + BackendPlan + QueryPlan bundle (stage) +POST one atomic PrecomputePlan + TransmissionPlan + BackendPlan + QueryPlan bundle (stage) | v publish target-specific CollectorPlans over OpAMP @@ -83,8 +83,8 @@ another Collector's plan. ## Backend wire contract -The matching BackendPlan protobuf is embedded with the typed PrecomputePlan -and QueryPlan in `POST /api/v1/physical-plan`. The backend validates all shared +The matching BackendPlan protobuf is embedded with the typed PrecomputePlan, +TransmissionPlan, and QueryPlan in `POST /api/v1/physical-plan`. The backend validates all shared identities, fingerprints, schemas, parameters, producers, and lifecycle fields before returning `staged`; `POST /api/v1/physical-plan/activate` performs the single immutable-snapshot swap. From e742bf822f31118311328a9fb8adfb4ada76f6bb Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 3 Sep 2026 14:08:02 -0600 Subject: [PATCH 2/3] fix(transport): make full-frame ingest fail closed --- control_plane/src/physical/compiler.rs | 6 + data_plane/src/drivers/ingest/otel.rs | 280 +++++++++++++++--- .../control-plane/physical-compiler.md | 11 +- 3 files changed, 251 insertions(+), 46 deletions(-) diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index d8b52b6b3..a6489165a 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -595,6 +595,10 @@ pub struct SummaryFrameIdentity { pub schema_id: String, pub producer_id: String, pub producer_epoch: String, + /// Canonical identity of the output series inside the materialization. + /// This remains stable when the sender switches between attribute-bearing + /// and SID-only frames, and prevents two series from sharing a receipt. + pub series_fingerprint: String, pub window_start_unix_nano: u64, pub window_end_unix_nano: u64, pub sequence: u64, @@ -721,6 +725,7 @@ impl TransmissionPlan { || frame.plan_version != self.envelope.plan_version || frame.backend_compat != self.envelope.backend_compat || frame.producer_epoch.is_empty() + || frame.series_fingerprint.is_empty() || frame.sequence == 0 || frame.window_start_unix_nano >= frame.window_end_unix_nano || (frame.kind == SummaryFrameKind::Full @@ -1574,6 +1579,7 @@ mod tests { schema_id: rule.schema_id.clone(), producer_id: rule.producer_id.clone(), producer_epoch: "boot-1".into(), + series_fingerprint: "service=api".into(), window_start_unix_nano: 1, window_end_unix_nano: 2, sequence: 1, diff --git a/data_plane/src/drivers/ingest/otel.rs b/data_plane/src/drivers/ingest/otel.rs index 3283f2460..f11cae03f 100644 --- a/data_plane/src/drivers/ingest/otel.rs +++ b/data_plane/src/drivers/ingest/otel.rs @@ -30,8 +30,8 @@ use crate::precompute_engine::IngestState; use crate::query_engines::routing::FreshnessProbeCache; use crate::storage_engines::types::AggregateCore; use asap_otel_proto::tonic::collector::metrics::v1::{ - metrics_service_server::MetricsService, ExportMetricsPartialSuccess, - ExportMetricsServiceRequest, ExportMetricsServiceResponse, + metrics_service_server::MetricsService, ExportMetricsServiceRequest, + ExportMetricsServiceResponse, }; use asap_otel_proto::tonic::common::v1::any_value::Value as AnyValueVariant; use asap_otel_proto::tonic::metrics::v1::number_data_point::Value as NumberValue; @@ -212,14 +212,16 @@ impl MetricsService for MetricsServiceImpl { let mut outcome = IngestOutcome::default(); if let Some(state) = &self.shared.ingest_state { route_otlp_to_precompute(&points, &sketch_payloads, state).await; - outcome = route_modified_otlp_sketches_to_precompute(&req, state).await; + outcome = route_modified_otlp_sketches_to_precompute(&req, state) + .await + .map_err(Status::invalid_argument)?; } debug!("OTLP sending response via gRPC"); Ok(Response::new(ExportMetricsServiceResponse { - partial_success: (outcome.rejected_frames > 0).then(|| ExportMetricsPartialSuccess { - rejected_data_points: outcome.rejected_frames as i64, - error_message: "summary frame identity rejected by active TransmissionPlan".into(), - }), + // A successful RPC is the transport ACK: every summary frame + // passed the active-plan gate and was applied. Contract failures + // return a non-OK gRPC status before any frame is written. + partial_success: None, // Sid bindings the sender should cache. Each entry maps an // `attributes_fingerprint` to the canonical sid the backend's // `SeriesIdResolver` minted (or returned from its cache). The @@ -320,7 +322,9 @@ async fn handle_otlp_http( let mut outcome = IngestOutcome::default(); if let Some(state) = &shared.ingest_state { route_otlp_to_precompute(&points, &sketch_payloads, state).await; - outcome = route_modified_otlp_sketches_to_precompute(&req, state).await; + outcome = route_modified_otlp_sketches_to_precompute(&req, state) + .await + .map_err(|error| (axum::http::StatusCode::UNPROCESSABLE_ENTITY, error))?; } debug!("OTLP sending response via HTTP"); // HTTP OTLP exporters don't generally read `series_assignments` @@ -331,7 +335,7 @@ async fn handle_otlp_http( // are the universal recovery primitive (see proto comment on // `ExportMetricsServiceResponse.unknown_series_ids`). Ok(Json(serde_json::json!({ - "rejected": outcome.rejected_frames, + "rejected": 0, "unknown_series_ids": outcome.unknown_series_ids, "series_assignments_count": outcome.series_assignments.len(), }))) @@ -860,13 +864,12 @@ async fn route_otlp_to_precompute( pub(crate) struct IngestOutcome { pub unknown_series_ids: Vec, pub series_assignments: Vec, - pub rejected_frames: u64, } async fn route_modified_otlp_sketches_to_precompute( request: &ExportMetricsServiceRequest, ingest_state: &Arc, -) -> IngestOutcome { +) -> Result { use asap_otel_proto::tonic::metrics::v1::metric::Data; let ingest_received_at = Instant::now(); @@ -874,6 +877,13 @@ async fn route_modified_otlp_sketches_to_precompute( let active_physical_plan = ingest_state .physical_plan_snapshot() .filter(|plan| plan.backend_plan.plan_id != 0); + if let Some(active) = active_physical_plan.as_ref() { + // Validate the complete request before mutating the SID registry, + // sketch store, snapshot cache, or worker queues. This makes the + // OTLP request the atomic full-frame publication unit: 2xx/OK means + // the batch was accepted, while a contract error applies none of it. + preflight_summary_frames(request, ingest_state, active)?; + } let agg_configs = snap.get_all_aggregation_configs(); // Schema retirement #5 — agg_id-keyed registry retired; sid-level // reconcile is the only path going forward. See the raw-OTLP @@ -888,7 +898,6 @@ async fn route_modified_otlp_sketches_to_precompute( let mut routed = 0usize; let mut decoded_failed = 0usize; let mut unconfigured = 0usize; - let mut rejected_frames = 0u64; // CQ-2 — the legacy routing-side WorkerMessage push (the DEPRECATED // dual-write that clones the accumulator into the worker under a // bucket-sid, in tandem with the Phase-5 SketchStore append above) is @@ -1086,11 +1095,9 @@ async fn route_modified_otlp_sketches_to_precompute( Ok(frame) }) { Ok(frame) => Some(frame), - Err(error) => { - rejected_frames = rejected_frames.saturating_add(1); - warn!(metric = %metric.name, %error, "rejected summary frame"); - continue; - } + Err(error) => unreachable!( + "summary frame changed after successful request preflight: {error}" + ), } } else { None @@ -1261,14 +1268,9 @@ async fn route_modified_otlp_sketches_to_precompute( ) }); if observed_policy != frame.materialization { - rejected_frames = rejected_frames.saturating_add(1); - warn!( - sid, - declared = frame.materialization.0, - observed = observed_policy.0, - "rejected summary frame with mismatched materialization" + unreachable!( + "materialization changed after successful request preflight" ); - continue; } } @@ -1770,11 +1772,10 @@ async fn route_modified_otlp_sketches_to_precompute( ); } - IngestOutcome { + Ok(IngestOutcome { unknown_series_ids: unknown_sids, series_assignments: new_assignments, - rejected_frames, - } + }) } /// Map `SketchAlgorithm` to the corresponding wire-format @@ -2081,6 +2082,173 @@ struct ModifiedOtlpSketchDp { container_config: crate::storage_engines::sketch_db::index::SketchConfig, } +/// Validate every first-class summary frame in an OTLP request before the +/// ingest loop performs any externally visible mutation. The transport +/// response is therefore the acknowledgement boundary; there is no second +/// application-level ACK protocol. +fn preflight_summary_frames( + request: &ExportMetricsServiceRequest, + ingest_state: &IngestState, + active: &crate::storage_engines::types::ActivePhysicalPlan, +) -> Result<(), String> { + use asap_otel_proto::tonic::metrics::v1::metric::Data; + + fn validate_one( + metric_name: &str, + mut dp: ModifiedOtlpSketchDp, + ingest_state: &IngestState, + active: &crate::storage_engines::types::ActivePhysicalPlan, + ) -> Result<(), String> { + let canonical_name = canonical_sketch_metric_name(metric_name, dp.kind); + let frame = + take_summary_frame_identity(&mut dp.attrs, dp.start_time_unix_nano, dp.time_unix_nano)?; + if state_encoding_for_wire(dp.encoding) != Some(frame.encoding.clone()) { + return Err(format!( + "summary frame for {metric_name} declares an encoding different from its payload" + )); + } + active + .transmission_plan + .validate_frame(&frame) + .map_err(|error| error.to_string())?; + + if !dp.attrs.is_empty() { + let pairs: Vec<(&str, &str)> = dp + .attrs + .iter() + .map(|(key, value)| (key.as_str(), value.as_str())) + .collect(); + let observed_series = crate::drivers::ingest::canonical_attrs_fingerprint(&pairs); + if observed_series != frame.series_fingerprint { + return Err(format!( + "summary frame for {metric_name} declares a series fingerprint different from its labels" + )); + } + } + + // A malformed full snapshot must not be discovered after an earlier + // frame in the request has already reached SketchStore. + if frame.kind == control_plane::physical::compiler::SummaryFrameKind::Full { + decode_modified_otlp_sketch_bytes(dp.kind, dp.encoding, &dp.sketch) + .map_err(|error| format!("invalid full frame for {metric_name}: {error}"))?; + } + + // Attribute-elided retries can recover the policy from their known + // SID. Attribute-bearing frames derive it from the active physical + // schema. Either route must agree with the declared materialization. + let observed = if dp.series_id != 0 && dp.attrs.is_empty() { + ingest_state + .sketch_index + .instance(dp.series_id) + .map(|metadata| metadata.policy_fp) + .ok_or_else(|| { + format!( + "summary frame for {metric_name} references unknown sid {} without labels", + dp.series_id + ) + })? + } else { + derive_sketch_policy_fp( + ingest_state, + canonical_name, + sketch_kind_handle_for(&dp), + &dp.container_config, + &dp.attrs.keys().cloned().collect(), + ) + }; + if observed != frame.materialization { + return Err(format!( + "summary frame for {metric_name} declares materialization {} but active schema resolves {}", + frame.materialization.0, observed.0 + )); + } + Ok(()) + } + + for resource_metrics in &request.resource_metrics { + let resource_attrs = resource_metrics + .resource + .as_ref() + .map(|resource| attributes_to_map(&resource.attributes)) + .unwrap_or_default(); + for scope_metrics in &resource_metrics.scope_metrics { + let scope_attrs = scope_metrics + .scope + .as_ref() + .map(|scope| attributes_to_map(&scope.attributes)) + .unwrap_or_default(); + for metric in &scope_metrics.metrics { + let base_labels: HashMap = scope_attrs + .iter() + .chain(resource_attrs.iter()) + .map(|(key, value)| (key.clone(), value.clone())) + .collect(); + macro_rules! validate_points { + ($points:expr, $kind:expr, $config:expr) => {{ + let config = $config; + for point in &$points { + validate_one( + &metric.name, + ModifiedOtlpSketchDp { + kind: $kind, + attrs: merge_point_attributes(&base_labels, &point.attributes), + time_unix_nano: point.time_unix_nano, + sketch: point.sketch.clone(), + encoding: point.encoding, + series_id: point.series_id, + start_time_unix_nano: point.start_time_unix_nano, + container_config: config.clone(), + }, + ingest_state, + active, + )?; + } + }}; + } + match &metric.data { + Some(Data::Ddsketch(data)) => validate_points!( + data.data_points, + SketchKind::DdSketch, + crate::storage_engines::sketch_db::index::SketchConfig::DDSketch { + relative_accuracy: data.relative_accuracy, + } + ), + Some(Data::Kllsketch(data)) => validate_points!( + data.data_points, + SketchKind::Kll, + crate::storage_engines::sketch_db::index::SketchConfig::Kll { k: data.k } + ), + Some(Data::Countsketch(data)) => validate_points!( + data.data_points, + SketchKind::CountSketch, + crate::storage_engines::sketch_db::index::SketchConfig::CountSketch { + rows: data.rows, + cols: data.cols, + } + ), + Some(Data::Countminsketch(data)) => validate_points!( + data.data_points, + SketchKind::CountMin, + crate::storage_engines::sketch_db::index::SketchConfig::CountMin { + rows: data.rows, + cols: data.cols, + } + ), + Some(Data::Hllsketch(data)) => validate_points!( + data.data_points, + SketchKind::Hll, + crate::storage_engines::sketch_db::index::SketchConfig::Hll { + precision: data.precision, + } + ), + _ => {} + } + } + } + } + Ok(()) +} + fn take_summary_frame_identity( attrs: &mut HashMap, window_start_unix_nano: u64, @@ -2112,6 +2280,7 @@ fn take_summary_frame_identity( let schema_id = required(attrs, "asap.frame.schema_id")?; let producer_id = required(attrs, "asap.frame.producer_id")?; let producer_epoch = required(attrs, "asap.frame.producer_epoch")?; + let series_fingerprint = required(attrs, "asap.frame.series_fingerprint")?; let sequence = number(attrs, "asap.frame.sequence")?; let kind = match required(attrs, "asap.frame.kind")?.as_str() { "full" => SummaryFrameKind::Full, @@ -2140,6 +2309,7 @@ fn take_summary_frame_identity( schema_id, producer_id, producer_epoch, + series_fingerprint, window_start_unix_nano, window_end_unix_nano, sequence, @@ -3332,7 +3502,9 @@ mod sid_resolution_tests { }; let req = build_request("http_latency_ms", dp); - let outcome = route_modified_otlp_sketches_to_precompute(&req, &state).await; + let outcome = route_modified_otlp_sketches_to_precompute(&req, &state) + .await + .expect("ingest succeeds"); assert!( outcome.unknown_series_ids.is_empty(), "no unknown sids on a fresh-attrs DP" @@ -3379,7 +3551,9 @@ mod sid_resolution_tests { }; let req = build_request("http_latency_ms", dp); - let outcome = route_modified_otlp_sketches_to_precompute(&req, &state).await; + let outcome = route_modified_otlp_sketches_to_precompute(&req, &state) + .await + .expect("ingest succeeds"); assert_eq!(outcome.unknown_series_ids, vec![7777]); assert!( outcome.series_assignments.is_empty(), @@ -3410,7 +3584,8 @@ mod sid_resolution_tests { &build_request("http_latency_ms", dp_seed), &state, ) - .await; + .await + .expect("seed ingest succeeds"); let assigned_sid = seed_outcome.series_assignments[0].series_id; assert!( state.sketch_index.instance(assigned_sid).is_some(), @@ -3435,7 +3610,8 @@ mod sid_resolution_tests { &build_request("http_latency_ms", dp_disagree), &state, ) - .await; + .await + .expect("ingest succeeds"); assert_eq!( outcome.unknown_series_ids, vec![stale], @@ -3517,7 +3693,8 @@ mod sid_resolution_tests { ), &state, ) - .await; + .await + .expect("first ingest succeeds"); // ── Window 1: delta (SAME window_start). Adds +3 to bucket 0, // +7 to bucket 1. Within the window this accumulates onto the @@ -3542,7 +3719,8 @@ mod sid_resolution_tests { ), &state, ) - .await; + .await + .expect("second ingest succeeds"); { let entry = state @@ -3583,7 +3761,8 @@ mod sid_resolution_tests { ), &state, ) - .await; + .await + .expect("ingest succeeds"); { let entry = state @@ -3661,7 +3840,8 @@ mod sid_resolution_tests { &build_request("http_latency_ms", dp_first), &state, ) - .await; + .await + .expect("first ingest succeeds"); let cached_sid = first.series_assignments[0].series_id; let dp_second = DdSketchDataPoint { @@ -3678,7 +3858,8 @@ mod sid_resolution_tests { &build_request("http_latency_ms", dp_second), &state, ) - .await; + .await + .expect("second ingest succeeds"); assert!( second.unknown_series_ids.is_empty(), "cached sid + no attrs hits the same SketchStore instance" @@ -3827,7 +4008,9 @@ mod sid_resolution_tests { ); // No prior full frame for this series — pre-fix this DP was // dropped (decoded_failed). Post-fix it bootstraps + applies. - route_modified_otlp_sketches_to_precompute(&req, &state).await; + route_modified_otlp_sketches_to_precompute(&req, &state) + .await + .expect("ingest succeeds"); // The per-series base is now cached, holding the window's // reconstructed matrix. @@ -3901,7 +4084,9 @@ mod sid_resolution_tests { WIN_START, 11_000_000, ); - route_modified_otlp_sketches_to_precompute(&req, &state).await; + route_modified_otlp_sketches_to_precompute(&req, &state) + .await + .expect("ingest succeeds"); let mut attrs = HashMap::new(); attrs.insert("svc".to_string(), "auth".to_string()); @@ -3965,7 +4150,8 @@ mod sid_resolution_tests { series_id: 0, }; route_modified_otlp_sketches_to_precompute(&build_request("dd_latency_ms", dp), &state) - .await; + .await + .expect("ingest succeeds"); let mut attrs = HashMap::new(); attrs.insert("zone".to_string(), "z0".to_string()); @@ -4019,7 +4205,9 @@ mod sid_resolution_tests { 1_000_000, 11_000_000, ); - let out1 = route_modified_otlp_sketches_to_precompute(&req1, &state).await; + let out1 = route_modified_otlp_sketches_to_precompute(&req1, &state) + .await + .expect("ingest succeeds"); let sid = out1.series_assignments[0].series_id; let meta1 = state.sketch_index.instance(sid).expect("sid registered"); assert_eq!( @@ -4047,7 +4235,9 @@ mod sid_resolution_tests { 1_000_000, 12_000_000, ); - route_modified_otlp_sketches_to_precompute(&req2, &state).await; + route_modified_otlp_sketches_to_precompute(&req2, &state) + .await + .expect("ingest succeeds"); let meta2 = state .sketch_index @@ -4076,7 +4266,9 @@ mod sid_resolution_tests { 1_000_000, 13_000_000, ); - route_modified_otlp_sketches_to_precompute(&req3, &state).await; + route_modified_otlp_sketches_to_precompute(&req3, &state) + .await + .expect("ingest succeeds"); let meta3 = state .sketch_index .instance(sid) @@ -4117,7 +4309,8 @@ mod sid_resolution_tests { &build_request("global_latency_ms", dp), &state, ) - .await; + .await + .expect("ingest succeeds"); // Pre-fix: dropped (sid=0 + no attrs → None). Post-fix: resolver // mints a stable sid for (metric, "", agg_kind) and echoes an @@ -4409,6 +4602,7 @@ mod sid_bucketing_tests { ("asap.frame.schema_id".into(), "schema-99".into()), ("asap.frame.producer_id".into(), "edge-a".into()), ("asap.frame.producer_epoch".into(), "boot-7".into()), + ("asap.frame.series_fingerprint".into(), "service=api".into()), ("asap.frame.sequence".into(), "8".into()), ("asap.frame.kind".into(), "full".into()), ("asap.frame.encoding".into(), "sketchlib_protobuf_v1".into()), diff --git a/docs/developer_docs/control-plane/physical-compiler.md b/docs/developer_docs/control-plane/physical-compiler.md index c79e70e47..4f980c550 100644 --- a/docs/developer_docs/control-plane/physical-compiler.md +++ b/docs/developer_docs/control-plane/physical-compiler.md @@ -280,10 +280,15 @@ matching raw transmission and ingest/archive declarations. Until modified OTLP has dedicated identity fields, Collector attaches reserved data-point attributes under `asap.frame.*`: identity version, plan ID/version, backend compatibility, materialization and schema IDs, producer ID/epoch, -sequence, full/delta kind, encoding, and checkpoint/base IDs. Window start/end +canonical series fingerprint, sequence, full/delta kind, encoding, and +checkpoint/base IDs. Window start/end remain the typed data-point timestamps. The backend removes reserved attributes -before building the series label key and rejects any frame that does not match -the active TransmissionPlan. +before building the series label key and rejects the complete request before +writing any frame when one identity, schema, encoding, materialization, or full +payload does not match the active TransmissionPlan. HTTP 2xx / gRPC OK is the +delivery acknowledgement. Retrying the same full frame is idempotent because +the identity selects the same SID, label set, and window replacement; no second +application-level ACK or transport WAL is part of this contract. The compiler error must identify an unsupported capability, invalid placement, window incompatibility, identity conflict, or invalid selected guarantee. It From 56e8a878af1bf5cbfe7f86384ff0b6bbf831ebd5 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 3 Sep 2026 20:57:26 -0600 Subject: [PATCH 3/3] fix(stack): validate frames with canonical sketch algorithms --- data_plane/src/drivers/ingest/otel.rs | 22 +++++++++++----------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/data_plane/src/drivers/ingest/otel.rs b/data_plane/src/drivers/ingest/otel.rs index f11cae03f..d28f2c345 100644 --- a/data_plane/src/drivers/ingest/otel.rs +++ b/data_plane/src/drivers/ingest/otel.rs @@ -1262,7 +1262,7 @@ async fn route_modified_otlp_sketches_to_precompute( derive_sketch_policy_fp( ingest_state, &canonical_name, - sketch_kind_handle_for(&dp), + sketch_algorithm_for(&dp), &dp.container_config, &dp.attrs.keys().cloned().collect(), ) @@ -2099,7 +2099,7 @@ fn preflight_summary_frames( ingest_state: &IngestState, active: &crate::storage_engines::types::ActivePhysicalPlan, ) -> Result<(), String> { - let canonical_name = canonical_sketch_metric_name(metric_name, dp.kind); + let canonical_name = canonical_sketch_metric_name(metric_name, dp.algorithm.clone()); let frame = take_summary_frame_identity(&mut dp.attrs, dp.start_time_unix_nano, dp.time_unix_nano)?; if state_encoding_for_wire(dp.encoding) != Some(frame.encoding.clone()) { @@ -2129,7 +2129,7 @@ fn preflight_summary_frames( // A malformed full snapshot must not be discovered after an earlier // frame in the request has already reached SketchStore. if frame.kind == control_plane::physical::compiler::SummaryFrameKind::Full { - decode_modified_otlp_sketch_bytes(dp.kind, dp.encoding, &dp.sketch) + decode_modified_otlp_sketch_bytes(dp.algorithm.clone(), dp.encoding, &dp.sketch) .map_err(|error| format!("invalid full frame for {metric_name}: {error}"))?; } @@ -2151,7 +2151,7 @@ fn preflight_summary_frames( derive_sketch_policy_fp( ingest_state, canonical_name, - sketch_kind_handle_for(&dp), + sketch_algorithm_for(&dp), &dp.container_config, &dp.attrs.keys().cloned().collect(), ) @@ -2184,13 +2184,13 @@ fn preflight_summary_frames( .map(|(key, value)| (key.clone(), value.clone())) .collect(); macro_rules! validate_points { - ($points:expr, $kind:expr, $config:expr) => {{ + ($points:expr, $algorithm:expr, $config:expr) => {{ let config = $config; for point in &$points { validate_one( &metric.name, ModifiedOtlpSketchDp { - kind: $kind, + algorithm: $algorithm, attrs: merge_point_attributes(&base_labels, &point.attributes), time_unix_nano: point.time_unix_nano, sketch: point.sketch.clone(), @@ -2208,19 +2208,19 @@ fn preflight_summary_frames( match &metric.data { Some(Data::Ddsketch(data)) => validate_points!( data.data_points, - SketchKind::DdSketch, + SketchAlgorithm::DDSketch, crate::storage_engines::sketch_db::index::SketchConfig::DDSketch { relative_accuracy: data.relative_accuracy, } ), Some(Data::Kllsketch(data)) => validate_points!( data.data_points, - SketchKind::Kll, + SketchAlgorithm::Kll, crate::storage_engines::sketch_db::index::SketchConfig::Kll { k: data.k } ), Some(Data::Countsketch(data)) => validate_points!( data.data_points, - SketchKind::CountSketch, + SketchAlgorithm::CountSketch, crate::storage_engines::sketch_db::index::SketchConfig::CountSketch { rows: data.rows, cols: data.cols, @@ -2228,7 +2228,7 @@ fn preflight_summary_frames( ), Some(Data::Countminsketch(data)) => validate_points!( data.data_points, - SketchKind::CountMin, + SketchAlgorithm::Cms, crate::storage_engines::sketch_db::index::SketchConfig::CountMin { rows: data.rows, cols: data.cols, @@ -2236,7 +2236,7 @@ fn preflight_summary_frames( ), Some(Data::Hllsketch(data)) => validate_points!( data.data_points, - SketchKind::Hll, + SketchAlgorithm::Hll, crate::storage_engines::sketch_db::index::SketchConfig::Hll { precision: data.precision, }