diff --git a/data_plane/tests/backend_process_e2e.rs b/data_plane/tests/backend_process_e2e.rs index a0beba65..338f2e9e 100644 --- a/data_plane/tests/backend_process_e2e.rs +++ b/data_plane/tests/backend_process_e2e.rs @@ -67,18 +67,63 @@ async fn wait_http(client: &reqwest::Client, url: &str, child: &mut Child, name: panic!("{name} did not become ready at {url}"); } -fn ddsketch_export(metric: &str, timestamp_ns: u64, values: &[f64], alpha: f64) -> Vec { +fn ddsketch_export( + metric: &str, + timestamp_ns: u64, + values: &[f64], + alpha: f64, + plan: &serde_json::Value, + sequence: u64, +) -> Vec { let mut sketch = asap_sketchlib::DdSketch::new(alpha); for value in values { sketch.update(*value); } - let point = DdSketchDataPoint { - attributes: vec![KeyValue { - key: "service".into(), + let materialization = plan["materializations"][0]["materialization"] + .as_u64() + .unwrap(); + let mut attributes = vec![KeyValue { + key: "service".into(), + value: Some(AnyValue { + value: Some(any_value::Value::StringValue("whole-e2e".into())), + }), + }]; + for (key, value) in [ + ("identity_version", "1".into()), + ("plan_id", plan["envelope"]["plan_id"].to_string()), + ("plan_version", plan["envelope"]["plan_version"].to_string()), + ( + "backend_compat", + control_plane::backend_plan::BACKEND_COMPAT.into(), + ), + ("materialization", materialization.to_string()), + ( + "series_identity", + data_plane::drivers::ingest::canonical_attrs_fingerprint(&[("service", "whole-e2e")]), + ), + ( + "schema_id", + format!( + "{}:summary-state:v1:{materialization}", + control_plane::backend_plan::BACKEND_COMPAT + ), + ), + ("producer_id", "whole-e2e-collector".into()), + ("producer_epoch", "process-e2e".into()), + ("sequence", sequence.to_string()), + ("kind", "full".into()), + ("encoding", "sketchlib_protobuf_v1".into()), + ("checkpoint_id", format!("checkpoint-{sequence}")), + ] { + attributes.push(KeyValue { + key: format!("asap.frame.{key}"), value: Some(AnyValue { - value: Some(any_value::Value::StringValue("whole-e2e".into())), + value: Some(any_value::Value::StringValue(value)), }), - }], + }); + } + let point = DdSketchDataPoint { + attributes, start_time_unix_nano: timestamp_ns.saturating_sub(1_000_000_000), time_unix_nano: timestamp_ns, sketch: DdSketchState { @@ -178,7 +223,13 @@ async fn send_agent_message( .expect("send OpAMP AgentToServer"); } -async fn apply_next_collector_plan(address: String) -> serde_json::Value { +type CollectorSocket = + tokio_tungstenite::WebSocketStream>; + +async fn apply_next_collector_plan( + address: String, + accept: bool, +) -> (serde_json::Value, CollectorSocket) { let mut socket = connect_collector(&address).await; send_agent_message( &mut socket, @@ -190,7 +241,13 @@ async fn apply_next_collector_plan(address: String) -> serde_json::Value { }, ) .await; + respond_next_collector_plan(socket, accept).await +} +async fn respond_next_collector_plan( + mut socket: CollectorSocket, + accept: bool, +) -> (serde_json::Value, CollectorSocket) { let frame = tokio::time::timeout(Duration::from_secs(10), socket.next()) .await .expect("controller did not publish a collector plan") @@ -205,6 +262,12 @@ async fn apply_next_collector_plan(address: String) -> serde_json::Value { .expect("collector-plan custom message"); assert_eq!(custom.capability, COLLECTOR_PLAN_CAPABILITY); assert_eq!(custom.r#type, COLLECTOR_PLAN_MESSAGE); + let decoded = asap_precompute_rs::collector_plan::CollectorPlan::from_json( + &custom.data, + "whole-e2e-collector", + ) + .expect("actual Collector validator accepts the emitted plan"); + assert_eq!(decoded.to_precompute_config_set().unwrap().configs.len(), 1); let plan: serde_json::Value = serde_json::from_slice(&custom.data).expect("decode collector physical plan"); let plan_id = plan["envelope"]["plan_id"] @@ -217,8 +280,12 @@ async fn apply_next_collector_plan(address: String) -> serde_json::Value { let status = serde_json::to_vec(&CollectorPlanStatus { plan_id, plan_version, - status: CollectorPlanStatusKind::Applied, - error: None, + status: if accept { + CollectorPlanStatusKind::Staged + } else { + CollectorPlanStatusKind::Failed + }, + error: (!accept).then(|| "injected Collector staging failure".into()), }) .expect("encode applied status"); send_agent_message( @@ -233,7 +300,7 @@ async fn apply_next_collector_plan(address: String) -> serde_json::Value { }, ) .await; - plan + (plan, socket) } #[tokio::test] @@ -312,45 +379,65 @@ async fn production_control_plane_to_data_plane_otlp_to_promql() { ) .await; - let collector = tokio::spawn(apply_next_collector_plan(control_opamp.clone())); + let collector = tokio::spawn(apply_next_collector_plan(control_opamp.clone(), true)); let observed_at_ms = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .expect("system clock") .as_millis() as u64; + let mut request = serde_json::json!({ + "queries": [{ + "query_id": "whole-process-e2e-query", + "query_string": "quantile_over_time(0.99, whole_process_e2e_latency_ms[1s])", + "metric": "whole_process_e2e_latency_ms", + "window_secs": 1, + "group_by": ["service"], + "accuracy": {"Epsilon": 0.01}, + "window_implementations": [{ + "implementation_id": "collector-tumbling-v1", "framework": "tumbling", + "window_secs": 1, "pane_secs": 1, "state_layout": "anchored-pane-v1", + "cost": { + "model_version": "process-e2e-v1", "workload_fingerprint": "shared-quantiles", + "observed_at_unix_ms": observed_at_ms, "valid_for_ms": 60000, + "horizon_seconds": 300.0, "cpu_cost": 1.0, "weighted_cost": 1.0, + "peak_memory_bytes": 4096, "network_bytes": 1024, "storage_bytes": 2048, + "source_scan_bytes": 0 + } + }], + "lifecycle": { + "evaluation_interval_ms": 1000, + "ingestion_rate_per_second": 100.0, + "evidence_observed_at_unix_ms": observed_at_ms, + "evidence_valid_for_ms": 60000, + "horizon_seconds": 300.0, + "costs": { + "build": 10.0, + "maintenance_per_update": 0.001, + "read": 0.1, + "retention_per_second": 0.001, + "retirement": 1.0 + } + } + }], + "collector_ids": ["whole-e2e-collector"], + "capability_snapshot_id": "whole-e2e-capabilities", + "evidence": {}, + "planner_revision": PLANNER_REVISION, + "max_evidence_age_ms": 60000, + "plan_version": 1, + "activation_unix_ms": observed_at_ms, + "expiry_unix_ms": null, + "backend_compat": control_plane::backend_plan::BACKEND_COMPAT, + "apply_timeout_ms": 10000 + }); + let mut second = request["queries"][0].clone(); + second["query_id"] = "whole-process-e2e-median".into(); + second["query_string"] = "quantile_over_time(0.5, whole_process_e2e_latency_ms[1s])".into(); + request["queries"].as_array_mut().unwrap().push(second); let publication_response = client .post(format!( "{control_base}/api/v1/physical-plan/compile-and-publish" )) - .json(&serde_json::json!({ - "queries": [{ - "query_id": "whole-process-e2e-query", - "query_string": "quantile_over_time(0.99, whole_process_e2e_latency_ms[30s])", - "metric": "whole_process_e2e_latency_ms", - "window_secs": 1, - "group_by": ["service"], - "accuracy": {"Epsilon": 0.01}, - "lifecycle": { - "evaluation_interval_ms": 1000, - "ingestion_rate_per_second": 100.0, - "evidence_observed_at_unix_ms": observed_at_ms, - "evidence_valid_for_ms": 60000, - "horizon_seconds": 300.0, - "costs": { - "build": 10.0, - "maintenance_per_update": 0.001, - "read": 0.1, - "retention_per_second": 0.001, - "retirement": 1.0 - } - } - }], - "collector_ids": ["whole-e2e-collector"], - "capability_snapshot_id": "whole-e2e-capabilities", - "evidence": {}, - "planner_revision": PLANNER_REVISION, - "max_evidence_age_ms": 60000, - "apply_timeout_ms": 10000 - })) + .json(&request) .send() .await .expect("request physical-plan publication"); @@ -365,7 +452,7 @@ async fn production_control_plane_to_data_plane_otlp_to_promql() { ); let publication: serde_json::Value = serde_json::from_str(&publication_body).expect("decode publication response"); - let collector_plan = collector.await.expect("collector task completed"); + let (collector_plan, collector_socket) = collector.await.expect("collector task completed"); assert_eq!( publication["plan_id"], collector_plan["envelope"]["plan_id"] @@ -407,10 +494,14 @@ async fn production_control_plane_to_data_plane_otlp_to_promql() { let now = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .expect("system clock"); - let sample_ns = now.as_nanos() as u64; - let raw_values = (1..=100).map(|value| value as f64).collect::>(); + let window_ms = planned_window_secs * 1000; + let window_end_ms = (now.as_millis() as u64 / window_ms) * window_ms; + let sample_ns = window_end_ms * 1_000_000; + // These ranks are integral under both PromQL interpolation and sketch + // order-statistic readout; interpolation coverage is a separate contract. + let raw_values = (1..=101).map(|value| value as f64).collect::>(); let reference_p99 = exact_quantile(&raw_values, 0.99); - client + let ingestion = client .post(format!("http://{otlp_http}/v1/metrics")) .header("content-type", "application/x-protobuf") .body(ddsketch_export( @@ -418,23 +509,19 @@ async fn production_control_plane_to_data_plane_otlp_to_promql() { sample_ns, &raw_values, planned_alpha, + &collector_plan, + 1, )) .send() .await - .expect("POST OTLP to production data plane") - .error_for_status() - .expect("data plane accepted OTLP"); + .expect("POST OTLP to production data plane"); + let status = ingestion.status(); + let body = ingestion.text().await.unwrap(); + assert!(status.is_success(), "OTLP rejected: {status}: {body}"); - // Advance event time after the controller-selected tumbling window has - // really ended. The E2E uses no artificial future timestamp here. - let window_ms = planned_window_secs * 1000; - let now_ms = now.as_millis() as u64; - let window_end_ms = (now_ms / window_ms + 1) * window_ms; - tokio::time::sleep(Duration::from_millis(window_end_ms - now_ms + 100)).await; - let watermark_ns = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .expect("system clock") - .as_nanos() as u64; + // The sealed full frame already carries its exact window. A subsequent + // checkpoint exercises the same producer's next window independently. + let watermark_ns = sample_ns + window_ms * 1_000_000; client .post(format!("http://{otlp_http}/v1/metrics")) .header("content-type", "application/x-protobuf") @@ -443,6 +530,8 @@ async fn production_control_plane_to_data_plane_otlp_to_promql() { watermark_ns, &[], planned_alpha, + &collector_plan, + 2, )) .send() .await @@ -450,12 +539,15 @@ async fn production_control_plane_to_data_plane_otlp_to_promql() { .error_for_status() .expect("data plane accepted watermark"); - let query = "quantile_over_time(0.99, whole_process_e2e_latency_ms[30s])"; + let query = "quantile_over_time(0.99, whole_process_e2e_latency_ms[1s])"; let mut last_response = serde_json::Value::Null; for _ in 0..50 { let response: serde_json::Value = client .get(format!("{data_base}/api/v1/query")) - .query(&[("query", query)]) + .query(&[ + ("query", query.to_string()), + ("time", (window_end_ms as f64 / 1000.0).to_string()), + ]) .send() .await .expect("query production data plane") @@ -474,6 +566,96 @@ async fn production_control_plane_to_data_plane_otlp_to_promql() { response["data"]["result"][0]["metric"]["service"], "whole-e2e" ); + let median: serde_json::Value = client + .get(format!("{data_base}/api/v1/query")) + .query(&[ + ( + "query", + "quantile_over_time(0.5, whole_process_e2e_latency_ms[1s])".to_string(), + ), + ("time", (window_end_ms as f64 / 1000.0).to_string()), + ]) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + let median_value = first_scalar(&median).expect("second consumer is warm"); + let reference = exact_quantile(&raw_values, 0.5); + assert!( + (median_value - reference).abs() / reference <= planned_alpha * 1.05, + "{median}" + ); + // A rejected successor must not replace the active generation. + request["plan_version"] = 2.into(); + request["activation_unix_ms"] = (std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_millis() as u64 + + 1000) + .into(); + let collector = tokio::spawn(respond_next_collector_plan(collector_socket, false)); + let failed = client + .post(format!( + "{control_base}/api/v1/physical-plan/compile-and-publish" + )) + .json(&request) + .send() + .await + .unwrap(); + let failed_status = failed.status().as_u16(); + let failed_body = failed.text().await.unwrap(); + assert_eq!(failed_status, 502, "{failed_body}"); + assert!( + failed_body.contains("injected Collector staging failure"), + "{failed_body}" + ); + let (rejected_plan, _collector_socket) = collector.await.unwrap(); + let rejected_frame = client + .post(format!("http://{otlp_http}/v1/metrics")) + .header("content-type", "application/x-protobuf") + .body(ddsketch_export( + "whole_process_e2e_latency_ms", + sample_ns, + &[9999.0], + planned_alpha, + &rejected_plan, + 1, + )) + .send() + .await + .unwrap(); + assert_eq!( + rejected_frame.status().as_u16(), + 422, + "inactive generation frame was accepted" + ); + let still_active: serde_json::Value = client + .get(format!("{data_base}/api/v1/backend-plan")) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + assert_eq!( + still_active, backend_plan, + "failed rollout changed active plan" + ); + let still_warm: serde_json::Value = client + .get(format!("{data_base}/api/v1/query")) + .query(&[ + ("query", query.to_string()), + ("time", (window_end_ms as f64 / 1000.0).to_string()), + ]) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + assert_eq!(first_scalar(&still_warm), Some(value)); return; } last_response = response;