rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-08-18 12:08:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(protocol): validate native histogram ingestion (#8775)

Browse Source 
Signed-off-by: shuiyisong <xixing.sys@gmail.com>
This commit is contained in:
shuiyisong
2026-08-07 15:23:43 +08:00
committed by GitHub
parent 97ca104129
commit ec51113ab6
5 changed files with 313 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/operator/src/insert.rs
+61 -1
View File
@@ -1496,7 +1496,7 @@ mod tests {
}
#[tokio::test]
async fn test_accommodate_existing_schema_logic() {
async fn test_accommodate_existing_schema_and_reject_kind_changes() {
let ts_name = "my_ts";
let field_name = "my_field";
let table =
@@ -1548,6 +1548,66 @@ mod tests {
let req_schema = req.rows.as_ref().unwrap().schema.clone();
assert_eq!(req_schema[0].column_name, ts_name);
assert_eq!(req_schema[1].column_name, field_name);
let (datatype, datatype_extension) =
ColumnDataTypeWrapper::try_from(native_histogram_value_type().clone())
.unwrap()
.into_parts();
let mut histogram_req = RowInsertRequest {
table_name: "test_table".to_string(),
rows: Some(Rows {
schema: vec![
time_index_column_schema("ts", ColumnDataType::TimestampMillisecond),
api::v1::ColumnSchema {
column_name: greptime_native_histogram().to_string(),
datatype: datatype as i32,
semantic_type: SemanticType::Field as i32,
datatype_extension,
options: None,
},
],
rows: vec![],
}),
};
let error = inserter
.get_alter_table_expr_on_demand(&mut histogram_req, &table, &ctx, true, true, true)
.unwrap_err();
assert!(
error
.to_string()
.contains("cannot mix native histogram and float sample fields")
);
let histogram_table = make_table_ref_with_schema(
"ts",
greptime_native_histogram(),
native_histogram_value_type().clone(),
);
let mut sample_req = RowInsertRequest {
table_name: "test_table".to_string(),
rows: Some(Rows {
schema: vec![
time_index_column_schema("ts", ColumnDataType::TimestampMillisecond),
field_column_schema(greptime_value(), ColumnDataType::Float64),
],
rows: vec![],
}),
};
let error = inserter
.get_alter_table_expr_on_demand(
&mut sample_req,
&histogram_table,
&ctx,
true,
true,
true,
)
.unwrap_err();
assert!(
error
.to_string()
.contains("cannot mix native histogram and float sample fields")
);
}
#[test]
src/servers/src/otlp/metrics.rs
+2 -7
View File
@@ -374,7 +374,8 @@ fn encode_metrics(
metric_ctx,
)?;
}
// TODO(sunng87) leave ExponentialHistogram for next release
// TODO: Convert OTLP exponential histograms into the canonical native-histogram
// Struct. See docs/rfcs/2026-08-04-native-histograms.md.
metric::Data::ExponentialHistogram(_hist) => {}
}
}
@@ -714,12 +715,6 @@ fn encode_histogram(
Ok(())
}
#[allow(dead_code)]
fn encode_exponential_histogram(_name: &str, _hist: &ExponentialHistogram) -> Result<()> {
// TODO(sunng87): implement this using a prometheus compatible way
Ok(())
}
fn encode_summary(
table_writer: &mut MultiTableData,
name: &str,
src/servers/src/prom_remote_write/decode.rs
+147 -9
View File
@@ -56,6 +56,12 @@ pub(crate) struct PromTimeSeries {
impl Clear for PromTimeSeries {
fn clear(&mut self) {
self.table_name.clear();
self.schema.clear();
self.physical_table.clear();
// Labels borrow from the request buffer, which may be replaced after this reset.
for label in self.labels.iter_mut() {
label.clear();
}
self.labels.clear();
self.samples.clear();
}
@@ -93,27 +99,31 @@ impl PromTimeSeries {
}
#[allow(deprecated)]
match label.name {
let is_special_label = match label.name {
METRIC_NAME_LABEL_BYTES => {
self.table_name = prom_validation_mode.decode_string(label.value)?;
self.labels.truncate(self.labels.len() - 1);
true
}
SCHEMA_LABEL_BYTES => {
self.schema = Some(prom_validation_mode.decode_string(label.value)?);
self.labels.truncate(self.labels.len() - 1);
true
}
DATABASE_LABEL_BYTES | DATABASE_LABEL_ALT_BYTES => {
if self.schema.is_none() {
self.schema = Some(prom_validation_mode.decode_string(label.value)?);
}
self.labels.truncate(self.labels.len() - 1);
true
}
PHYSICAL_TABLE_LABEL_BYTES | PHYSICAL_TABLE_LABEL_ALT_BYTES => {
self.physical_table =
Some(prom_validation_mode.decode_string(label.value)?);
self.labels.truncate(self.labels.len() - 1);
true
}
_ => {}
_ => false,
};
if is_special_label {
label.clear();
self.labels.truncate(self.labels.len() - 1);
}
Ok(())
@@ -129,6 +139,9 @@ impl PromTimeSeries {
Ok(())
}
3u32 => prost::encoding::skip_field(wire_type, tag, buf, Default::default()),
4u32 => Err(DecodeError::new(
"remote write v1 native histogram ingestion is unsupported; use remote write v2",
)),
_ => prost::encoding::skip_field(wire_type, tag, buf, Default::default()),
}
}
@@ -170,6 +183,7 @@ pub struct PromWriteRequest<'a> {
impl<'a> Clear for PromWriteRequest<'a> {
fn clear(&mut self) {
self.series.clear();
self.table_data.clear();
}
}
@@ -186,6 +200,7 @@ impl<'a> PromWriteRequest<'a> {
processor: &mut PromSeriesProcessor,
) -> Result<(), DecodeError> {
const STRUCT_NAME: &str = "PromWriteRequest";
self.clear();
self.table_data.set_raw_data(buf);
let mut offset = 0;
while offset < self.table_data.raw_data.len() {
@@ -247,8 +262,7 @@ impl<'a> PromWriteRequest<'a> {
}
if decoded_timeseries {
self.series.labels.clear();
self.series.samples.clear();
self.series.clear();
}
}
@@ -371,7 +385,7 @@ impl PromSeriesProcessor {
mod tests {
use std::collections::HashMap;
use api::prom_store::remote::WriteRequest;
use api::prom_store::remote::{Histogram, Label, Sample, TimeSeries, WriteRequest};
use api::v1::{Row, RowInsertRequests, Rows};
use bytes::Bytes;
use prost::Message;
@@ -453,6 +467,130 @@ mod tests {
}
}
#[test]
fn test_decode_rejects_remote_write_v1_native_histograms() {
for samples in [
Vec::new(),
vec![Sample {
value: 1.0,
timestamp: 1000,
}],
] {
let request = WriteRequest {
timeseries: vec![TimeSeries {
samples,
histograms: vec![Histogram::default()],
..Default::default()
}],
..Default::default()
};
let mut processor = PromSeriesProcessor::default_processor();
let mut write_request = PromWriteRequest::default();
let error = write_request
.decode(
request.encode_to_vec(),
PromValidationMode::Strict,
&mut processor,
)
.unwrap_err();
assert!(error.to_string().contains(
"remote write v1 native histogram ingestion is unsupported; use remote write v2"
));
assert_eq!(
write_request.as_row_insert_requests().ref_all_req().count(),
0
);
}
}
#[test]
fn test_decode_clears_state_after_error() {
let label = |name: &str, value: &str| Label {
name: name.to_string(),
value: value.to_string(),
};
let sample = Sample {
value: 1.0,
timestamp: 1000,
};
let failed_request = WriteRequest {
timeseries: vec![
TimeSeries {
labels: vec![
label("__name__", "stale_metric"),
label("stale_label", "stale_value"),
],
samples: vec![sample],
..Default::default()
},
TimeSeries {
labels: vec![
label("__name__", "rejected_metric"),
label("rejected_label", "rejected_value"),
label("__schema__", "rejected_schema"),
label("x_greptime_physical_table", "rejected_physical_table"),
],
samples: vec![sample],
histograms: vec![Histogram::default()],
..Default::default()
},
],
..Default::default()
};
let successful_request = WriteRequest {
timeseries: vec![TimeSeries {
labels: vec![
label("__name__", "fresh_metric"),
label("fresh_label", "fresh_value"),
],
samples: vec![sample],
..Default::default()
}],
..Default::default()
};
let mut processor = PromSeriesProcessor::default_processor();
let mut write_request = PromWriteRequest::default();
write_request
.decode(
failed_request.encode_to_vec(),
PromValidationMode::Strict,
&mut processor,
)
.unwrap_err();
write_request
.decode(
successful_request.encode_to_vec(),
PromValidationMode::Strict,
&mut processor,
)
.unwrap();
assert_eq!(write_request.table_data.tables.len(), 1);
let (prom_ctx, tables) = write_request.table_data.tables.iter().next().unwrap();
assert_eq!(prom_ctx.schema, None);
assert_eq!(prom_ctx.physical_table, None);
assert_eq!(tables.len(), 1);
assert!(tables.contains_key("fresh_metric"));
let requests = write_request
.as_row_insert_requests()
.all_req()
.collect::<Vec<_>>();
assert_eq!(requests.len(), 1);
assert_eq!(requests[0].table_name, "fresh_metric");
let rows = requests[0].rows.as_ref().unwrap();
assert_eq!(rows.rows.len(), 1);
assert_eq!(
rows.schema
.iter()
.map(|column| column.column_name.as_str())
.collect::<Vec<_>>(),
vec![greptime_timestamp(), greptime_value(), "fresh_label"]
);
}
#[test]
fn test_decode_string_strict_mode_valid_utf8() {
let valid_utf8 = Bytes::from("hello world");
src/servers/src/prom_remote_write/v2.rs
+101 -25
View File
@@ -45,7 +45,6 @@ use crate::row_writer::{self, TableData};
type PromTags = Vec<(String, String)>;
type ResolvedSeriesLabels = (PromCtx, String, PromTags);
const MIN_REMOTE_WRITE_V2_SCHEMA: i32 = -4;
const MAX_REMOTE_WRITE_V2_SCHEMA: i32 = 8;
const MAX_REDUCIBLE_REMOTE_WRITE_V2_SCHEMA: i32 = 52;
@@ -346,7 +345,7 @@ fn native_histogram_struct_value(histogram: &Histogram) -> Result<ValueData> {
}
fn validate_native_histogram(histogram: &Histogram, uses_float_counts: bool) -> Result<()> {
validate_native_histogram_schema(histogram.schema)?;
let exponential_overflow_index = validate_native_histogram_schema(histogram.schema)?;
validate_native_histogram_custom_values(histogram)?;
if histogram.schema == CUSTOM_BUCKETS_SCHEMA {
@@ -379,8 +378,11 @@ fn validate_native_histogram(histogram: &Histogram, uses_float_counts: bool) ->
histogram.negative_deltas.len(),
)
};
let custom_max_index = if histogram.schema == CUSTOM_BUCKETS_SCHEMA {
Some(
let bucket_index_range = if let Some(overflow_index) = exponential_overflow_index {
(i32::MIN, overflow_index)
} else {
(
0,
i32::try_from(histogram.custom_values.len()).ok().context(
error::InvalidPromRemoteRequestSnafu {
msg: "remote write v2 custom native histogram has too many custom_values"
@@ -388,30 +390,30 @@ fn validate_native_histogram(histogram: &Histogram, uses_float_counts: bool) ->
},
)?,
)
} else {
None
};
validate_native_histogram_spans(
"positive",
&histogram.positive_spans,
positive_buckets,
custom_max_index,
bucket_index_range,
)?;
validate_native_histogram_spans(
"negative",
&histogram.negative_spans,
negative_buckets,
None,
bucket_index_range,
)?;
Ok(())
}
fn validate_native_histogram_schema(schema: i32) -> Result<()> {
if schema == CUSTOM_BUCKETS_SCHEMA
|| (MIN_REMOTE_WRITE_V2_SCHEMA..=MAX_REMOTE_WRITE_V2_SCHEMA).contains(&schema)
{
return Ok(());
fn validate_native_histogram_schema(schema: i32) -> Result<Option<i32>> {
if schema == CUSTOM_BUCKETS_SCHEMA {
return Ok(None);
}
if let Some(overflow_index) = exponential_overflow_bucket_index(schema) {
return Ok(Some(overflow_index));
}
if (MAX_REMOTE_WRITE_V2_SCHEMA + 1..=MAX_REDUCIBLE_REMOTE_WRITE_V2_SCHEMA).contains(&schema) {
@@ -467,7 +469,7 @@ fn validate_native_histogram_spans(
name: &str,
spans: &[BucketSpan],
bucket_count: usize,
custom_max_index: Option<i32>,
bucket_index_range: (i32, i32),
) -> Result<()> {
let span_len = spans
.iter()
@@ -487,7 +489,7 @@ fn validate_native_histogram_spans(
let mut current_index = 0i32;
for (span_index, span) in spans.iter().enumerate() {
ensure!(
span.offset >= 0 || (span_index == 0 && custom_max_index.is_none()),
span.offset >= 0 || (span_index == 0 && bucket_index_range.0 == i32::MIN),
error::InvalidPromRemoteRequestSnafu {
msg: format!(
"remote write v2 native histogram {name} span {} has negative offset {}",
@@ -509,16 +511,14 @@ fn validate_native_histogram_spans(
};
for _ in 0..span.length {
if let Some(max_index) = custom_max_index {
ensure!(
(0..=max_index).contains(&current_index),
error::InvalidPromRemoteRequestSnafu {
msg: format!(
"remote write v2 custom native histogram {name} bucket index {current_index} is out of range"
),
}
);
}
ensure!(
(bucket_index_range.0..=bucket_index_range.1).contains(&current_index),
error::InvalidPromRemoteRequestSnafu {
msg: format!(
"remote write v2 native histogram {name} bucket index {current_index} is out of range"
),
}
);
current_index =
current_index
.checked_add(1)
@@ -1473,6 +1473,82 @@ mod tests {
);
}
#[test]
fn test_into_context_req_rejects_metric_kind_conflict_across_label_sets() {
let request = Request {
symbols: vec![
"".to_string(),
METRIC_NAME_LABEL.to_string(),
"metric".to_string(),
"job".to_string(),
"api".to_string(),
"worker".to_string(),
],
timeseries: vec![
TimeSeries {
labels_refs: vec![1, 2, 3, 4],
samples: vec![Sample {
value: 1.0,
timestamp: 1000,
start_timestamp: 0,
}],
..Default::default()
},
TimeSeries {
labels_refs: vec![1, 2, 3, 5],
histograms: vec![Histogram::default()],
..Default::default()
},
],
};
assert_invalid(
"same metric kind conflict across label sets",
request,
"contains both samples and native histograms",
);
}
#[test]
fn test_into_context_req_validates_exponential_overflow_bucket_index() {
for schema in [-4, 0, 8] {
let max_index = exponential_overflow_bucket_index(schema).unwrap();
for positive in [true, false] {
let mut histogram = Histogram {
schema,
count: Some(Count::CountInt(1)),
..Default::default()
};
if positive {
histogram.positive_spans = vec![BucketSpan {
offset: max_index,
length: 1,
}];
histogram.positive_deltas = vec![1];
} else {
histogram.negative_spans = vec![BucketSpan {
offset: max_index,
length: 1,
}];
histogram.negative_deltas = vec![1];
}
into_write_requests(request_with_histogram(histogram.clone())).unwrap();
let beyond = max_index + 1;
if positive {
histogram.positive_spans[0].offset = beyond;
} else {
histogram.negative_spans[0].offset = beyond;
}
assert_invalid(
"exponential overflow bucket index",
request_with_histogram(histogram),
&format!("bucket index {beyond} is out of range"),
);
}
}
}
#[test]
fn test_into_context_req_converts_histograms_and_ignores_exemplars() {
let request = Request {
src/servers/src/prom_store.rs
+2
View File
@@ -381,6 +381,8 @@ fn recordbatch_to_timeseries(table: &str, recordbatch: RecordBatch) -> Result<Ve
),
})?;
// TODO: Add native-histogram encoding when Prometheus Remote Read support is prioritized.
// The current path intentionally returns scalar samples only.
let field_column = recordbatch.column_by_name(greptime_value()).context(
error::InvalidPromRemoteReadQueryResultSnafu {
msg: "missing greptime_value column in query result",