diff --git a/src/config/src/utils/async_file.rs b/src/config/src/utils/async_file.rs index ce63379329..321cdd6927 100644 --- a/src/config/src/utils/async_file.rs +++ b/src/config/src/utils/async_file.rs @@ -90,6 +90,8 @@ pub async fn put_file_contents( // Write to temporary file first let mut file_handle = File::create(&temp_file).await?; file_handle.write_all(contents).await?; + // tokio writes on a blocking task; without flush, rename can publish a short file + file_handle.flush().await?; // Atomically rename the temp file to the target file // This ensures we either have the old file or the new file, never a partially written file @@ -368,6 +370,21 @@ mod tests { use super::*; + #[tokio::test(flavor = "multi_thread")] + async fn test_put_file_contents_is_complete_when_it_returns() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("large.bin"); + // Past tokio's 2 MiB write buffer, so an unflushed tail is still in flight at the rename. + let content = vec![7u8; 4 << 20]; + for _ in 0..20 { + put_file_contents(&path, &content).await.unwrap(); + assert_eq!( + std::fs::metadata(&path).unwrap().len(), + content.len() as u64 + ); + } + } + #[tokio::test] async fn test_get_file_contents_with_range() { let content = b"Hello World"; diff --git a/src/core/src/traces/red_insights.rs b/src/core/src/traces/red_insights.rs index bd4539e2b7..49aa8b4b93 100644 --- a/src/core/src/traces/red_insights.rs +++ b/src/core/src/traces/red_insights.rs @@ -29,7 +29,7 @@ pub const MAX_CREATES: usize = 15; const HISTOGRAM_INTERVAL: &str = "5m"; const SCHEDULE_INTERVAL: &str = "1h"; -const DETECTION_WINDOW_SECONDS: i64 = 3_900; +const DETECTION_WINDOW_SECONDS: i64 = 10_800; const MIN_BUCKET_REQUESTS: u32 = 20; const NAME_SEPARATOR: &str = " ยท "; /// Alert names forbid `/`; U+2215 reads the same and the web strip maps it back. diff --git a/src/infra/src/cache/file_data/disk.rs b/src/infra/src/cache/file_data/disk.rs index 6328c219da..9690968b8a 100644 --- a/src/infra/src/cache/file_data/disk.rs +++ b/src/infra/src/cache/file_data/disk.rs @@ -695,7 +695,12 @@ pub async fn get_opts(file: &str, options: GetOptions) -> object_store::Result r.as_range(meta.size).unwrap(), + Some(r) => r + .as_range(meta.size) + .map_err(|e| object_store::Error::Precondition { + path: file.to_string(), + source: Box::new(e), + })?, None => 0..meta.size, }; @@ -1537,6 +1542,28 @@ mod tests { )); } + #[cfg(unix)] + #[tokio::test(flavor = "multi_thread")] + async fn get_opts_rejects_a_range_past_a_short_cached_file() { + let key = format!( + "files/default/logs/short/2026/10/15/10/{}.parquet", + config::ider::uuid() + ); + set(&key, Bytes::from_static(b"0123456789abcdef")) + .await + .unwrap(); + std::fs::write(get_file_path(&key).unwrap(), b"").unwrap(); + let options = GetOptions { + range: Some((8..16).into()), + ..Default::default() + }; + assert!(matches!( + get_opts(&key, options).await, + Err(object_store::Error::Precondition { .. }) + )); + remove(&key).await.unwrap(); + } + #[test] fn key_escapes_rejects_traversal_and_absolute() { assert!(key_escapes("../etc/passwd")); diff --git a/src/infra/src/cluster/mod.rs b/src/infra/src/cluster/mod.rs index 49d7675592..f171af6658 100644 --- a/src/infra/src/cluster/mod.rs +++ b/src/infra/src/cluster/mod.rs @@ -426,15 +426,13 @@ pub async fn check_nodes_status(client: &reqwest::Client) -> Result<()> { node.name, node.http_addr ); - let mut w = NODES_HEALTH_CHECK.write().await; - let Some(entry) = w.get_mut(&node.uuid) else { - // node haven't been added to the cluster yet, when the health check first succeed, - // it will be added to the check map - continue; + // count never-healthy peers too, or one that died during our startup is never removed + let times = { + let mut w = NODES_HEALTH_CHECK.write().await; + let entry = w.entry(node.uuid.clone()).or_insert(0); + *entry += 1; + *entry }; - *entry += 1; - let times = *entry; - drop(w); if times >= cfg.health_check.failed_times { log::error!( @@ -808,4 +806,30 @@ mod tests { assert_eq!(nodes.len(), 3); assert_eq!(nodes, vec!["node-q-2", "node-q-5", "node-q-8"]); } + + #[tokio::test] + async fn test_check_nodes_status_removes_never_healthy_node() { + let uuid = "test-never-healthy-node".to_string(); + let dead = Node { + uuid: uuid.clone(), + name: "dead-ingester".to_string(), + http_addr: "http://127.0.0.1:1".to_string(), + role: vec![Role::Ingester], + status: NodeStatus::Online, + ..LOCAL_NODE.clone() + }; + add_node_to_cache(dead).await; + assert!(!NODES_HEALTH_CHECK.read().await.contains_key(&uuid)); + + let client = reqwest::Client::new(); + let failed_times = get_config().health_check.failed_times; + for _ in 1..failed_times { + check_nodes_status(&client).await.unwrap(); + } + assert!(get_node_by_uuid(&uuid).await.is_some()); + + check_nodes_status(&client).await.unwrap(); + assert!(get_node_by_uuid(&uuid).await.is_none()); + assert!(!NODES_HEALTH_CHECK.read().await.contains_key(&uuid)); + } } diff --git a/src/promql_service/src/search/cache/mod.rs b/src/promql_service/src/search/cache/mod.rs index 34c4e5d150..9b61c4ccfc 100644 --- a/src/promql_service/src/search/cache/mod.rs +++ b/src/promql_service/src/search/cache/mod.rs @@ -223,6 +223,9 @@ pub async fn get( } } + // a series whose readings all fall before the window would reach callers with no samples + resp.series.retain(|series| !series.samples.is_empty()); + // if new_start == start, it means we have no data in cache, so we need to return None if new_start == start { return Ok(None); diff --git a/src/promql_service/src/search/mod.rs b/src/promql_service/src/search/mod.rs index 256594a0de..fbc2e9f5cf 100644 --- a/src/promql_service/src/search/mod.rs +++ b/src/promql_service/src/search/mod.rs @@ -620,6 +620,8 @@ async fn merge_matrix_query(series: &[cluster_rpc::Series], org_id: &str) -> Res } let mut merged_data = merged_data .into_iter() + // Prometheus never answers a range query with a series that has no samples + .filter(|(_, samples)| !samples.is_empty()) .map(|(sig, samples)| { let mut samples = samples .into_iter() @@ -873,6 +875,37 @@ mod tests { } } + #[tokio::test] + async fn test_merge_matrix_drops_series_without_samples() { + let mut response = cluster_rpc::MetricsQueryResponse::default(); + grpc::add_value( + &mut response, + Value::Range(RangeValue::new( + vec![Arc::new(Label::new("pod", "recovered"))], + Vec::::new(), + )), + ); + grpc::add_value( + &mut response, + Value::Range(RangeValue::new( + vec![Arc::new(Label::new("pod", "waiting"))], + vec![Sample::new(1_000_000, 1.0)], + )), + ); + + let value = merge_matrix_query(&response.series, "test_merge_empty_series") + .await + .unwrap(); + let Value::Matrix(matrix) = value else { + panic!("expected matrix result"); + }; + assert_eq!(matrix.len(), 1); + assert_eq!( + matrix[0].labels, + vec![Arc::new(Label::new("pod", "waiting"))] + ); + } + #[tokio::test] async fn test_merge_exemplars_skips_series_without_exemplars() { let labels = vec![Arc::new(Label::new("job", "test"))]; diff --git a/src/search/src/datafusion/distributed_plan/remote_scan_exec.rs b/src/search/src/datafusion/distributed_plan/remote_scan_exec.rs index 51634071ef..6026f888a8 100644 --- a/src/search/src/datafusion/distributed_plan/remote_scan_exec.rs +++ b/src/search/src/datafusion/distributed_plan/remote_scan_exec.rs @@ -26,7 +26,7 @@ use config::{ use datafusion::{ common::{DataFusionError, Result, tree_node::TreeNodeRecursion}, execution::{SendableRecordBatchStream, TaskContext}, - physical_expr::{EquivalenceProperties, Partitioning, PhysicalExpr}, + physical_expr::{AcrossPartitions, EquivalenceProperties, Partitioning, PhysicalExpr}, physical_plan::{ DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties, execution_plan::{Boundedness, EmissionType}, @@ -77,7 +77,7 @@ impl RemoteScanExec { // Sharing happens on the leader only, a follower must never see a marker. let input = strip_shared_subplan_markers(input)?; let output_partitions = remote_scan_node.nodes.len(); - let cache = Self::compute_properties(Arc::clone(&input.schema()), output_partitions); + let cache = Self::compute_properties(&input, output_partitions)?; // serialize the input plan and set it as the plan for the remote scan node let proto = get_physical_extension_codec(); @@ -162,17 +162,29 @@ impl RemoteScanExec { /// This function creates the cache object that stores the plan properties such as schema, /// equivalence properties, ordering, partitioning, etc. - fn compute_properties(schema: SchemaRef, n_partitions: usize) -> Arc { - let eq_properties = EquivalenceProperties::new(schema); + fn compute_properties( + input: &Arc, + n_partitions: usize, + ) -> Result> { + let mut eq_properties = EquivalenceProperties::new(input.schema()); + // Followers each run their own copy of the input; only literal constants hold on all. + eq_properties.add_constants( + input + .properties() + .eq_properties + .constants() + .into_iter() + .filter(|c| matches!(c.across_partitions, AcrossPartitions::Uniform(Some(_)))), + )?; let output_partitioning = Self::output_partitioning_helper(n_partitions); - Arc::new(PlanProperties::new( + Ok(Arc::new(PlanProperties::new( eq_properties, // Output Partitioning output_partitioning, // Execution Mode EmissionType::Incremental, Boundedness::Bounded, - )) + ))) } pub fn set_analyze(mut self) -> Self { diff --git a/src/search/src/datafusion/optimizer/physical_optimizer/remote_scan.rs b/src/search/src/datafusion/optimizer/physical_optimizer/remote_scan.rs index ea40d5ebc0..8657c68d94 100644 --- a/src/search/src/datafusion/optimizer/physical_optimizer/remote_scan.rs +++ b/src/search/src/datafusion/optimizer/physical_optimizer/remote_scan.rs @@ -973,4 +973,109 @@ mod tests { assert_ne!(plan.name(), "RemoteScanExec"); }); } + + async fn cluster_plan(sql: &str) -> (SessionContext, Arc) { + use tracing_opentelemetry::OpenTelemetrySpanExt; + let schema = Arc::new(Schema::new(vec![ + Field::new("_timestamp", DataType::Int64, false), + Field::new("k8s_cluster", DataType::Utf8, true), + Field::new("k8s_resource_name", DataType::Utf8, true), + Field::new("k8s_namespace_name", DataType::Utf8, true), + Field::new("event_name", DataType::Utf8, true), + Field::new("value", DataType::Int64, true), + ])); + let mut file_id_lists = HashMap::new(); + file_id_lists.insert(TableReference::from("t"), vec![]); + let nodes = (0..3) + .map(|_| Arc::new(config::meta::cluster::Node::default()) as Arc) + .collect(); + let remote_scan_nodes = Arc::new(RemoteScanNodes::new( + Request::default(), + nodes, + file_id_lists, + HashMap::new(), + false, + tracing::Span::current().context(), + None, + )); + let rule = RemoteScanRule { + remote_scan_nodes, + single_node_optimizer_enable: false, + }; + let state = SessionStateBuilder::new() + .with_config(SessionConfig::new().with_target_partitions(4)) + .with_runtime_env(Arc::new(RuntimeEnvBuilder::new().build().unwrap())) + .with_physical_optimizer_rule(Arc::new(rule)) + .with_default_features() + .build(); + let ctx = SessionContext::new_with_state(state); + let table = NewEmptyTable::new("t", schema) + .with_partitions(4) + .with_sort_order(crate::datafusion::sort_order::FileSortOrder::TimestampDesc); + ctx.register_table("t", Arc::new(table)).unwrap(); + let logical = ctx.state().create_logical_plan(sql).await.unwrap(); + let plan = ctx.state().create_physical_plan(&logical).await.unwrap(); + (ctx, plan) + } + + #[tokio::test] + async fn test_cluster_window_partitioned_by_filtered_constants_executes() { + use datafusion::physical_plan::windows::BoundedWindowAggExec; + let latest = "SELECT event_name, _timestamp FROM (SELECT event_name, _timestamp, \ + ROW_NUMBER() OVER (PARTITION BY k8s_cluster, k8s_resource_name, \ + COALESCE(k8s_namespace_name, ''), event_name ORDER BY _timestamp DESC) AS rn \ + FROM t WHERE k8s_resource_name = 'nodes' AND k8s_cluster = 'c1'{terms}) \ + WHERE rn = 1 ORDER BY _timestamp DESC LIMIT {limit}"; + for (terms, limit) in [ + ("", "5000"), + (" AND event_name = 'n1' AND k8s_namespace_name = 'ns1'", "1"), + ] { + let sql = latest.replace("{terms}", terms).replace("{limit}", limit); + let (ctx, plan) = cluster_plan(&sql).await; + let mut windows = vec![]; + plan.apply(|node| { + if node.downcast_ref::().is_some() { + windows.push(Arc::clone(node)); + } + Ok(TreeNodeRecursion::Continue) + }) + .unwrap(); + assert_eq!(windows.len(), 1, "{sql}"); + assert!(decode_every_remote_scan(&ctx, &plan) >= 1, "{sql}"); + if let Err(e) = windows[0].execute(0, ctx.task_ctx()) { + panic!("{sql}: {e}"); + } + } + } + + #[tokio::test] + async fn test_cluster_union_literal_column_keeps_sort_requirement() { + let sql = "SELECT event_name, value, 'a' AS origin FROM t WHERE value > 1 \ + UNION ALL SELECT event_name, value, 'b' AS origin FROM t WHERE value < 0 \ + ORDER BY origin, value DESC LIMIT 15"; + let (ctx, plan) = cluster_plan(sql).await; + assert!(decode_every_remote_scan(&ctx, &plan) >= 2); + plan.apply(|node| { + for (child, required) in node + .children() + .into_iter() + .zip(node.required_input_ordering()) + { + // A merge over a remote scan trusts each follower's stream to be sorted. + let Some(required) = + required.filter(|_| child.downcast_ref::().is_none()) + else { + continue; + }; + let (alternatives, _) = required.into_alternatives(); + let eq = child.equivalence_properties(); + let satisfied = alternatives + .iter() + .any(|r| eq.ordering_satisfy_requirement(r.iter().cloned()).unwrap()); + assert!(satisfied, "{} <- {}", node.name(), child.name()); + } + Ok(TreeNodeRecursion::Continue) + }) + .unwrap(); + } }