fix(prom): return empty remote read results for missing metrics (#9613)

* fix: return empty remote read results for missing metrics Signed-off-by: fys <fengys1996@gmail.com> * test(prom): cover empty reads for missing databases Signed-off-by: fys <fengys1996@gmail.com> * test(prom): simplify remote read regression cases Signed-off-by: fys <fengys1996@gmail.com> * fix(prom): log missing tables during remote read Signed-off-by: fys <fengys1996@gmail.com> --------- Signed-off-by: fys <fengys1996@gmail.com>
2026-10-21 11:22:31 +00:00 · 2026-10-15 12:16:14 +00:00
parent 60097a0628
commit cf0f6c308a
4 changed files with 180 additions and 11 deletions
@@ -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<RemoteQueryOutput> {
-        let table = self
+    ) -> Result<Option<RemoteQueryOutput>> {
+        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<Vec<RemoteQueryOutput>> {
+    ) -> ServerResult<Vec<Option<RemoteQueryOutput>>> {
        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,
@@ -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<String> {
        .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(
@@ -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<Query>) -> 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
+    }
 }
@@ -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;
 }