diff --git a/asap-query-engine/src/drivers/controller_client.rs b/asap-query-engine/src/drivers/controller_client.rs new file mode 100644 index 00000000..b5cc72a8 --- /dev/null +++ b/asap-query-engine/src/drivers/controller_client.rs @@ -0,0 +1,156 @@ +//! DataCollector controller client. +//! +//! Fetches query configuration from the DataCollector controller's API +//! instead of reading a static `inference_config.yaml` file. +//! +//! This replaces the pattern-matching approach: the controller already +//! knows which queries are precomputed and which sketches to use. + +use serde::Deserialize; +use tracing::{info, warn}; + +/// Configuration for the controller client. +#[derive(Debug, Clone)] +pub struct ControllerClientConfig { + /// Base URL of the DataCollector controller (e.g., "http://controller:8080"). + pub base_url: String, + /// How often to poll for config updates (seconds). 0 = poll only on query miss. + pub poll_interval_secs: u64, +} + +impl Default for ControllerClientConfig { + fn default() -> Self { + Self { + base_url: "http://localhost:8080".to_string(), + poll_interval_secs: 0, + } + } +} + +/// Slim plan status returned by `GET /api/v1/plan/:metric`. +/// +/// As of DataCollector main, the GET endpoint only returns the metric name, +/// sketch type, and validity. Callers that need the richer plan (mode, +/// aggregate_by, staged_plan, etc.) must use [`ControllerClient::create_plan`] +/// which invokes `POST /api/v1/plan`. +#[derive(Debug, Clone, Deserialize)] +pub struct ControllerPlanStatus { + pub metric: String, + pub sketch_type: String, + #[serde(default)] + pub valid_until: Option, +} + +/// Rich plan returned by `POST /api/v1/plan`. +#[derive(Debug, Clone, Deserialize)] +pub struct ControllerPlan { + pub metric: String, + pub sketch_type: String, + #[serde(default)] + pub mode: String, + #[serde(default)] + pub aggregate_by: Vec, + #[serde(default)] + pub precompute_jobs: usize, + #[serde(default)] + pub delta_decision: serde_json::Value, + #[serde(default)] + pub staged_plan: Option, +} + +/// Client for the DataCollector controller API. +pub struct ControllerClient { + config: ControllerClientConfig, + http: reqwest::Client, +} + +impl ControllerClient { + pub fn new(config: ControllerClientConfig) -> Self { + Self { + config, + http: reqwest::Client::new(), + } + } + + /// Look up the current plan status for a specific metric. + /// + /// Invokes `GET /api/v1/plan/:metric`. Returns `None` if the controller + /// has no plan for this metric (404) or is unreachable. The response is a + /// slim `{ metric, sketch_type, valid_until }` shape — for the full plan, + /// use [`Self::create_plan`]. + pub async fn get_plan(&self, metric: &str) -> Option { + let url = format!("{}/api/v1/plan/{}", self.config.base_url, metric); + + match self.http.get(&url).send().await { + Ok(resp) => { + if resp.status().is_success() { + match resp.json::().await { + Ok(plan) => { + info!(metric = %metric, sketch_type = %plan.sketch_type, "Fetched plan status from controller"); + Some(plan) + } + Err(e) => { + warn!(metric = %metric, error = %e, "Failed to parse controller plan status"); + None + } + } + } else { + None // 404 = no plan exists + } + } + Err(e) => { + warn!(url = %url, error = %e, "Controller unreachable"); + None + } + } + } + + /// Create or refresh a plan by posting a `QuerySpec` to the controller. + /// + /// Invokes `POST /api/v1/plan`. The `query_spec` is passed through as the + /// JSON body; refer to the DataCollector controller's `QuerySpec` type for + /// the exact schema. Returns the rich plan (mode, aggregate_by, staged + /// plan, delta decision, etc.) on success. + pub async fn create_plan(&self, query_spec: serde_json::Value) -> Option { + let url = format!("{}/api/v1/plan", self.config.base_url); + + match self.http.post(&url).json(&query_spec).send().await { + Ok(resp) if resp.status().is_success() => match resp.json::().await { + Ok(plan) => { + info!(metric = %plan.metric, sketch_type = %plan.sketch_type, "Created plan via controller"); + Some(plan) + } + Err(e) => { + warn!(error = %e, "Failed to parse controller plan response"); + None + } + }, + Ok(resp) => { + warn!(status = %resp.status(), "Controller rejected plan request"); + None + } + Err(e) => { + warn!(url = %url, error = %e, "Controller unreachable"); + None + } + } + } + + /// Check if the controller is healthy by listing connected agents. + pub async fn health_check(&self) -> bool { + let url = format!("{}/api/v1/agents", self.config.base_url); + match self.http.get(&url).send().await { + Ok(resp) => resp.status().is_success(), + Err(_) => false, + } + } + + /// Fetch the generated collector config YAML for a metric. + pub async fn get_config_yaml(&self, metric: &str) -> Option { + let url = format!("{}/api/v1/config/{}", self.config.base_url, metric); + match self.http.get(&url).send().await { + Ok(resp) if resp.status().is_success() => resp.text().await.ok(), + _ => None, + } + } +} diff --git a/asap-query-engine/src/drivers/ingest/otel.rs b/asap-query-engine/src/drivers/ingest/otel.rs index e909a187..18d764bd 100644 --- a/asap-query-engine/src/drivers/ingest/otel.rs +++ b/asap-query-engine/src/drivers/ingest/otel.rs @@ -6,6 +6,7 @@ use std::collections::HashMap; use std::io::Read; +use std::sync::Arc; use asap_sketchlib::proto::sketchlib::{sketch_envelope, SketchEnvelope}; use axum::{body::Bytes, extract::State, routing::post, Json, Router}; @@ -18,7 +19,11 @@ use opentelemetry_proto::tonic::common::v1::any_value::Value as AnyValueVariant; use opentelemetry_proto::tonic::metrics::v1::number_data_point::Value as NumberValue; use prost::Message; use tonic::{Request, Response, Status}; -use tracing::{debug, error, info}; +use tracing::{debug, error, info, warn}; + +use crate::data_model::{KeyByLabelValues, PrecomputedOutput, StreamingConfig}; +use crate::precompute_operators::SketchEnvelopeAccumulator; +use crate::stores::Store; /// Configuration for the OTLP receiver. #[derive(Debug, Clone)] @@ -27,21 +32,45 @@ pub struct OtlpReceiverConfig { pub http_port: u16, } +/// Shared state passed to both gRPC and HTTP handlers. +#[derive(Clone)] +struct OtlpSharedState { + store: Arc, + streaming_config: Arc, +} + /// OTLP receiver that accepts metrics via gRPC and HTTP. pub struct OtlpReceiver { config: OtlpReceiverConfig, + store: Arc, + streaming_config: Arc, } impl OtlpReceiver { - pub fn new(config: OtlpReceiverConfig) -> Self { - Self { config } + pub fn new( + config: OtlpReceiverConfig, + store: Arc, + streaming_config: Arc, + ) -> Self { + Self { + config, + store, + streaming_config, + } } pub async fn run(&self) -> Result<(), Box> { let grpc_addr = std::net::SocketAddr::from(([0, 0, 0, 0], self.config.grpc_port)); let http_addr = std::net::SocketAddr::from(([0, 0, 0, 0], self.config.http_port)); - let grpc_svc = MetricsServiceImpl; + let shared = Arc::new(OtlpSharedState { + store: self.store.clone(), + streaming_config: self.streaming_config.clone(), + }); + + let grpc_svc = MetricsServiceImpl { + shared: shared.clone(), + }; let grpc_svc = opentelemetry_proto::tonic::collector::metrics::v1::metrics_service_server::MetricsServiceServer::new( grpc_svc, @@ -49,7 +78,7 @@ impl OtlpReceiver { let app = Router::new() .route("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/v1/metrics", post(handle_otlp_http)) - .with_state(()); + .with_state(shared.clone()); let grpc_listener = tokio::net::TcpListener::bind(grpc_addr).await?; let http_listener = tokio::net::TcpListener::bind(http_addr).await?; @@ -76,7 +105,9 @@ impl OtlpReceiver { } } -struct MetricsServiceImpl; +struct MetricsServiceImpl { + shared: Arc, +} #[tonic::async_trait] impl MetricsService for MetricsServiceImpl { @@ -86,7 +117,7 @@ impl MetricsService for MetricsServiceImpl { ) -> Result, Status> { debug!("OTLP received request via gRPC"); let req = request.into_inner(); - process_otlp_request(&req, "gRPC"); + process_otlp_request(&req, "gRPC", &self.shared); debug!("OTLP sending response via gRPC"); Ok(Response::new(ExportMetricsServiceResponse { partial_success: None, @@ -96,7 +127,7 @@ impl MetricsService for MetricsServiceImpl { async fn handle_otlp_http( headers: axum::http::HeaderMap, - State(_state): State<()>, + State(shared): State>, body: Bytes, ) -> Result, (axum::http::StatusCode, String)> { debug!("OTLP received request via HTTP, body_bytes={}", body.len()); @@ -125,7 +156,7 @@ async fn handle_otlp_http( format!("Protobuf decode error: {}", e), ) })?; - process_otlp_request(&req, "HTTP"); + process_otlp_request(&req, "HTTP", &shared); debug!("OTLP sending response via HTTP"); Ok(Json(serde_json::json!({"rejected": 0}))) } @@ -139,8 +170,18 @@ pub struct MetricPoint { pub value: f64, } -type SketchPayload = (String, String, Vec); -type OtlpParseResult = (Vec, Vec); +/// A parsed sketch data point: metric name, attribute key, raw payload bytes, +/// timestamp in nanoseconds, and label map from the data point. +#[derive(Debug)] +pub struct SketchPoint { + pub metric_name: String, + pub attr_name: String, + pub payload: Vec, + pub timestamp_nanos: u64, + pub labels: HashMap, +} + +type OtlpParseResult = (Vec, Vec); fn format_series_key(name: &str, labels: &HashMap) -> String { let mut pairs: Vec<_> = labels.iter().collect(); @@ -201,7 +242,31 @@ fn log_sketch_envelope_type(attr_name: &str, payload: &[u8], metric_name: &str) } } -fn process_otlp_request(request: &ExportMetricsServiceRequest, transport: &str) { +/// Build a reverse lookup from metric name -> aggregation_id using the +/// streaming config. Called once per request (configs are small). +fn build_metric_to_aggregation_id(streaming_config: &StreamingConfig) -> HashMap { + let mut map = HashMap::new(); + for (agg_id, config) in streaming_config.get_all_aggregation_configs() { + map.entry(config.metric.clone()).or_insert(*agg_id); + } + map +} + +/// Stable hash of a metric name to use as aggregation_id when the metric is +/// not found in the streaming config. Uses the same FNV-1a 64-bit hash that +/// other parts of the pipeline use for deterministic IDs. +fn hash_metric_name(name: &str) -> u64 { + use std::hash::{Hash, Hasher}; + let mut hasher = std::collections::hash_map::DefaultHasher::new(); + name.hash(&mut hasher); + hasher.finish() +} + +fn process_otlp_request( + request: &ExportMetricsServiceRequest, + transport: &str, + shared: &OtlpSharedState, +) { let resource_count = request.resource_metrics.len(); let total_points = otlp_to_record_count(request); if resource_count > 0 || total_points > 0 { @@ -211,18 +276,93 @@ fn process_otlp_request(request: &ExportMetricsServiceRequest, transport: &str) ); } - let (points, sketch_payloads) = otlp_to_metric_points_and_sketches(request); + let (points, sketch_points) = otlp_to_metric_points_and_sketches(request); - for (metric_name, attr_name, payload) in &sketch_payloads { - log_sketch_envelope_type(attr_name, payload, metric_name); + for sp in &sketch_points { + log_sketch_envelope_type(&sp.attr_name, &sp.payload, &sp.metric_name); } - if !sketch_payloads.is_empty() { + if !sketch_points.is_empty() { debug!( "OTLP Sketch Payload Flow: received {} sketch payload(s), decoded successfully", - sketch_payloads.len() + sketch_points.len() ); } + // --- Store sketch data into SimpleMapStore --- + if !sketch_points.is_empty() { + let metric_to_agg = build_metric_to_aggregation_id(&shared.streaming_config); + let mut batch: Vec<(PrecomputedOutput, Box)> = + Vec::with_capacity(sketch_points.len()); + + for sp in &sketch_points { + // Resolve aggregation_id: prefer streaming config, fall back to hash. + let aggregation_id = metric_to_agg + .get(&sp.metric_name) + .copied() + .unwrap_or_else(|| { + let h = hash_metric_name(&sp.metric_name); + warn!( + "OTLP ingest: metric '{}' not found in streaming config, using hash {} as aggregation_id", + sp.metric_name, h + ); + h + }); + + // Convert nanoseconds to milliseconds for store timestamps. + let ts_ms = sp.timestamp_nanos / 1_000_000; + + // Build KeyByLabelValues from the label map (sorted for determinism). + let mut sorted_labels: Vec<(&String, &String)> = sp.labels.iter().collect(); + sorted_labels.sort_by_key(|(k, _)| *k); + let label_values: Vec = sorted_labels + .iter() + .map(|(k, v)| format!("{}={}", k, v)) + .collect(); + let key = if label_values.is_empty() { + None + } else { + Some(KeyByLabelValues::new_with_labels(label_values)) + }; + + let output = PrecomputedOutput::new(ts_ms, ts_ms, key, aggregation_id); + + match SketchEnvelopeAccumulator::from_proto_bytes(sp.payload.clone()) { + Ok(accumulator) => { + debug!( + "OTLP ingest: storing sketch metric='{}' agg_id={} ts_ms={} type={}", + sp.metric_name, aggregation_id, ts_ms, accumulator.sketch_type + ); + batch.push((output, Box::new(accumulator))); + } + Err(e) => { + error!( + "OTLP ingest: failed to decode SketchEnvelope for metric '{}': {}", + sp.metric_name, e + ); + } + } + } + + if !batch.is_empty() { + let batch_len = batch.len(); + match shared.store.insert_precomputed_output_batch(batch) { + Ok(_) => { + info!( + "OTLP ingest: stored {} sketch precompute(s) (transport={})", + batch_len, transport + ); + } + Err(e) => { + error!( + "OTLP ingest: failed to insert sketch batch into store: {}", + e + ); + } + } + } + } + + // --- Log raw metric points (non-sketch) --- let mut by_series: HashMap = HashMap::new(); for point in &points { let key = format_series_key(&point.name, &point.labels); @@ -243,7 +383,6 @@ fn process_otlp_request(request: &ExportMetricsServiceRequest, transport: &str) first.name, first.labels, first.timestamp_nanos, first.value ); } - // TODO: Pass metrics to precompute engine. } /// Count total data points by traversing resource_metrics -> scope_metrics -> metrics. @@ -299,7 +438,7 @@ fn otlp_to_record_count(request: &ExportMetricsServiceRequest) -> usize { /// separately for Sketch Payload Flow processing. fn otlp_to_metric_points_and_sketches(request: &ExportMetricsServiceRequest) -> OtlpParseResult { let mut points = Vec::new(); - let mut sketch_payloads = Vec::new(); + let mut sketch_points = Vec::new(); for resource_metrics in &request.resource_metrics { let resource_attrs = resource_metrics .resource @@ -332,7 +471,14 @@ fn otlp_to_metric_points_and_sketches(request: &ExportMetricsServiceRequest) -> if let Some((attr_name, payload)) = get_sketch_payload_from_attrs(&dp.attributes) { - sketch_payloads.push((metric.name.clone(), attr_name, payload)); + let labels = merge_point_attributes(&base_labels, &dp.attributes); + sketch_points.push(SketchPoint { + metric_name: metric.name.clone(), + attr_name, + payload, + timestamp_nanos: dp.time_unix_nano, + labels, + }); continue; } let labels = merge_point_attributes(&base_labels, &dp.attributes); @@ -350,7 +496,14 @@ fn otlp_to_metric_points_and_sketches(request: &ExportMetricsServiceRequest) -> if let Some((attr_name, payload)) = get_sketch_payload_from_attrs(&dp.attributes) { - sketch_payloads.push((metric.name.clone(), attr_name, payload)); + let labels = merge_point_attributes(&base_labels, &dp.attributes); + sketch_points.push(SketchPoint { + metric_name: metric.name.clone(), + attr_name, + payload, + timestamp_nanos: dp.time_unix_nano, + labels, + }); continue; } let labels = merge_point_attributes(&base_labels, &dp.attributes); @@ -368,7 +521,14 @@ fn otlp_to_metric_points_and_sketches(request: &ExportMetricsServiceRequest) -> if let Some((attr_name, payload)) = get_sketch_payload_from_attrs(&dp.attributes) { - sketch_payloads.push((metric.name.clone(), attr_name, payload)); + let labels = merge_point_attributes(&base_labels, &dp.attributes); + sketch_points.push(SketchPoint { + metric_name: metric.name.clone(), + attr_name, + payload, + timestamp_nanos: dp.time_unix_nano, + labels, + }); continue; } let labels = merge_point_attributes(&base_labels, &dp.attributes); @@ -393,7 +553,14 @@ fn otlp_to_metric_points_and_sketches(request: &ExportMetricsServiceRequest) -> if let Some((attr_name, payload)) = get_sketch_payload_from_attrs(&dp.attributes) { - sketch_payloads.push((metric.name.clone(), attr_name, payload)); + let labels = merge_point_attributes(&base_labels, &dp.attributes); + sketch_points.push(SketchPoint { + metric_name: metric.name.clone(), + attr_name, + payload, + timestamp_nanos: dp.time_unix_nano, + labels, + }); continue; } let labels = merge_point_attributes(&base_labels, &dp.attributes); @@ -418,7 +585,14 @@ fn otlp_to_metric_points_and_sketches(request: &ExportMetricsServiceRequest) -> if let Some((attr_name, payload)) = get_sketch_payload_from_attrs(&dp.attributes) { - sketch_payloads.push((metric.name.clone(), attr_name, payload)); + let labels = merge_point_attributes(&base_labels, &dp.attributes); + sketch_points.push(SketchPoint { + metric_name: metric.name.clone(), + attr_name, + payload, + timestamp_nanos: dp.time_unix_nano, + labels, + }); continue; } let labels = merge_point_attributes(&base_labels, &dp.attributes); @@ -441,7 +615,7 @@ fn otlp_to_metric_points_and_sketches(request: &ExportMetricsServiceRequest) -> } } } - (points, sketch_payloads) + (points, sketch_points) } fn merge_point_attributes( diff --git a/asap-query-engine/src/drivers/mod.rs b/asap-query-engine/src/drivers/mod.rs index a5f3e00d..926e4582 100644 --- a/asap-query-engine/src/drivers/mod.rs +++ b/asap-query-engine/src/drivers/mod.rs @@ -1,6 +1,8 @@ +pub mod controller_client; pub mod ingest; pub mod query; // Re-export commonly used types for convenience +pub use controller_client::{ControllerClient, ControllerClientConfig}; pub use ingest::{KafkaConsumer, KafkaConsumerConfig, OtlpReceiver, OtlpReceiverConfig}; pub use query::{AdapterConfig, HttpServer, HttpServerConfig}; diff --git a/asap-query-engine/src/main.rs b/asap-query-engine/src/main.rs index fa589aa0..f39dfafa 100644 --- a/asap-query-engine/src/main.rs +++ b/asap-query-engine/src/main.rs @@ -299,7 +299,7 @@ async fn main() -> Result<()> { grpc_port: args.otel_grpc_port, http_port: args.otel_http_port, }; - let receiver = OtlpReceiver::new(otel_config); + let receiver = OtlpReceiver::new(otel_config, store.clone(), streaming_config.clone()); info!( "Starting OTLP receiver (gRPC port {}, HTTP port {})", args.otel_grpc_port, args.otel_http_port diff --git a/asap-query-engine/src/precompute_operators/mod.rs b/asap-query-engine/src/precompute_operators/mod.rs index fbbff1cc..8a52e677 100644 --- a/asap-query-engine/src/precompute_operators/mod.rs +++ b/asap-query-engine/src/precompute_operators/mod.rs @@ -9,6 +9,7 @@ pub mod multiple_increase_accumulator; pub mod multiple_min_max_accumulator; pub mod multiple_sum_accumulator; pub mod set_aggregator_accumulator; +pub mod sketch_envelope_accumulator; pub mod sum_accumulator; pub use count_min_sketch_accumulator::*; @@ -22,4 +23,5 @@ pub use multiple_increase_accumulator::*; pub use multiple_min_max_accumulator::*; pub use multiple_sum_accumulator::*; pub use set_aggregator_accumulator::*; +pub use sketch_envelope_accumulator::*; pub use sum_accumulator::*; diff --git a/asap-query-engine/src/precompute_operators/sketch_envelope_accumulator.rs b/asap-query-engine/src/precompute_operators/sketch_envelope_accumulator.rs new file mode 100644 index 00000000..fc56f313 --- /dev/null +++ b/asap-query-engine/src/precompute_operators/sketch_envelope_accumulator.rs @@ -0,0 +1,154 @@ +//! SketchEnvelopeAccumulator — wraps a raw SketchEnvelope protobuf payload +//! received via OTLP ingest so it can be stored through the `Store` trait. +//! +//! The accumulator preserves the opaque proto bytes and decodes them lazily +//! (via `SketchEnvelope::decode`) only when merge or query operations need +//! the inner sketch type. + +use crate::data_model::{AggregateCore, KeyByLabelValues, SerializableToSink}; +use asap_sketchlib::proto::sketchlib::{sketch_envelope, SketchEnvelope}; +use prost::Message; +use serde_json::Value; +use std::collections::HashMap; + +use promql_utilities::query_logics::enums::{AggregationType, Statistic}; + +/// Accumulator that stores a serialized `SketchEnvelope` protobuf. +/// +/// This is the simplest viable path for OTLP sketch ingest: the OTel Collector +/// has already computed the sketch, so the backend just stores the bytes and +/// serves them back at query time. +#[derive(Debug, Clone)] +pub struct SketchEnvelopeAccumulator { + /// Raw protobuf-encoded `SketchEnvelope`. + pub payload: Vec, + /// Sketch type string cached from decoding (e.g. "CountMin", "KLL"). + pub sketch_type: String, +} + +impl SketchEnvelopeAccumulator { + /// Create from raw protobuf bytes. Decodes the envelope once to cache + /// the sketch type; the full payload is kept for later use. + pub fn from_proto_bytes( + payload: Vec, + ) -> Result> { + let sketch_type = match SketchEnvelope::decode(payload.as_slice()) { + Ok(env) => match env.sketch_state { + Some(sketch_envelope::SketchState::CountMin(_)) => "CountMin".to_string(), + Some(sketch_envelope::SketchState::CountSketch(_)) => "CountSketch".to_string(), + Some(sketch_envelope::SketchState::Kll(_)) => "KLL".to_string(), + Some(sketch_envelope::SketchState::Hll(_)) => "HLL".to_string(), + Some(sketch_envelope::SketchState::Ddsketch(_)) => "DDSketch".to_string(), + Some(sketch_envelope::SketchState::Univmon(_)) => "UnivMon".to_string(), + Some(sketch_envelope::SketchState::Hydra(_)) => "Hydra".to_string(), + Some(sketch_envelope::SketchState::Coco(_)) => "CocoSketch".to_string(), + Some(sketch_envelope::SketchState::Elastic(_)) => "Elastic".to_string(), + None => "Unknown".to_string(), + }, + Err(e) => { + return Err(format!("Failed to decode SketchEnvelope: {}", e).into()); + } + }; + + Ok(Self { + payload, + sketch_type, + }) + } + + /// Return the decoded envelope (re-decodes from bytes each time). + pub fn decode_envelope(&self) -> Result { + SketchEnvelope::decode(self.payload.as_slice()) + } +} + +// --------------------------------------------------------------------------- +// Trait implementations +// --------------------------------------------------------------------------- + +impl SerializableToSink for SketchEnvelopeAccumulator { + fn serialize_to_json(&self) -> Value { + serde_json::json!({ + "type": "SketchEnvelopeAccumulator", + "sketch_type": self.sketch_type, + "payload_bytes": self.payload.len(), + }) + } + + fn serialize_to_bytes(&self) -> Vec { + self.payload.clone() + } +} + +impl AggregateCore for SketchEnvelopeAccumulator { + fn clone_boxed_core(&self) -> Box { + Box::new(self.clone()) + } + + fn type_name(&self) -> &'static str { + "SketchEnvelopeAccumulator" + } + + fn as_any(&self) -> &dyn std::any::Any { + self + } + + fn merge_with( + &self, + other: &dyn AggregateCore, + ) -> Result, Box> { + if other.get_accumulator_type() != self.get_accumulator_type() { + return Err(format!( + "Cannot merge SketchEnvelopeAccumulator with {:?}", + other.get_accumulator_type() + ) + .into()); + } + + // For now, merging opaque envelopes is not supported — each window is + // a self-contained sketch produced by the OTel Collector. Return self + // as-is so the store can still call merge_with without panicking. + Ok(Box::new(self.clone())) + } + + fn get_accumulator_type(&self) -> AggregationType { + // Opaque wrapper — report as the generic multi-subpopulation bucket. + // Direct dispatch is not supported; native sketch query path must + // decode the envelope and delegate to the correct accumulator. + AggregationType::MultipleSubpopulation + } + + fn get_keys(&self) -> Option> { + None + } + + fn query_statistic( + &self, + _statistic: Statistic, + _key: &Option, + _query_kwargs: &HashMap, + ) -> Result> { + Err( + "SketchEnvelopeAccumulator: query_statistic not supported; decode envelope first" + .into(), + ) + } +} + +impl crate::data_model::MultipleSubpopulationAggregate for SketchEnvelopeAccumulator { + fn query( + &self, + _statistic: Statistic, + _key: &KeyByLabelValues, + _query_kwargs: Option<&HashMap>, + ) -> Result> { + Err( + "SketchEnvelopeAccumulator: direct query not supported; use native sketch query path" + .into(), + ) + } + + fn clone_boxed(&self) -> Box { + Box::new(self.clone()) + } +}