From cf0f6c308ae092ccedf1f505c04d4ca1a747a2f7 Mon Sep 17 00:00:00 2001 From: fys <40801205+fengys1996@users.noreply.github.com> Date: Thu, 15 Oct 2026 12:16:14 +0000 Subject: [PATCH] fix(prom): return empty remote read results for missing metrics (#9613) * fix: return empty remote read results for missing metrics Signed-off-by: fys * test(prom): cover empty reads for missing databases Signed-off-by: fys * test(prom): simplify remote read regression cases Signed-off-by: fys * fix(prom): log missing tables during remote read Signed-off-by: fys --------- Signed-off-by: fys --- src/frontend/src/instance/prom_store.rs | 32 ++++-- src/servers/src/prom_store.rs | 26 ++++- tests-integration/src/prom_store.rs | 127 ++++++++++++++++++++++++ tests-integration/tests/http.rs | 6 +- 4 files changed, 180 insertions(+), 11 deletions(-) diff --git a/src/frontend/src/instance/prom_store.rs b/src/frontend/src/instance/prom_store.rs index d972abc9112..875cbaa873d 100644 --- a/src/frontend/src/instance/prom_store.rs +++ b/src/frontend/src/instance/prom_store.rs @@ -61,7 +61,7 @@ use tracing::instrument; use crate::error::{ AmbiguousValueColumnSnafu, CatalogSnafu, ColumnNotFoundSnafu, ExecLogicalPlanSnafu, - PromStoreRemoteQueryPlanSnafu, ReadTableSnafu, Result, TableNotFoundSnafu, + PromStoreRemoteQueryPlanSnafu, ReadTableSnafu, Result, }; use crate::instance::Instance; @@ -207,15 +207,21 @@ impl Instance { schema_name: &str, table_name: &str, query: &Query, - ) -> Result { - let table = self + ) -> Result> { + let Some(table) = self .catalog_manager .table(catalog_name, schema_name, table_name, Some(ctx)) .await .context(CatalogSnafu)? - .with_context(|| TableNotFoundSnafu { - table_name: format_full_table_name(catalog_name, schema_name, table_name), - })?; + else { + debug!( + catalog = catalog_name, + schema = schema_name, + table = table_name, + "Table not found during remote read, returning empty result" + ); + return Ok(None); + }; let timestamp_column_name = table .schema() @@ -250,12 +256,12 @@ impl Instance { .await .context(ExecLogicalPlanSnafu)?; - Ok(RemoteQueryOutput { + Ok(Some(RemoteQueryOutput { table_name: table_name.to_string(), timestamp_column_name, value_column_name, output, - }) + })) } #[tracing::instrument(skip_all)] @@ -264,7 +270,7 @@ impl Instance { ctx: QueryContextRef, queries: &[Query], query_targets: &[PermissionTableTarget], - ) -> ServerResult> { + ) -> ServerResult>> { let mut results = Vec::with_capacity(queries.len()); for (query, target) in queries.iter().zip(query_targets) { @@ -599,6 +605,10 @@ impl PromStoreProtocolHandler for Instance { let response_type = negotiate_response_type(&request.accepted_response_types)?; + for query in &request.queries { + prom_store::validate_matchers(&query.matchers)?; + } + // TODO(dennis): use read_hints to speedup query if possible let results = self .handle_remote_queries(ctx, &request.queries, &query_targets) @@ -609,6 +619,10 @@ impl PromStoreProtocolHandler for Instance { let mut query_results = Vec::with_capacity(results.len()); let mut map = HashMap::new(); for result in results { + let Some(result) = result else { + query_results.push(QueryResult { timeseries: vec![] }); + continue; + }; let RemoteQueryOutput { table_name, timestamp_column_name, diff --git a/src/servers/src/prom_store.rs b/src/servers/src/prom_store.rs index ff82f8f933b..0568d967010 100644 --- a/src/servers/src/prom_store.rs +++ b/src/servers/src/prom_store.rs @@ -21,7 +21,7 @@ use std::hash::{Hash, Hasher}; use std::sync::Arc; use api::prom_store::remote::label_matcher::Type as MatcherType; -use api::prom_store::remote::{Label, Query, Sample, TimeSeries, WriteRequest}; +use api::prom_store::remote::{Label, LabelMatcher, Query, Sample, TimeSeries, WriteRequest}; use api::v1::RowInsertRequests; use arrow::array::{ Array, ArrayRef, AsArray, DictionaryArray, LargeStringArray, StringArray, StringViewArray, @@ -138,6 +138,30 @@ pub fn extract_schema_from_query(query: &Query) -> Option { .map(|matcher| matcher.value.clone()) } +/// Validate matcher syntax independently of whether the metric exists. +pub fn validate_matchers(matchers: &[LabelMatcher]) -> Result<()> { + for matcher in matchers { + let matcher_type = MatcherType::try_from(matcher.r#type).map_err(|e| { + error::InvalidPromRemoteRequestSnafu { + msg: format!("invalid LabelMatcher type, decode error: {e}"), + } + .build() + })?; + if matches!(matcher_type, MatcherType::Re | MatcherType::Nre) { + // Validate using the same regex engine as query execution. + // Its syntax differs from Prometheus's; e.g. `(?x) a` is accepted here + // but rejected by Prometheus. + regex::Regex::new(&matcher.value).map_err(|e| { + error::InvalidPromRemoteRequestSnafu { + msg: format!("invalid regex for label '{}': {e}", matcher.name), + } + .build() + })?; + } + } + Ok(()) +} + /// Create a DataFrame from a remote Query #[tracing::instrument(skip_all)] pub fn query_to_plan( diff --git a/tests-integration/src/prom_store.rs b/tests-integration/src/prom_store.rs index af901120454..80da82c0e89 100644 --- a/tests-integration/src/prom_store.rs +++ b/tests-integration/src/prom_store.rs @@ -34,6 +34,8 @@ mod tests { use frontend::instance::Instance; use prost::Message; use servers::http::prom_store::PHYSICAL_TABLE_PARAM; + use servers::http::test_helpers::{TestClient, TestResponse}; + use servers::http::{HttpOptions, HttpServerBuilder}; use servers::prom_store; use servers::prom_store::to_grpc_row_insert_requests; use servers::query_handler::PromStoreProtocolHandler; @@ -663,10 +665,135 @@ mod tests { ] ); + instance + .do_query( + "CREATE TABLE invalid_metric (ts TIMESTAMP TIME INDEX, a DOUBLE, b DOUBLE)", + ctx.clone(), + ) + .await[0] + .as_ref() + .unwrap(); + let server = HttpServerBuilder::new(HttpOptions::default()) + .with_prom_handler(instance.clone(), None, true, Default::default(), None) + .build(); + let client = TestClient::new(server.build(server.make_app()).unwrap()).await; + + // Cover missing schemas/catalogs and each database label alias. + for (selector, database) in [ + ("__schema__", "unknown_schema"), + ("__database__", "unknown_catalog-prometheus"), + ("x_greptime_database", "unknown_catalog-information_schema"), + ] { + let mut query = remote_query("missing_metric"); + query.matchers.push(LabelMatcher { + name: selector.to_string(), + value: database.to_string(), + r#type: MatcherType::Eq as i32, + }); + let response = post_remote_read(&client, vec![query]).await; + assert_eq!(response.status(), 200, "{selector}={database}"); + let body = prom_store::snappy_decompress(&response.bytes().await).unwrap(); + let response = ReadResponse::decode(&body[..]).unwrap(); + assert_eq!(response.results.len(), 1); + assert!(response.results[0].timeseries.is_empty()); + } + + // Matcher syntax must be checked even when the metric does not exist. + for metric in ["metric1", "missing_metric"] { + for (matcher_type, value, expected_status) in [ + (99, "spark", 400), + (MatcherType::Re as i32, "[", 400), + (MatcherType::Nre as i32, "[", 400), + (MatcherType::Eq as i32, "[", 200), + (MatcherType::Neq as i32, "[", 200), + (MatcherType::Re as i32, "spark", 200), + (MatcherType::Nre as i32, "api.*", 200), + ] { + let mut query = remote_query(metric); + query.matchers.push(LabelMatcher { + name: "job".to_string(), + value: value.to_string(), + r#type: matcher_type, + }); + let response = post_remote_read(&client, vec![query]).await; + assert_eq!( + response.status().as_u16(), + expected_status, + "metric={metric}, type={matcher_type}, value={value}" + ); + } + } + + // Missing metrics must retain their result slots in multi-query responses. + for metrics in [ + vec!["missing_metric"], + vec!["missing_metric", "another_missing_metric"], + vec![ + "missing_metric", + "metric1", + "missing_metric", + "metric3", + "missing_metric", + ], + vec!["missing_metric", "invalid_metric"], + ] { + let queries = metrics.iter().map(|metric| remote_query(metric)).collect(); + let response = post_remote_read(&client, queries).await; + if metrics.contains(&"invalid_metric") { + assert!(!response.status().is_success()); + assert!(response.text().await.contains("Ambiguous value column")); + continue; + } + assert_eq!(response.status(), 200, "{}", response.text().await); + assert_eq!(response.headers()["content-type"], "application/x-protobuf"); + assert_eq!(response.headers()["content-encoding"], "snappy"); + let body = prom_store::snappy_decompress(&response.bytes().await).unwrap(); + let response = ReadResponse::decode(&body[..]).unwrap(); + assert_eq!(response.results.len(), metrics.len()); + for (result, metric) in response.results.iter().zip(metrics) { + if metric.ends_with("missing_metric") { + assert!(result.timeseries.is_empty()); + continue; + } + assert!(!result.timeseries.is_empty()); + for series in &result.timeseries { + assert!(series.labels.iter().any(|label| { + label.name == prom_store::METRIC_NAME_LABEL && label.value == metric + })); + assert_eq!(series.samples.len(), 2); + } + } + } + // check physical table if provided if let Some(physical_table) = physical_table { let sql = format!("DESC TABLE {physical_table};"); instance.do_query(&sql, ctx).await[0].as_ref().unwrap(); } } + + fn remote_query(metric: &str) -> Query { + Query { + start_timestamp_ms: 1000, + end_timestamp_ms: 2000, + matchers: vec![LabelMatcher { + name: prom_store::METRIC_NAME_LABEL.to_string(), + value: metric.to_string(), + r#type: MatcherType::Eq as i32, + }], + ..Default::default() + } + } + + async fn post_remote_read(client: &TestClient, queries: Vec) -> TestResponse { + let request = ReadRequest { + queries, + ..Default::default() + }; + client + .post("/v1/prometheus/read?db=prometheus") + .body(prom_store::snappy_compress(&request.encode_to_vec()).unwrap()) + .send() + .await + } } diff --git a/tests-integration/tests/http.rs b/tests-integration/tests/http.rs index 5e006c5d80d..20d6f6dde7f 100644 --- a/tests-integration/tests/http.rs +++ b/tests-integration/tests/http.rs @@ -5463,7 +5463,11 @@ pub async fn test_prometheus_remote_schema_labels(store_type: StorageType) { .body(compressed_unknown_read_request) .send() .await; - assert_eq!(unknown_result.status(), StatusCode::BAD_REQUEST); + assert_eq!(unknown_result.status(), StatusCode::OK); + let body = prom_store::snappy_decompress(&unknown_result.bytes().await).unwrap(); + let response = ReadResponse::decode(&body[..]).unwrap(); + assert_eq!(response.results.len(), 1); + assert!(response.results[0].timeseries.is_empty()); guard.remove_all().await; }