diff --git a/src/servers/benches/prometheus_response.rs b/src/servers/benches/prometheus_response.rs index 5d9325253a2..83895465112 100644 --- a/src/servers/benches/prometheus_response.rs +++ b/src/servers/benches/prometheus_response.rs @@ -122,7 +122,6 @@ fn bench_prometheus_response(c: &mut Criterion) { Ok(Output::new_with_record_batches( RecordBatches::try_new(schema.clone(), batches.clone()).unwrap(), )), - Some("metric".to_string()), ValueType::Matrix, None, )) @@ -137,8 +136,7 @@ fn bench_prometheus_response(c: &mut Criterion) { }; assert_eq!(series.len(), SERIES); for (index, series) in series.iter().enumerate() { - assert_eq!(series.metric.len(), labels + 1); - assert_eq!(series.metric["__name__"], "metric"); + assert_eq!(series.metric.len(), labels); for label in 0..labels { assert_eq!( series.metric[&format!("label_{label}")], @@ -181,7 +179,6 @@ fn bench_prometheus_single_point(c: &mut Criterion) { Ok(Output::new_with_record_batches( RecordBatches::try_new(schema.clone(), batches.clone()).unwrap(), )), - Some("metric".to_string()), result_type, None, )) @@ -195,7 +192,7 @@ fn bench_prometheus_single_point(c: &mut Criterion) { PromQueryResult::Matrix(series) => { assert_eq!(series.len(), series_count); assert!(series.iter().all(|series| { - series.metric.len() == LABELS + 1 + series.metric.len() == LABELS && series.values.len() == 1 && series.values[0].0 == 0.0 && matches!(series.values[0].1, PromSampleValue::Number(0.0)) @@ -205,7 +202,7 @@ fn bench_prometheus_single_point(c: &mut Criterion) { PromQueryResult::Vector(series) => { assert_eq!(series.len(), series_count); assert!(series.iter().all(|series| { - series.metric.len() == LABELS + 1 + series.metric.len() == LABELS && series.value.as_ref().is_some_and(|(timestamp, value)| { *timestamp == 0.0 && value == "0.0" }) diff --git a/src/servers/src/grpc/prom_query_gateway.rs b/src/servers/src/grpc/prom_query_gateway.rs index abec69604f6..9c36acf18d9 100644 --- a/src/servers/src/grpc/prom_query_gateway.rs +++ b/src/servers/src/grpc/prom_query_gateway.rs @@ -134,7 +134,7 @@ impl PrometheusGatewayService { return PrometheusJsonResponse::error(err.status_code(), err.output_msg()); } }; - let (metric_name, mut result_type) = retrieve_metric_name_and_result_type(query.expr()); + let (_, mut result_type) = retrieve_metric_name_and_result_type(query.expr()); let query_id = ctx.remote_query_id().map(str::to_string); // A range query only returns a matrix, and matrix serialization sorts // samples and series, so execution order never reaches the response. @@ -146,12 +146,6 @@ impl PrometheusGatewayService { }; let result = self.handler.do_query_parsed(query, ctx).await; - PrometheusJsonResponse::from_query_result( - result, - metric_name, - result_type, - query_id.as_deref(), - ) - .await + PrometheusJsonResponse::from_query_result(result, result_type, query_id.as_deref()).await } } diff --git a/src/servers/src/http/prometheus.rs b/src/servers/src/http/prometheus.rs index e417044ef43..55ae9358da2 100644 --- a/src/servers/src/http/prometheus.rs +++ b/src/servers/src/http/prometheus.rs @@ -492,11 +492,10 @@ async fn do_instant_query( prom_query: ParsedPromQuery, query_ctx: QueryContextRef, ) -> PrometheusJsonResponse { - let (metric_name, result_type) = retrieve_metric_name_and_result_type(prom_query.expr()); + let (_, result_type) = retrieve_metric_name_and_result_type(prom_query.expr()); let query_id = query_ctx.remote_query_id().map(str::to_string); let result = handler.do_query_parsed(prom_query, query_ctx).await; - PrometheusJsonResponse::from_query_result(result, metric_name, result_type, query_id.as_deref()) - .await + PrometheusJsonResponse::from_query_result(result, result_type, query_id.as_deref()).await } #[derive(Debug, Default, Serialize, Deserialize)] @@ -628,20 +627,13 @@ async fn do_range_query( prom_query: ParsedPromQuery, query_ctx: QueryContextRef, ) -> PrometheusJsonResponse { - let (metric_name, _) = retrieve_metric_name_and_result_type(prom_query.expr()); let query_id = query_ctx.remote_query_id().map(str::to_string); // Matrix serialization sorts samples and series, so execution order never // reaches the response. let result = handler .do_query_parsed(prom_query.with_unordered_output(), query_ctx) .await; - PrometheusJsonResponse::from_query_result( - result, - metric_name, - ValueType::Matrix, - query_id.as_deref(), - ) - .await + PrometheusJsonResponse::from_query_result(result, ValueType::Matrix, query_id.as_deref()).await } #[derive(Debug, Default, Serialize)] diff --git a/src/servers/src/http/result/prometheus_resp.rs b/src/servers/src/http/result/prometheus_resp.rs index 16d7c293539..cfa880b03b1 100644 --- a/src/servers/src/http/result/prometheus_resp.rs +++ b/src/servers/src/http/result/prometheus_resp.rs @@ -228,7 +228,6 @@ impl PrometheusJsonResponse { /// Convert from `Result` pub async fn from_query_result( result: Result, - metric_name: Option, result_type: ValueType, query_id: Option<&str>, ) -> Self { @@ -236,19 +235,18 @@ impl PrometheusJsonResponse { let collector = query_id.and_then(get_promql_annotation_collector); let response: Result = async { let result = result?; - let mut resp = - match result.data { - OutputData::RecordBatches(batches) => Self::success( - Self::record_batches_to_data(batches, metric_name, result_type)?, - ), - OutputData::Stream(stream) => Self::success( - Self::consume_stream_to_data(stream, metric_name, result_type).await?, - ), - OutputData::AffectedRows(_) => Self::error( - StatusCode::Unexpected, - "expected data result, but got affected rows", - ), - }; + let mut resp = match result.data { + OutputData::RecordBatches(batches) => { + Self::success(Self::record_batches_to_data(batches, result_type)?) + } + OutputData::Stream(stream) => { + Self::success(Self::consume_stream_to_data(stream, result_type).await?) + } + OutputData::AffectedRows(_) => Self::error( + StatusCode::Unexpected, + "expected data result, but got affected rows", + ), + }; if let Some(physical_plan) = result.meta.plan { let mut result_map = HashMap::new(); @@ -293,7 +291,6 @@ impl PrometheusJsonResponse { /// Convert [RecordBatches] to [PromData] fn record_batches_to_data( batches: RecordBatches, - metric_name: Option, result_type: ValueType, ) -> Result { // Return empty result if no batches @@ -308,7 +305,7 @@ impl PrometheusJsonResponse { let mut buffer = IndexMap::, PromSeriesSamples>::new(); for batch in batches.iter() { - merge_batch(&mut buffer, &layout, batch, metric_name.as_deref())?; + merge_batch(&mut buffer, &layout, batch)?; } samples_to_data(buffer, result_type) @@ -321,7 +318,6 @@ impl PrometheusJsonResponse { /// never coexist in memory. async fn consume_stream_to_data( mut stream: SendableRecordBatchStream, - metric_name: Option, result_type: ValueType, ) -> Result { // An empty stream carries no batch to infer the column roles from, and @@ -336,12 +332,12 @@ impl PrometheusJsonResponse { // Tag order matters, e.g., after sorc and sort_desc, the output order must be kept. let mut buffer = IndexMap::, PromSeriesSamples>::new(); - merge_batch(&mut buffer, &layout, &first_batch, metric_name.as_deref())?; + merge_batch(&mut buffer, &layout, &first_batch)?; // Release the batch before polling the next one. drop(first_batch); while let Some(batch) = stream.try_next().await.context(CollectRecordbatchSnafu)? { - merge_batch(&mut buffer, &layout, &batch, metric_name.as_deref())?; + merge_batch(&mut buffer, &layout, &batch)?; } samples_to_data(buffer, result_type) @@ -435,13 +431,25 @@ impl ColumnLayout { .fail(); } - if metric_name_column_index.is_some() { - tag_column_indices.retain(|index| schema.column_name_by_index(*index) != METRIC_NAME); - } - + // Only the marked identity supplies `__name__`; retain an ordinary physical + // column under an unoccupied label so distinct series are never merged. let tag_names = tag_column_indices .iter() - .map(|index| schema.column_name_by_index(*index).to_string()) + .map(|index| { + let name = schema.column_name_by_index(*index); + if name != METRIC_NAME { + return name.to_string(); + } + let mut exported = "exported_name".to_string(); + while schema + .column_schemas() + .iter() + .any(|column| column.name == exported) + { + exported.push('_'); + } + exported + }) .collect::>(); Ok(Self { @@ -481,7 +489,6 @@ fn merge_batch( buffer: &mut IndexMap, PromSeriesSamples>, layout: &ColumnLayout, batch: &RecordBatch, - metric_name: Option<&str>, ) -> Result<()> { // Scratch buffer of the labels of the series being read, borrowed from // `batch` and `layout` and refilled for every series that is not in @@ -576,14 +583,7 @@ fn merge_batch( let marked_name = metric_name_column .and_then(|column| string_array_value_at_index(column, row_index)) .filter(|name| !name.is_empty()); - // A schema with the marker takes the name from the marked - // column alone; a null or empty marker never falls back to - // the caller-provided name. - let name = if metric_name_column.is_some() { - marked_name - } else { - metric_name - }; + let name = marked_name; if let Some(name) = name { tags.push((METRIC_NAME, name)); } @@ -820,7 +820,7 @@ mod tests { let stream = RecordBatches::try_new(schema, vec![]).unwrap().as_stream(); let response = - PrometheusJsonResponse::consume_stream_to_data(stream, None, ValueType::Matrix).await; + PrometheusJsonResponse::consume_stream_to_data(stream, ValueType::Matrix).await; assert!(matches!( response.unwrap(), @@ -843,10 +843,9 @@ mod tests { } #[test] - fn marked_metric_name_overrides_external_fallback_and_ordinary_name_tag() { - // The marked column is the renamed physical identity, while both the - // caller-provided name and an ordinary `__name__` label column would - // otherwise supply the metric name. + fn marked_metric_name_overrides_ordinary_name_tag() { + // The marked column is the renamed physical identity; the ordinary `__name__` + // label column must not override it. let schema = Arc::new(Schema::new(vec![ ColumnSchema::new( "timestamp", @@ -871,7 +870,6 @@ mod tests { .unwrap(); let response = PrometheusJsonResponse::record_batches_to_data( RecordBatches::try_new(schema, vec![batch]).unwrap(), - Some("external_fallback".to_string()), ValueType::Vector, ) .unwrap(); @@ -891,6 +889,108 @@ mod tests { series[0].metric.get("host").map(String::as_str), Some("host-a") ); + assert_eq!( + series[0].metric.get("exported_name").map(String::as_str), + Some("ordinary_name") + ); + } + + #[tokio::test] + async fn unmarked_ordinary_name_never_restores_metric_identity() { + let schema = Arc::new(Schema::new(vec![ + ColumnSchema::new( + "timestamp", + ConcreteDataType::timestamp_millisecond_datatype(), + false, + ), + ColumnSchema::new("__name__", ConcreteDataType::string_datatype(), true), + ColumnSchema::new("exported_name", ConcreteDataType::string_datatype(), true), + ColumnSchema::new("value", ConcreteDataType::float64_datatype(), false), + ])); + let batch = RecordBatch::new( + schema.clone(), + vec![ + Arc::new(TimestampMillisecondVector::from_values([1_000, 1_000])) as _, + Arc::new(StringVector::from(vec![ + Some("physical_a"), + Some("physical_b"), + ])) as _, + Arc::new(StringVector::from(vec![Some("other"), Some("other")])) as _, + Arc::new(Float64Vector::from_values([1.0, 2.0])) as _, + ], + ) + .unwrap(); + for result_type in [ValueType::Vector, ValueType::Matrix] { + let buffered = PrometheusJsonResponse::record_batches_to_data( + RecordBatches::try_new(schema.clone(), vec![batch.clone()]).unwrap(), + result_type, + ) + .unwrap(); + let streamed = PrometheusJsonResponse::consume_stream_to_data( + RecordBatches::try_new( + schema.clone(), + vec![batch.slice(0, 1).unwrap(), batch.slice(1, 1).unwrap()], + ) + .unwrap() + .as_stream(), + result_type, + ) + .await + .unwrap(); + assert_eq!( + serde_json::to_value(&buffered).unwrap(), + serde_json::to_value(&streamed).unwrap() + ); + let PrometheusResponse::PromData(PromData { result, .. }) = buffered else { + panic!("expected Prometheus data"); + }; + let (metrics, values) = match &result { + PromQueryResult::Vector(series) => ( + series + .iter() + .map(|series| &series.metric) + .collect::>(), + series + .iter() + .map(|series| series.value.as_ref().unwrap().1.clone()) + .collect::>(), + ), + PromQueryResult::Matrix(series) => ( + series + .iter() + .map(|series| &series.metric) + .collect::>(), + series + .iter() + .map(|series| { + serde_json::to_value(&series.values[0].1) + .unwrap() + .as_str() + .unwrap() + .to_string() + }) + .collect::>(), + ), + _ => panic!("expected vector or matrix"), + }; + assert_eq!(metrics.len(), 2); + for metric in &metrics { + assert!(!metric.contains_key(METRIC_NAME)); + assert_eq!( + metric.get("exported_name").map(String::as_str), + Some("other") + ); + } + assert_eq!( + metrics[0].get("exported_name_").map(String::as_str), + Some("physical_a") + ); + assert_eq!( + metrics[1].get("exported_name_").map(String::as_str), + Some("physical_b") + ); + assert_eq!(values, vec!["1.0", "2.0"]); + } } #[tokio::test] @@ -925,16 +1025,12 @@ mod tests { ) .unwrap(); let batches = RecordBatches::try_new(schema.clone(), vec![batch.clone()]).unwrap(); - let streamed = PrometheusJsonResponse::consume_stream_to_data( - batches.as_stream(), - None, - ValueType::Matrix, - ) - .await - .unwrap(); + let streamed = + PrometheusJsonResponse::consume_stream_to_data(batches.as_stream(), ValueType::Matrix) + .await + .unwrap(); let eager = PrometheusJsonResponse::record_batches_to_data( RecordBatches::try_new(schema, vec![batch]).unwrap(), - None, ValueType::Matrix, ) .unwrap(); @@ -963,8 +1059,8 @@ mod tests { } #[test] - fn marked_null_or_empty_metric_name_does_not_restore_fallback() { - // Both rows share their ordinary labels, so a fallback name would merge + fn marked_null_or_empty_metric_name_is_not_emitted() { + // Both rows share their ordinary labels; a fabricated name would merge // them into one named series instead of leaving the name unset. let schema = Arc::new(Schema::new(vec![ ColumnSchema::new( @@ -988,7 +1084,6 @@ mod tests { .unwrap(); let response = PrometheusJsonResponse::record_batches_to_data( RecordBatches::try_new(schema, vec![batch]).unwrap(), - Some("fallback".to_string()), ValueType::Matrix, ) .unwrap(); @@ -1093,14 +1188,12 @@ mod tests { RecordBatches::try_new(schema.clone(), batches.clone()) .unwrap() .as_stream(), - Some("metric".to_string()), ValueType::Matrix, ) .await .unwrap(); let eager = PrometheusJsonResponse::record_batches_to_data( RecordBatches::try_new(schema, batches).unwrap(), - Some("metric".to_string()), ValueType::Matrix, ) .unwrap(); @@ -1187,14 +1280,12 @@ mod tests { RecordBatches::try_new(schema.clone(), vec![first.clone(), second.clone()]) .unwrap() .as_stream(), - Some("mixed_metric".to_string()), ValueType::Vector, ) .await .unwrap(); let eager = PrometheusJsonResponse::record_batches_to_data( RecordBatches::try_new(schema, vec![first, second]).unwrap(), - Some("mixed_metric".to_string()), ValueType::Vector, ) .unwrap(); @@ -1310,12 +1401,9 @@ mod tests { ])); let stream = LazyDropCheckingStream::new(schema); - let response = PrometheusJsonResponse::consume_stream_to_data( - Box::pin(stream), - None, - ValueType::Matrix, - ) - .await; + let response = + PrometheusJsonResponse::consume_stream_to_data(Box::pin(stream), ValueType::Matrix) + .await; assert!(matches!( response.unwrap(), @@ -1334,7 +1422,6 @@ mod tests { left.record_info("left info"); let mut response = PrometheusJsonResponse::from_query_result( Ok(Output::new_with_record_batches(RecordBatches::empty())), - None, ValueType::Vector, Some(query_id), ) @@ -1528,7 +1615,6 @@ mod tests { .unwrap(); let actual = PrometheusJsonResponse::from_query_result( Ok(Output::new_with_record_batches(batches)), - None, ValueType::Matrix, None, ) @@ -1577,8 +1663,7 @@ mod tests { let batches = RecordBatches::try_new(schema, vec![batch]).unwrap(); let response = - PrometheusJsonResponse::record_batches_to_data(batches, None, ValueType::Matrix) - .unwrap(); + PrometheusJsonResponse::record_batches_to_data(batches, ValueType::Matrix).unwrap(); let PrometheusResponse::PromData(data) = response else { panic!("expected Prometheus data response"); }; @@ -1623,8 +1708,7 @@ mod tests { let batches = RecordBatches::try_new(schema, vec![batch]).unwrap(); let response = - PrometheusJsonResponse::record_batches_to_data(batches, None, ValueType::Matrix) - .unwrap(); + PrometheusJsonResponse::record_batches_to_data(batches, ValueType::Matrix).unwrap(); let PrometheusResponse::PromData(PromData { result: PromQueryResult::Matrix(series), .. @@ -1684,8 +1768,7 @@ mod tests { let batches = RecordBatches::try_new(schema, vec![batch]).unwrap(); let response = - PrometheusJsonResponse::record_batches_to_data(batches, None, ValueType::Matrix) - .unwrap(); + PrometheusJsonResponse::record_batches_to_data(batches, ValueType::Matrix).unwrap(); let PrometheusResponse::PromData(PromData { result: PromQueryResult::Matrix(series), .. @@ -1742,12 +1825,8 @@ mod tests { .unwrap(); let batches = RecordBatches::try_new(schema, vec![batch]).unwrap(); - let response = PrometheusJsonResponse::record_batches_to_data( - batches, - Some("metric".to_string()), - ValueType::Vector, - ) - .unwrap(); + let response = + PrometheusJsonResponse::record_batches_to_data(batches, ValueType::Vector).unwrap(); let PrometheusResponse::PromData(PromData { result: PromQueryResult::Vector(series), .. @@ -1833,7 +1912,6 @@ mod tests { } let response = PrometheusJsonResponse::record_batches_to_data( RecordBatches::try_new(schema, batches).unwrap(), - None, ValueType::Matrix, ) .unwrap(); @@ -1910,7 +1988,6 @@ mod tests { let response = PrometheusJsonResponse::record_batches_to_data( RecordBatches::try_new(schema, batches).unwrap(), - None, ValueType::Matrix, ) .unwrap(); @@ -2007,7 +2084,6 @@ mod tests { } let response = PrometheusJsonResponse::record_batches_to_data( RecordBatches::try_new(schema.clone(), batches).unwrap(), - Some("metric".to_string()), ValueType::Matrix, ) .unwrap(); @@ -2032,11 +2108,11 @@ mod tests { assert_eq!( serde_json::to_value(&clustered[..4]).unwrap(), serde_json::json!([ - {"metric": {"__name__": "metric"}, "values": [[6.0, "6.0"], [8.0, "8.0"]]}, - {"metric": {"__name__": "metric", "host": ""}, "values": [[7.0, "7.0"]]}, - {"metric": {"__name__": "metric", "host": "a"}, + {"metric": {}, "values": [[6.0, "6.0"], [8.0, "8.0"]]}, + {"metric": {"host": ""}, "values": [[7.0, "7.0"]]}, + {"metric": {"host": "a"}, "values": [[4.0, "4.0"], [5.0, "5.0"]]}, - {"metric": {"__name__": "metric", "host": "a", "rack": "r"}, + {"metric": {"host": "a", "rack": "r"}, "values": [[1.0, "1.0"], [2.0, "2.0"], [3.0, "3.0"]]}, ]) ); @@ -2075,12 +2151,8 @@ mod tests { .unwrap(); let batches = RecordBatches::try_new(schema, vec![batch]).unwrap(); - let response = PrometheusJsonResponse::record_batches_to_data( - batches, - Some("mixed_metric".to_string()), - ValueType::Vector, - ) - .unwrap(); + let response = + PrometheusJsonResponse::record_batches_to_data(batches, ValueType::Vector).unwrap(); let PrometheusResponse::PromData(PromData { result: PromQueryResult::Vector(series), .. @@ -2138,12 +2210,8 @@ mod tests { .unwrap(); let batches = RecordBatches::try_new(schema, vec![batch]).unwrap(); - let response = PrometheusJsonResponse::record_batches_to_data( - batches, - Some("label_replace_repro".to_string()), - ValueType::Vector, - ) - .unwrap(); + let response = + PrometheusJsonResponse::record_batches_to_data(batches, ValueType::Vector).unwrap(); let PrometheusResponse::PromData(PromData { result: PromQueryResult::Vector(series), .. @@ -2153,7 +2221,7 @@ mod tests { }; assert_eq!(series.len(), 1); - assert_eq!(series[0].metric["__name__"], "label_replace_repro"); + assert!(!series[0].metric.contains_key(METRIC_NAME)); assert_eq!(series[0].metric["host"], "server-01"); assert_eq!(series[0].metric["host_copy"], "server-01"); assert_eq!(series[0].value, Some((1.0, "1.0".to_string()))); @@ -2184,8 +2252,7 @@ mod tests { let batches = RecordBatches::try_new(schema, vec![batch]).unwrap(); let response = - PrometheusJsonResponse::record_batches_to_data(batches, None, ValueType::Matrix) - .unwrap(); + PrometheusJsonResponse::record_batches_to_data(batches, ValueType::Matrix).unwrap(); let PrometheusResponse::PromData(PromData { result: PromQueryResult::Matrix(series), .. diff --git a/src/servers/tests/http/prom_count_values_test.rs b/src/servers/tests/http/prom_count_values_test.rs index ab17ef8791f..d37bd81ffcb 100644 --- a/src/servers/tests/http/prom_count_values_test.rs +++ b/src/servers/tests/http/prom_count_values_test.rs @@ -152,7 +152,7 @@ async fn count_values_generated_label_is_reported_as_a_prometheus_label() { let output = query_engine.execute(plan, query_ctx).await.unwrap(); let response = - PrometheusJsonResponse::from_query_result(Ok(output), None, ValueType::Vector, None).await; + PrometheusJsonResponse::from_query_result(Ok(output), ValueType::Vector, None).await; let PrometheusResponse::PromData(PromData { result: PromQueryResult::Vector(series), ..