mirror of
https://github.com/GreptimeTeam/greptimedb.git
synced 2026-10-20 02:42:29 +00:00
feat: persist Prometheus remote write v2 native histograms (#8382)
* chore: add histogram decode Signed-off-by: shuiyisong <xixing.sys@gmail.com> * chore: add convert func Signed-off-by: shuiyisong <xixing.sys@gmail.com> * feat: implement native histogram persistency Signed-off-by: shuiyisong <xixing.sys@gmail.com> * chore: add test Signed-off-by: shuiyisong <xixing.sys@gmail.com> * chore: refactor Signed-off-by: shuiyisong <xixing.sys@gmail.com> * chore: add README for v2 Signed-off-by: shuiyisong <xixing.sys@gmail.com> * chore: fix CR issues Signed-off-by: shuiyisong <xixing.sys@gmail.com> * fix: cr issue Signed-off-by: shuiyisong <xixing.sys@gmail.com> * fix: cr issue Signed-off-by: shuiyisong <xixing.sys@gmail.com> * feat: implement metric engine Signed-off-by: shuiyisong <xixing.sys@gmail.com> * chore: add comments Signed-off-by: shuiyisong <xixing.sys@gmail.com> * chore: fix test Signed-off-by: shuiyisong <xixing.sys@gmail.com> * chore: store nh in one struct Signed-off-by: shuiyisong <xixing.sys@gmail.com> * chore: fix CR issues Signed-off-by: shuiyisong <xixing.sys@gmail.com> * chore: fix CR issues Signed-off-by: shuiyisong <xixing.sys@gmail.com> * chore: fix CR issues Signed-off-by: shuiyisong <xixing.sys@gmail.com> * chore: fix CR issues Signed-off-by: shuiyisong <xixing.sys@gmail.com> * chore: fix test Signed-off-by: shuiyisong <xixing.sys@gmail.com> * fix: cr issue Signed-off-by: shuiyisong <xixing.sys@gmail.com> * chore: fix CR issues and add feature gate Signed-off-by: shuiyisong <xixing.sys@gmail.com> * chore: fix CR issues Signed-off-by: shuiyisong <xixing.sys@gmail.com> * fix: add config option Signed-off-by: shuiyisong <xixing.sys@gmail.com> --------- Signed-off-by: shuiyisong <xixing.sys@gmail.com>
This commit is contained in:
shuiyisong
@@ -32,6 +32,7 @@
|
||||
| `http.enable_cors` | Bool | `true` | HTTP CORS support, it's turned on by default<br/>This allows browser to access http APIs without CORS restrictions |
|
||||
| `http.cors_allowed_origins` | Array | Unset | Customize allowed origins for HTTP CORS. |
|
||||
| `http.prom_validation_mode` | String | `strict` | Whether to enable validation for Prometheus remote write requests.<br/>Available options:<br/>- strict: deny invalid UTF-8 strings (default).<br/>- lossy: allow invalid UTF-8 strings, replace invalid characters with REPLACEMENT_CHARACTER(U+FFFD).<br/>- unchecked: do not valid strings. |
|
||||
| `http.experimental_enable_prometheus_native_histogram` | Bool | `false` | Experimental: enable Prometheus remote write v2 native histogram ingestion. |
|
||||
| `http.experimental_enable_explain_analyze_stream` | Bool | `false` | Experimental: enable POST /v1/sql/analyze/stream for streaming EXPLAIN ANALYZE VERBOSE metrics. |
|
||||
| `grpc` | -- | -- | The gRPC server options. |
|
||||
| `grpc.bind_addr` | String | `127.0.0.1:4001` | The address to bind the gRPC server. |
|
||||
@@ -248,6 +249,7 @@
|
||||
| `http.enable_cors` | Bool | `true` | HTTP CORS support, it's turned on by default<br/>This allows browser to access http APIs without CORS restrictions |
|
||||
| `http.cors_allowed_origins` | Array | Unset | Customize allowed origins for HTTP CORS. |
|
||||
| `http.prom_validation_mode` | String | `strict` | Whether to enable validation for Prometheus remote write requests.<br/>Available options:<br/>- strict: deny invalid UTF-8 strings (default).<br/>- lossy: allow invalid UTF-8 strings, replace invalid characters with REPLACEMENT_CHARACTER(U+FFFD).<br/>- unchecked: do not valid strings. |
|
||||
| `http.experimental_enable_prometheus_native_histogram` | Bool | `false` | Experimental: enable Prometheus remote write v2 native histogram ingestion. |
|
||||
| `http.experimental_enable_explain_analyze_stream` | Bool | `false` | Experimental: enable POST /v1/sql/analyze/stream for streaming EXPLAIN ANALYZE VERBOSE metrics. |
|
||||
| `grpc` | -- | -- | The gRPC server options. |
|
||||
| `grpc.bind_addr` | String | `127.0.0.1:4001` | The address to bind the gRPC server. |
|
||||
|
||||
@@ -57,6 +57,8 @@ cors_allowed_origins = ["https://example.com"]
|
||||
## - lossy: allow invalid UTF-8 strings, replace invalid characters with REPLACEMENT_CHARACTER(U+FFFD).
|
||||
## - unchecked: do not valid strings.
|
||||
prom_validation_mode = "strict"
|
||||
## Experimental: enable Prometheus remote write v2 native histogram ingestion.
|
||||
experimental_enable_prometheus_native_histogram = false
|
||||
## Experimental: enable POST /v1/sql/analyze/stream for streaming EXPLAIN ANALYZE VERBOSE metrics.
|
||||
experimental_enable_explain_analyze_stream = false
|
||||
|
||||
|
||||
@@ -71,6 +71,8 @@ cors_allowed_origins = ["https://example.com"]
|
||||
## - lossy: allow invalid UTF-8 strings, replace invalid characters with REPLACEMENT_CHARACTER(U+FFFD).
|
||||
## - unchecked: do not valid strings.
|
||||
prom_validation_mode = "strict"
|
||||
## Experimental: enable Prometheus remote write v2 native histogram ingestion.
|
||||
experimental_enable_prometheus_native_histogram = false
|
||||
## Experimental: enable POST /v1/sql/analyze/stream for streaming EXPLAIN ANALYZE VERBOSE metrics.
|
||||
experimental_enable_explain_analyze_stream = false
|
||||
|
||||
|
||||
@@ -15,6 +15,7 @@
|
||||
pub mod columnar_value;
|
||||
pub mod error;
|
||||
pub mod logical_plan;
|
||||
pub mod native_histogram;
|
||||
pub mod prelude;
|
||||
pub mod request;
|
||||
pub mod stream;
|
||||
|
||||
@@ -0,0 +1,111 @@
|
||||
// Copyright 2023 Greptime Team
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
//! Shared native histogram field contract.
|
||||
//!
|
||||
//! Prom remote-write v2 stores these names as children of one Struct field, while
|
||||
//! metric-engine uses the same contract to recognize native histogram tables.
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use datatypes::data_type::ConcreteDataType;
|
||||
use datatypes::types::{StructField, StructType};
|
||||
use once_cell::sync::Lazy;
|
||||
|
||||
pub const NATIVE_HISTOGRAM_FIELD: &str = "greptime_native_histogram";
|
||||
pub const SCHEMA_FIELD: &str = "schema";
|
||||
pub const ZERO_THRESHOLD_FIELD: &str = "zero_threshold";
|
||||
pub const SUM_FIELD: &str = "sum";
|
||||
pub const RESET_HINT_FIELD: &str = "reset_hint";
|
||||
pub const START_TIMESTAMP_FIELD: &str = "start_timestamp";
|
||||
pub const CUSTOM_VALUES_FIELD: &str = "custom_values";
|
||||
pub const POSITIVE_SPAN_OFFSETS_FIELD: &str = "positive_span_offsets";
|
||||
pub const POSITIVE_SPAN_LENGTHS_FIELD: &str = "positive_span_lengths";
|
||||
pub const NEGATIVE_SPAN_OFFSETS_FIELD: &str = "negative_span_offsets";
|
||||
pub const NEGATIVE_SPAN_LENGTHS_FIELD: &str = "negative_span_lengths";
|
||||
pub const COUNT_U64_FIELD: &str = "count_u64";
|
||||
pub const ZERO_COUNT_U64_FIELD: &str = "zero_count_u64";
|
||||
pub const POSITIVE_BUCKETS_I64_FIELD: &str = "positive_buckets_i64";
|
||||
pub const NEGATIVE_BUCKETS_I64_FIELD: &str = "negative_buckets_i64";
|
||||
pub const COUNT_F64_FIELD: &str = "count_f64";
|
||||
pub const ZERO_COUNT_F64_FIELD: &str = "zero_count_f64";
|
||||
pub const POSITIVE_BUCKETS_F64_FIELD: &str = "positive_buckets_f64";
|
||||
pub const NEGATIVE_BUCKETS_F64_FIELD: &str = "negative_buckets_f64";
|
||||
|
||||
// Keep int and float payloads in separate columns. The populated family is the
|
||||
// type signal, so we don't need to persist an extra histogram-type tag.
|
||||
pub const NATIVE_HISTOGRAM_FIELD_NAMES: &[&str] = &[
|
||||
SCHEMA_FIELD,
|
||||
ZERO_THRESHOLD_FIELD,
|
||||
SUM_FIELD,
|
||||
RESET_HINT_FIELD,
|
||||
START_TIMESTAMP_FIELD,
|
||||
CUSTOM_VALUES_FIELD,
|
||||
POSITIVE_SPAN_OFFSETS_FIELD,
|
||||
POSITIVE_SPAN_LENGTHS_FIELD,
|
||||
NEGATIVE_SPAN_OFFSETS_FIELD,
|
||||
NEGATIVE_SPAN_LENGTHS_FIELD,
|
||||
COUNT_U64_FIELD,
|
||||
ZERO_COUNT_U64_FIELD,
|
||||
POSITIVE_BUCKETS_I64_FIELD,
|
||||
NEGATIVE_BUCKETS_I64_FIELD,
|
||||
COUNT_F64_FIELD,
|
||||
ZERO_COUNT_F64_FIELD,
|
||||
POSITIVE_BUCKETS_F64_FIELD,
|
||||
NEGATIVE_BUCKETS_F64_FIELD,
|
||||
];
|
||||
|
||||
static NATIVE_HISTOGRAM_VALUE_TYPE: Lazy<ConcreteDataType> = Lazy::new(|| {
|
||||
let fields = NATIVE_HISTOGRAM_FIELD_NAMES
|
||||
.iter()
|
||||
.filter_map(|name| {
|
||||
let data_type = native_histogram_field_type(name)?;
|
||||
Some(StructField::new((*name).to_string(), data_type, true))
|
||||
})
|
||||
.collect();
|
||||
ConcreteDataType::struct_datatype(StructType::new(Arc::new(fields)))
|
||||
});
|
||||
|
||||
/// Returns the exact Greptime type for a persisted native histogram field.
|
||||
pub fn native_histogram_field_type(name: &str) -> Option<ConcreteDataType> {
|
||||
match name {
|
||||
SCHEMA_FIELD | RESET_HINT_FIELD => Some(ConcreteDataType::int32_datatype()),
|
||||
ZERO_THRESHOLD_FIELD | SUM_FIELD | COUNT_F64_FIELD | ZERO_COUNT_F64_FIELD => {
|
||||
Some(ConcreteDataType::float64_datatype())
|
||||
}
|
||||
START_TIMESTAMP_FIELD => Some(ConcreteDataType::timestamp_millisecond_datatype()),
|
||||
CUSTOM_VALUES_FIELD | POSITIVE_BUCKETS_F64_FIELD | NEGATIVE_BUCKETS_F64_FIELD => Some(
|
||||
ConcreteDataType::list_datatype(Arc::new(ConcreteDataType::float64_datatype())),
|
||||
),
|
||||
POSITIVE_SPAN_OFFSETS_FIELD | NEGATIVE_SPAN_OFFSETS_FIELD => Some(
|
||||
ConcreteDataType::list_datatype(Arc::new(ConcreteDataType::int32_datatype())),
|
||||
),
|
||||
POSITIVE_SPAN_LENGTHS_FIELD | NEGATIVE_SPAN_LENGTHS_FIELD => Some(
|
||||
ConcreteDataType::list_datatype(Arc::new(ConcreteDataType::uint32_datatype())),
|
||||
),
|
||||
COUNT_U64_FIELD | ZERO_COUNT_U64_FIELD => Some(ConcreteDataType::uint64_datatype()),
|
||||
POSITIVE_BUCKETS_I64_FIELD | NEGATIVE_BUCKETS_I64_FIELD => Some(
|
||||
ConcreteDataType::list_datatype(Arc::new(ConcreteDataType::int64_datatype())),
|
||||
),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn native_histogram_value_type() -> &'static ConcreteDataType {
|
||||
&NATIVE_HISTOGRAM_VALUE_TYPE
|
||||
}
|
||||
|
||||
pub fn is_native_histogram_value_schema(name: &str, data_type: &ConcreteDataType) -> bool {
|
||||
name == NATIVE_HISTOGRAM_FIELD && data_type == native_histogram_value_type()
|
||||
}
|
||||
@@ -59,7 +59,9 @@ Mito2 data region. Queries against a physical region pass straight through.
|
||||
`store-api`'s metric engine consts; keep them in sync with `engine.rs`.
|
||||
- **Metadata K-V encoding** (`metadata_region.rs`): changes the on-disk metadata layout.
|
||||
- **RegionId group mapping** (`utils.rs`): data vs metadata region derivation.
|
||||
- **Physical column rules** (`engine/alter/`): a physical region allows only one field column.
|
||||
- **Physical column rules** (`engine/alter/`): ordinary metric fields stay on
|
||||
logical tables; shared physical field adds are only for the native histogram
|
||||
Struct field.
|
||||
|
||||
## Testing
|
||||
|
||||
|
||||
@@ -13,6 +13,7 @@
|
||||
// limitations under the License.
|
||||
|
||||
use api::v1::SemanticType;
|
||||
use common_query::native_histogram::is_native_histogram_value_schema;
|
||||
use common_telemetry::{debug, info};
|
||||
use datatypes::schema::{SkippingIndexOptions, SkippingIndexType};
|
||||
use mito2::engine::MitoEngine;
|
||||
@@ -123,42 +124,55 @@ impl DataRegion {
|
||||
.into_iter()
|
||||
.enumerate()
|
||||
.map(|(delta, mut c)| {
|
||||
if c.semantic_type == SemanticType::Tag {
|
||||
if !c.column_schema.data_type.is_string() {
|
||||
return ColumnTypeMismatchSnafu {
|
||||
expect: ConcreteDataType::string_datatype(),
|
||||
actual: c.column_schema.data_type.clone(),
|
||||
match c.semantic_type {
|
||||
SemanticType::Tag => {
|
||||
if !c.column_schema.data_type.is_string() {
|
||||
return ColumnTypeMismatchSnafu {
|
||||
expect: ConcreteDataType::string_datatype(),
|
||||
actual: c.column_schema.data_type.clone(),
|
||||
}
|
||||
.fail();
|
||||
}
|
||||
}
|
||||
// Field columns can only be added to the shared physical
|
||||
// table for native histograms; ordinary metric fields are
|
||||
// created with the logical table.
|
||||
SemanticType::Field
|
||||
if is_native_histogram_value_schema(
|
||||
&c.column_schema.name,
|
||||
&c.column_schema.data_type,
|
||||
) => {}
|
||||
_ => {
|
||||
return AddingFieldColumnSnafu {
|
||||
name: &c.column_schema.name,
|
||||
}
|
||||
.fail();
|
||||
}
|
||||
} else {
|
||||
return AddingFieldColumnSnafu {
|
||||
name: &c.column_schema.name,
|
||||
}
|
||||
.fail();
|
||||
};
|
||||
}
|
||||
|
||||
c.column_id = new_column_id_start + delta as u32;
|
||||
c.column_schema.set_nullable();
|
||||
match index_options {
|
||||
IndexOptions::None => {}
|
||||
IndexOptions::Inverted => {
|
||||
c.column_schema.set_inverted_index(true);
|
||||
}
|
||||
IndexOptions::Skipping {
|
||||
granularity,
|
||||
false_positive_rate,
|
||||
} => {
|
||||
c.column_schema
|
||||
.set_skipping_options(
|
||||
&SkippingIndexOptions::new(
|
||||
granularity,
|
||||
false_positive_rate,
|
||||
SkippingIndexType::BloomFilter,
|
||||
if c.semantic_type == SemanticType::Tag {
|
||||
match index_options {
|
||||
IndexOptions::None => {}
|
||||
IndexOptions::Inverted => {
|
||||
c.column_schema.set_inverted_index(true);
|
||||
}
|
||||
IndexOptions::Skipping {
|
||||
granularity,
|
||||
false_positive_rate,
|
||||
} => {
|
||||
c.column_schema
|
||||
.set_skipping_options(
|
||||
&SkippingIndexOptions::new(
|
||||
granularity,
|
||||
false_positive_rate,
|
||||
SkippingIndexType::BloomFilter,
|
||||
)
|
||||
.context(SetSkippingIndexOptionSnafu)?,
|
||||
)
|
||||
.context(SetSkippingIndexOptionSnafu)?,
|
||||
)
|
||||
.context(SetSkippingIndexOptionSnafu)?;
|
||||
.context(SetSkippingIndexOptionSnafu)?;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -17,6 +17,8 @@ mod validate;
|
||||
|
||||
use std::collections::{BTreeSet, HashMap, HashSet};
|
||||
|
||||
use api::v1::SemanticType;
|
||||
use common_query::native_histogram::is_native_histogram_value_schema;
|
||||
use extract_new_columns::extract_new_columns;
|
||||
use snafu::{OptionExt, ResultExt, ensure};
|
||||
use store_api::metadata::ColumnMetadata;
|
||||
@@ -27,8 +29,8 @@ use validate::validate_alter_region_requests;
|
||||
|
||||
use crate::engine::MetricEngineInner;
|
||||
use crate::error::{
|
||||
LogicalRegionNotFoundSnafu, PhysicalRegionNotFoundSnafu, Result, SerializeColumnMetadataSnafu,
|
||||
UnexpectedRequestSnafu,
|
||||
AddingFieldColumnSnafu, LogicalRegionNotFoundSnafu, PhysicalRegionNotFoundSnafu, Result,
|
||||
SerializeColumnMetadataSnafu, UnexpectedRequestSnafu,
|
||||
};
|
||||
use crate::utils::{append_manifest_info, encode_manifest_info_to_extensions, to_data_region_id};
|
||||
|
||||
@@ -117,6 +119,8 @@ impl MetricEngineInner {
|
||||
) -> Result<AffectedRows> {
|
||||
// Checks all alter requests are add columns.
|
||||
validate_alter_region_requests(&requests)?;
|
||||
self.validate_logical_field_alters(physical_region_id, &requests)
|
||||
.await?;
|
||||
|
||||
// Finds new columns to add
|
||||
let mut new_column_names = HashSet::new();
|
||||
@@ -209,6 +213,52 @@ impl MetricEngineInner {
|
||||
Ok(0)
|
||||
}
|
||||
|
||||
async fn validate_logical_field_alters(
|
||||
&self,
|
||||
physical_region_id: RegionId,
|
||||
requests: &[(RegionId, RegionAlterRequest)],
|
||||
) -> Result<()> {
|
||||
// Logical metric tables have one field column. Native histograms are a
|
||||
// special struct field, so field alters must leave exactly that field.
|
||||
for (region_id, request) in requests {
|
||||
let AlterKind::AddColumns { columns } = &request.kind else {
|
||||
unreachable!()
|
||||
};
|
||||
let added_fields = columns
|
||||
.iter()
|
||||
.filter(|col| col.column_metadata.semantic_type == SemanticType::Field)
|
||||
.collect::<Vec<_>>();
|
||||
let Some(&first_added_field) = added_fields.first() else {
|
||||
continue;
|
||||
};
|
||||
|
||||
let mut fields = self
|
||||
.load_logical_columns(physical_region_id, *region_id)
|
||||
.await?
|
||||
.into_iter()
|
||||
.filter(|col| col.semantic_type == SemanticType::Field)
|
||||
.collect::<Vec<_>>();
|
||||
fields.extend(
|
||||
added_fields
|
||||
.into_iter()
|
||||
.map(|col| col.column_metadata.clone()),
|
||||
);
|
||||
|
||||
ensure!(
|
||||
fields.len() == 1
|
||||
&& is_native_histogram_value_schema(
|
||||
&fields[0].column_schema.name,
|
||||
&fields[0].column_schema.data_type
|
||||
),
|
||||
AddingFieldColumnSnafu {
|
||||
name: first_added_field.column_metadata.column_schema.name.clone(),
|
||||
}
|
||||
);
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn alter_physical_region(
|
||||
&self,
|
||||
region_id: RegionId,
|
||||
|
||||
@@ -15,6 +15,7 @@
|
||||
use std::collections::{HashMap, HashSet};
|
||||
|
||||
use api::v1::SemanticType;
|
||||
use common_query::native_histogram::is_native_histogram_value_schema;
|
||||
use snafu::ensure;
|
||||
use store_api::metadata::ColumnMetadata;
|
||||
use store_api::region_request::{AlterKind, RegionAlterRequest};
|
||||
@@ -43,7 +44,11 @@ pub fn extract_new_columns<'a>(
|
||||
&& !new_column_names.contains(column_name)
|
||||
{
|
||||
ensure!(
|
||||
col.column_metadata.semantic_type != SemanticType::Field,
|
||||
col.column_metadata.semantic_type != SemanticType::Field
|
||||
|| is_native_histogram_value_schema(
|
||||
&col.column_metadata.column_schema.name,
|
||||
&col.column_metadata.column_schema.data_type
|
||||
),
|
||||
AddingFieldColumnSnafu {
|
||||
name: column_name.to_string(),
|
||||
}
|
||||
|
||||
@@ -17,6 +17,7 @@ mod extract_new_columns;
|
||||
use std::collections::{HashMap, HashSet};
|
||||
|
||||
use api::v1::SemanticType;
|
||||
use common_query::native_histogram::is_native_histogram_value_schema;
|
||||
use common_telemetry::info;
|
||||
use common_time::{FOREVER, Timestamp};
|
||||
use datatypes::data_type::ConcreteDataType;
|
||||
@@ -390,8 +391,8 @@ impl MetricEngineInner {
|
||||
ConflictRegionOptionSnafu {}
|
||||
);
|
||||
|
||||
// check if only one field column is declared, and all tag columns are string
|
||||
let mut field_col: Option<&ColumnMetadata> = None;
|
||||
// check if field columns are either a normal metric value or a native histogram.
|
||||
let mut field_cols = Vec::new();
|
||||
for col in &request.column_metadatas {
|
||||
// Verified in above steps.
|
||||
if is_metric_engine_internal_column(&col.column_schema.name) {
|
||||
@@ -406,21 +407,30 @@ impl MetricEngineInner {
|
||||
}
|
||||
),
|
||||
SemanticType::Field => {
|
||||
if let Some(field_col) = field_col {
|
||||
MultipleFieldColumnSnafu {
|
||||
previous: field_col.column_schema.name.clone(),
|
||||
current: col.column_schema.name.clone(),
|
||||
}
|
||||
.fail()?;
|
||||
}
|
||||
field_col = Some(col)
|
||||
field_cols.push(col);
|
||||
}
|
||||
SemanticType::Timestamp => {}
|
||||
}
|
||||
}
|
||||
let field_col = field_col.context(NoFieldColumnSnafu)?;
|
||||
let [field_col] = field_cols.as_slice() else {
|
||||
if field_cols.is_empty() {
|
||||
NoFieldColumnSnafu.fail()?;
|
||||
}
|
||||
return MultipleFieldColumnSnafu {
|
||||
previous: field_cols[0].column_schema.name.clone(),
|
||||
current: field_cols[1].column_schema.name.clone(),
|
||||
}
|
||||
.fail();
|
||||
};
|
||||
|
||||
// make sure the field column is float64 type
|
||||
if is_native_histogram_value_schema(
|
||||
&field_col.column_schema.name,
|
||||
&field_col.column_schema.data_type,
|
||||
) {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
// make sure the normal field column is float64 type
|
||||
ensure!(
|
||||
field_col.column_schema.data_type == ConcreteDataType::float64_datatype(),
|
||||
ColumnTypeMismatchSnafu {
|
||||
@@ -653,6 +663,7 @@ pub(crate) fn region_options_for_metadata_region(
|
||||
mod test {
|
||||
use common_meta::ddl::test_util::assert_column_name_and_id;
|
||||
use common_meta::ddl::utils::{parse_column_metadatas, parse_manifest_infos_from_extensions};
|
||||
use common_query::native_histogram::{NATIVE_HISTOGRAM_FIELD, native_histogram_value_type};
|
||||
use common_query::prelude::{greptime_timestamp, greptime_value};
|
||||
use store_api::metric_engine_consts::{METRIC_ENGINE_NAME, PHYSICAL_TABLE_METADATA_KEY};
|
||||
use store_api::region_request::{BatchRegionDdlRequest, RegionRequirements};
|
||||
@@ -845,6 +856,96 @@ mod test {
|
||||
MetricEngineInner::verify_region_create_request(&request).unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_verify_region_create_request_native_histogram_fields() {
|
||||
let native_histogram_columns = vec![
|
||||
ColumnMetadata {
|
||||
column_id: 0,
|
||||
semantic_type: SemanticType::Timestamp,
|
||||
column_schema: ColumnSchema::new(
|
||||
greptime_timestamp(),
|
||||
ConcreteDataType::timestamp_millisecond_datatype(),
|
||||
false,
|
||||
),
|
||||
},
|
||||
ColumnMetadata {
|
||||
column_id: 1,
|
||||
semantic_type: SemanticType::Tag,
|
||||
column_schema: ColumnSchema::new("job", ConcreteDataType::string_datatype(), true),
|
||||
},
|
||||
ColumnMetadata {
|
||||
column_id: 2,
|
||||
semantic_type: SemanticType::Field,
|
||||
column_schema: ColumnSchema::new(
|
||||
NATIVE_HISTOGRAM_FIELD,
|
||||
native_histogram_value_type().clone(),
|
||||
true,
|
||||
),
|
||||
},
|
||||
];
|
||||
let request = RegionCreateRequest {
|
||||
column_metadatas: native_histogram_columns,
|
||||
table_dir: "test_dir".to_string(),
|
||||
path_type: PathType::Bare,
|
||||
engine: METRIC_ENGINE_NAME.to_string(),
|
||||
primary_key: vec![],
|
||||
options: [(
|
||||
LOGICAL_TABLE_METADATA_KEY.to_string(),
|
||||
"physical".to_string(),
|
||||
)]
|
||||
.into_iter()
|
||||
.collect(),
|
||||
partition_expr_json: Some("".to_string()),
|
||||
requirements: Default::default(),
|
||||
};
|
||||
MetricEngineInner::verify_region_create_request(&request).unwrap();
|
||||
|
||||
let request = RegionCreateRequest {
|
||||
column_metadatas: vec![
|
||||
ColumnMetadata {
|
||||
column_id: 0,
|
||||
semantic_type: SemanticType::Timestamp,
|
||||
column_schema: ColumnSchema::new(
|
||||
greptime_timestamp(),
|
||||
ConcreteDataType::timestamp_millisecond_datatype(),
|
||||
false,
|
||||
),
|
||||
},
|
||||
ColumnMetadata {
|
||||
column_id: 1,
|
||||
semantic_type: SemanticType::Field,
|
||||
column_schema: ColumnSchema::new(
|
||||
"value_a",
|
||||
ConcreteDataType::float64_datatype(),
|
||||
true,
|
||||
),
|
||||
},
|
||||
ColumnMetadata {
|
||||
column_id: 2,
|
||||
semantic_type: SemanticType::Field,
|
||||
column_schema: ColumnSchema::new(
|
||||
"value_b",
|
||||
ConcreteDataType::float64_datatype(),
|
||||
true,
|
||||
),
|
||||
},
|
||||
],
|
||||
table_dir: "test_dir".to_string(),
|
||||
path_type: PathType::Bare,
|
||||
engine: METRIC_ENGINE_NAME.to_string(),
|
||||
primary_key: vec![],
|
||||
options: [(
|
||||
LOGICAL_TABLE_METADATA_KEY.to_string(),
|
||||
"physical".to_string(),
|
||||
)]
|
||||
.into_iter()
|
||||
.collect(),
|
||||
partition_expr_json: Some("".to_string()),
|
||||
requirements: Default::default(),
|
||||
};
|
||||
assert!(MetricEngineInner::verify_region_create_request(&request).is_err());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_create_request_for_physical_regions() {
|
||||
// original request
|
||||
|
||||
@@ -15,6 +15,7 @@
|
||||
use std::collections::{HashMap, HashSet};
|
||||
|
||||
use api::v1::SemanticType;
|
||||
use common_query::native_histogram::is_native_histogram_value_schema;
|
||||
use snafu::ensure;
|
||||
use store_api::metadata::ColumnMetadata;
|
||||
use store_api::region_request::RegionCreateRequest;
|
||||
@@ -35,7 +36,11 @@ pub fn extract_new_columns<'a>(
|
||||
&& !new_column_names.contains(&col.column_schema.name.as_str())
|
||||
{
|
||||
ensure!(
|
||||
col.semantic_type != SemanticType::Field,
|
||||
col.semantic_type != SemanticType::Field
|
||||
|| is_native_histogram_value_schema(
|
||||
&col.column_schema.name,
|
||||
&col.column_schema.data_type
|
||||
),
|
||||
AddingFieldColumnSnafu {
|
||||
name: col.column_schema.name.clone(),
|
||||
}
|
||||
|
||||
@@ -195,12 +195,10 @@ impl MetricEngineInner {
|
||||
requests: Vec<(RegionId, RegionPutRequest)>,
|
||||
) -> Result<(RegionPutRequest, AffectedRows)> {
|
||||
let total_rows: usize = requests.iter().map(|(_, req)| req.rows.rows.len()).sum();
|
||||
let mut merged_rows = Vec::with_capacity(total_rows);
|
||||
let mut modified_requests = Vec::with_capacity(requests.len());
|
||||
let mut total_affected_rows: AffectedRows = 0;
|
||||
let mut output_schema: Option<Vec<ColumnSchema>> = None;
|
||||
let mut merged_version: Option<u64> = None;
|
||||
|
||||
// Modify and collect rows from each request
|
||||
for (logical_region_id, mut request) in requests {
|
||||
if let Some(request_version) = request.partition_expr_version {
|
||||
if let Some(merged_version) = merged_version {
|
||||
@@ -224,17 +222,15 @@ impl MetricEngineInner {
|
||||
|
||||
let row_count = request.rows.rows.len();
|
||||
total_affected_rows += row_count as AffectedRows;
|
||||
|
||||
// Capture the output schema from the first modified request
|
||||
if output_schema.is_none() {
|
||||
output_schema = Some(request.rows.schema.clone());
|
||||
}
|
||||
|
||||
merged_rows.extend(request.rows.rows);
|
||||
modified_requests.push(request.rows);
|
||||
}
|
||||
|
||||
// Safe to unwrap: requests is guaranteed non-empty by caller
|
||||
let schema = output_schema.unwrap();
|
||||
let schema =
|
||||
Self::build_union_schema(modified_requests.iter().map(|rows| rows.schema.as_slice()));
|
||||
let mut merged_rows = Vec::with_capacity(total_rows);
|
||||
for rows in modified_requests {
|
||||
merged_rows.extend(Self::align_rows_to_schema(rows, &schema));
|
||||
}
|
||||
|
||||
let merged_request = RegionPutRequest {
|
||||
rows: Rows {
|
||||
@@ -261,7 +257,8 @@ impl MetricEngineInner {
|
||||
requests: Vec<(RegionId, RegionPutRequest)>,
|
||||
) -> Result<(RegionPutRequest, AffectedRows)> {
|
||||
// Build union schema from all requests
|
||||
let merged_schema = Self::build_union_schema(&requests);
|
||||
let merged_schema =
|
||||
Self::build_union_schema(requests.iter().map(|(_, req)| req.rows.schema.as_slice()));
|
||||
|
||||
// Align all rows to the merged schema and collect table_ids
|
||||
let (merged_rows, table_ids, merged_version) =
|
||||
@@ -302,17 +299,21 @@ impl MetricEngineInner {
|
||||
Ok((merged_request, table_ids.len() as AffectedRows))
|
||||
}
|
||||
|
||||
/// Builds a union schema containing all columns from all requests.
|
||||
fn build_union_schema(requests: &[(RegionId, RegionPutRequest)]) -> Vec<ColumnSchema> {
|
||||
let mut schema_map: HashMap<&str, ColumnSchema> = HashMap::new();
|
||||
for (_, request) in requests {
|
||||
for col in &request.rows.schema {
|
||||
schema_map
|
||||
.entry(col.column_name.as_str())
|
||||
.or_insert_with(|| col.clone());
|
||||
fn build_union_schema<'a>(
|
||||
schemas: impl IntoIterator<Item = &'a [ColumnSchema]>,
|
||||
) -> Vec<ColumnSchema> {
|
||||
let mut schema = Vec::new();
|
||||
for columns in schemas {
|
||||
for col in columns {
|
||||
if !schema
|
||||
.iter()
|
||||
.any(|existing: &ColumnSchema| existing.column_name == col.column_name)
|
||||
{
|
||||
schema.push(col.clone());
|
||||
}
|
||||
}
|
||||
}
|
||||
schema_map.into_values().collect()
|
||||
schema
|
||||
}
|
||||
|
||||
fn align_requests_to_schema(
|
||||
@@ -325,8 +326,6 @@ impl MetricEngineInner {
|
||||
let mut table_ids = Vec::with_capacity(total_rows);
|
||||
let mut merged_version: Option<u64> = None;
|
||||
|
||||
let null_value = Value { value_data: None };
|
||||
|
||||
for (logical_region_id, request) in requests {
|
||||
if let Some(request_version) = request.partition_expr_version {
|
||||
if let Some(merged_version) = merged_version {
|
||||
@@ -342,47 +341,54 @@ impl MetricEngineInner {
|
||||
}
|
||||
}
|
||||
let table_id = logical_region_id.table_id();
|
||||
|
||||
// Build column name to index mapping once per request
|
||||
let col_name_to_idx: FxHashMap<&str, usize> = request
|
||||
.rows
|
||||
.schema
|
||||
.iter()
|
||||
.enumerate()
|
||||
.map(|(idx, col)| (col.column_name.as_str(), idx))
|
||||
.collect();
|
||||
|
||||
// Build column mapping array once per request
|
||||
// col_mapping[i] = Some(idx) means merged_schema[i] is at request.schema[idx]
|
||||
// col_mapping[i] = None means merged_schema[i] doesn't exist in request.schema
|
||||
let col_mapping: Vec<Option<usize>> = merged_schema
|
||||
.iter()
|
||||
.map(|merged_col| {
|
||||
col_name_to_idx
|
||||
.get(merged_col.column_name.as_str())
|
||||
.copied()
|
||||
})
|
||||
.collect();
|
||||
|
||||
// Apply the mapping to all rows
|
||||
for mut row in request.rows.rows {
|
||||
let mut aligned_values = Vec::with_capacity(merged_schema.len());
|
||||
for &opt_idx in &col_mapping {
|
||||
aligned_values.push(match opt_idx {
|
||||
Some(idx) => std::mem::take(&mut row.values[idx]),
|
||||
None => null_value.clone(),
|
||||
});
|
||||
}
|
||||
merged_rows.push(Row {
|
||||
values: aligned_values,
|
||||
});
|
||||
table_ids.push(table_id);
|
||||
}
|
||||
let row_count = request.rows.rows.len();
|
||||
merged_rows.extend(Self::align_rows_to_schema(request.rows, merged_schema));
|
||||
table_ids.extend(std::iter::repeat_n(table_id, row_count));
|
||||
}
|
||||
|
||||
Ok((merged_rows, table_ids, merged_version))
|
||||
}
|
||||
|
||||
fn align_rows_to_schema(rows: Rows, merged_schema: &[ColumnSchema]) -> Vec<Row> {
|
||||
let Rows { schema, rows } = rows;
|
||||
if schema.len() == merged_schema.len()
|
||||
&& schema
|
||||
.iter()
|
||||
.zip(merged_schema)
|
||||
.all(|(left, right)| left.column_name == right.column_name)
|
||||
{
|
||||
return rows;
|
||||
}
|
||||
|
||||
let col_name_to_idx: FxHashMap<&str, usize> = schema
|
||||
.iter()
|
||||
.enumerate()
|
||||
.map(|(idx, col)| (col.column_name.as_str(), idx))
|
||||
.collect();
|
||||
let col_mapping: Vec<Option<usize>> = merged_schema
|
||||
.iter()
|
||||
.map(|merged_col| {
|
||||
col_name_to_idx
|
||||
.get(merged_col.column_name.as_str())
|
||||
.copied()
|
||||
})
|
||||
.collect();
|
||||
let null_value = Value { value_data: None };
|
||||
|
||||
rows.into_iter()
|
||||
.map(|mut row| {
|
||||
let values = col_mapping
|
||||
.iter()
|
||||
.map(|opt_idx| match opt_idx {
|
||||
Some(idx) => std::mem::take(&mut row.values[*idx]),
|
||||
None => null_value.clone(),
|
||||
})
|
||||
.collect();
|
||||
Row { values }
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Find the physical region id for a logical region.
|
||||
fn find_physical_region_id(&self, logical_region_id: RegionId) -> Result<RegionId> {
|
||||
let state = self.state.read().unwrap();
|
||||
@@ -519,7 +525,7 @@ impl MetricEngineInner {
|
||||
/// - Check if every column in the request exists in the physical region
|
||||
/// - Check each column's datatype and semantic type match the physical region's schema
|
||||
/// - Check the time index column is present
|
||||
/// - When `check_fields` is true, check every physical field column is present.
|
||||
/// - When `check_fields` is true, check every logical field column is present.
|
||||
/// Set this to `false` for delete requests, which legitimately carry only
|
||||
/// the primary key + timestamp.
|
||||
async fn verify_rows(
|
||||
@@ -531,23 +537,29 @@ impl MetricEngineInner {
|
||||
) -> Result<()> {
|
||||
// Check if the region exists
|
||||
let data_region_id = to_data_region_id(physical_region_id);
|
||||
let state = self.state.read().unwrap();
|
||||
if !state.is_logical_region_exist(logical_region_id) {
|
||||
error!("Trying to write to an nonexistent region {logical_region_id}");
|
||||
return LogicalRegionNotFoundSnafu {
|
||||
region_id: logical_region_id,
|
||||
let (physical_columns, ts_name) = {
|
||||
let state = self.state.read().unwrap();
|
||||
if !state.is_logical_region_exist(logical_region_id) {
|
||||
error!("Trying to write to an nonexistent region {logical_region_id}");
|
||||
return LogicalRegionNotFoundSnafu {
|
||||
region_id: logical_region_id,
|
||||
}
|
||||
.fail();
|
||||
}
|
||||
.fail();
|
||||
}
|
||||
|
||||
let physical_state = state
|
||||
.physical_region_states()
|
||||
.get(&data_region_id)
|
||||
.context(PhysicalRegionNotFoundSnafu {
|
||||
region_id: data_region_id,
|
||||
})?;
|
||||
(
|
||||
physical_state.physical_columns().clone(),
|
||||
physical_state.time_index_column_name().to_string(),
|
||||
)
|
||||
};
|
||||
|
||||
// Type + semantic check on every column in the request schema.
|
||||
let physical_state = state
|
||||
.physical_region_states()
|
||||
.get(&data_region_id)
|
||||
.context(PhysicalRegionNotFoundSnafu {
|
||||
region_id: data_region_id,
|
||||
})?;
|
||||
let physical_columns = physical_state.physical_columns();
|
||||
for col in &rows.schema {
|
||||
let info = physical_columns
|
||||
.get(&col.column_name)
|
||||
@@ -593,7 +605,6 @@ impl MetricEngineInner {
|
||||
);
|
||||
}
|
||||
|
||||
let ts_name = physical_state.time_index_column_name();
|
||||
ensure!(
|
||||
rows.schema.iter().any(|col| col.column_name == ts_name),
|
||||
InvalidRequestSnafu {
|
||||
@@ -602,17 +613,56 @@ impl MetricEngineInner {
|
||||
}
|
||||
);
|
||||
|
||||
let logical_columns = self
|
||||
.load_logical_columns(physical_region_id, logical_region_id)
|
||||
.await?;
|
||||
let logical_fields = logical_columns
|
||||
.iter()
|
||||
.filter(|col| col.semantic_type == SemanticType::Field)
|
||||
.map(|col| (col.column_schema.name.as_str(), col))
|
||||
.collect::<HashMap<_, _>>();
|
||||
|
||||
for col in &rows.schema {
|
||||
if api::helper::is_semantic_type_eq(col.semantic_type, SemanticType::Field) {
|
||||
ensure!(
|
||||
logical_fields.contains_key(col.column_name.as_str()),
|
||||
InvalidRequestSnafu {
|
||||
region_id: logical_region_id,
|
||||
reason: format!(
|
||||
"field column {} does not belong to logical region {logical_region_id}",
|
||||
col.column_name,
|
||||
),
|
||||
}
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
if check_fields {
|
||||
let field_name = physical_state.field_column_name();
|
||||
if !rows.schema.iter().any(|col| col.column_name == field_name) {
|
||||
let field_meta =
|
||||
physical_columns
|
||||
.get(field_name)
|
||||
.with_context(|| ColumnNotFoundSnafu {
|
||||
name: field_name,
|
||||
region_id: logical_region_id,
|
||||
})?;
|
||||
Self::fill_missing_field_column(logical_region_id, field_name, field_meta, rows)?;
|
||||
// Sparse logical writes may omit nullable field columns. Fill them
|
||||
// before the rows are rewritten for the shared physical table.
|
||||
for (field_name, field_meta) in logical_fields {
|
||||
if !rows.schema.iter().any(|col| col.column_name == field_name) {
|
||||
Self::fill_missing_field_column(
|
||||
logical_region_id,
|
||||
field_name,
|
||||
field_meta,
|
||||
rows,
|
||||
)?;
|
||||
}
|
||||
}
|
||||
|
||||
for (field_name, field_meta) in physical_columns
|
||||
.iter()
|
||||
.filter(|(_, col)| col.semantic_type == SemanticType::Field)
|
||||
{
|
||||
if !rows.schema.iter().any(|col| col.column_name == *field_name) {
|
||||
Self::fill_missing_field_column(
|
||||
logical_region_id,
|
||||
field_name,
|
||||
field_meta,
|
||||
rows,
|
||||
)?;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -625,6 +675,8 @@ impl MetricEngineInner {
|
||||
field_meta: &ColumnMetadata,
|
||||
rows: &mut Rows,
|
||||
) -> Result<()> {
|
||||
// This is only for schema columns with a concrete default, usually NULL
|
||||
// for field columns from other logical tables sharing this physical table.
|
||||
ensure!(
|
||||
!field_meta.column_schema.is_default_impure(),
|
||||
UnexpectedRequestSnafu {
|
||||
@@ -715,6 +767,8 @@ mod tests {
|
||||
use common_error::ext::ErrorExt;
|
||||
use common_error::status_code::StatusCode;
|
||||
use common_function::utils::partition_expr_version;
|
||||
use common_query::native_histogram::NATIVE_HISTOGRAM_FIELD;
|
||||
use common_query::prelude::{greptime_timestamp, greptime_value};
|
||||
use common_recordbatch::RecordBatches;
|
||||
use datatypes::prelude::ConcreteDataType;
|
||||
use datatypes::schema::{ColumnDefaultConstraint, ColumnSchema};
|
||||
@@ -814,6 +868,13 @@ mod tests {
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
fn column_index(rows: &Rows, name: &str) -> usize {
|
||||
rows.schema
|
||||
.iter()
|
||||
.position(|col| col.column_name == name)
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
async fn run_batch_write_with_schema_variants(
|
||||
env: &TestEnv,
|
||||
physical_region_id: RegionId,
|
||||
@@ -929,6 +990,103 @@ mod tests {
|
||||
assert_eq!(batches.iter().map(|b| b.num_rows()).sum::<usize>(), 5);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_sparse_batch_aligns_mixed_field_order() {
|
||||
let primary_key = PbColumnSchema {
|
||||
column_name: PRIMARY_KEY_COLUMN_NAME.to_string(),
|
||||
datatype: ColumnDataType::Binary as i32,
|
||||
semantic_type: SemanticType::Tag as _,
|
||||
datatype_extension: None,
|
||||
options: None,
|
||||
};
|
||||
let timestamp = PbColumnSchema {
|
||||
column_name: greptime_timestamp().to_string(),
|
||||
datatype: ColumnDataType::TimestampMillisecond as i32,
|
||||
semantic_type: SemanticType::Timestamp as _,
|
||||
datatype_extension: None,
|
||||
options: None,
|
||||
};
|
||||
let value = PbColumnSchema {
|
||||
column_name: greptime_value().to_string(),
|
||||
datatype: ColumnDataType::Float64 as i32,
|
||||
semantic_type: SemanticType::Field as _,
|
||||
datatype_extension: None,
|
||||
options: None,
|
||||
};
|
||||
let histogram = PbColumnSchema {
|
||||
column_name: NATIVE_HISTOGRAM_FIELD.to_string(),
|
||||
datatype: ColumnDataType::Struct as i32,
|
||||
semantic_type: SemanticType::Field as _,
|
||||
datatype_extension: None,
|
||||
options: None,
|
||||
};
|
||||
|
||||
let sample_rows = Rows {
|
||||
schema: vec![
|
||||
primary_key.clone(),
|
||||
timestamp.clone(),
|
||||
value.clone(),
|
||||
histogram.clone(),
|
||||
],
|
||||
rows: vec![Row {
|
||||
values: vec![
|
||||
ValueData::BinaryValue(vec![1]).into(),
|
||||
ValueData::TimestampMillisecondValue(0).into(),
|
||||
ValueData::F64Value(1.0).into(),
|
||||
Value { value_data: None },
|
||||
],
|
||||
}],
|
||||
};
|
||||
let histogram_rows = Rows {
|
||||
schema: vec![primary_key, timestamp, histogram, value],
|
||||
rows: vec![Row {
|
||||
values: vec![
|
||||
ValueData::BinaryValue(vec![2]).into(),
|
||||
ValueData::TimestampMillisecondValue(0).into(),
|
||||
ValueData::StructValue(api::v1::StructValue { items: vec![] }).into(),
|
||||
Value { value_data: None },
|
||||
],
|
||||
}],
|
||||
};
|
||||
|
||||
let schema = MetricEngineInner::build_union_schema([
|
||||
sample_rows.schema.as_slice(),
|
||||
histogram_rows.schema.as_slice(),
|
||||
]);
|
||||
let merged_rows = MetricEngineInner::align_rows_to_schema(sample_rows, &schema)
|
||||
.into_iter()
|
||||
.chain(MetricEngineInner::align_rows_to_schema(
|
||||
histogram_rows,
|
||||
&schema,
|
||||
))
|
||||
.collect();
|
||||
let merged_request = Rows {
|
||||
schema,
|
||||
rows: merged_rows,
|
||||
};
|
||||
|
||||
let value_idx = column_index(&merged_request, greptime_value());
|
||||
let histogram_idx = column_index(&merged_request, NATIVE_HISTOGRAM_FIELD);
|
||||
assert!(matches!(
|
||||
merged_request.rows[0].values[value_idx].value_data,
|
||||
Some(ValueData::F64Value(_))
|
||||
));
|
||||
assert!(
|
||||
merged_request.rows[0].values[histogram_idx]
|
||||
.value_data
|
||||
.is_none()
|
||||
);
|
||||
assert!(
|
||||
merged_request.rows[1].values[value_idx]
|
||||
.value_data
|
||||
.is_none()
|
||||
);
|
||||
assert!(matches!(
|
||||
merged_request.rows[1].values[histogram_idx].value_data,
|
||||
Some(ValueData::StructValue(_))
|
||||
));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_write_logical_region() {
|
||||
let env = TestEnv::new().await;
|
||||
|
||||
@@ -17,7 +17,6 @@
|
||||
use std::collections::{HashMap, HashSet};
|
||||
|
||||
use api::v1::SemanticType;
|
||||
use common_telemetry::warn;
|
||||
use common_time::timestamp::TimeUnit;
|
||||
use snafu::OptionExt;
|
||||
use store_api::codec::PrimaryKeyEncoding;
|
||||
@@ -37,10 +36,6 @@ pub struct PhysicalRegionState {
|
||||
/// row batch. The time index is fixed at region creation and never
|
||||
/// changes, so this stays in sync with `physical_columns`.
|
||||
time_index_column_name: String,
|
||||
/// Name of the field column. Metric regions have exactly one field column
|
||||
/// verified at creation time, so the write path can validate completeness
|
||||
/// without consulting per-logical-region metadata.
|
||||
field_column_name: String,
|
||||
primary_key_encoding: PrimaryKeyEncoding,
|
||||
options: PhysicalRegionOptions,
|
||||
time_index_unit: TimeUnit,
|
||||
@@ -61,16 +56,10 @@ impl PhysicalRegionState {
|
||||
.find(|(_, meta)| meta.semantic_type == SemanticType::Timestamp)
|
||||
.map(|(name, _)| name.clone())
|
||||
.unwrap_or_default();
|
||||
let field_column_name = physical_columns
|
||||
.iter()
|
||||
.find(|(_, meta)| meta.semantic_type == SemanticType::Field)
|
||||
.map(|(name, _)| name.clone())
|
||||
.unwrap_or_default();
|
||||
Self {
|
||||
logical_regions: HashSet::new(),
|
||||
physical_columns,
|
||||
time_index_column_name,
|
||||
field_column_name,
|
||||
primary_key_encoding,
|
||||
options,
|
||||
time_index_unit,
|
||||
@@ -92,11 +81,6 @@ impl PhysicalRegionState {
|
||||
&self.time_index_column_name
|
||||
}
|
||||
|
||||
/// Returns the cached name of the field column.
|
||||
pub fn field_column_name(&self) -> &str {
|
||||
&self.field_column_name
|
||||
}
|
||||
|
||||
/// Returns a reference to the physical region options.
|
||||
pub fn options(&self) -> &PhysicalRegionOptions {
|
||||
&self.options
|
||||
@@ -160,12 +144,6 @@ impl MetricEngineState {
|
||||
SemanticType::Timestamp,
|
||||
"unexpected time index column {col} added to an existing physical region"
|
||||
);
|
||||
if meta.semantic_type == SemanticType::Field {
|
||||
warn!(
|
||||
"Unexpected field column {col} added to physical region {physical_region_id}; cached field column remains {}",
|
||||
state.field_column_name
|
||||
);
|
||||
}
|
||||
state.physical_columns.insert(col, meta);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -37,6 +37,9 @@ use common_meta::cache::TableFlownodeSetCacheRef;
|
||||
use common_meta::node_manager::{AffectedRows, NodeManagerRef};
|
||||
use common_meta::peer::Peer;
|
||||
use common_query::Output;
|
||||
use common_query::native_histogram::{
|
||||
NATIVE_HISTOGRAM_FIELD, is_native_histogram_value_schema, native_histogram_value_type,
|
||||
};
|
||||
use common_query::prelude::{greptime_timestamp, greptime_value};
|
||||
use common_telemetry::tracing_context::TracingContext;
|
||||
use common_telemetry::{error, info, warn};
|
||||
@@ -907,6 +910,10 @@ impl Inserter {
|
||||
let table_name = table.table_info().name.clone();
|
||||
|
||||
let request_schema = req.rows.as_ref().unwrap().schema.as_slice();
|
||||
let request_field_count = request_schema
|
||||
.iter()
|
||||
.filter(|col| col.semantic_type == SemanticType::Field as i32)
|
||||
.count();
|
||||
let column_exprs = ColumnExpr::from_column_schemas(request_schema);
|
||||
let add_columns = expr_helper::extract_add_columns_expr(&table.schema(), column_exprs)?;
|
||||
let Some(mut add_columns) = add_columns else {
|
||||
@@ -915,12 +922,22 @@ impl Inserter {
|
||||
|
||||
// If accommodate_existing_schema is true, update request schema for Timestamp/Field columns
|
||||
if accommodate_existing_schema {
|
||||
let request_is_native_histogram = request_is_native_histogram(request_schema);
|
||||
let table_is_native_histogram = table_is_native_histogram(table);
|
||||
ensure!(
|
||||
request_is_native_histogram == table_is_native_histogram,
|
||||
InvalidInsertRequestSnafu {
|
||||
reason: format!(
|
||||
"Table `{table_name}` cannot mix native histogram and float sample fields"
|
||||
),
|
||||
}
|
||||
);
|
||||
let table_schema = table.schema();
|
||||
// Find timestamp column name
|
||||
let ts_col_name = table_schema.timestamp_column().map(|c| c.name.clone());
|
||||
// Find field column name if there is only one and `is_single_value` is true.
|
||||
let mut field_col_name = None;
|
||||
if is_single_value {
|
||||
if is_single_value && request_field_count <= 1 {
|
||||
let mut multiple_field_cols = false;
|
||||
table.field_columns().for_each(|col| {
|
||||
if field_col_name.is_none() {
|
||||
@@ -1063,6 +1080,32 @@ impl Inserter {
|
||||
}
|
||||
}
|
||||
|
||||
fn request_is_native_histogram(request_schema: &[ColumnSchema]) -> bool {
|
||||
let mut fields = request_schema
|
||||
.iter()
|
||||
.filter(|col| col.semantic_type == SemanticType::Field as i32);
|
||||
let Some(col) = fields.next() else {
|
||||
return false;
|
||||
};
|
||||
|
||||
fields.next().is_none()
|
||||
&& col.column_name == NATIVE_HISTOGRAM_FIELD
|
||||
&& api::helper::is_column_type_value_eq(
|
||||
col.datatype,
|
||||
col.datatype_extension.clone(),
|
||||
native_histogram_value_type(),
|
||||
)
|
||||
}
|
||||
|
||||
fn table_is_native_histogram(table: &TableRef) -> bool {
|
||||
let mut fields = table.field_columns();
|
||||
let Some(col) = fields.next() else {
|
||||
return false;
|
||||
};
|
||||
|
||||
fields.next().is_none() && is_native_histogram_value_schema(&col.name, &col.data_type)
|
||||
}
|
||||
|
||||
fn validate_column_count_match(requests: &RowInsertRequests) -> Result<()> {
|
||||
for request in &requests.inserts {
|
||||
let rows = request.rows.as_ref().unwrap();
|
||||
|
||||
@@ -163,6 +163,9 @@ pub struct HttpOptions {
|
||||
/// Validation mode while decoding Prometheus remote write requests.
|
||||
pub prom_validation_mode: PromValidationMode,
|
||||
|
||||
/// Enables experimental Prometheus remote write v2 native histogram ingestion.
|
||||
pub experimental_enable_prometheus_native_histogram: bool,
|
||||
|
||||
pub cors_allowed_origins: Vec<String>,
|
||||
|
||||
pub enable_cors: bool,
|
||||
@@ -180,6 +183,7 @@ impl Default for HttpOptions {
|
||||
cors_allowed_origins: Vec::new(),
|
||||
enable_cors: true,
|
||||
prom_validation_mode: PromValidationMode::Strict,
|
||||
experimental_enable_prometheus_native_histogram: false,
|
||||
experimental_enable_explain_analyze_stream: false,
|
||||
}
|
||||
}
|
||||
@@ -601,6 +605,9 @@ impl HttpServerBuilder {
|
||||
pipeline_handler,
|
||||
prom_store_with_metric_engine,
|
||||
prom_validation_mode,
|
||||
experimental_enable_prometheus_native_histogram: self
|
||||
.options
|
||||
.experimental_enable_prometheus_native_histogram,
|
||||
pending_rows_batcher,
|
||||
};
|
||||
|
||||
|
||||
@@ -45,7 +45,7 @@ use crate::http::header::{
|
||||
use crate::pending_rows_batcher::PendingRowsBatcher;
|
||||
use crate::prom_remote_write::decode::PromSeriesProcessor;
|
||||
use crate::prom_remote_write::decode_remote_write_request;
|
||||
use crate::prom_remote_write::v2::{RemoteWriteV2RequestExt, decode_remote_write_v2_request};
|
||||
use crate::prom_remote_write::v2::{decode_remote_write_v2_request, into_write_requests};
|
||||
use crate::prom_remote_write::validation::PromValidationMode;
|
||||
use crate::prom_store::{extract_schema_from_read_request, snappy_decompress};
|
||||
use crate::query_handler::{PipelineHandlerRef, PromStoreProtocolHandlerRef, PromStoreResponse};
|
||||
@@ -71,6 +71,7 @@ pub struct PromStoreState {
|
||||
pub pipeline_handler: Option<PipelineHandlerRef>,
|
||||
pub prom_store_with_metric_engine: bool,
|
||||
pub prom_validation_mode: PromValidationMode,
|
||||
pub experimental_enable_prometheus_native_histogram: bool,
|
||||
pub pending_rows_batcher: Option<Arc<PendingRowsBatcher>>,
|
||||
}
|
||||
|
||||
@@ -142,6 +143,7 @@ async fn remote_write_v1(
|
||||
pipeline_handler,
|
||||
prom_store_with_metric_engine,
|
||||
prom_validation_mode,
|
||||
experimental_enable_prometheus_native_histogram: _,
|
||||
pending_rows_batcher,
|
||||
} = state;
|
||||
|
||||
@@ -177,15 +179,22 @@ async fn remote_write_v1(
|
||||
req.as_insert_requests()
|
||||
};
|
||||
|
||||
let outcome = write_prometheus_rows(
|
||||
let outcome = match write_prometheus_rows_with_progress(
|
||||
prom_store_handler,
|
||||
pending_rows_batcher,
|
||||
prom_store_with_metric_engine,
|
||||
&db,
|
||||
query_ctx,
|
||||
req,
|
||||
)
|
||||
.await?;
|
||||
.await
|
||||
{
|
||||
Ok(outcome) => outcome,
|
||||
Err(error) => {
|
||||
record_remote_write_samples(&db, error.rows_written);
|
||||
return Err(error.error);
|
||||
}
|
||||
};
|
||||
record_remote_write_samples(&db, outcome.rows_written);
|
||||
|
||||
Ok((
|
||||
StatusCode::NO_CONTENT,
|
||||
@@ -207,6 +216,7 @@ async fn remote_write_v2(
|
||||
pipeline_handler: _,
|
||||
prom_store_with_metric_engine,
|
||||
prom_validation_mode: _,
|
||||
experimental_enable_prometheus_native_histogram,
|
||||
pending_rows_batcher,
|
||||
} = state;
|
||||
|
||||
@@ -225,38 +235,96 @@ async fn remote_write_v2(
|
||||
Ok(request) => request,
|
||||
Err(error) => return Ok(remote_write_v2_error_response(error, 0, 0, 0)),
|
||||
};
|
||||
let req = match request.into_context_req() {
|
||||
if !experimental_enable_prometheus_native_histogram && request_has_native_histograms(&request) {
|
||||
return Ok(remote_write_v2_error_response(
|
||||
error::InvalidPromRemoteRequestSnafu {
|
||||
msg: "prometheus remote write v2 native histogram ingestion is experimental; set http.experimental_enable_prometheus_native_histogram = true to enable it"
|
||||
.to_string(),
|
||||
}
|
||||
.build(),
|
||||
0,
|
||||
0,
|
||||
0,
|
||||
));
|
||||
}
|
||||
let req = match into_write_requests(request) {
|
||||
Ok(req) => req,
|
||||
Err(error) => return Ok(remote_write_v2_error_response(error, 0, 0, 0)),
|
||||
};
|
||||
|
||||
let outcome = match write_prometheus_rows_with_progress(
|
||||
prom_store_handler,
|
||||
pending_rows_batcher,
|
||||
prom_store_with_metric_engine,
|
||||
&db,
|
||||
query_ctx,
|
||||
req,
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(outcome) => outcome,
|
||||
Err(error) => {
|
||||
return Ok(remote_write_v2_error_response(
|
||||
error.error,
|
||||
error.rows_written,
|
||||
0,
|
||||
0,
|
||||
));
|
||||
let outcome = if req.sample_count > 0 {
|
||||
match write_prometheus_rows_with_progress(
|
||||
prom_store_handler.clone(),
|
||||
pending_rows_batcher.clone(),
|
||||
prom_store_with_metric_engine,
|
||||
query_ctx.clone(),
|
||||
req.samples,
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(outcome) => outcome,
|
||||
Err(error) => {
|
||||
record_remote_write_samples(&db, error.rows_written);
|
||||
return Ok(remote_write_v2_error_response(
|
||||
error.error,
|
||||
error.rows_written,
|
||||
0,
|
||||
0,
|
||||
));
|
||||
}
|
||||
}
|
||||
} else {
|
||||
PromWriteOutcome {
|
||||
write_cost: 0,
|
||||
rows_written: 0,
|
||||
}
|
||||
};
|
||||
let samples_written = outcome.rows_written;
|
||||
record_remote_write_samples(&db, samples_written);
|
||||
let mut histograms_written = 0;
|
||||
let mut write_cost = outcome.write_cost;
|
||||
|
||||
let mut headers = write_cost_header_map(outcome.write_cost);
|
||||
append_remote_write_v2_written_headers(&mut headers, outcome.rows_written, 0, 0);
|
||||
if req.histogram_count > 0 {
|
||||
let outcome = match write_prometheus_rows_with_progress(
|
||||
prom_store_handler,
|
||||
None,
|
||||
prom_store_with_metric_engine,
|
||||
query_ctx,
|
||||
req.histograms,
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(outcome) => outcome,
|
||||
Err(error) => {
|
||||
record_remote_write_histograms(&db, error.rows_written);
|
||||
return Ok(remote_write_v2_error_response(
|
||||
error.error,
|
||||
samples_written,
|
||||
error.rows_written,
|
||||
0,
|
||||
));
|
||||
}
|
||||
};
|
||||
histograms_written = outcome.rows_written;
|
||||
record_remote_write_histograms(&db, histograms_written);
|
||||
write_cost += outcome.write_cost;
|
||||
}
|
||||
|
||||
let mut headers = write_cost_header_map(write_cost);
|
||||
append_remote_write_v2_written_headers(&mut headers, samples_written, histograms_written, 0);
|
||||
|
||||
Ok((StatusCode::NO_CONTENT, headers).into_response())
|
||||
}
|
||||
|
||||
fn request_has_native_histograms(
|
||||
request: &api::greptime_proto::io::prometheus::write::v2::Request,
|
||||
) -> bool {
|
||||
request
|
||||
.timeseries
|
||||
.iter()
|
||||
.any(|series| !series.histograms.is_empty())
|
||||
}
|
||||
|
||||
fn vm_proto_version_response(params: &RemoteWriteQuery) -> Option<axum::response::Response> {
|
||||
params
|
||||
.get_vm_proto_version
|
||||
@@ -302,31 +370,14 @@ struct PromWriteError {
|
||||
rows_written: u64,
|
||||
}
|
||||
|
||||
async fn write_prometheus_rows(
|
||||
prom_store_handler: PromStoreProtocolHandlerRef,
|
||||
pending_rows_batcher: Option<Arc<PendingRowsBatcher>>,
|
||||
prom_store_with_metric_engine: bool,
|
||||
db: &str,
|
||||
query_ctx: Arc<QueryContext>,
|
||||
req: ContextReq,
|
||||
) -> Result<PromWriteOutcome> {
|
||||
write_prometheus_rows_with_progress(
|
||||
prom_store_handler,
|
||||
pending_rows_batcher,
|
||||
prom_store_with_metric_engine,
|
||||
db,
|
||||
query_ctx,
|
||||
req,
|
||||
)
|
||||
.await
|
||||
.map_err(|error| error.error)
|
||||
}
|
||||
|
||||
/// Writes one decoded PRW batch and keeps the number of persisted rows on error.
|
||||
///
|
||||
/// The v2 handler uses that partial progress to return Prometheus' written
|
||||
/// sample/histogram headers even when a later table write fails.
|
||||
async fn write_prometheus_rows_with_progress(
|
||||
prom_store_handler: PromStoreProtocolHandlerRef,
|
||||
pending_rows_batcher: Option<Arc<PendingRowsBatcher>>,
|
||||
prom_store_with_metric_engine: bool,
|
||||
db: &str,
|
||||
query_ctx: Arc<QueryContext>,
|
||||
req: ContextReq,
|
||||
) -> std::result::Result<PromWriteOutcome, PromWriteError> {
|
||||
@@ -347,9 +398,6 @@ async fn write_prometheus_rows_with_progress(
|
||||
error,
|
||||
rows_written,
|
||||
})?;
|
||||
crate::metrics::PROM_STORE_REMOTE_WRITE_SAMPLES
|
||||
.with_label_values(&[db])
|
||||
.inc_by(rows);
|
||||
rows_written += rows;
|
||||
}
|
||||
return Ok(PromWriteOutcome {
|
||||
@@ -373,9 +421,6 @@ async fn write_prometheus_rows_with_progress(
|
||||
error,
|
||||
rows_written,
|
||||
})?;
|
||||
crate::metrics::PROM_STORE_REMOTE_WRITE_SAMPLES
|
||||
.with_label_values(&[db])
|
||||
.inc_by(cnt);
|
||||
write_cost += output.meta.cost;
|
||||
rows_written += cnt;
|
||||
}
|
||||
@@ -386,6 +431,24 @@ async fn write_prometheus_rows_with_progress(
|
||||
})
|
||||
}
|
||||
|
||||
fn record_remote_write_samples(db: &str, rows: u64) {
|
||||
if rows == 0 {
|
||||
return;
|
||||
}
|
||||
crate::metrics::PROM_STORE_REMOTE_WRITE_SAMPLES
|
||||
.with_label_values(&[db])
|
||||
.inc_by(rows);
|
||||
}
|
||||
|
||||
fn record_remote_write_histograms(db: &str, rows: u64) {
|
||||
if rows == 0 {
|
||||
return;
|
||||
}
|
||||
crate::metrics::PROM_STORE_REMOTE_WRITE_HISTOGRAMS
|
||||
.with_label_values(&[db])
|
||||
.inc_by(rows);
|
||||
}
|
||||
|
||||
fn remote_write_v2_error_response(
|
||||
error: error::Error,
|
||||
samples: u64,
|
||||
@@ -647,6 +710,7 @@ mod tests {
|
||||
pipeline_handler: None,
|
||||
prom_store_with_metric_engine: false,
|
||||
prom_validation_mode: PromValidationMode::Strict,
|
||||
experimental_enable_prometheus_native_histogram: false,
|
||||
pending_rows_batcher: None,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -128,6 +128,13 @@ lazy_static! {
|
||||
&[METRIC_DB_LABEL]
|
||||
)
|
||||
.unwrap();
|
||||
/// The native histograms count of Prometheus remote write.
|
||||
pub static ref PROM_STORE_REMOTE_WRITE_HISTOGRAMS: IntCounterVec = register_int_counter_vec!(
|
||||
"greptime_servers_prometheus_remote_write_histograms",
|
||||
"frontend prometheus remote write native histograms",
|
||||
&[METRIC_DB_LABEL]
|
||||
)
|
||||
.unwrap();
|
||||
pub static ref PENDING_BATCHES: IntGauge = register_int_gauge!(
|
||||
"greptime_prom_store_pending_batches",
|
||||
"Number of pending batches waiting to be flushed"
|
||||
|
||||
@@ -303,7 +303,7 @@ pub(crate) fn write_attributes(
|
||||
mod tests {
|
||||
use api::v1::value::ValueData;
|
||||
use opentelemetry_proto::tonic::common::v1::any_value::Value as OtlpValue;
|
||||
use opentelemetry_proto::tonic::common::v1::{AnyValue, KeyValue};
|
||||
use opentelemetry_proto::tonic::common::v1::{AnyValue, ArrayValue, KeyValue};
|
||||
|
||||
use super::*;
|
||||
use crate::otlp::trace::TraceAuxData;
|
||||
@@ -533,6 +533,42 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_keep_mixed_binary_and_json_values_until_frontend_reconciliation() {
|
||||
let mut writer = TableData::new(4, 2);
|
||||
|
||||
let attrs1 = Attributes::from(vec![make_kv(
|
||||
"val",
|
||||
OtlpValue::BytesValue(vec![1_u8, 2, 3]),
|
||||
)]);
|
||||
let mut row1 = writer.alloc_one_row();
|
||||
write_attributes(&mut writer, "attr", attrs1, &mut row1).unwrap();
|
||||
writer.add_row(row1);
|
||||
|
||||
let attrs2 = Attributes::from(vec![make_kv(
|
||||
"val",
|
||||
OtlpValue::ArrayValue(ArrayValue {
|
||||
values: vec![AnyValue {
|
||||
value: Some(OtlpValue::IntValue(1)),
|
||||
}],
|
||||
}),
|
||||
)]);
|
||||
let mut row2 = writer.alloc_one_row();
|
||||
write_attributes(&mut writer, "attr", attrs2, &mut row2).unwrap();
|
||||
writer.add_row(row2);
|
||||
|
||||
let (schema, rows) = writer.into_schema_and_rows();
|
||||
let col_idx = schema
|
||||
.iter()
|
||||
.position(|c| c.column_name == "attr.val")
|
||||
.unwrap();
|
||||
assert_eq!(schema[col_idx].datatype, ColumnDataType::Binary as i32);
|
||||
assert!(matches!(
|
||||
rows[1].values[col_idx].value_data.as_ref(),
|
||||
Some(ValueData::BinaryValue(_))
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_build_aux_table_requests_deduplicates_services_and_operations() {
|
||||
let spans = vec![
|
||||
|
||||
@@ -0,0 +1,54 @@
|
||||
# Prometheus Remote Write
|
||||
|
||||
This module decodes Prometheus remote write requests and converts them into
|
||||
Greptime row insert requests.
|
||||
|
||||
## Remote Write V2
|
||||
|
||||
Remote write v2 enters through `remote_write_v2` in
|
||||
`src/servers/src/http/prom_store.rs`.
|
||||
|
||||
```mermaid
|
||||
flowchart TD
|
||||
A["HTTP /v1/prometheus/write"] --> B["remote_write_v2"]
|
||||
B --> C["decode_remote_write_v2_request"]
|
||||
C --> D["into_write_requests"]
|
||||
|
||||
D --> E["samples ContextReq"]
|
||||
D --> F["histograms ContextReq"]
|
||||
|
||||
E --> G["write_prometheus_rows_with_progress"]
|
||||
G --> H["metric engine / pending batcher when enabled"]
|
||||
|
||||
F --> I["histogram ContextOpt"]
|
||||
I --> J["write_prometheus_rows_with_progress"]
|
||||
J --> K["same metric-engine flag as samples, no batcher"]
|
||||
K --> L["table: <metric>"]
|
||||
|
||||
L --> M["field: greptime_native_histogram Struct"]
|
||||
M --> N["struct children: counts, spans, buckets, sum, schema"]
|
||||
H --> P["written headers and counters"]
|
||||
K --> P
|
||||
```
|
||||
|
||||
The conversion step splits one v2 request into two `ContextReq`s:
|
||||
|
||||
- samples keep the existing sample table name and can use the metric-engine
|
||||
physical table path and pending rows batcher;
|
||||
- native histograms keep the existing metric table name.
|
||||
|
||||
Native histogram rows follow the same metric-engine switch as samples. They do
|
||||
not use the pending rows batcher yet because the batcher assumes the classic
|
||||
timestamp + Float64 value + string tags shape.
|
||||
|
||||
Each histogram row stores `greptime_native_histogram` as one Struct field:
|
||||
|
||||
- common scalar children: `schema`, `zero_threshold`, `sum`, `reset_hint`,
|
||||
`start_timestamp`;
|
||||
- count children: `count_u64` / `zero_count_u64` or `count_f64` / `zero_count_f64`;
|
||||
- list children for custom values, spans, and positive/negative buckets;
|
||||
- original Prometheus labels as Greptime tags.
|
||||
|
||||
The v2 response always reports written sample, histogram, and exemplar counts in
|
||||
Prometheus remote-write headers. Exemplars are currently ignored and reported as
|
||||
zero.
|
||||
@@ -13,14 +13,18 @@
|
||||
// limitations under the License.
|
||||
|
||||
use std::collections::hash_map::Entry;
|
||||
use std::collections::{HashMap, HashSet};
|
||||
|
||||
use ahash::{HashMap, HashMapExt, HashSet, HashSetExt};
|
||||
use api::greptime_proto::io::prometheus::write::v2::histogram::{Count, ZeroCount};
|
||||
#[cfg(test)]
|
||||
use api::greptime_proto::io::prometheus::write::v2::{Exemplar, Histogram, Metadata, metadata};
|
||||
use api::greptime_proto::io::prometheus::write::v2::{Request, Sample, TimeSeries};
|
||||
use api::v1::{RowInsertRequest, Rows, Value};
|
||||
use api::greptime_proto::io::prometheus::write::v2::{Exemplar, Metadata, metadata};
|
||||
use api::greptime_proto::io::prometheus::write::v2::{Histogram, Request, Sample, TimeSeries};
|
||||
use api::helper::ColumnDataTypeWrapper;
|
||||
use api::v1::value::ValueData;
|
||||
use api::v1::{ColumnSchema, ListValue, RowInsertRequest, Rows, SemanticType, Value};
|
||||
use bytes::Bytes;
|
||||
use common_grpc::precision::Precision;
|
||||
use common_query::native_histogram::*;
|
||||
use common_query::prelude::{greptime_timestamp, greptime_value};
|
||||
use pipeline::{ContextOpt, ContextReq};
|
||||
use prost::Message;
|
||||
@@ -29,6 +33,7 @@ use snafu::{OptionExt, ResultExt, ensure};
|
||||
use crate::error::{self, Result};
|
||||
use crate::prom_remote_write::row_builder::PromCtx;
|
||||
use crate::prom_remote_write::try_decompress;
|
||||
use crate::prom_remote_write::validation::validate_label_name;
|
||||
#[allow(deprecated)]
|
||||
use crate::prom_store::{
|
||||
DATABASE_LABEL, DATABASE_LABEL_ALT, METRIC_NAME_LABEL, PHYSICAL_TABLE_LABEL,
|
||||
@@ -53,49 +58,132 @@ pub(crate) fn decode_remote_write_v2_request(is_zstd: bool, body: Bytes) -> Resu
|
||||
Request::decode(&buf[..]).context(error::DecodePromRemoteRequestSnafu)
|
||||
}
|
||||
|
||||
pub(crate) trait RemoteWriteV2RequestExt {
|
||||
fn into_context_req(self) -> Result<ContextReq>;
|
||||
pub(crate) struct RemoteWriteV2WriteRequests {
|
||||
pub samples: ContextReq,
|
||||
pub histograms: ContextReq,
|
||||
pub sample_count: u64,
|
||||
pub histogram_count: u64,
|
||||
}
|
||||
|
||||
impl RemoteWriteV2RequestExt for Request {
|
||||
fn into_context_req(self) -> Result<ContextReq> {
|
||||
let _timer = crate::metrics::METRIC_HTTP_PROM_STORE_CONVERT_ELAPSED.start_timer();
|
||||
let Request {
|
||||
symbols,
|
||||
timeseries,
|
||||
} = self;
|
||||
/// Converts a PRW v2 request into normal sample writes and native histogram writes.
|
||||
///
|
||||
/// A metric name may appear in only one payload kind per request because the
|
||||
/// metric-engine logical table is either a float metric or a native histogram.
|
||||
pub(crate) fn into_write_requests(request: Request) -> Result<RemoteWriteV2WriteRequests> {
|
||||
let _timer = crate::metrics::METRIC_HTTP_PROM_STORE_CONVERT_ELAPSED.start_timer();
|
||||
let Request {
|
||||
symbols,
|
||||
timeseries,
|
||||
} = request;
|
||||
|
||||
ensure!(
|
||||
symbols.first().map(|s| s.as_str()) == Some(""),
|
||||
error::InvalidPromRemoteRequestSnafu {
|
||||
msg: "remote write v2 symbols must start with an empty string".to_string(),
|
||||
}
|
||||
);
|
||||
ensure!(
|
||||
symbols.first().map(|s| s.as_str()) == Some(""),
|
||||
error::InvalidPromRemoteRequestSnafu {
|
||||
msg: "remote write v2 symbols must start with an empty string".to_string(),
|
||||
}
|
||||
);
|
||||
|
||||
let mut tables = HashMap::<PromCtx, HashMap<String, TableData>>::new();
|
||||
let mut sample_tables = HashMap::<PromCtx, HashMap<String, TableData>>::new();
|
||||
let mut histogram_tables = HashMap::<PromCtx, HashMap<String, TableData>>::new();
|
||||
let mut sample_metrics = HashSet::<(PromCtx, String)>::new();
|
||||
let mut histogram_metrics = HashSet::<(PromCtx, String)>::new();
|
||||
let mut sample_count_total = 0;
|
||||
let mut histogram_count_total = 0;
|
||||
|
||||
for series in timeseries {
|
||||
// Native histograms and exemplars are intentionally ignored for now.
|
||||
// They will be converted into Greptime rows once their ingestion path is implemented.
|
||||
let sample_count = series.samples.len();
|
||||
if sample_count == 0 {
|
||||
continue;
|
||||
}
|
||||
|
||||
let (prom_ctx, table_name, tags) = resolve_series_labels(&symbols, &series)?;
|
||||
let table_data = match tables.entry(prom_ctx).or_default().entry(table_name) {
|
||||
Entry::Occupied(entry) => {
|
||||
let table_data = entry.into_mut();
|
||||
table_data.reserve_rows(sample_count);
|
||||
table_data
|
||||
}
|
||||
Entry::Vacant(entry) => entry.insert(TableData::new(tags.len() + 2, sample_count)),
|
||||
};
|
||||
|
||||
write_samples(table_data, series.samples, tags)?;
|
||||
for series in timeseries {
|
||||
// Exemplars are intentionally ignored for now.
|
||||
let sample_count = series.samples.len();
|
||||
let histogram_count = series.histograms.len();
|
||||
if sample_count == 0 && histogram_count == 0 {
|
||||
continue;
|
||||
}
|
||||
|
||||
Ok(into_context_req(tables))
|
||||
let (prom_ctx, table_name, tags) = resolve_series_labels(&symbols, &series)?;
|
||||
ensure_no_internal_histogram_labels(&tags)?;
|
||||
let metric_key = (prom_ctx.clone(), table_name.clone());
|
||||
if sample_count > 0 {
|
||||
ensure!(
|
||||
!histogram_metrics.contains(&metric_key),
|
||||
error::InvalidPromRemoteRequestSnafu {
|
||||
msg: format!(
|
||||
"remote write v2 metric `{table_name}` contains both samples and native histograms"
|
||||
),
|
||||
}
|
||||
);
|
||||
sample_metrics.insert(metric_key.clone());
|
||||
}
|
||||
if histogram_count > 0 {
|
||||
ensure!(
|
||||
!sample_metrics.contains(&metric_key),
|
||||
error::InvalidPromRemoteRequestSnafu {
|
||||
msg: format!(
|
||||
"remote write v2 metric `{table_name}` contains both samples and native histograms"
|
||||
),
|
||||
}
|
||||
);
|
||||
histogram_metrics.insert(metric_key);
|
||||
}
|
||||
|
||||
if sample_count > 0 && histogram_count == 0 {
|
||||
// Fast path for regular sample-only series. Move the resolved labels into
|
||||
// the sample writer instead of cloning them for a histogram path we won't use.
|
||||
let table_data = get_or_create_table_data(
|
||||
&mut sample_tables,
|
||||
prom_ctx,
|
||||
table_name,
|
||||
tags.len() + 2,
|
||||
sample_count,
|
||||
);
|
||||
|
||||
write_samples(table_data, series.samples, tags)?;
|
||||
sample_count_total += sample_count as u64;
|
||||
// The owned labels were moved above, so skip the mixed-series path below.
|
||||
continue;
|
||||
}
|
||||
|
||||
if histogram_count > 0 {
|
||||
let table_data = get_or_create_table_data(
|
||||
&mut histogram_tables,
|
||||
prom_ctx,
|
||||
table_name,
|
||||
tags.len() + 2,
|
||||
histogram_count,
|
||||
);
|
||||
|
||||
let mut histograms = series.histograms;
|
||||
let Some(last_histogram) = histograms.pop() else {
|
||||
continue;
|
||||
};
|
||||
for histogram in &histograms {
|
||||
write_native_histogram(table_data, histogram, tags.iter().cloned())?;
|
||||
}
|
||||
write_native_histogram(table_data, &last_histogram, tags.into_iter())?;
|
||||
histogram_count_total += histogram_count as u64;
|
||||
}
|
||||
}
|
||||
|
||||
Ok(RemoteWriteV2WriteRequests {
|
||||
samples: into_context_req(sample_tables),
|
||||
histograms: into_context_req(histogram_tables),
|
||||
sample_count: sample_count_total,
|
||||
histogram_count: histogram_count_total,
|
||||
})
|
||||
}
|
||||
|
||||
fn get_or_create_table_data(
|
||||
tables: &mut HashMap<PromCtx, HashMap<String, TableData>>,
|
||||
prom_ctx: PromCtx,
|
||||
table_name: String,
|
||||
column_count: usize,
|
||||
row_count: usize,
|
||||
) -> &mut TableData {
|
||||
match tables.entry(prom_ctx).or_default().entry(table_name) {
|
||||
Entry::Occupied(entry) => {
|
||||
let table_data = entry.into_mut();
|
||||
table_data.reserve_rows(row_count);
|
||||
table_data
|
||||
}
|
||||
Entry::Vacant(entry) => entry.insert(TableData::new(column_count, row_count)),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -135,6 +223,195 @@ fn write_sample(
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn write_native_histogram(
|
||||
table_data: &mut TableData,
|
||||
histogram: &Histogram,
|
||||
tags: impl Iterator<Item = (String, String)>,
|
||||
) -> Result<()> {
|
||||
// Persist both int and float families into the logical table schema. Only one
|
||||
// family is populated per row; the other is written as NULL so PromQL can
|
||||
// infer the original histogram flavor without a separate type column.
|
||||
let mut row = table_data.alloc_one_row();
|
||||
row_writer::write_ts_to_millis(
|
||||
table_data,
|
||||
greptime_timestamp(),
|
||||
Some(histogram.timestamp),
|
||||
Precision::Millisecond,
|
||||
&mut row,
|
||||
)?;
|
||||
|
||||
write_native_histogram_value(table_data, histogram, &mut row)?;
|
||||
|
||||
row_writer::write_tags(table_data, tags, &mut row)?;
|
||||
table_data.add_row(row);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn write_native_histogram_value(
|
||||
table_data: &mut TableData,
|
||||
histogram: &Histogram,
|
||||
row: &mut Vec<Value>,
|
||||
) -> Result<()> {
|
||||
let column_schema = native_histogram_column_schema();
|
||||
let value = native_histogram_struct_value(histogram)?;
|
||||
|
||||
row_writer::write_by_schema(
|
||||
table_data,
|
||||
std::iter::once((column_schema, Some(value))),
|
||||
row,
|
||||
)
|
||||
}
|
||||
|
||||
fn native_histogram_column_schema() -> ColumnSchema {
|
||||
let (datatype, datatype_extension) =
|
||||
ColumnDataTypeWrapper::try_from(native_histogram_value_type().clone())
|
||||
.expect("native histogram type is convertible to protobuf")
|
||||
.into_parts();
|
||||
|
||||
ColumnSchema {
|
||||
column_name: NATIVE_HISTOGRAM_FIELD.to_string(),
|
||||
datatype: datatype as i32,
|
||||
semantic_type: SemanticType::Field as i32,
|
||||
datatype_extension,
|
||||
options: None,
|
||||
}
|
||||
}
|
||||
|
||||
fn native_histogram_struct_value(histogram: &Histogram) -> Result<ValueData> {
|
||||
let mut items = Vec::with_capacity(NATIVE_HISTOGRAM_FIELD_NAMES.len());
|
||||
items.extend([
|
||||
pb_value(ValueData::I32Value(histogram.schema)),
|
||||
pb_value(ValueData::F64Value(histogram.zero_threshold)),
|
||||
pb_value(ValueData::F64Value(histogram.sum)),
|
||||
pb_value(ValueData::I32Value(histogram.reset_hint)),
|
||||
optional_pb_value((histogram.start_timestamp != 0).then_some(
|
||||
ValueData::TimestampMillisecondValue(histogram.start_timestamp),
|
||||
)),
|
||||
f64_list_value(histogram.custom_values.iter().copied()),
|
||||
i32_list_value(histogram.positive_spans.iter().map(|span| span.offset)),
|
||||
u32_list_value(histogram.positive_spans.iter().map(|span| span.length)),
|
||||
i32_list_value(histogram.negative_spans.iter().map(|span| span.offset)),
|
||||
u32_list_value(histogram.negative_spans.iter().map(|span| span.length)),
|
||||
]);
|
||||
|
||||
let int_counts = match (&histogram.count, &histogram.zero_count) {
|
||||
(Some(Count::CountInt(count)), Some(ZeroCount::ZeroCountInt(zero_count))) => {
|
||||
(*count, *zero_count)
|
||||
}
|
||||
(Some(Count::CountInt(count)), _) => (*count, 0),
|
||||
(_, Some(ZeroCount::ZeroCountInt(zero_count))) => (0, *zero_count),
|
||||
_ => (0, 0),
|
||||
};
|
||||
let float_counts = match (&histogram.count, &histogram.zero_count) {
|
||||
(Some(Count::CountFloat(count)), Some(ZeroCount::ZeroCountFloat(zero_count))) => {
|
||||
Some((*count, *zero_count))
|
||||
}
|
||||
(Some(Count::CountFloat(count)), _) => Some((*count, 0.0)),
|
||||
_ => None,
|
||||
};
|
||||
|
||||
if let Some(counts) = float_counts {
|
||||
items.extend([
|
||||
null_pb_value(),
|
||||
null_pb_value(),
|
||||
i64_list_value(std::iter::empty()),
|
||||
i64_list_value(std::iter::empty()),
|
||||
pb_value(ValueData::F64Value(counts.0)),
|
||||
pb_value(ValueData::F64Value(counts.1)),
|
||||
f64_list_value(histogram.positive_counts.iter().copied()),
|
||||
f64_list_value(histogram.negative_counts.iter().copied()),
|
||||
]);
|
||||
} else {
|
||||
let positive_buckets = bucket_counts_from_deltas(&histogram.positive_deltas)?;
|
||||
let negative_buckets = bucket_counts_from_deltas(&histogram.negative_deltas)?;
|
||||
items.extend([
|
||||
pb_value(ValueData::U64Value(int_counts.0)),
|
||||
pb_value(ValueData::U64Value(int_counts.1)),
|
||||
i64_list_value(positive_buckets.iter().copied()),
|
||||
i64_list_value(negative_buckets.iter().copied()),
|
||||
null_pb_value(),
|
||||
null_pb_value(),
|
||||
f64_list_value(std::iter::empty()),
|
||||
f64_list_value(std::iter::empty()),
|
||||
]);
|
||||
}
|
||||
|
||||
Ok(ValueData::StructValue(api::v1::StructValue { items }))
|
||||
}
|
||||
|
||||
fn pb_value(value_data: ValueData) -> Value {
|
||||
optional_pb_value(Some(value_data))
|
||||
}
|
||||
|
||||
fn null_pb_value() -> Value {
|
||||
optional_pb_value(None)
|
||||
}
|
||||
|
||||
fn optional_pb_value(value_data: Option<ValueData>) -> Value {
|
||||
Value { value_data }
|
||||
}
|
||||
|
||||
fn list_value(values: impl IntoIterator<Item = ValueData>) -> Value {
|
||||
pb_value(ValueData::ListValue(ListValue {
|
||||
items: values.into_iter().map(pb_value).collect(),
|
||||
}))
|
||||
}
|
||||
|
||||
fn i32_list_value(values: impl IntoIterator<Item = i32>) -> Value {
|
||||
list_value(values.into_iter().map(ValueData::I32Value))
|
||||
}
|
||||
|
||||
fn u32_list_value(values: impl IntoIterator<Item = u32>) -> Value {
|
||||
list_value(values.into_iter().map(ValueData::U32Value))
|
||||
}
|
||||
|
||||
fn i64_list_value(values: impl IntoIterator<Item = i64>) -> Value {
|
||||
list_value(values.into_iter().map(ValueData::I64Value))
|
||||
}
|
||||
|
||||
fn f64_list_value(values: impl IntoIterator<Item = f64>) -> Value {
|
||||
list_value(values.into_iter().map(ValueData::F64Value))
|
||||
}
|
||||
|
||||
fn bucket_counts_from_deltas(deltas: &[i64]) -> Result<Vec<i64>> {
|
||||
let mut count = 0_i64;
|
||||
let mut buckets = Vec::with_capacity(deltas.len());
|
||||
|
||||
for delta in deltas {
|
||||
count = count
|
||||
.checked_add(*delta)
|
||||
.context(error::InvalidPromRemoteRequestSnafu {
|
||||
msg: "remote write v2 native histogram bucket count overflows i64".to_string(),
|
||||
})?;
|
||||
ensure!(
|
||||
count >= 0,
|
||||
error::InvalidPromRemoteRequestSnafu {
|
||||
msg: "remote write v2 native histogram bucket count is negative".to_string(),
|
||||
}
|
||||
);
|
||||
buckets.push(count);
|
||||
}
|
||||
|
||||
Ok(buckets)
|
||||
}
|
||||
|
||||
fn ensure_no_internal_histogram_labels(tags: &PromTags) -> Result<()> {
|
||||
// The histogram field column is generated from the protobuf payload.
|
||||
for (name, _) in tags {
|
||||
ensure!(
|
||||
name != NATIVE_HISTOGRAM_FIELD,
|
||||
error::InvalidPromRemoteRequestSnafu {
|
||||
msg: format!(
|
||||
"remote write v2 label `{name}` conflicts with an internal native histogram label"
|
||||
),
|
||||
}
|
||||
);
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn resolve_series_labels(symbols: &[String], series: &TimeSeries) -> Result<ResolvedSeriesLabels> {
|
||||
ensure!(
|
||||
series.labels_refs.len().is_multiple_of(2),
|
||||
@@ -151,7 +428,7 @@ fn resolve_series_labels(symbols: &[String], series: &TimeSeries) -> Result<Reso
|
||||
for pair in series.labels_refs.chunks_exact(2) {
|
||||
let name = symbol_ref(symbols, pair[0], "label name")?;
|
||||
let value = symbol_ref(symbols, pair[1], "label value")?;
|
||||
validate_label(name, value)?;
|
||||
validate_label(name)?;
|
||||
ensure!(
|
||||
label_names.insert(name),
|
||||
error::InvalidPromRemoteRequestSnafu {
|
||||
@@ -173,21 +450,21 @@ fn resolve_series_labels(symbols: &[String], series: &TimeSeries) -> Result<Reso
|
||||
let table_name = table_name.context(error::InvalidPromRemoteRequestSnafu {
|
||||
msg: "missing '__name__' label in time-series".to_string(),
|
||||
})?;
|
||||
ensure!(
|
||||
!table_name.is_empty(),
|
||||
error::InvalidPromRemoteRequestSnafu {
|
||||
msg: "remote write v2 label `__name__` value must not be empty".to_string(),
|
||||
}
|
||||
);
|
||||
|
||||
Ok((prom_ctx, table_name, tags))
|
||||
}
|
||||
|
||||
fn validate_label(name: &str, value: &str) -> Result<()> {
|
||||
fn validate_label(name: &str) -> Result<()> {
|
||||
ensure!(
|
||||
!name.is_empty(),
|
||||
validate_label_name(name.as_bytes()),
|
||||
error::InvalidPromRemoteRequestSnafu {
|
||||
msg: "remote write v2 label names must not be empty".to_string(),
|
||||
}
|
||||
);
|
||||
ensure!(
|
||||
!value.is_empty(),
|
||||
error::InvalidPromRemoteRequestSnafu {
|
||||
msg: format!("remote write v2 label `{name}` value must not be empty"),
|
||||
msg: format!("remote write v2 invalid label name `{name}`"),
|
||||
}
|
||||
);
|
||||
|
||||
@@ -266,6 +543,10 @@ fn table_data_to_row_insert_request(table_name: String, table_data: TableData) -
|
||||
#[cfg(any(test, feature = "testing"))]
|
||||
pub mod test_util {
|
||||
use api::greptime_proto::io::prometheus::write::v2::{Histogram, Request, Sample, TimeSeries};
|
||||
use api::v1::RowInsertRequest;
|
||||
use bytes::Bytes;
|
||||
|
||||
use crate::error::Result;
|
||||
|
||||
pub fn request_with_labels_and_samples(
|
||||
labels: Vec<(&str, &str)>,
|
||||
@@ -281,6 +562,22 @@ pub mod test_util {
|
||||
request_with_labels(labels, Vec::new(), histograms)
|
||||
}
|
||||
|
||||
pub fn decode_request(is_zstd: bool, body: Bytes) -> Result<Request> {
|
||||
super::decode_remote_write_v2_request(is_zstd, body)
|
||||
}
|
||||
|
||||
pub fn write_requests(
|
||||
request: Request,
|
||||
) -> Result<(Vec<RowInsertRequest>, Vec<RowInsertRequest>, u64, u64)> {
|
||||
let requests = super::into_write_requests(request)?;
|
||||
Ok((
|
||||
requests.samples.all_req().collect(),
|
||||
requests.histograms.all_req().collect(),
|
||||
requests.sample_count,
|
||||
requests.histogram_count,
|
||||
))
|
||||
}
|
||||
|
||||
pub fn histogram(timestamp: i64) -> Histogram {
|
||||
Histogram {
|
||||
timestamp,
|
||||
@@ -375,7 +672,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_into_context_req_samples() {
|
||||
let ctx_req = test_util::request_with_labels_and_samples(
|
||||
let ctx_req = into_write_requests(test_util::request_with_labels_and_samples(
|
||||
vec![
|
||||
(METRIC_NAME_LABEL, "http_requests_total"),
|
||||
("job", "api"),
|
||||
@@ -393,11 +690,13 @@ mod tests {
|
||||
start_timestamp: 0,
|
||||
},
|
||||
],
|
||||
)
|
||||
.into_context_req()
|
||||
))
|
||||
.unwrap();
|
||||
|
||||
let mut inserts = ctx_req.all_req().collect::<Vec<_>>();
|
||||
assert_eq!(ctx_req.sample_count, 2);
|
||||
assert_eq!(ctx_req.histogram_count, 0);
|
||||
assert_eq!(ctx_req.histograms.all_req().count(), 0);
|
||||
let mut inserts = ctx_req.samples.all_req().collect::<Vec<_>>();
|
||||
assert_eq!(inserts.len(), 1);
|
||||
|
||||
let request = inserts.pop().unwrap();
|
||||
@@ -437,51 +736,9 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_into_context_req_accepts_utf8_label_names() {
|
||||
let ctx_req = test_util::request_with_labels_and_samples(
|
||||
vec![
|
||||
(METRIC_NAME_LABEL, "http_requests_total"),
|
||||
("service.name", "api"),
|
||||
("区域", "华东"),
|
||||
],
|
||||
vec![Sample {
|
||||
value: 42.0,
|
||||
timestamp: 1000,
|
||||
start_timestamp: 0,
|
||||
}],
|
||||
)
|
||||
.into_context_req()
|
||||
.unwrap();
|
||||
|
||||
let mut inserts = ctx_req.all_req().collect::<Vec<_>>();
|
||||
assert_eq!(inserts.len(), 1);
|
||||
let rows = inserts.pop().unwrap().rows.unwrap();
|
||||
assert_eq!(
|
||||
rows.schema
|
||||
.iter()
|
||||
.map(|col| col.column_name.as_str())
|
||||
.collect::<Vec<_>>(),
|
||||
vec![
|
||||
greptime_timestamp(),
|
||||
greptime_value(),
|
||||
"service.name",
|
||||
"区域"
|
||||
]
|
||||
);
|
||||
assert_eq!(
|
||||
rows.rows[0].values[2].value_data,
|
||||
Some(ValueData::StringValue("api".to_string()))
|
||||
);
|
||||
assert_eq!(
|
||||
rows.rows[0].values[3].value_data,
|
||||
Some(ValueData::StringValue("华东".to_string()))
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_into_context_req_special_labels() {
|
||||
let ctx_req = test_util::request_with_labels_and_samples(
|
||||
let ctx_req = into_write_requests(test_util::request_with_labels_and_samples(
|
||||
vec![
|
||||
(METRIC_NAME_LABEL, "cpu_usage"),
|
||||
(DATABASE_LABEL, "tenant_a"),
|
||||
@@ -493,11 +750,12 @@ mod tests {
|
||||
timestamp: 1000,
|
||||
start_timestamp: 0,
|
||||
}],
|
||||
)
|
||||
.into_context_req()
|
||||
))
|
||||
.unwrap();
|
||||
|
||||
let mut iter = ctx_req.as_req_iter(Arc::new(QueryContext::with("greptime", "public")));
|
||||
let mut iter = ctx_req
|
||||
.samples
|
||||
.as_req_iter(Arc::new(QueryContext::with("greptime", "public")));
|
||||
let (ctx, reqs) = iter.next().unwrap();
|
||||
assert!(iter.next().is_none());
|
||||
|
||||
@@ -565,13 +823,40 @@ mod tests {
|
||||
cases.push((
|
||||
"empty label name",
|
||||
request_with_sample(vec![(METRIC_NAME_LABEL, "metric"), ("", "api")]),
|
||||
"label names must not be empty",
|
||||
"invalid label name",
|
||||
));
|
||||
|
||||
cases.push((
|
||||
"empty label value",
|
||||
request_with_sample(vec![(METRIC_NAME_LABEL, "metric"), ("job", "")]),
|
||||
"label `job` value must not be empty",
|
||||
"invalid label name",
|
||||
request_with_sample(vec![(METRIC_NAME_LABEL, "metric"), ("has-dash", "api")]),
|
||||
"invalid label name",
|
||||
));
|
||||
|
||||
cases.push((
|
||||
"dotted label name",
|
||||
request_with_sample(vec![(METRIC_NAME_LABEL, "metric"), ("service.name", "api")]),
|
||||
"invalid label name",
|
||||
));
|
||||
|
||||
cases.push((
|
||||
"non-ascii label name",
|
||||
request_with_sample(vec![(METRIC_NAME_LABEL, "metric"), ("区域", "api")]),
|
||||
"invalid label name",
|
||||
));
|
||||
|
||||
cases.push((
|
||||
"empty metric name",
|
||||
request_with_sample(vec![(METRIC_NAME_LABEL, "")]),
|
||||
"label `__name__` value must not be empty",
|
||||
));
|
||||
|
||||
cases.push((
|
||||
"internal histogram label on samples",
|
||||
request_with_sample(vec![
|
||||
(METRIC_NAME_LABEL, "metric"),
|
||||
(NATIVE_HISTOGRAM_FIELD, "user_value"),
|
||||
]),
|
||||
"conflicts with an internal native histogram label",
|
||||
));
|
||||
|
||||
for (name, request, expected) in cases {
|
||||
@@ -580,7 +865,27 @@ mod tests {
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_into_context_req_ignores_histograms_and_exemplars() {
|
||||
fn test_into_context_req_allows_empty_label_values() {
|
||||
let ctx_req = into_write_requests(test_util::request_with_labels_and_samples(
|
||||
vec![(METRIC_NAME_LABEL, "metric"), ("job", "")],
|
||||
vec![Sample {
|
||||
value: 1.0,
|
||||
timestamp: 1000,
|
||||
start_timestamp: 0,
|
||||
}],
|
||||
))
|
||||
.unwrap();
|
||||
|
||||
let rows = ctx_req.samples.all_req().next().unwrap().rows.unwrap();
|
||||
let job_idx = column_index(&rows.schema, "job");
|
||||
assert_eq!(
|
||||
rows.rows[0].values[job_idx].value_data,
|
||||
Some(ValueData::StringValue(String::new()))
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_into_context_req_rejects_same_metric_samples_and_histograms() {
|
||||
let mut request = test_util::request_with_labels_and_samples(
|
||||
vec![(METRIC_NAME_LABEL, "metric")],
|
||||
vec![Sample {
|
||||
@@ -590,22 +895,213 @@ mod tests {
|
||||
}],
|
||||
);
|
||||
request.timeseries[0].histograms.push(Histogram::default());
|
||||
request.timeseries[0].exemplars.push(Exemplar::default());
|
||||
|
||||
let ctx_req = request.into_context_req().unwrap();
|
||||
|
||||
assert_eq!(ctx_req.all_req().count(), 1);
|
||||
assert_invalid(
|
||||
"same metric samples and histograms",
|
||||
request,
|
||||
"contains both samples and native histograms",
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_into_context_req_skips_histogram_only_series() {
|
||||
fn test_into_context_req_converts_histograms_and_ignores_exemplars() {
|
||||
let request = Request {
|
||||
symbols: vec![
|
||||
"".to_string(),
|
||||
METRIC_NAME_LABEL.to_string(),
|
||||
"sample_metric".to_string(),
|
||||
"histogram_metric".to_string(),
|
||||
],
|
||||
timeseries: vec![
|
||||
TimeSeries {
|
||||
labels_refs: vec![1, 2],
|
||||
samples: vec![Sample {
|
||||
value: 1.0,
|
||||
timestamp: 1000,
|
||||
start_timestamp: 0,
|
||||
}],
|
||||
..Default::default()
|
||||
},
|
||||
TimeSeries {
|
||||
labels_refs: vec![1, 3],
|
||||
histograms: vec![Histogram::default()],
|
||||
exemplars: vec![Exemplar::default()],
|
||||
..Default::default()
|
||||
},
|
||||
],
|
||||
};
|
||||
|
||||
let ctx_req = into_write_requests(request).unwrap();
|
||||
|
||||
assert_eq!(ctx_req.sample_count, 1);
|
||||
assert_eq!(ctx_req.histogram_count, 1);
|
||||
assert_eq!(ctx_req.samples.all_req().count(), 1);
|
||||
assert_eq!(ctx_req.histograms.all_req().count(), 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_into_context_req_converts_histogram_only_series() {
|
||||
let mut request =
|
||||
test_util::request_with_labels_and_samples(vec![(METRIC_NAME_LABEL, "metric")], vec![]);
|
||||
request.timeseries[0].histograms.push(Histogram::default());
|
||||
|
||||
let ctx_req = request.into_context_req().unwrap();
|
||||
let ctx_req = into_write_requests(request).unwrap();
|
||||
|
||||
assert_eq!(ctx_req.all_req().count(), 0);
|
||||
assert_eq!(ctx_req.sample_count, 0);
|
||||
assert_eq!(ctx_req.histogram_count, 1);
|
||||
assert_eq!(ctx_req.samples.all_req().count(), 0);
|
||||
let mut inserts = ctx_req.histograms.all_req().collect::<Vec<_>>();
|
||||
assert_eq!(inserts.len(), 1);
|
||||
|
||||
let request = inserts.pop().unwrap();
|
||||
assert_eq!(request.table_name, "metric");
|
||||
let rows = request.rows.unwrap();
|
||||
assert_eq!(rows.rows.len(), 1);
|
||||
assert_eq!(
|
||||
rows.schema
|
||||
.iter()
|
||||
.map(|col| col.column_name.as_str())
|
||||
.collect::<Vec<_>>(),
|
||||
vec![greptime_timestamp(), NATIVE_HISTOGRAM_FIELD]
|
||||
);
|
||||
assert_eq!(
|
||||
rows.rows[0].values[0].value_data,
|
||||
Some(ValueData::TimestampMillisecondValue(0))
|
||||
);
|
||||
assert_eq!(
|
||||
histogram_field_value(&rows, 0, SCHEMA_FIELD),
|
||||
Some(ValueData::I32Value(0))
|
||||
);
|
||||
assert_eq!(
|
||||
histogram_field_value(&rows, 0, COUNT_U64_FIELD),
|
||||
Some(ValueData::U64Value(0))
|
||||
);
|
||||
assert_eq!(histogram_field_value(&rows, 0, COUNT_F64_FIELD), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_into_context_req_preserves_histogram_start_timestamp() {
|
||||
let ctx_req = into_write_requests(test_util::request_with_labels_and_histograms(
|
||||
vec![(METRIC_NAME_LABEL, "metric")],
|
||||
vec![Histogram {
|
||||
timestamp: 2000,
|
||||
start_timestamp: 1000,
|
||||
..Default::default()
|
||||
}],
|
||||
))
|
||||
.unwrap();
|
||||
|
||||
let mut inserts = ctx_req.histograms.all_req().collect::<Vec<_>>();
|
||||
let rows = inserts.pop().unwrap().rows.unwrap();
|
||||
|
||||
assert_eq!(
|
||||
histogram_field_value(&rows, 0, START_TIMESTAMP_FIELD),
|
||||
Some(ValueData::TimestampMillisecondValue(1000))
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_into_context_req_rejects_internal_histogram_labels() {
|
||||
let mut request = test_util::request_with_labels_and_samples(
|
||||
vec![
|
||||
(METRIC_NAME_LABEL, "metric"),
|
||||
(NATIVE_HISTOGRAM_FIELD, "user_value"),
|
||||
],
|
||||
vec![],
|
||||
);
|
||||
request.timeseries[0].histograms.push(Histogram::default());
|
||||
|
||||
let err = match into_write_requests(request) {
|
||||
Ok(_) => panic!("expected invalid request error"),
|
||||
Err(err) => err,
|
||||
};
|
||||
assert_eq!(
|
||||
err.to_string(),
|
||||
"Invalid prometheus remote request, msg: remote write v2 label `greptime_native_histogram` conflicts with an internal native histogram label"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_into_context_req_converts_int_and_float_histograms_to_one_schema() {
|
||||
let float_histogram = Histogram {
|
||||
count: Some(api::greptime_proto::io::prometheus::write::v2::histogram::Count::CountFloat(3.5)),
|
||||
zero_count: Some(
|
||||
api::greptime_proto::io::prometheus::write::v2::histogram::ZeroCount::ZeroCountFloat(
|
||||
0.5,
|
||||
),
|
||||
),
|
||||
positive_counts: vec![2.0, 3.5],
|
||||
positive_spans: vec![api::greptime_proto::io::prometheus::write::v2::BucketSpan {
|
||||
offset: 3,
|
||||
length: 2,
|
||||
}],
|
||||
timestamp: 2000,
|
||||
..Default::default()
|
||||
};
|
||||
let request = Request {
|
||||
symbols: vec![
|
||||
"".to_string(),
|
||||
METRIC_NAME_LABEL.to_string(),
|
||||
"metric".to_string(),
|
||||
],
|
||||
timeseries: vec![
|
||||
TimeSeries {
|
||||
labels_refs: vec![1, 2],
|
||||
histograms: vec![test_util::histogram(1000)],
|
||||
..Default::default()
|
||||
},
|
||||
TimeSeries {
|
||||
labels_refs: vec![1, 2],
|
||||
histograms: vec![float_histogram],
|
||||
..Default::default()
|
||||
},
|
||||
],
|
||||
};
|
||||
|
||||
let ctx_req = into_write_requests(request).unwrap();
|
||||
|
||||
assert_eq!(ctx_req.histogram_count, 2);
|
||||
let mut inserts = ctx_req.histograms.all_req().collect::<Vec<_>>();
|
||||
assert_eq!(inserts.len(), 1);
|
||||
let rows = inserts.pop().unwrap().rows.unwrap();
|
||||
assert_eq!(rows.rows.len(), 2);
|
||||
assert_eq!(
|
||||
rows.schema
|
||||
.iter()
|
||||
.map(|col| col.column_name.as_str())
|
||||
.collect::<Vec<_>>(),
|
||||
vec![greptime_timestamp(), NATIVE_HISTOGRAM_FIELD]
|
||||
);
|
||||
|
||||
assert_eq!(
|
||||
histogram_field_value(&rows, 0, COUNT_U64_FIELD),
|
||||
Some(ValueData::U64Value(0))
|
||||
);
|
||||
assert_eq!(histogram_field_value(&rows, 0, COUNT_F64_FIELD), None);
|
||||
assert!(matches!(
|
||||
histogram_field_value(&rows, 0, POSITIVE_BUCKETS_I64_FIELD),
|
||||
Some(ValueData::ListValue(_))
|
||||
));
|
||||
assert!(is_empty_list(histogram_field_value(
|
||||
&rows,
|
||||
0,
|
||||
POSITIVE_BUCKETS_F64_FIELD
|
||||
)));
|
||||
|
||||
assert_eq!(histogram_field_value(&rows, 1, COUNT_U64_FIELD), None);
|
||||
assert_eq!(
|
||||
histogram_field_value(&rows, 1, COUNT_F64_FIELD),
|
||||
Some(ValueData::F64Value(3.5))
|
||||
);
|
||||
assert!(is_empty_list(histogram_field_value(
|
||||
&rows,
|
||||
1,
|
||||
POSITIVE_BUCKETS_I64_FIELD
|
||||
)));
|
||||
assert!(matches!(
|
||||
histogram_field_value(&rows, 1, POSITIVE_BUCKETS_F64_FIELD),
|
||||
Some(ValueData::ListValue(_))
|
||||
));
|
||||
}
|
||||
|
||||
fn request_with_sample(labels: Vec<(&str, &str)>) -> Request {
|
||||
@@ -620,7 +1116,10 @@ mod tests {
|
||||
}
|
||||
|
||||
fn assert_invalid(name: &str, request: Request, expected: &str) {
|
||||
let err = request.into_context_req().unwrap_err();
|
||||
let err = match into_write_requests(request) {
|
||||
Ok(_) => panic!("{name}: expected invalid request error"),
|
||||
Err(err) => err,
|
||||
};
|
||||
assert!(
|
||||
matches!(err, error::Error::InvalidPromRemoteRequest { .. }),
|
||||
"{name}: expected invalid request error, got {err}"
|
||||
@@ -630,4 +1129,29 @@ mod tests {
|
||||
"{name}: expected error containing {expected:?}, got {err}"
|
||||
);
|
||||
}
|
||||
|
||||
fn column_index(schema: &[ColumnSchema], column_name: &str) -> usize {
|
||||
schema
|
||||
.iter()
|
||||
.position(|column| column.column_name == column_name)
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
fn histogram_field_value(rows: &Rows, row_idx: usize, field_name: &str) -> Option<ValueData> {
|
||||
let histogram_idx = column_index(&rows.schema, NATIVE_HISTOGRAM_FIELD);
|
||||
let Some(ValueData::StructValue(histogram)) =
|
||||
&rows.rows[row_idx].values[histogram_idx].value_data
|
||||
else {
|
||||
panic!("expected native histogram struct value");
|
||||
};
|
||||
let field_idx = NATIVE_HISTOGRAM_FIELD_NAMES
|
||||
.iter()
|
||||
.position(|name| *name == field_name)
|
||||
.unwrap();
|
||||
histogram.items[field_idx].value_data.clone()
|
||||
}
|
||||
|
||||
fn is_empty_list(value: Option<ValueData>) -> bool {
|
||||
matches!(value, Some(ValueData::ListValue(list)) if list.items.is_empty())
|
||||
}
|
||||
}
|
||||
|
||||
@@ -263,7 +263,7 @@ pub fn write_json(
|
||||
)
|
||||
}
|
||||
|
||||
fn write_by_schema(
|
||||
pub(crate) fn write_by_schema(
|
||||
table_data: &mut TableData,
|
||||
kv_iter: impl Iterator<Item = (ColumnSchema, Option<ValueData>)>,
|
||||
one_row: &mut Vec<Value>,
|
||||
|
||||
@@ -28,7 +28,8 @@ use async_trait::async_trait;
|
||||
use axum::Router;
|
||||
use axum::http::HeaderMap;
|
||||
use common_query::Output;
|
||||
use common_query::prelude::{greptime_timestamp, greptime_value};
|
||||
use common_query::native_histogram::NATIVE_HISTOGRAM_FIELD;
|
||||
use common_query::prelude::{GREPTIME_PHYSICAL_TABLE, greptime_timestamp, greptime_value};
|
||||
use common_test_util::ports;
|
||||
use datafusion_expr::LogicalPlan;
|
||||
use prost::Message;
|
||||
@@ -62,6 +63,7 @@ struct DummyInstance {
|
||||
struct RemoteWriteCapture {
|
||||
schema: String,
|
||||
physical_table: Option<String>,
|
||||
with_metric_engine: bool,
|
||||
request: RowInsertRequests,
|
||||
}
|
||||
|
||||
@@ -71,7 +73,7 @@ impl PromStoreProtocolHandler for DummyInstance {
|
||||
&self,
|
||||
request: RowInsertRequests,
|
||||
ctx: QueryContextRef,
|
||||
_with_metric_engine: bool,
|
||||
with_metric_engine: bool,
|
||||
) -> Result<Output> {
|
||||
let write_call = self.write_calls.fetch_add(1, Ordering::SeqCst) + 1;
|
||||
if self.fail_write_call == Some(write_call) {
|
||||
@@ -86,6 +88,7 @@ impl PromStoreProtocolHandler for DummyInstance {
|
||||
.send(RemoteWriteCapture {
|
||||
schema: ctx.current_schema(),
|
||||
physical_table: ctx.extension(PHYSICAL_TABLE_PARAM).map(ToString::to_string),
|
||||
with_metric_engine,
|
||||
request,
|
||||
})
|
||||
.await;
|
||||
@@ -166,13 +169,38 @@ fn make_test_app_with_write_capture(
|
||||
make_test_app_with_write_failure(read_tx, write_tx, None)
|
||||
}
|
||||
|
||||
fn make_test_app_with_native_histogram_write_capture(
|
||||
read_tx: mpsc::Sender<(String, Vec<u8>)>,
|
||||
write_tx: mpsc::Sender<RemoteWriteCapture>,
|
||||
) -> Router {
|
||||
make_test_app_with_write_failure_inner(read_tx, write_tx, None, true)
|
||||
}
|
||||
|
||||
fn make_test_app_with_write_failure(
|
||||
read_tx: mpsc::Sender<(String, Vec<u8>)>,
|
||||
write_tx: mpsc::Sender<RemoteWriteCapture>,
|
||||
fail_write_call: Option<usize>,
|
||||
) -> Router {
|
||||
make_test_app_with_write_failure_inner(read_tx, write_tx, fail_write_call, false)
|
||||
}
|
||||
|
||||
fn make_test_app_with_native_histogram_write_failure(
|
||||
read_tx: mpsc::Sender<(String, Vec<u8>)>,
|
||||
write_tx: mpsc::Sender<RemoteWriteCapture>,
|
||||
fail_write_call: Option<usize>,
|
||||
) -> Router {
|
||||
make_test_app_with_write_failure_inner(read_tx, write_tx, fail_write_call, true)
|
||||
}
|
||||
|
||||
fn make_test_app_with_write_failure_inner(
|
||||
read_tx: mpsc::Sender<(String, Vec<u8>)>,
|
||||
write_tx: mpsc::Sender<RemoteWriteCapture>,
|
||||
fail_write_call: Option<usize>,
|
||||
experimental_enable_prometheus_native_histogram: bool,
|
||||
) -> Router {
|
||||
let http_opts = HttpOptions {
|
||||
addr: format!("127.0.0.1:{}", ports::get_port()),
|
||||
experimental_enable_prometheus_native_histogram,
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
@@ -327,6 +355,59 @@ async fn test_prometheus_remote_write_v2_write_error_has_partial_written_headers
|
||||
assert!(write_rx.try_recv().is_err());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_prometheus_remote_write_v2_histogram_write_error_has_partial_written_headers() {
|
||||
common_telemetry::init_default_ut_logging();
|
||||
let (read_tx, _read_rx) = mpsc::channel(100);
|
||||
let (write_tx, mut write_rx) = mpsc::channel(100);
|
||||
|
||||
let app = make_test_app_with_native_histogram_write_failure(read_tx, write_tx, Some(2));
|
||||
let client = TestClient::new(app).await;
|
||||
|
||||
let mut write_request = remote_write_v2::request_with_labels_and_samples(
|
||||
vec![(prom_store::METRIC_NAME_LABEL, "http_requests_total")],
|
||||
vec![RemoteWriteV2Sample {
|
||||
value: 42.0,
|
||||
timestamp: 1000,
|
||||
start_timestamp: 0,
|
||||
}],
|
||||
);
|
||||
let metric_name_ref = write_request
|
||||
.symbols
|
||||
.iter()
|
||||
.position(|symbol| symbol == prom_store::METRIC_NAME_LABEL)
|
||||
.unwrap() as u32;
|
||||
let histogram_metric_ref = write_request.symbols.len() as u32;
|
||||
write_request
|
||||
.symbols
|
||||
.push("http_request_duration_seconds".to_string());
|
||||
write_request.timeseries.push(RemoteWriteV2TimeSeries {
|
||||
labels_refs: vec![metric_name_ref, histogram_metric_ref],
|
||||
histograms: vec![remote_write_v2::histogram(1000)],
|
||||
..Default::default()
|
||||
});
|
||||
|
||||
let result = post_remote_write_v2(&client, &write_request).await;
|
||||
|
||||
assert_eq!(result.status(), 400);
|
||||
assert_remote_write_v2_written_headers_with_histograms(&result.headers(), "1", "0");
|
||||
assert!(
|
||||
result
|
||||
.text()
|
||||
.await
|
||||
.contains("injected prometheus remote write failure")
|
||||
);
|
||||
|
||||
let captured = write_rx.recv().await.unwrap();
|
||||
assert!(captured.with_metric_engine);
|
||||
assert_eq!(1, captured.request.inserts.len());
|
||||
assert_eq!(
|
||||
"http_requests_total",
|
||||
captured.request.inserts[0].table_name
|
||||
);
|
||||
assert!(write_rx.try_recv().is_err());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_prometheus_remote_write_read() {
|
||||
common_telemetry::init_default_ut_logging();
|
||||
@@ -528,7 +609,50 @@ async fn test_prometheus_remote_write_v2_samples() {
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_prometheus_remote_write_v2_ignores_histogram_only_series() {
|
||||
async fn test_prometheus_remote_write_v2_writes_histogram_only_series() {
|
||||
common_telemetry::init_default_ut_logging();
|
||||
let (read_tx, _read_rx) = mpsc::channel(100);
|
||||
let (write_tx, mut write_rx) = mpsc::channel(100);
|
||||
|
||||
let app = make_test_app_with_native_histogram_write_capture(read_tx, write_tx);
|
||||
let client = TestClient::new(app).await;
|
||||
|
||||
let write_request = remote_write_v2::request_with_labels_and_histograms(
|
||||
vec![(
|
||||
prom_store::METRIC_NAME_LABEL,
|
||||
"http_request_duration_seconds",
|
||||
)],
|
||||
vec![remote_write_v2::histogram(1000)],
|
||||
);
|
||||
|
||||
let result = post_remote_write_v2(&client, &write_request).await;
|
||||
|
||||
assert_eq!(result.status(), 204);
|
||||
assert_remote_write_v2_written_headers_with_histograms(&result.headers(), "0", "1");
|
||||
assert!(result.text().await.is_empty());
|
||||
|
||||
let captured = write_rx.recv().await.unwrap();
|
||||
assert!(captured.with_metric_engine);
|
||||
assert_eq!(
|
||||
Some(GREPTIME_PHYSICAL_TABLE.to_string()),
|
||||
captured.physical_table
|
||||
);
|
||||
assert_eq!(1, captured.request.inserts.len());
|
||||
let insert = &captured.request.inserts[0];
|
||||
assert_eq!("http_request_duration_seconds", insert.table_name);
|
||||
let rows = insert.rows.as_ref().unwrap();
|
||||
assert_eq!(1, rows.rows.len());
|
||||
assert!(
|
||||
rows.schema
|
||||
.iter()
|
||||
.any(|column| column.column_name == NATIVE_HISTOGRAM_FIELD
|
||||
&& column.datatype == ColumnDataType::Struct as i32)
|
||||
);
|
||||
assert!(write_rx.try_recv().is_err());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_prometheus_remote_write_v2_rejects_native_histogram_when_disabled() {
|
||||
common_telemetry::init_default_ut_logging();
|
||||
let (read_tx, _read_rx) = mpsc::channel(100);
|
||||
let (write_tx, mut write_rx) = mpsc::channel(100);
|
||||
@@ -546,9 +670,14 @@ async fn test_prometheus_remote_write_v2_ignores_histogram_only_series() {
|
||||
|
||||
let result = post_remote_write_v2(&client, &write_request).await;
|
||||
|
||||
assert_eq!(result.status(), 204);
|
||||
assert_remote_write_v2_written_headers(&result.headers(), "0");
|
||||
assert!(result.text().await.is_empty());
|
||||
assert_eq!(result.status(), 400);
|
||||
assert_remote_write_v2_written_headers_with_histograms(&result.headers(), "0", "0");
|
||||
assert!(
|
||||
result
|
||||
.text()
|
||||
.await
|
||||
.contains("native histogram ingestion is experimental")
|
||||
);
|
||||
assert!(write_rx.try_recv().is_err());
|
||||
}
|
||||
|
||||
@@ -606,6 +735,14 @@ async fn test_prometheus_remote_write_v2_rejects_unsupported_content_encoding()
|
||||
}
|
||||
|
||||
fn assert_remote_write_v2_written_headers(headers: &HeaderMap, samples: &str) {
|
||||
assert_remote_write_v2_written_headers_with_histograms(headers, samples, "0")
|
||||
}
|
||||
|
||||
fn assert_remote_write_v2_written_headers_with_histograms(
|
||||
headers: &HeaderMap,
|
||||
samples: &str,
|
||||
histograms: &str,
|
||||
) {
|
||||
assert_eq!(
|
||||
Some(samples),
|
||||
headers
|
||||
@@ -613,7 +750,7 @@ fn assert_remote_write_v2_written_headers(headers: &HeaderMap, samples: &str) {
|
||||
.map(|x| x.to_str().unwrap())
|
||||
);
|
||||
assert_eq!(
|
||||
Some("0"),
|
||||
Some(histograms),
|
||||
headers
|
||||
.get("X-Prometheus-Remote-Write-Histograms-Written")
|
||||
.map(|x| x.to_str().unwrap())
|
||||
|
||||
@@ -41,6 +41,7 @@ mod http;
|
||||
mod interceptor;
|
||||
mod mysql;
|
||||
mod postgres;
|
||||
mod prom_remote_write_v2_test;
|
||||
|
||||
pub struct DummyInstance {
|
||||
query_engine: QueryEngineRef,
|
||||
|
||||
@@ -0,0 +1,174 @@
|
||||
// Copyright 2023 Greptime Team
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use api::greptime_proto::io::prometheus::write::v2::Metadata;
|
||||
use api::greptime_proto::io::prometheus::write::v2::histogram::{Count, ZeroCount};
|
||||
use api::v1::value::ValueData;
|
||||
use api::v1::{ColumnSchema, Rows};
|
||||
use bytes::Bytes;
|
||||
use common_query::native_histogram::{
|
||||
COUNT_U64_FIELD, NATIVE_HISTOGRAM_FIELD, NATIVE_HISTOGRAM_FIELD_NAMES,
|
||||
POSITIVE_BUCKETS_F64_FIELD, POSITIVE_BUCKETS_I64_FIELD, POSITIVE_SPAN_OFFSETS_FIELD,
|
||||
SCHEMA_FIELD,
|
||||
};
|
||||
use servers::prom_remote_write::v2::test_util as remote_write_v2;
|
||||
|
||||
#[test]
|
||||
fn test_decode_remote_write_v2_native_histogram_dump() {
|
||||
const BODY: &[u8] = include_bytes!(
|
||||
"testdata/prom_remote_write/remote_write_v2_native_hist_1782358162264510000.raw.snappy"
|
||||
);
|
||||
|
||||
let decoded = remote_write_v2::decode_request(false, Bytes::from_static(BODY)).unwrap();
|
||||
|
||||
assert_eq!(
|
||||
decoded.symbols,
|
||||
vec![
|
||||
"",
|
||||
"__name__",
|
||||
"received_from_a_http_request_duration_seconds",
|
||||
"handler",
|
||||
"/api/v1/write",
|
||||
"instance",
|
||||
"localhost:9090",
|
||||
"job",
|
||||
"prometheus",
|
||||
"source_prometheus",
|
||||
"A",
|
||||
]
|
||||
);
|
||||
assert_eq!(decoded.timeseries.len(), 1);
|
||||
|
||||
let series = &decoded.timeseries[0];
|
||||
assert_eq!(series.labels_refs, vec![1, 2, 3, 4, 5, 6, 7, 8, 9, 10]);
|
||||
assert!(series.samples.is_empty());
|
||||
assert!(series.exemplars.is_empty());
|
||||
assert_eq!(series.metadata, Some(Metadata::default()));
|
||||
assert_eq!(series.histograms.len(), 1);
|
||||
|
||||
let histogram = &series.histograms[0];
|
||||
assert_eq!(histogram.count, Some(Count::CountInt(24)));
|
||||
assert_eq!(histogram.sum, 0.205418879);
|
||||
assert_eq!(histogram.schema, 3);
|
||||
assert_eq!(histogram.zero_threshold, 2.938735877055719e-39);
|
||||
assert_eq!(histogram.zero_count, Some(ZeroCount::ZeroCountInt(0)));
|
||||
assert_eq!(histogram.positive_spans.len(), 3);
|
||||
assert_eq!(histogram.positive_spans[0].offset, -63);
|
||||
assert_eq!(histogram.positive_spans[0].length, 1);
|
||||
assert_eq!(histogram.positive_spans[1].offset, 1);
|
||||
assert_eq!(histogram.positive_spans[1].length, 14);
|
||||
assert_eq!(histogram.positive_spans[2].offset, 5);
|
||||
assert_eq!(histogram.positive_spans[2].length, 1);
|
||||
assert_eq!(
|
||||
histogram.positive_deltas,
|
||||
vec![2, -1, 3, -3, 3, -3, 0, 0, 0, 0, 0, 0, 0, 1, -1, 0]
|
||||
);
|
||||
assert_eq!(histogram.timestamp, 1782358160412);
|
||||
|
||||
let (sample_inserts, histogram_inserts, sample_count, histogram_count) =
|
||||
remote_write_v2::write_requests(decoded).unwrap();
|
||||
assert!(sample_inserts.is_empty());
|
||||
assert_eq!(sample_count, 0);
|
||||
assert_eq!(histogram_count, 1);
|
||||
assert_eq!(histogram_inserts.len(), 1);
|
||||
|
||||
let insert = &histogram_inserts[0];
|
||||
assert_eq!(
|
||||
insert.table_name,
|
||||
"received_from_a_http_request_duration_seconds"
|
||||
);
|
||||
let rows = insert.rows.as_ref().unwrap();
|
||||
assert_eq!(rows.rows.len(), 1);
|
||||
let row = &rows.rows[0];
|
||||
assert_eq!(
|
||||
row.values[column_index(&rows.schema, "greptime_timestamp")].value_data,
|
||||
Some(ValueData::TimestampMillisecondValue(1782358160412))
|
||||
);
|
||||
assert_eq!(
|
||||
histogram_field_value(rows, 0, SCHEMA_FIELD),
|
||||
Some(ValueData::I32Value(3))
|
||||
);
|
||||
assert_eq!(
|
||||
histogram_field_value(rows, 0, COUNT_U64_FIELD),
|
||||
Some(ValueData::U64Value(24))
|
||||
);
|
||||
assert_eq!(
|
||||
list_i32_values(histogram_field_value(rows, 0, POSITIVE_SPAN_OFFSETS_FIELD)),
|
||||
vec![-63, 1, 5]
|
||||
);
|
||||
assert_eq!(
|
||||
list_i64_values(histogram_field_value(rows, 0, POSITIVE_BUCKETS_I64_FIELD)),
|
||||
vec![2, 1, 4, 1, 4, 1, 1, 1, 1, 1, 1, 1, 1, 2, 1, 1]
|
||||
);
|
||||
assert!(is_empty_list(histogram_field_value(
|
||||
rows,
|
||||
0,
|
||||
POSITIVE_BUCKETS_F64_FIELD
|
||||
)));
|
||||
}
|
||||
|
||||
fn column_index(schema: &[ColumnSchema], column_name: &str) -> usize {
|
||||
schema
|
||||
.iter()
|
||||
.position(|column| column.column_name == column_name)
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
fn histogram_field_value(rows: &Rows, row_idx: usize, field_name: &str) -> Option<ValueData> {
|
||||
let histogram_idx = column_index(&rows.schema, NATIVE_HISTOGRAM_FIELD);
|
||||
let Some(ValueData::StructValue(histogram)) =
|
||||
&rows.rows[row_idx].values[histogram_idx].value_data
|
||||
else {
|
||||
panic!("expected native histogram struct value");
|
||||
};
|
||||
let field_idx = NATIVE_HISTOGRAM_FIELD_NAMES
|
||||
.iter()
|
||||
.position(|name| *name == field_name)
|
||||
.unwrap();
|
||||
histogram.items[field_idx].value_data.clone()
|
||||
}
|
||||
|
||||
fn list_i32_values(value: Option<ValueData>) -> Vec<i32> {
|
||||
let Some(ValueData::ListValue(list)) = value else {
|
||||
panic!("expected list value");
|
||||
};
|
||||
list.items
|
||||
.iter()
|
||||
.map(|value| {
|
||||
let Some(ValueData::I32Value(value)) = value.value_data.as_ref() else {
|
||||
panic!("expected i32 list item");
|
||||
};
|
||||
*value
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn list_i64_values(value: Option<ValueData>) -> Vec<i64> {
|
||||
let Some(ValueData::ListValue(list)) = value else {
|
||||
panic!("expected list value");
|
||||
};
|
||||
list.items
|
||||
.iter()
|
||||
.map(|value| {
|
||||
let Some(ValueData::I64Value(value)) = value.value_data.as_ref() else {
|
||||
panic!("expected i64 list item");
|
||||
};
|
||||
*value
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn is_empty_list(value: Option<ValueData>) -> bool {
|
||||
matches!(value, Some(ValueData::ListValue(list)) if list.items.is_empty())
|
||||
}
|
||||
BIN
Binary file not shown.
@@ -567,7 +567,14 @@ pub async fn setup_test_prom_app_with_frontend(
|
||||
store_type: StorageType,
|
||||
name: &str,
|
||||
) -> (Router, TestGuard) {
|
||||
setup_test_prom_app_with_frontend_inner(store_type, name, false).await
|
||||
setup_test_prom_app_with_frontend_inner(store_type, name, false, false).await
|
||||
}
|
||||
|
||||
pub async fn setup_test_prom_app_with_frontend_native_histogram(
|
||||
store_type: StorageType,
|
||||
name: &str,
|
||||
) -> (Router, TestGuard) {
|
||||
setup_test_prom_app_with_frontend_inner(store_type, name, false, true).await
|
||||
}
|
||||
|
||||
/// Like [`setup_test_prom_app_with_frontend`] but enables the pending-rows batcher,
|
||||
@@ -577,13 +584,14 @@ pub async fn setup_test_prom_app_with_frontend_batched(
|
||||
store_type: StorageType,
|
||||
name: &str,
|
||||
) -> (Router, TestGuard) {
|
||||
setup_test_prom_app_with_frontend_inner(store_type, name, true).await
|
||||
setup_test_prom_app_with_frontend_inner(store_type, name, true, false).await
|
||||
}
|
||||
|
||||
async fn setup_test_prom_app_with_frontend_inner(
|
||||
store_type: StorageType,
|
||||
name: &str,
|
||||
enable_batcher: bool,
|
||||
experimental_enable_prometheus_native_histogram: bool,
|
||||
) -> (Router, TestGuard) {
|
||||
unsafe {
|
||||
std::env::set_var("TZ", "UTC");
|
||||
@@ -634,6 +642,7 @@ async fn setup_test_prom_app_with_frontend_inner(
|
||||
|
||||
let http_opts = HttpOptions {
|
||||
addr: format!("127.0.0.1:{}", ports::get_port()),
|
||||
experimental_enable_prometheus_native_histogram,
|
||||
..Default::default()
|
||||
};
|
||||
let frontend_ref = instance.fe_instance().clone();
|
||||
|
||||
+150
-13
@@ -17,7 +17,10 @@ use std::io::Write;
|
||||
use std::str::FromStr;
|
||||
use std::time::Duration;
|
||||
|
||||
use api::greptime_proto::io::prometheus::write::v2::Sample as RemoteWriteV2Sample;
|
||||
use api::greptime_proto::io::prometheus::write::v2::histogram::{Count, ZeroCount};
|
||||
use api::greptime_proto::io::prometheus::write::v2::{
|
||||
BucketSpan, Histogram, Sample as RemoteWriteV2Sample, TimeSeries as RemoteWriteV2TimeSeries,
|
||||
};
|
||||
use api::prom_store::remote::label_matcher::Type as MatcherType;
|
||||
use api::prom_store::remote::{
|
||||
Label, LabelMatcher, Query, ReadRequest, ReadResponse, Sample, TimeSeries, WriteRequest,
|
||||
@@ -74,7 +77,7 @@ use tests_integration::test_util::{
|
||||
StorageType, setup_test_http_app, setup_test_http_app_with_frontend,
|
||||
setup_test_http_app_with_frontend_and_slow_query_threshold,
|
||||
setup_test_http_app_with_frontend_and_user_provider, setup_test_prom_app_with_frontend,
|
||||
setup_test_prom_app_with_frontend_batched,
|
||||
setup_test_prom_app_with_frontend_batched, setup_test_prom_app_with_frontend_native_histogram,
|
||||
};
|
||||
use urlencoding::encode;
|
||||
use yaml_rust::YamlLoader;
|
||||
@@ -122,6 +125,7 @@ macro_rules! http_tests {
|
||||
test_dashboard_api,
|
||||
test_prometheus_remote_write,
|
||||
test_prometheus_remote_write_v2,
|
||||
test_prometheus_remote_write_v2_native_histogram,
|
||||
test_prometheus_remote_write_batched,
|
||||
test_prometheus_remote_special_labels,
|
||||
test_prometheus_remote_schema_labels,
|
||||
@@ -279,6 +283,14 @@ fn basic_auth(username: &str, password: &str) -> String {
|
||||
}
|
||||
|
||||
fn assert_remote_write_v2_written_headers(headers: &HeaderMap, samples: &str) {
|
||||
assert_remote_write_v2_written_headers_with_histograms(headers, samples, "0")
|
||||
}
|
||||
|
||||
fn assert_remote_write_v2_written_headers_with_histograms(
|
||||
headers: &HeaderMap,
|
||||
samples: &str,
|
||||
histograms: &str,
|
||||
) {
|
||||
assert_eq!(
|
||||
Some(samples),
|
||||
headers
|
||||
@@ -286,7 +298,7 @@ fn assert_remote_write_v2_written_headers(headers: &HeaderMap, samples: &str) {
|
||||
.map(|x| x.to_str().unwrap())
|
||||
);
|
||||
assert_eq!(
|
||||
Some("0"),
|
||||
Some(histograms),
|
||||
headers
|
||||
.get("X-Prometheus-Remote-Write-Histograms-Written")
|
||||
.map(|x| x.to_str().unwrap())
|
||||
@@ -1805,6 +1817,7 @@ addr = "127.0.0.1:4000"
|
||||
timeout = "0s"
|
||||
body_limit = "64MiB"
|
||||
prom_validation_mode = "strict"
|
||||
experimental_enable_prometheus_native_histogram = false
|
||||
cors_allowed_origins = []
|
||||
enable_cors = true
|
||||
experimental_enable_explain_analyze_stream = false
|
||||
@@ -2344,7 +2357,7 @@ pub async fn test_prometheus_remote_write_v2(store_type: StorageType) {
|
||||
"select greptime_timestamp, greptime_value, job, instance from remote_write_v2_total order by greptime_timestamp;",
|
||||
"[[1000,42.0,\"api\",\"localhost:9090\"],[2000,43.0,\"api\",\"localhost:9090\"]]",
|
||||
)
|
||||
.await;
|
||||
.await;
|
||||
|
||||
validate_data(
|
||||
"prometheus_remote_write_v2_semantic_identity",
|
||||
@@ -2357,13 +2370,113 @@ pub async fn test_prometheus_remote_write_v2(store_type: StorageType) {
|
||||
)
|
||||
.await;
|
||||
|
||||
let write_request = remote_write_v2::request_with_labels_and_histograms(
|
||||
vec![(
|
||||
prom_store::METRIC_NAME_LABEL,
|
||||
"remote_write_v2_histogram_only",
|
||||
)],
|
||||
vec![remote_write_v2::histogram(3000)],
|
||||
guard.remove_all().await;
|
||||
}
|
||||
|
||||
pub async fn test_prometheus_remote_write_v2_native_histogram(store_type: StorageType) {
|
||||
common_telemetry::init_default_ut_logging();
|
||||
let (app, mut guard) = setup_test_prom_app_with_frontend_native_histogram(
|
||||
store_type,
|
||||
"prometheus_remote_write_v2_native_histogram",
|
||||
)
|
||||
.await;
|
||||
let client = TestClient::new(app).await;
|
||||
|
||||
let mut write_request = remote_write_v2::request_with_labels_and_samples(
|
||||
vec![
|
||||
(
|
||||
prom_store::METRIC_NAME_LABEL,
|
||||
"remote_write_v2_sample_seconds",
|
||||
),
|
||||
("job", "api"),
|
||||
("instance", "localhost:9090"),
|
||||
],
|
||||
vec![RemoteWriteV2Sample {
|
||||
value: 5.0,
|
||||
timestamp: 2500,
|
||||
start_timestamp: 0,
|
||||
}],
|
||||
);
|
||||
let name_ref = write_request
|
||||
.symbols
|
||||
.iter()
|
||||
.position(|symbol| symbol == prom_store::METRIC_NAME_LABEL)
|
||||
.unwrap() as u32;
|
||||
let job_ref = write_request
|
||||
.symbols
|
||||
.iter()
|
||||
.position(|symbol| symbol == "job")
|
||||
.unwrap() as u32;
|
||||
let api_ref = write_request
|
||||
.symbols
|
||||
.iter()
|
||||
.position(|symbol| symbol == "api")
|
||||
.unwrap() as u32;
|
||||
let instance_ref = write_request
|
||||
.symbols
|
||||
.iter()
|
||||
.position(|symbol| symbol == "instance")
|
||||
.unwrap() as u32;
|
||||
let localhost_ref = write_request
|
||||
.symbols
|
||||
.iter()
|
||||
.position(|symbol| symbol == "localhost:9090")
|
||||
.unwrap() as u32;
|
||||
let histogram_metric_ref = write_request.symbols.len() as u32;
|
||||
write_request
|
||||
.symbols
|
||||
.push("remote_write_v2_latency_seconds".to_string());
|
||||
write_request.timeseries.push(RemoteWriteV2TimeSeries {
|
||||
labels_refs: vec![
|
||||
name_ref,
|
||||
histogram_metric_ref,
|
||||
job_ref,
|
||||
api_ref,
|
||||
instance_ref,
|
||||
localhost_ref,
|
||||
],
|
||||
histograms: vec![
|
||||
Histogram {
|
||||
count: Some(Count::CountInt(4)),
|
||||
sum: 10.0,
|
||||
schema: 1,
|
||||
zero_threshold: 0.001,
|
||||
zero_count: Some(ZeroCount::ZeroCountInt(1)),
|
||||
negative_spans: vec![BucketSpan {
|
||||
offset: -2,
|
||||
length: 1,
|
||||
}],
|
||||
negative_deltas: vec![1],
|
||||
positive_spans: vec![BucketSpan {
|
||||
offset: 0,
|
||||
length: 3,
|
||||
}],
|
||||
positive_deltas: vec![1, 2, -1],
|
||||
reset_hint: 2,
|
||||
timestamp: 3000,
|
||||
start_timestamp: 1500,
|
||||
custom_values: vec![0.5, 1.5],
|
||||
..Default::default()
|
||||
},
|
||||
Histogram {
|
||||
count: Some(Count::CountFloat(3.5)),
|
||||
sum: 20.0,
|
||||
schema: 2,
|
||||
zero_threshold: 0.002,
|
||||
zero_count: Some(ZeroCount::ZeroCountFloat(0.5)),
|
||||
positive_spans: vec![BucketSpan {
|
||||
offset: 3,
|
||||
length: 2,
|
||||
}],
|
||||
positive_counts: vec![2.0, 3.5],
|
||||
reset_hint: 3,
|
||||
timestamp: 4000,
|
||||
start_timestamp: 2500,
|
||||
..Default::default()
|
||||
},
|
||||
],
|
||||
..Default::default()
|
||||
});
|
||||
let serialized_request = write_request.encode_to_vec();
|
||||
let compressed_request =
|
||||
prom_store::snappy_compress(&serialized_request).expect("failed to encode snappy");
|
||||
@@ -2380,13 +2493,37 @@ pub async fn test_prometheus_remote_write_v2(store_type: StorageType) {
|
||||
.await;
|
||||
assert_eq!(res.status(), StatusCode::NO_CONTENT);
|
||||
let headers = res.headers();
|
||||
assert_remote_write_v2_written_headers(&headers, "0");
|
||||
assert_remote_write_v2_written_headers_with_histograms(&headers, "1", "2");
|
||||
assert!(res.text().await.is_empty());
|
||||
|
||||
validate_data(
|
||||
"prometheus_remote_write_v2_histogram_only",
|
||||
"prometheus_remote_write_v2_native_histogram_sample_rows",
|
||||
&client,
|
||||
"select count(*) from information_schema.tables where table_name = 'remote_write_v2_histogram_only';",
|
||||
"select greptime_timestamp, greptime_value, job, instance from remote_write_v2_sample_seconds order by greptime_timestamp;",
|
||||
"[[2500,5.0,\"api\",\"localhost:9090\"]]",
|
||||
)
|
||||
.await;
|
||||
|
||||
validate_data(
|
||||
"prometheus_remote_write_v2_native_histogram_rows",
|
||||
&client,
|
||||
"select greptime_timestamp, greptime_native_histogram, job, instance from remote_write_v2_latency_seconds order by greptime_timestamp;",
|
||||
"[[3000,{\"count_f64\":null,\"count_u64\":4,\"custom_values\":[0.5,1.5],\"negative_buckets_f64\":[],\"negative_buckets_i64\":[1],\"negative_span_lengths\":[1],\"negative_span_offsets\":[-2],\"positive_buckets_f64\":[],\"positive_buckets_i64\":[1,3,2],\"positive_span_lengths\":[3],\"positive_span_offsets\":[0],\"reset_hint\":2,\"schema\":1,\"start_timestamp\":1500,\"sum\":10.0,\"zero_count_f64\":null,\"zero_count_u64\":1,\"zero_threshold\":0.001},\"api\",\"localhost:9090\"],[4000,{\"count_f64\":3.5,\"count_u64\":null,\"custom_values\":[],\"negative_buckets_f64\":[],\"negative_buckets_i64\":[],\"negative_span_lengths\":[],\"negative_span_offsets\":[],\"positive_buckets_f64\":[2.0,3.5],\"positive_buckets_i64\":[],\"positive_span_lengths\":[2],\"positive_span_offsets\":[3],\"reset_hint\":3,\"schema\":2,\"start_timestamp\":2500,\"sum\":20.0,\"zero_count_f64\":0.5,\"zero_count_u64\":null,\"zero_threshold\":0.002},\"api\",\"localhost:9090\"]]",
|
||||
)
|
||||
.await;
|
||||
|
||||
validate_data(
|
||||
"prometheus_remote_write_v2_native_histogram_table_created",
|
||||
&client,
|
||||
"select count(*) from information_schema.tables where table_name = 'remote_write_v2_latency_seconds';",
|
||||
"[[1]]",
|
||||
)
|
||||
.await;
|
||||
|
||||
validate_data(
|
||||
"prometheus_remote_write_v2_native_histogram_suffix_table_not_created",
|
||||
&client,
|
||||
"select count(*) from information_schema.tables where table_name = 'remote_write_v2_latency_seconds_native_histogram';",
|
||||
"[[0]]",
|
||||
)
|
||||
.await;
|
||||
|
||||
Reference in New Issue
Block a user