mirror of
https://github.com/GreptimeTeam/greptimedb.git
synced 2026-10-29 07:06:20 +00:00
feat: support native histograms in sampled remote read (#9971)
* feat: support native histograms in sampled remote read Signed-off-by: shuiyisong <xixing.sys@gmail.com> * fix: align native histogram ingestion with remote read validation Signed-off-by: shuiyisong <xixing.sys@gmail.com> --------- Signed-off-by: shuiyisong <xixing.sys@gmail.com>
This commit is contained in:
shuiyisong
@@ -16,8 +16,8 @@ Struct before persistence and are queried only as native histograms.
|
||||
Prometheus Remote Write 2.0 native histograms and cumulative OTLP/HTTP
|
||||
exponential histograms are enabled by default.
|
||||
Remote Write 1.0 histogram payloads are rejected instead of being acknowledged and dropped.
|
||||
Native-histogram Remote Read is deferred; the existing Remote Read path
|
||||
continues to return scalar samples only.
|
||||
Remote Read returns scalar samples and native histograms through `SAMPLES`,
|
||||
including exact stored integer counts. Streamed responses remain deferred.
|
||||
|
||||
# Goals
|
||||
|
||||
@@ -25,13 +25,15 @@ continues to return scalar samples only.
|
||||
2. Keep one stable Prometheus-compatible persisted representation across
|
||||
Prometheus and OTLP ingestion.
|
||||
3. Match Prometheus query behavior where it is observable and practical.
|
||||
4. State intentional limitations explicitly so incomplete behavior is not
|
||||
4. Return stored native histograms through sampled Remote Read without rounding
|
||||
integer counts or losing supported histogram fields.
|
||||
5. State intentional limitations explicitly so incomplete behavior is not
|
||||
mistaken for support.
|
||||
|
||||
# Non-Goals
|
||||
|
||||
- Supporting native histograms in Remote Write 1.0.
|
||||
- Returning native histograms through Prometheus Remote Read.
|
||||
- Streaming Remote Read responses (`STREAMED_XOR_CHUNKS`).
|
||||
- Persisting Remote Write metric metadata.
|
||||
- Persisting exemplars.
|
||||
- Converting OTLP delta values to cumulative values during ingestion. Supported
|
||||
@@ -107,19 +109,57 @@ beyond it. The overflow index is `(1024 << schema) + 1` for non-negative schemas
|
||||
and `(1024 >> -schema) + 1` for negative schemas. Inputs beyond that index are
|
||||
rejected rather than clamped or merged into infinity.
|
||||
|
||||
Ingestion and sampled Remote Read share histogram layout validation. An initial
|
||||
exponential span offset of `i32::MIN` is rejected on either side, including for
|
||||
empty spans; other negative initial offsets remain supported. For custom
|
||||
histograms with `N` bounds, bucket index `N` is the overflow bucket, and empty
|
||||
spans may end at `N + 1`. Span extents beyond that boundary are rejected.
|
||||
|
||||
This tightens ingestion acceptance: these layouts were previously accepted on
|
||||
write but rejected by Remote Read. A validation failure rejects the entire
|
||||
Remote Write 2.0 request with HTTP 400 and all written-count headers set to zero,
|
||||
before any samples or histograms are persisted. Healthy samples in the same
|
||||
request are also rejected; senders that do not retry HTTP 400 can drop them.
|
||||
|
||||
Exemplars and declared metric metadata are accepted on the wire but are not
|
||||
persisted by this feature.
|
||||
|
||||
## Remote Read
|
||||
|
||||
Native-histogram Remote Read is deferred. GreptimeDB's existing Remote Read path
|
||||
continues to negotiate `SAMPLES` responses and serialize scalar samples only.
|
||||
`/v1/prometheus/read` returns scalar samples and integer or float native
|
||||
histograms through the existing Snappy-compressed `SAMPLES` protobuf response.
|
||||
Omitting accepted types, requesting `SAMPLES`, or requesting
|
||||
`[STREAMED_XOR_CHUNKS, SAMPLES]` selects this format. Streaming-only requests
|
||||
remain unsupported; the response is still buffered as a whole.
|
||||
|
||||
A follow-up can map the stored Struct into `TimeSeries.histograms`. It must
|
||||
preserve integer counts without passing through `f64`, retain the stored
|
||||
histogram fields, and decide sampled versus streamed response coverage. The
|
||||
sampled protobuf has no start-timestamp field, so that omission must remain
|
||||
explicit if sampled responses are implemented.
|
||||
One request can query both scalar and histogram metrics. A histogram series
|
||||
can contain integer and float histogram samples, but the one-sample-kind-per-
|
||||
metric-table rule remains unchanged. The selected field must be `Float64` or
|
||||
the canonical native-histogram Struct, regardless of its stored column name.
|
||||
|
||||
Integer counts remain exact within the existing nonnegative `i64` storage
|
||||
range, including above `2^53`. Stored absolute integer buckets become signed
|
||||
wire deltas, retaining their sequence across span boundaries. Float bucket
|
||||
counts remain absolute. Schema, spans, custom bounds, zero threshold, sum
|
||||
(including stale-NaN bits), reset hint, and millisecond timestamps are retained.
|
||||
The sampled protobuf has no `start_timestamp` field, so that value remains in
|
||||
storage but is omitted from the response. Raw-delta OTLP histograms retain
|
||||
their temporality labels and gauge hints; Remote Read does not accumulate them
|
||||
or reconstruct the original OTLP point.
|
||||
|
||||
Malformed or wire-incompatible non-null stored histograms fail the entire
|
||||
`ReadRequest` with HTTP 500, including healthy queries in that request. Read
|
||||
validation remains necessary for older rows and typed inserts that bypass the
|
||||
histogram ingestion encoder. Incompatible stored rows are not repaired by the
|
||||
stricter ingestion checks. Null histograms remain absent samples.
|
||||
|
||||
Custom-bucket interoperability is exercised against Prometheus revision
|
||||
`e471c7708c6b0b13e2050222bbd5a907b36f4113`, including `AtHistogram` and
|
||||
`AtFloatHistogram` on integer custom buckets. This is a tested revision, not
|
||||
a minimum-release claim; older readers can lose custom bounds during integer-
|
||||
to-float conversion. Integer payloads are not downgraded for those readers.
|
||||
The [wire fixture and iterator check](../../tests/compatibility/wire/native_histogram_remote_read/README.md)
|
||||
record the independent expectations and reproduction commands.
|
||||
|
||||
## OTLP
|
||||
|
||||
@@ -340,16 +380,20 @@ partial-success behavior, kind exclusivity,
|
||||
stale-marker selector semantics, synthetic-zero rates, incompatible empty
|
||||
layouts, overflow-safe averages, layout-sensitive equality, infinite custom
|
||||
midpoints, and exponential overflow indices for schemas `-4`, `0`, and `8`.
|
||||
Behavioral coverage also verifies frontend PromQL warning and info responses.
|
||||
Behavioral coverage also verifies frontend PromQL warning and info responses,
|
||||
sampled Remote Read before and after flush in standalone and distributed
|
||||
setups, exact large integers, stale-marker bits, timestamp units, and OTLP
|
||||
normalization. Frozen upstream protobuf fixtures guard the wire encoder.
|
||||
|
||||
This work does not change GreptimeDB's persisted Struct or protobuf
|
||||
dependencies. It adds the disabled-by-default public OTLP gate; existing
|
||||
native-histogram data remains readable.
|
||||
dependencies. Remote Read uses the existing endpoint and negotiation without
|
||||
a new configuration option; existing native-histogram data remains readable.
|
||||
|
||||
# Future Work
|
||||
|
||||
- Native-histogram Remote Read, including exact integer round-trips and streamed
|
||||
chunks with start timestamps.
|
||||
- TODO: Add native histogram and float histogram chunks to
|
||||
`STREAMED_XOR_CHUNKS`, with checksummed framing and bounded buffering. Decide
|
||||
separately whether to include start-timestamp-bearing chunk encodings.
|
||||
- Raw-delta OTLP exponential-histogram storage and temporality-aware native
|
||||
histogram queries, as described above; no delta-to-cumulative conversion.
|
||||
- Persistent Remote Write metadata and accurate help/unit updates.
|
||||
|
||||
@@ -36,7 +36,10 @@ use datafusion::arrow::datatypes::{
|
||||
use datafusion_common::{DataFusionError, Result as DfResult};
|
||||
use datatypes::data_type::{ConcreteDataType, DataType};
|
||||
use datatypes::types::{StructField, StructType};
|
||||
pub use encoding::{NativeHistogramError, encode_native_histogram, native_histogram_column_schema};
|
||||
pub use encoding::{
|
||||
NativeHistogramError, decode_native_histogram_for_remote_read, encode_native_histogram,
|
||||
native_histogram_column_schema,
|
||||
};
|
||||
use once_cell::sync::Lazy;
|
||||
|
||||
use crate::prelude::greptime_native_histogram;
|
||||
@@ -136,6 +139,37 @@ pub fn is_native_histogram_value_schema(name: &str, data_type: &ConcreteDataType
|
||||
name == greptime_native_histogram() && is_native_histogram_value_type(data_type)
|
||||
}
|
||||
|
||||
#[cfg(any(test, feature = "testing"))]
|
||||
pub mod testing {
|
||||
use api::helper::pb_value_to_value_ref;
|
||||
use api::v1::{Value, value::ValueData};
|
||||
use datafusion::arrow::array::{AsArray, StructArray};
|
||||
use datatypes::vectors::Helper;
|
||||
|
||||
use crate::native_histogram::{native_histogram_column_schema, native_histogram_value_type};
|
||||
|
||||
/// Builds stored arrays through the same protobuf-to-vector path as ingestion.
|
||||
pub fn histogram_array(values: &[Option<ValueData>]) -> StructArray {
|
||||
let schema = native_histogram_column_schema().unwrap();
|
||||
let rows = values
|
||||
.iter()
|
||||
.map(|value_data| {
|
||||
datatypes::value::Value::from(pb_value_to_value_ref(
|
||||
&Value {
|
||||
value_data: value_data.clone(),
|
||||
},
|
||||
schema.datatype_extension.as_ref(),
|
||||
))
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
Helper::try_from_row_into_vector(&rows, native_histogram_value_type())
|
||||
.unwrap()
|
||||
.to_arrow_array()
|
||||
.as_struct()
|
||||
.clone()
|
||||
}
|
||||
}
|
||||
|
||||
/// Schema identifier for native histograms with explicit custom bucket bounds.
|
||||
pub const CUSTOM_BUCKETS_SCHEMA: i32 = -53;
|
||||
const MIN_EXPONENTIAL_SCHEMA: i32 = -4;
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -850,6 +850,24 @@ mod tests {
|
||||
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn test_native_histogram_remote_read_response_negotiation() {
|
||||
for accepted in [
|
||||
vec![],
|
||||
vec![ResponseType::Samples as i32],
|
||||
vec![
|
||||
ResponseType::StreamedXorChunks as i32,
|
||||
ResponseType::Samples as i32,
|
||||
],
|
||||
] {
|
||||
assert_eq!(
|
||||
negotiate_response_type(&accepted).unwrap(),
|
||||
ResponseType::Samples
|
||||
);
|
||||
}
|
||||
assert!(negotiate_response_type(&[ResponseType::StreamedXorChunks as i32]).is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_remote_read_label_types() {
|
||||
use datatypes::schema::ColumnSchema;
|
||||
@@ -864,6 +882,10 @@ mod tests {
|
||||
(ConcreteDataType::vector_datatype(3), false),
|
||||
(ConcreteDataType::int64_datatype(), false),
|
||||
(ConcreteDataType::boolean_datatype(), false),
|
||||
(
|
||||
common_query::native_histogram::native_histogram_value_type().clone(),
|
||||
false,
|
||||
),
|
||||
] {
|
||||
let schema = Schema::new(vec![
|
||||
ColumnSchema::new(
|
||||
|
||||
@@ -155,6 +155,7 @@ catalog = { workspace = true, features = ["testing"] }
|
||||
client = { workspace = true, features = ["testing"] }
|
||||
common-base.workspace = true
|
||||
common-meta = { workspace = true, features = ["testing"] }
|
||||
common-query = { workspace = true, features = ["testing"] }
|
||||
common-test-util.workspace = true
|
||||
criterion = "0.5"
|
||||
json5 = "0.4"
|
||||
|
||||
+331
-11
@@ -33,6 +33,7 @@ use arrow::datatypes::{
|
||||
};
|
||||
use common_grpc::precision::Precision;
|
||||
use common_memory_manager::MemoryGuard;
|
||||
use common_query::native_histogram;
|
||||
use common_query::prelude::{greptime_timestamp, greptime_value};
|
||||
use common_recordbatch::{RecordBatch, RecordBatches};
|
||||
use common_telemetry::{tracing, warn};
|
||||
@@ -458,6 +459,7 @@ pub fn recordbatches_to_timeseries(
|
||||
|
||||
for ts in &mut timeseries {
|
||||
ts.samples.sort_unstable_by_key(|s| s.timestamp);
|
||||
ts.histograms.sort_unstable_by_key(|h| h.timestamp);
|
||||
}
|
||||
|
||||
timeseries
|
||||
@@ -531,21 +533,26 @@ fn append_recordbatch_to_timeseries(
|
||||
),
|
||||
})?;
|
||||
|
||||
// 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(value_column_name)
|
||||
.with_context(|| error::InvalidPromRemoteReadQueryResultSnafu {
|
||||
msg: format!("missing value column '{value_column_name}' in query result"),
|
||||
})?;
|
||||
let field_column = field_column
|
||||
.as_primitive_opt::<Float64Type>()
|
||||
.with_context(|| error::InvalidPromRemoteReadQueryResultSnafu {
|
||||
let scalar_column = field_column.as_primitive_opt::<Float64Type>();
|
||||
let histogram_column = recordbatch
|
||||
.schema
|
||||
.column_schema_by_name(value_column_name)
|
||||
.filter(|column| native_histogram::is_native_histogram_value_type(&column.data_type))
|
||||
.and_then(|_| field_column.as_struct_opt());
|
||||
ensure!(
|
||||
scalar_column.is_some() || histogram_column.is_some(),
|
||||
error::InvalidPromRemoteReadQueryResultSnafu {
|
||||
msg: format!(
|
||||
"Expect value column of datatype Float64, actual {:?}",
|
||||
"Expect value column of datatype Float64 or canonical native histogram Struct, actual {:?}",
|
||||
field_column.data_type()
|
||||
),
|
||||
})?;
|
||||
}
|
||||
);
|
||||
|
||||
let columns = label_columns(&recordbatch, timestamp_column_name, value_column_name)?;
|
||||
let mut previous_timeseries: Option<usize> = None;
|
||||
@@ -577,11 +584,25 @@ fn append_recordbatch_to_timeseries(
|
||||
continue;
|
||||
}
|
||||
|
||||
let value = field_column.value(row);
|
||||
let timestamp = ts_column.value(row);
|
||||
let sample = Sample { value, timestamp };
|
||||
|
||||
timeseries[timeseries_index].samples.push(sample);
|
||||
if let Some(column) = scalar_column {
|
||||
timeseries[timeseries_index].samples.push(Sample {
|
||||
value: column.value(row),
|
||||
timestamp,
|
||||
});
|
||||
} else if let Some(column) = histogram_column {
|
||||
let histogram =
|
||||
native_histogram::decode_native_histogram_for_remote_read(column, row, timestamp)
|
||||
.map_err(|error| {
|
||||
error::InvalidPromRemoteReadQueryResultSnafu {
|
||||
msg: error.to_string(),
|
||||
}
|
||||
.build()
|
||||
})?;
|
||||
if let Some(histogram) = histogram {
|
||||
timeseries[timeseries_index].histograms.push(histogram);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Ok(())
|
||||
@@ -1540,6 +1561,305 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_native_histogram_remote_read_wire_fixture() {
|
||||
use api::greptime_proto::io::prometheus::write::v2::{
|
||||
BucketSpan, Histogram,
|
||||
histogram::{Count, ZeroCount},
|
||||
};
|
||||
use api::prom_store::remote::{QueryResult, ReadResponse};
|
||||
use common_query::native_histogram::testing::histogram_array;
|
||||
use prost::Message;
|
||||
|
||||
let histograms = vec![
|
||||
Histogram {
|
||||
count: Some(Count::CountInt((1 << 53) + 13)),
|
||||
zero_count: Some(ZeroCount::ZeroCountInt((1 << 53) + 1)),
|
||||
sum: 10.0,
|
||||
schema: 1,
|
||||
zero_threshold: 0.001,
|
||||
reset_hint: 2,
|
||||
start_timestamp: 123,
|
||||
positive_spans: vec![
|
||||
BucketSpan {
|
||||
offset: -2,
|
||||
length: 2,
|
||||
},
|
||||
BucketSpan {
|
||||
offset: 3,
|
||||
length: 1,
|
||||
},
|
||||
],
|
||||
positive_deltas: vec![1, 2, -1],
|
||||
negative_spans: vec![
|
||||
BucketSpan {
|
||||
offset: -3,
|
||||
length: 1,
|
||||
},
|
||||
BucketSpan {
|
||||
offset: 2,
|
||||
length: 2,
|
||||
},
|
||||
],
|
||||
negative_deltas: vec![1, 2, -1],
|
||||
..Default::default()
|
||||
},
|
||||
Histogram {
|
||||
count: Some(Count::CountFloat(6.0)),
|
||||
zero_count: Some(ZeroCount::ZeroCountFloat(0.5)),
|
||||
sum: 20.0,
|
||||
schema: 2,
|
||||
zero_threshold: 0.002,
|
||||
reset_hint: 3,
|
||||
positive_spans: vec![BucketSpan {
|
||||
offset: 3,
|
||||
length: 2,
|
||||
}],
|
||||
positive_counts: vec![2.0, 3.5],
|
||||
..Default::default()
|
||||
},
|
||||
Histogram {
|
||||
count: Some(Count::CountInt(1)),
|
||||
sum: 1.0,
|
||||
schema: -53,
|
||||
custom_values: vec![1.0, 2.0],
|
||||
positive_spans: vec![BucketSpan {
|
||||
offset: 0,
|
||||
length: 1,
|
||||
}],
|
||||
positive_deltas: vec![1],
|
||||
..Default::default()
|
||||
},
|
||||
Histogram {
|
||||
count: Some(Count::CountFloat(3.5)),
|
||||
sum: 4.0,
|
||||
schema: -53,
|
||||
custom_values: vec![1.0, 2.0],
|
||||
reset_hint: 3,
|
||||
positive_spans: vec![BucketSpan {
|
||||
offset: 0,
|
||||
length: 2,
|
||||
}],
|
||||
positive_counts: vec![1.0, 2.5],
|
||||
..Default::default()
|
||||
},
|
||||
Histogram {
|
||||
count: Some(Count::CountInt(0)),
|
||||
..Default::default()
|
||||
},
|
||||
Histogram {
|
||||
count: Some(Count::CountFloat(0.0)),
|
||||
..Default::default()
|
||||
},
|
||||
Histogram {
|
||||
count: Some(Count::CountInt(8)),
|
||||
sum: f64::from_bits(0x7ff8_0000_0000_0123),
|
||||
positive_spans: vec![BucketSpan {
|
||||
offset: 0,
|
||||
length: 2,
|
||||
}],
|
||||
positive_deltas: vec![3, -1],
|
||||
..Default::default()
|
||||
},
|
||||
Histogram {
|
||||
count: Some(Count::CountInt(0)),
|
||||
sum: f64::from_bits(0x7ff0_0000_0000_0002),
|
||||
..Default::default()
|
||||
},
|
||||
Histogram {
|
||||
count: Some(Count::CountInt(i64::MAX as u64)),
|
||||
zero_count: Some(ZeroCount::ZeroCountInt(i64::MAX as u64)),
|
||||
..Default::default()
|
||||
},
|
||||
];
|
||||
let values = histograms
|
||||
.iter()
|
||||
.map(|h| Some(native_histogram::encode_native_histogram(h).unwrap()))
|
||||
.collect::<Vec<_>>();
|
||||
let array = histogram_array(&values);
|
||||
let buckets = array
|
||||
.column_by_name("positive_buckets_i64")
|
||||
.unwrap()
|
||||
.as_list::<i32>()
|
||||
.value(0);
|
||||
assert_eq!(
|
||||
buckets
|
||||
.as_primitive::<arrow::datatypes::Int64Type>()
|
||||
.values()
|
||||
.as_ref(),
|
||||
&[1, 3, 2]
|
||||
);
|
||||
let arrow_schema = Arc::new(ArrowSchema::new(vec![
|
||||
Field::new(
|
||||
"ts",
|
||||
ArrowDataType::Timestamp(TimeUnit::Millisecond, None),
|
||||
false,
|
||||
),
|
||||
Field::new("histogram", array.data_type().clone(), true),
|
||||
]));
|
||||
let schema = Arc::new(Schema::try_from(arrow_schema.clone()).unwrap());
|
||||
let batch = DfRecordBatch::try_new(
|
||||
arrow_schema,
|
||||
vec![
|
||||
Arc::new(TimestampMillisecondArray::from(
|
||||
(1..=9).map(|i| i * 1000).collect::<Vec<_>>(),
|
||||
)),
|
||||
Arc::new(array),
|
||||
],
|
||||
)
|
||||
.unwrap();
|
||||
let batches = RecordBatches::try_new(
|
||||
schema.clone(),
|
||||
vec![RecordBatch::from_df_record_batch(schema, batch)],
|
||||
)
|
||||
.unwrap();
|
||||
let response = ReadResponse {
|
||||
results: vec![QueryResult {
|
||||
timeseries: recordbatches_to_timeseries(
|
||||
"native_histogram_remote_read",
|
||||
"ts",
|
||||
"histogram",
|
||||
batches,
|
||||
)
|
||||
.unwrap(),
|
||||
}],
|
||||
};
|
||||
let body = snappy_compress(&response.encode_to_vec()).unwrap();
|
||||
let mut decoded =
|
||||
ReadResponse::decode(snappy_decompress(&body).unwrap().as_slice()).unwrap();
|
||||
let mut expected = ReadResponse::decode(
|
||||
include_bytes!(
|
||||
"../../../tests/compatibility/wire/native_histogram_remote_read/response.pb"
|
||||
)
|
||||
.as_slice(),
|
||||
)
|
||||
.unwrap();
|
||||
let actual_histograms = &mut decoded.results[0].timeseries[0].histograms;
|
||||
assert_eq!(actual_histograms[0].positive_deltas, vec![1, 2, -1]);
|
||||
assert_eq!(actual_histograms[0].negative_deltas, vec![1, 2, -1]);
|
||||
assert_eq!(actual_histograms[6].positive_deltas, vec![3, -1]);
|
||||
for (actual, expected) in actual_histograms
|
||||
.iter_mut()
|
||||
.zip(&mut expected.results[0].timeseries[0].histograms)
|
||||
{
|
||||
assert_eq!(actual.sum.to_bits(), expected.sum.to_bits());
|
||||
actual.sum = 0.0;
|
||||
expected.sum = 0.0;
|
||||
}
|
||||
assert_eq!(decoded, expected);
|
||||
// The optional output is consumed by the pinned Prometheus iterator check.
|
||||
if let Some(path) = std::env::var_os("GREPTIME_NATIVE_HISTOGRAM_REMOTE_READ_OUTPUT") {
|
||||
std::fs::write(path, body).unwrap();
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_native_histogram_remote_read_groups_and_sorts_batches() {
|
||||
use api::greptime_proto::io::prometheus::write::v2::{Histogram, histogram::Count};
|
||||
use common_query::native_histogram::testing::histogram_array;
|
||||
|
||||
let integer = native_histogram::encode_native_histogram(&Histogram {
|
||||
count: Some(Count::CountInt(0)),
|
||||
sum: f64::from_bits(0x7ff0_0000_0000_0002),
|
||||
..Default::default()
|
||||
})
|
||||
.unwrap();
|
||||
let float = native_histogram::encode_native_histogram(&Histogram {
|
||||
count: Some(Count::CountFloat(0.0)),
|
||||
..Default::default()
|
||||
})
|
||||
.unwrap();
|
||||
let histograms = histogram_array(&[Some(integer), Some(float.clone()), None, Some(float)]);
|
||||
let arrow_schema = Arc::new(ArrowSchema::new(vec![
|
||||
Field::new(
|
||||
"event_time",
|
||||
ArrowDataType::Timestamp(TimeUnit::Microsecond, None),
|
||||
true,
|
||||
),
|
||||
Field::new("legacy_histogram", histograms.data_type().clone(), true),
|
||||
Field::new_dictionary("instance", ArrowDataType::UInt32, ArrowDataType::Utf8, true),
|
||||
]));
|
||||
let schema = Arc::new(Schema::try_from(arrow_schema.clone()).unwrap());
|
||||
let batch = |timestamps: Vec<Option<i64>>, labels: Vec<Option<u32>>| {
|
||||
RecordBatch::from_df_record_batch(
|
||||
schema.clone(),
|
||||
DfRecordBatch::try_new(
|
||||
arrow_schema.clone(),
|
||||
vec![
|
||||
Arc::new(arrow::array::TimestampMicrosecondArray::from(timestamps)),
|
||||
Arc::new(histograms.clone()),
|
||||
Arc::new(DictionaryArray::<UInt32Type>::new(
|
||||
UInt32Array::from(labels),
|
||||
Arc::new(StringArray::from(vec![Some("b"), Some("a"), None])),
|
||||
)),
|
||||
],
|
||||
)
|
||||
.unwrap(),
|
||||
)
|
||||
};
|
||||
let batches = vec![
|
||||
batch(
|
||||
vec![Some(4000), Some(3000), Some(5000), Some(-1001)],
|
||||
vec![Some(0), None, Some(0), Some(0)],
|
||||
),
|
||||
batch(
|
||||
vec![Some(2000), Some(1000), Some(6000), None],
|
||||
vec![Some(0), Some(2), Some(1), Some(0)],
|
||||
),
|
||||
];
|
||||
let series = recordbatches_to_timeseries(
|
||||
"latency",
|
||||
"event_time",
|
||||
"legacy_histogram",
|
||||
RecordBatches::try_new(schema.clone(), batches).unwrap(),
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(series.len(), 3);
|
||||
assert_eq!(
|
||||
series[0].labels,
|
||||
vec![new_label(
|
||||
METRIC_NAME_LABEL.to_string(),
|
||||
"latency".to_string()
|
||||
)]
|
||||
);
|
||||
assert_eq!(
|
||||
series[0]
|
||||
.histograms
|
||||
.iter()
|
||||
.map(|h| h.timestamp)
|
||||
.collect::<Vec<_>>(),
|
||||
vec![1, 3]
|
||||
);
|
||||
assert_eq!(series[1].labels[1].value, "a");
|
||||
assert!(series[1].histograms.is_empty());
|
||||
assert_eq!(series[2].labels[1].value, "b");
|
||||
assert_eq!(
|
||||
series[2]
|
||||
.histograms
|
||||
.iter()
|
||||
.map(|h| h.timestamp)
|
||||
.collect::<Vec<_>>(),
|
||||
vec![-2, 2, 4]
|
||||
);
|
||||
assert!(matches!(
|
||||
series[2].histograms[0].count,
|
||||
Some(api::prom_store::remote::histogram::Count::CountFloat(0.0))
|
||||
));
|
||||
for histogram in &series[2].histograms[1..] {
|
||||
assert!(matches!(
|
||||
histogram.count,
|
||||
Some(api::prom_store::remote::histogram::Count::CountInt(0))
|
||||
));
|
||||
assert_eq!(histogram.sum.to_bits(), 0x7ff0_0000_0000_0002);
|
||||
}
|
||||
assert!(
|
||||
series
|
||||
.iter()
|
||||
.all(|s| s.samples.is_empty()
|
||||
&& s.labels.iter().all(|l| l.name != "legacy_histogram"))
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_recordbatches_to_timeseries() {
|
||||
let schema = Arc::new(Schema::new(vec![
|
||||
|
||||
@@ -16,8 +16,8 @@ use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
|
||||
use api::greptime_proto::io::prometheus::write::v2::{
|
||||
Request as RemoteWriteV2Request, Sample as RemoteWriteV2Sample,
|
||||
TimeSeries as RemoteWriteV2TimeSeries,
|
||||
BucketSpan as RemoteWriteV2BucketSpan, Request as RemoteWriteV2Request,
|
||||
Sample as RemoteWriteV2Sample, TimeSeries as RemoteWriteV2TimeSeries,
|
||||
};
|
||||
use api::prom_store::remote::{
|
||||
LabelMatcher, Query, QueryResult, ReadRequest, ReadResponse, WriteRequest,
|
||||
@@ -28,6 +28,7 @@ use async_trait::async_trait;
|
||||
use axum::Router;
|
||||
use axum::http::HeaderMap;
|
||||
use common_query::Output;
|
||||
use common_query::native_histogram::CUSTOM_BUCKETS_SCHEMA;
|
||||
use common_query::prelude::{
|
||||
GREPTIME_PHYSICAL_TABLE, greptime_native_histogram, greptime_timestamp, greptime_value,
|
||||
};
|
||||
@@ -317,6 +318,59 @@ async fn test_prometheus_remote_write_v2_convert_error_has_written_headers() {
|
||||
assert!(write_rx.try_recv().is_err());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_prometheus_remote_write_v2_rejects_incompatible_histogram_before_writes() {
|
||||
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_write_capture(read_tx, write_tx);
|
||||
let client = TestClient::new(app).await;
|
||||
|
||||
for (schema, offset, message) in [
|
||||
(0, i32::MIN, "initial span offset i32::MIN"),
|
||||
(
|
||||
CUSTOM_BUCKETS_SCHEMA,
|
||||
2,
|
||||
"span extent exceeds custom_values",
|
||||
),
|
||||
] {
|
||||
let mut histogram = remote_write_v2::histogram(2000);
|
||||
histogram.schema = schema;
|
||||
histogram.positive_spans = vec![RemoteWriteV2BucketSpan { offset, length: 0 }];
|
||||
let write_request = RemoteWriteV2Request {
|
||||
symbols: vec![
|
||||
String::new(),
|
||||
prom_store::METRIC_NAME_LABEL.to_string(),
|
||||
"http_requests_total".to_string(),
|
||||
"http_request_duration_seconds".to_string(),
|
||||
],
|
||||
timeseries: vec![
|
||||
RemoteWriteV2TimeSeries {
|
||||
labels_refs: vec![1, 2],
|
||||
samples: vec![RemoteWriteV2Sample {
|
||||
value: 42.0,
|
||||
timestamp: 1000,
|
||||
start_timestamp: 0,
|
||||
}],
|
||||
..Default::default()
|
||||
},
|
||||
RemoteWriteV2TimeSeries {
|
||||
labels_refs: vec![1, 3],
|
||||
histograms: vec![remote_write_v2::histogram(1000), histogram],
|
||||
..Default::default()
|
||||
},
|
||||
],
|
||||
};
|
||||
|
||||
let result = post_remote_write_v2(&client, &write_request).await;
|
||||
|
||||
assert_eq!(result.status(), 400);
|
||||
assert_remote_write_v2_written_headers(&result.headers(), "0");
|
||||
assert!(result.text().await.contains(message));
|
||||
assert!(write_rx.try_recv().is_err());
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_prometheus_remote_write_v2_write_error_has_partial_written_headers() {
|
||||
common_telemetry::init_default_ut_logging();
|
||||
|
||||
@@ -2838,6 +2838,32 @@ WITH(
|
||||
.remove(0)
|
||||
.unwrap();
|
||||
}
|
||||
let series =
|
||||
native_histogram_remote_read(&instance, "delta_distribution", ctx.clone()).await;
|
||||
assert_eq!(series.len(), 1);
|
||||
assert!(
|
||||
series[0]
|
||||
.labels
|
||||
.iter()
|
||||
.any(|l| l.name == "otlp_aggregation_temporality" && l.value == "delta")
|
||||
);
|
||||
assert_eq!(series[0].histograms.len(), 3);
|
||||
for (histogram, (timestamp, count)) in
|
||||
series[0]
|
||||
.histograms
|
||||
.iter()
|
||||
.zip([(60_000, 10), (120_000, 20), (180_000, 15)])
|
||||
{
|
||||
assert_eq!(histogram.timestamp, timestamp);
|
||||
assert_eq!(
|
||||
histogram.count,
|
||||
Some(api::prom_store::remote::histogram::Count::CountInt(count))
|
||||
);
|
||||
assert_eq!(histogram.positive_deltas, vec![count as i64]);
|
||||
assert_eq!(histogram.positive_spans[0].offset, 1);
|
||||
assert_eq!(histogram.sum, count as f64 * 1.5);
|
||||
assert_eq!(histogram.reset_hint, 3);
|
||||
}
|
||||
for (expression, expected) in [
|
||||
(
|
||||
"histogram_count(sum_over_time(delta_distribution[3m]))",
|
||||
@@ -3139,6 +3165,42 @@ WITH(
|
||||
}
|
||||
}
|
||||
|
||||
async fn native_histogram_remote_read(
|
||||
instance: &Arc<Instance>,
|
||||
table: &str,
|
||||
ctx: Arc<QueryContext>,
|
||||
) -> Vec<api::prom_store::remote::TimeSeries> {
|
||||
use api::prom_store::remote::{LabelMatcher, Query, ReadRequest, ReadResponse};
|
||||
use prost::Message;
|
||||
use servers::query_handler::PromStoreProtocolHandler;
|
||||
|
||||
let response = instance
|
||||
.read(
|
||||
ReadRequest {
|
||||
queries: vec![Query {
|
||||
start_timestamp_ms: 0,
|
||||
end_timestamp_ms: 200_000,
|
||||
matchers: vec![LabelMatcher {
|
||||
name: "__name__".to_string(),
|
||||
value: table.to_string(),
|
||||
r#type: 0,
|
||||
}],
|
||||
..Default::default()
|
||||
}],
|
||||
..Default::default()
|
||||
},
|
||||
ctx,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let body = servers::prom_store::snappy_decompress(&response.body).unwrap();
|
||||
ReadResponse::decode(body.as_slice())
|
||||
.unwrap()
|
||||
.results
|
||||
.remove(0)
|
||||
.timeseries
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
async fn test_otlp_delta_exponential_storage_precision_and_replacement() {
|
||||
use datatypes::arrow::datatypes::Int64Type;
|
||||
@@ -3194,6 +3256,24 @@ WITH(
|
||||
.remove(0)
|
||||
.unwrap();
|
||||
}
|
||||
let series =
|
||||
native_histogram_remote_read(instance, "interval", ctx.clone()).await;
|
||||
assert_eq!(series.len(), 1);
|
||||
assert!(series[0].labels.iter().any(|l| l.name == "otlp_aggregation_temporality" && l.value == "delta"));
|
||||
let histogram = &series[0].histograms[0];
|
||||
assert_eq!(histogram.timestamp, 60_000);
|
||||
assert_eq!(
|
||||
histogram.count,
|
||||
Some(api::prom_store::remote::histogram::Count::CountInt(count))
|
||||
);
|
||||
assert_eq!(
|
||||
histogram.zero_count,
|
||||
Some(api::prom_store::remote::histogram::ZeroCount::ZeroCountInt(
|
||||
count
|
||||
))
|
||||
);
|
||||
assert_eq!(histogram.reset_hint, 3);
|
||||
assert_eq!(histogram.sum, 0.0);
|
||||
let output = instance
|
||||
.do_query(
|
||||
"SELECT greptime_timestamp,greptime_native_histogram FROM interval",
|
||||
|
||||
@@ -778,6 +778,329 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
async fn test_standalone_native_histogram_remote_read() {
|
||||
let standalone =
|
||||
GreptimeDbStandaloneBuilder::new("standalone_native_histogram_remote_read")
|
||||
.build()
|
||||
.await;
|
||||
check_native_histogram_remote_read(standalone.fe_instance()).await;
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
async fn test_distributed_native_histogram_remote_read() {
|
||||
let cluster =
|
||||
tests::create_distributed_instance("distributed_native_histogram_remote_read").await;
|
||||
check_native_histogram_remote_read(&cluster.frontend()).await;
|
||||
}
|
||||
|
||||
async fn check_native_histogram_remote_read(instance: &Arc<Instance>) {
|
||||
use api::greptime_proto::io::prometheus::write::v2::{
|
||||
Histogram,
|
||||
histogram::{Count, ZeroCount},
|
||||
};
|
||||
use common_query::native_histogram::{
|
||||
encode_native_histogram, native_histogram_column_schema,
|
||||
};
|
||||
use servers::prom_remote_write::v2::test_util as remote_write_v2;
|
||||
|
||||
let ctx = Arc::new(QueryContext::with(DEFAULT_CATALOG_NAME, "prometheus"));
|
||||
instance
|
||||
.do_query("CREATE DATABASE prometheus", ctx.clone())
|
||||
.await
|
||||
.remove(0)
|
||||
.unwrap();
|
||||
let server = HttpServerBuilder::new(HttpOptions::default())
|
||||
.with_prom_handler(instance.clone(), None, true, Default::default(), None)
|
||||
.build();
|
||||
let client = TestClient::new(server.build(server.make_app()).unwrap()).await;
|
||||
let large = (1u64 << 53) + 1;
|
||||
let histograms = vec![
|
||||
Histogram {
|
||||
count: Some(Count::CountInt(6)),
|
||||
timestamp: 1000,
|
||||
positive_spans: vec![api::greptime_proto::io::prometheus::write::v2::BucketSpan {
|
||||
offset: 0,
|
||||
length: 3,
|
||||
}],
|
||||
positive_deltas: vec![1, 2, -1],
|
||||
sum: 10.0,
|
||||
..Default::default()
|
||||
},
|
||||
Histogram {
|
||||
count: Some(Count::CountInt(large)),
|
||||
zero_count: Some(ZeroCount::ZeroCountInt(large)),
|
||||
timestamp: 2000,
|
||||
..Default::default()
|
||||
},
|
||||
Histogram {
|
||||
count: Some(Count::CountFloat(6.0)),
|
||||
zero_count: Some(ZeroCount::ZeroCountFloat(0.5)),
|
||||
positive_spans: vec![api::greptime_proto::io::prometheus::write::v2::BucketSpan {
|
||||
offset: 3,
|
||||
length: 2,
|
||||
}],
|
||||
positive_counts: vec![2.0, 3.5],
|
||||
timestamp: 3000,
|
||||
..Default::default()
|
||||
},
|
||||
Histogram {
|
||||
count: Some(Count::CountInt(0)),
|
||||
sum: f64::from_bits(0x7ff0_0000_0000_0002),
|
||||
timestamp: 4000,
|
||||
..Default::default()
|
||||
},
|
||||
];
|
||||
let request = remote_write_v2::request_with_labels_and_histograms(
|
||||
vec![(prom_store::METRIC_NAME_LABEL, "latency"), ("job", "api")],
|
||||
histograms,
|
||||
);
|
||||
let response = client
|
||||
.post("/v1/prometheus/write?db=prometheus")
|
||||
.header(
|
||||
"content-type",
|
||||
"application/x-protobuf;proto=io.prometheus.write.v2.Request",
|
||||
)
|
||||
.header("content-encoding", "snappy")
|
||||
.body(prom_store::snappy_compress(&request.encode_to_vec()).unwrap())
|
||||
.send()
|
||||
.await;
|
||||
assert_eq!(response.status(), 204, "{}", response.text().await);
|
||||
let (scalar, _) = to_grpc_row_insert_requests(&WriteRequest {
|
||||
timeseries: vec![prom_store::mock_timeseries()[0].clone()],
|
||||
..Default::default()
|
||||
})
|
||||
.unwrap();
|
||||
instance.write(scalar, ctx.clone(), true).await.unwrap();
|
||||
|
||||
for flushed in [false, true] {
|
||||
if flushed {
|
||||
instance
|
||||
.do_query("ADMIN FLUSH_TABLE('latency')", ctx.clone())
|
||||
.await
|
||||
.remove(0)
|
||||
.unwrap();
|
||||
}
|
||||
for accepted_response_types in [vec![], vec![0], vec![1, 0]] {
|
||||
let mut query = remote_query("latency");
|
||||
query.end_timestamp_ms = 4000;
|
||||
query.matchers.push(LabelMatcher {
|
||||
name: "job".to_string(),
|
||||
value: "api".to_string(),
|
||||
r#type: MatcherType::Eq as i32,
|
||||
});
|
||||
let mut empty = query.clone();
|
||||
empty.start_timestamp_ms = 5000;
|
||||
empty.end_timestamp_ms = 6000;
|
||||
let request = ReadRequest {
|
||||
queries: vec![
|
||||
remote_query("metric1"),
|
||||
query,
|
||||
remote_query("missing_metric"),
|
||||
empty,
|
||||
],
|
||||
accepted_response_types,
|
||||
};
|
||||
let response = client
|
||||
.post("/v1/prometheus/read?db=prometheus")
|
||||
.body(prom_store::snappy_compress(&request.encode_to_vec()).unwrap())
|
||||
.send()
|
||||
.await;
|
||||
assert_eq!(response.status(), 200, "{}", response.text().await);
|
||||
assert_eq!(response.headers()["content-type"], "application/x-protobuf");
|
||||
assert_eq!(response.headers()["content-encoding"], "snappy");
|
||||
let body = prom_store::snappy_decompress(&response.bytes().await).unwrap();
|
||||
let response = ReadResponse::decode(body.as_slice()).unwrap();
|
||||
assert_eq!(response.results.len(), 4);
|
||||
assert!(!response.results[0].timeseries[0].samples.is_empty());
|
||||
assert!(response.results[2].timeseries.is_empty());
|
||||
assert!(response.results[3].timeseries.is_empty());
|
||||
let series = &response.results[1].timeseries[0];
|
||||
assert_eq!(
|
||||
series.labels,
|
||||
vec![
|
||||
Label {
|
||||
name: "__name__".to_string(),
|
||||
value: "latency".to_string()
|
||||
},
|
||||
Label {
|
||||
name: "job".to_string(),
|
||||
value: "api".to_string()
|
||||
}
|
||||
]
|
||||
);
|
||||
assert_eq!(series.histograms.len(), 4);
|
||||
assert_eq!(series.histograms[0].positive_deltas, vec![1, 2, -1]);
|
||||
assert_eq!(series.histograms[2].positive_counts, vec![2.0, 3.5]);
|
||||
assert_eq!(
|
||||
series.histograms[1].count,
|
||||
Some(api::prom_store::remote::histogram::Count::CountInt(large))
|
||||
);
|
||||
assert_eq!(
|
||||
series.histograms[1].zero_count,
|
||||
Some(api::prom_store::remote::histogram::ZeroCount::ZeroCountInt(
|
||||
large
|
||||
))
|
||||
);
|
||||
assert_eq!(
|
||||
series.histograms[2].count,
|
||||
Some(api::prom_store::remote::histogram::Count::CountFloat(6.0))
|
||||
);
|
||||
assert_eq!(series.histograms[3].sum.to_bits(), 0x7ff0_0000_0000_0002);
|
||||
}
|
||||
}
|
||||
let response = client
|
||||
.post("/v1/prometheus/read?db=prometheus")
|
||||
.body(
|
||||
prom_store::snappy_compress(
|
||||
&ReadRequest {
|
||||
queries: vec![remote_query("latency")],
|
||||
accepted_response_types: vec![1],
|
||||
}
|
||||
.encode_to_vec(),
|
||||
)
|
||||
.unwrap(),
|
||||
)
|
||||
.send()
|
||||
.await;
|
||||
assert!(!response.status().is_success());
|
||||
assert!(
|
||||
response
|
||||
.text()
|
||||
.await
|
||||
.contains("does not support any of the requested response types")
|
||||
);
|
||||
|
||||
// Typed inserts supply sub-millisecond pre-epoch times that neither OTLP
|
||||
// nor millisecond Remote Write can express, and use a legacy field name.
|
||||
for (table, datatype, timestamp) in [
|
||||
(
|
||||
"legacy_us",
|
||||
ColumnDataType::TimestampMicrosecond,
|
||||
ValueData::TimestampMicrosecondValue(-1001),
|
||||
),
|
||||
(
|
||||
"legacy_ns",
|
||||
ColumnDataType::TimestampNanosecond,
|
||||
ValueData::TimestampNanosecondValue(-1_001_000),
|
||||
),
|
||||
] {
|
||||
let mut histogram_schema = native_histogram_column_schema().unwrap();
|
||||
histogram_schema.column_name = "payload".to_string();
|
||||
let histogram = encode_native_histogram(&Histogram {
|
||||
count: Some(Count::CountInt(large)),
|
||||
zero_count: Some(ZeroCount::ZeroCountInt(large)),
|
||||
..Default::default()
|
||||
})
|
||||
.unwrap();
|
||||
instance
|
||||
.write(
|
||||
RowInsertRequests {
|
||||
inserts: vec![RowInsertRequest {
|
||||
table_name: table.to_string(),
|
||||
rows: Some(Rows {
|
||||
schema: vec![
|
||||
api::v1::ColumnSchema {
|
||||
column_name: "event_time".to_string(),
|
||||
datatype: datatype as i32,
|
||||
semantic_type: SemanticType::Timestamp as i32,
|
||||
..Default::default()
|
||||
},
|
||||
histogram_schema,
|
||||
],
|
||||
rows: vec![Row {
|
||||
values: vec![
|
||||
Value {
|
||||
value_data: Some(timestamp),
|
||||
},
|
||||
Value {
|
||||
value_data: Some(histogram),
|
||||
},
|
||||
],
|
||||
}],
|
||||
}),
|
||||
}],
|
||||
},
|
||||
ctx.clone(),
|
||||
false,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
for flushed in [false, true] {
|
||||
if flushed {
|
||||
instance
|
||||
.do_query(&format!("ADMIN FLUSH_TABLE('{table}')"), ctx.clone())
|
||||
.await
|
||||
.remove(0)
|
||||
.unwrap();
|
||||
}
|
||||
let mut query = remote_query(table);
|
||||
query.start_timestamp_ms = -2;
|
||||
query.end_timestamp_ms = 0;
|
||||
let response = post_remote_read(&client, vec![query]).await;
|
||||
assert_eq!(response.status(), 200, "{}", response.text().await);
|
||||
let body = prom_store::snappy_decompress(&response.bytes().await).unwrap();
|
||||
let response = ReadResponse::decode(body.as_slice()).unwrap();
|
||||
let series = &response.results[0].timeseries[0];
|
||||
assert_eq!(series.labels.len(), 1);
|
||||
assert_eq!(series.histograms[0].timestamp, -2);
|
||||
assert_eq!(
|
||||
series.histograms[0].count,
|
||||
Some(api::prom_store::remote::histogram::Count::CountInt(large))
|
||||
);
|
||||
}
|
||||
}
|
||||
// Bypass ingestion's histogram codec to exercise whole-request failure.
|
||||
let ValueData::StructValue(mut malformed) =
|
||||
encode_native_histogram(&Histogram::default()).unwrap()
|
||||
else {
|
||||
unreachable!()
|
||||
};
|
||||
malformed.items[14].value_data = Some(ValueData::F64Value(0.0));
|
||||
instance
|
||||
.write(
|
||||
RowInsertRequests {
|
||||
inserts: vec![RowInsertRequest {
|
||||
table_name: "malformed_histogram".to_string(),
|
||||
rows: Some(Rows {
|
||||
schema: vec![
|
||||
api::v1::ColumnSchema {
|
||||
column_name: "ts".to_string(),
|
||||
datatype: ColumnDataType::TimestampMillisecond as i32,
|
||||
semantic_type: SemanticType::Timestamp as i32,
|
||||
..Default::default()
|
||||
},
|
||||
native_histogram_column_schema().unwrap(),
|
||||
],
|
||||
rows: vec![Row {
|
||||
values: vec![
|
||||
Value {
|
||||
value_data: Some(ValueData::TimestampMillisecondValue(
|
||||
1000,
|
||||
)),
|
||||
},
|
||||
Value {
|
||||
value_data: Some(ValueData::StructValue(malformed)),
|
||||
},
|
||||
],
|
||||
}],
|
||||
}),
|
||||
}],
|
||||
},
|
||||
ctx.clone(),
|
||||
false,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
for queries in [
|
||||
vec![remote_query("metric1"), remote_query("malformed_histogram")],
|
||||
vec![remote_query("malformed_histogram"), remote_query("metric1")],
|
||||
] {
|
||||
let response = post_remote_read(&client, queries).await;
|
||||
assert_eq!(response.status(), 500);
|
||||
}
|
||||
}
|
||||
|
||||
fn remote_query(metric: &str) -> Query {
|
||||
Query {
|
||||
start_timestamp_ms: 1000,
|
||||
|
||||
@@ -4262,6 +4262,27 @@ pub async fn test_prometheus_remote_write_v2_native_histogram(store_type: Storag
|
||||
],
|
||||
..Default::default()
|
||||
});
|
||||
let large = (1u64 << 53) + 1;
|
||||
write_request.timeseries[1].histograms.extend([
|
||||
Histogram {
|
||||
count: Some(Count::CountInt(large)),
|
||||
zero_count: Some(ZeroCount::ZeroCountInt(large)),
|
||||
timestamp: 5000,
|
||||
..Default::default()
|
||||
},
|
||||
Histogram {
|
||||
count: Some(Count::CountInt(0)),
|
||||
sum: f64::from_bits(0x7ff0_0000_0000_0002),
|
||||
timestamp: 6000,
|
||||
..Default::default()
|
||||
},
|
||||
Histogram {
|
||||
count: Some(Count::CountFloat(0.0)),
|
||||
sum: f64::from_bits(0x7ff8_0000_0000_0123),
|
||||
timestamp: 7000,
|
||||
..Default::default()
|
||||
},
|
||||
]);
|
||||
let serialized_request = write_request.encode_to_vec();
|
||||
let compressed_request =
|
||||
prom_store::snappy_compress(&serialized_request).expect("failed to encode snappy");
|
||||
@@ -4278,7 +4299,7 @@ pub async fn test_prometheus_remote_write_v2_native_histogram(store_type: Storag
|
||||
.await;
|
||||
assert_eq!(res.status(), StatusCode::NO_CONTENT);
|
||||
let headers = res.headers();
|
||||
assert_remote_write_v2_written_headers_with_histograms(&headers, "1", "2");
|
||||
assert_remote_write_v2_written_headers_with_histograms(&headers, "1", "5");
|
||||
assert!(res.text().await.is_empty());
|
||||
|
||||
validate_data(
|
||||
@@ -4289,7 +4310,7 @@ pub async fn test_prometheus_remote_write_v2_native_histogram(store_type: Storag
|
||||
)
|
||||
.await;
|
||||
|
||||
let histogram_sql = "select greptime_timestamp, greptime_native_histogram, job, instance from remote_write_v2_latency_seconds order by greptime_timestamp;";
|
||||
let histogram_sql = "select greptime_timestamp, greptime_native_histogram, job, instance from remote_write_v2_latency_seconds where greptime_timestamp < 5000::TIMESTAMP order by greptime_timestamp;";
|
||||
let histogram_rows = "[[3000,{\"count_f64\":null,\"count_i64\":8,\"custom_values\":[],\"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_i64\":1,\"zero_threshold\":0.001},\"api\",\"localhost:9090\"],[4000,{\"count_f64\":6.0,\"count_i64\":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_i64\":null,\"zero_threshold\":0.002},\"api\",\"localhost:9090\"]]";
|
||||
validate_data(
|
||||
"prometheus_remote_write_v2_native_histogram_rows",
|
||||
@@ -4299,6 +4320,76 @@ pub async fn test_prometheus_remote_write_v2_native_histogram(store_type: Storag
|
||||
)
|
||||
.await;
|
||||
|
||||
for flushed in [false, true] {
|
||||
if flushed {
|
||||
flush_table(&client, "remote_write_v2_latency_seconds").await;
|
||||
}
|
||||
for accepted_response_types in [vec![], vec![0], vec![1, 0]] {
|
||||
let query = |name: &str| Query {
|
||||
start_timestamp_ms: 0,
|
||||
end_timestamp_ms: 10_000,
|
||||
matchers: vec![LabelMatcher {
|
||||
name: prom_store::METRIC_NAME_LABEL.to_string(),
|
||||
value: name.to_string(),
|
||||
r#type: 0,
|
||||
}],
|
||||
..Default::default()
|
||||
};
|
||||
let request = ReadRequest {
|
||||
queries: vec![
|
||||
query("remote_write_v2_sample_seconds"),
|
||||
query("missing_native_histogram"),
|
||||
query("remote_write_v2_latency_seconds"),
|
||||
],
|
||||
accepted_response_types,
|
||||
};
|
||||
let response = client
|
||||
.post("/v1/prometheus/read")
|
||||
.body(prom_store::snappy_compress(&request.encode_to_vec()).unwrap())
|
||||
.send()
|
||||
.await;
|
||||
assert_eq!(response.status(), StatusCode::OK);
|
||||
assert_eq!(response.headers()["content-type"], "application/x-protobuf");
|
||||
assert_eq!(response.headers()["content-encoding"], "snappy");
|
||||
let body = prom_store::snappy_decompress(&response.bytes().await).unwrap();
|
||||
let response = ReadResponse::decode(body.as_slice()).unwrap();
|
||||
assert_eq!(response.results.len(), 3);
|
||||
assert_eq!(
|
||||
response.results[0].timeseries[0].samples,
|
||||
vec![Sample {
|
||||
value: 5.0,
|
||||
timestamp: 2500
|
||||
}]
|
||||
);
|
||||
assert!(response.results[1].timeseries.is_empty());
|
||||
let series = &response.results[2].timeseries[0];
|
||||
assert!(series.samples.is_empty());
|
||||
assert_eq!(
|
||||
series
|
||||
.histograms
|
||||
.iter()
|
||||
.map(|h| h.timestamp)
|
||||
.collect::<Vec<_>>(),
|
||||
vec![3000, 4000, 5000, 6000, 7000]
|
||||
);
|
||||
assert_eq!(series.histograms[0].positive_deltas, vec![1, 2, -1]);
|
||||
assert_eq!(series.histograms[0].negative_deltas, vec![1]);
|
||||
assert_eq!(series.histograms[1].positive_counts, vec![2.0, 3.5]);
|
||||
assert_eq!(
|
||||
series.histograms[2].count,
|
||||
Some(api::prom_store::remote::histogram::Count::CountInt(large))
|
||||
);
|
||||
assert_eq!(
|
||||
series.histograms[2].zero_count,
|
||||
Some(api::prom_store::remote::histogram::ZeroCount::ZeroCountInt(
|
||||
large
|
||||
))
|
||||
);
|
||||
assert_eq!(series.histograms[3].sum.to_bits(), 0x7ff0_0000_0000_0002);
|
||||
assert_eq!(series.histograms[4].sum.to_bits(), 0x7ff8_0000_0000_0123);
|
||||
}
|
||||
}
|
||||
|
||||
// The native histogram column is a struct of nested lists, so its SST
|
||||
// encoding differs from its memtable representation.
|
||||
flush_table(&client, "remote_write_v2_latency_seconds").await;
|
||||
@@ -9133,9 +9224,17 @@ pub async fn test_otlp_exponential_histogram(store_type: StorageType) {
|
||||
)]
|
||||
};
|
||||
|
||||
let (app, mut guard) =
|
||||
setup_test_http_app_with_frontend(store_type, "test_otlp_exponential_histogram").await;
|
||||
let client = TestClient::new(app).await;
|
||||
let standalone = tests_integration::standalone::GreptimeDbStandaloneBuilder::new(
|
||||
"test_otlp_exponential_histogram",
|
||||
)
|
||||
.with_default_store_type(store_type)
|
||||
.build()
|
||||
.await;
|
||||
let server = build_test_prom_server(standalone.fe_instance().clone(), false, None)
|
||||
.with_otlp_handler(standalone.fe_instance().clone(), true)
|
||||
.build();
|
||||
let client = TestClient::new(server.build(server.make_app()).unwrap()).await;
|
||||
let mut guard = standalone.guard;
|
||||
let res = send_req(&client, headers(), "/v1/otlp/v1/metrics", body, false).await;
|
||||
assert_eq!(StatusCode::OK, res.status());
|
||||
let response = ExportMetricsServiceResponse::decode(res.bytes().await).unwrap();
|
||||
@@ -9158,6 +9257,42 @@ pub async fn test_otlp_exponential_histogram(store_type: StorageType) {
|
||||
)
|
||||
.await;
|
||||
|
||||
for flushed in [false, true] {
|
||||
if flushed {
|
||||
flush_table(&client, "otlp_exponential_latency").await;
|
||||
}
|
||||
let response = post_remote_read(
|
||||
&client,
|
||||
vec![Query {
|
||||
start_timestamp_ms: 0,
|
||||
end_timestamp_ms: 4000,
|
||||
matchers: vec![LabelMatcher {
|
||||
name: "__name__".to_string(),
|
||||
value: "otlp_exponential_latency".to_string(),
|
||||
r#type: 0,
|
||||
}],
|
||||
..Default::default()
|
||||
}],
|
||||
)
|
||||
.await;
|
||||
let histogram = &response.results[0].timeseries[0].histograms[0];
|
||||
assert_eq!(histogram.timestamp, 3000);
|
||||
assert_eq!(
|
||||
histogram.count,
|
||||
Some(api::prom_store::remote::histogram::Count::CountInt(4))
|
||||
);
|
||||
assert_eq!(
|
||||
histogram.zero_count,
|
||||
Some(api::prom_store::remote::histogram::ZeroCount::ZeroCountInt(
|
||||
1
|
||||
))
|
||||
);
|
||||
assert_eq!(histogram.positive_deltas, vec![1, 1]);
|
||||
assert_eq!(histogram.positive_spans[0].offset, 0);
|
||||
assert_eq!(histogram.reset_hint, 0);
|
||||
assert_eq!(histogram.sum, 8.0);
|
||||
}
|
||||
|
||||
for query in [
|
||||
"histogram_count(otlp_exponential_latency)",
|
||||
"histogram_sum(otlp_exponential_latency)",
|
||||
|
||||
@@ -15,6 +15,14 @@ read-preference defaulting and unrelated metadata preservation:
|
||||
cargo nextest run -p session -E 'test(test_legacy_query_context_wire_defaults_to_leader)'
|
||||
```
|
||||
|
||||
The [native histogram Remote Read fixture](wire/native_histogram_remote_read/README.md)
|
||||
checks the sampled encoder against independent Prometheus protobuf values,
|
||||
including exact integer counts, custom bounds, and NaN payloads:
|
||||
|
||||
```shell
|
||||
cargo nextest run -p servers -E 'test(native_histogram_remote_read_wire_fixture)'
|
||||
```
|
||||
|
||||
## Object store WAL fixtures
|
||||
|
||||
Frozen objects of the object store WAL are checked byte for byte by the format tests of
|
||||
|
||||
@@ -0,0 +1,184 @@
|
||||
# Native histogram sampled Remote Read wire compatibility
|
||||
|
||||
`response.pb` is a frozen `prompb.ReadResponse` generated from Prometheus
|
||||
revision `e471c7708c6b0b13e2050222bbd5a907b36f4113`, independently of
|
||||
`greptime-proto`. It contains one query result, one series named
|
||||
`native_histogram_remote_read`, and nine samples at 1000ms intervals.
|
||||
|
||||
| Timestamp (ms) | Independent expectation |
|
||||
| --- | --- |
|
||||
| 1000 | Integer count `2^53 + 13`, zero count `2^53 + 1`, both bucket sides `[1,3,2]`, wire deltas `[1,2,-1]` across span boundaries, schema 1, reset hint 2 |
|
||||
| 2000 | Float count 6, zero count 0.5, absolute buckets `[2,3.5]`, schema 2, gauge hint |
|
||||
| 3000 | Integer custom buckets `[1,2]`, one observation in `(-Inf,1]` |
|
||||
| 4000 | Float custom buckets `[1,2]`, absolute counts `[1,2.5]`, gauge hint |
|
||||
| 5000 | Empty integer histogram with explicit count and zero-count oneofs |
|
||||
| 6000 | Empty float histogram with explicit count and zero-count oneofs |
|
||||
| 7000 | Integer count 8, buckets `[3,2]`, wire deltas `[3,-1]`, ordinary NaN sum bits `0x7ff8000000000123` |
|
||||
| 8000 | Empty integer histogram, stale NaN sum bits `0x7ff0000000000002` |
|
||||
| 9000 | Integer count and zero count `i64::MAX` |
|
||||
|
||||
The 420-byte fixture has SHA-256 `f1414bfa6f0821e920a88e325964abfad0269b05b7c0643a84ca6e908729d1eb`.
|
||||
Generation used Go `go1.27.3 darwin/arm64`, the pinned Prometheus `go.mod` and
|
||||
`go.sum` (declared Go version `1.26.10`), and its generated `prompb` types.
|
||||
Direct wire dependencies were `github.com/gogo/protobuf v1.3.2` and
|
||||
`github.com/golang/snappy v1.0.0`. No local `protoc` or Greptime protobuf
|
||||
generator is involved.
|
||||
|
||||
## Automated regression
|
||||
|
||||
From the GreptimeDB checkout:
|
||||
|
||||
```shell
|
||||
cargo nextest run -p servers -E 'test(native_histogram_remote_read_wire_fixture)'
|
||||
```
|
||||
|
||||
This constructs stored arrays from independent Rust inputs, checks fixed
|
||||
absolute bucket counts, runs the actual record-batch encoder, Snappy-compresses
|
||||
and protobuf-decodes its output, and compares all fields with the upstream
|
||||
fixture. NaN payloads are compared by bits. It also checks fixed wire deltas;
|
||||
converting those deltas back with an upstream helper is not the expected-value
|
||||
generator. Tests never regenerate the frozen fixture.
|
||||
|
||||
## Reproduce and check with Prometheus
|
||||
|
||||
Use a scratch module to preserve the Prometheus checkout. Set the two checkout
|
||||
paths, create a scratch directory, and save the Go program below as `main.go`
|
||||
in that directory:
|
||||
|
||||
```shell
|
||||
PROMETHEUS_CHECKOUT=/path/to/prometheus
|
||||
GREPTIMEDB_CHECKOUT=/path/to/greptimedb
|
||||
NH_SCRATCH=$(mktemp -d)
|
||||
git -C "$PROMETHEUS_CHECKOUT" rev-parse HEAD
|
||||
cp "$PROMETHEUS_CHECKOUT/go.mod" "$PROMETHEUS_CHECKOUT/go.sum" "$NH_SCRATCH/"
|
||||
cd "$NH_SCRATCH"
|
||||
go mod edit -module=native-histogram-interop
|
||||
go mod edit -require=github.com/prometheus/prometheus@v0.0.0
|
||||
go mod edit -replace="github.com/prometheus/prometheus=$PROMETHEUS_CHECKOUT"
|
||||
# Save the Go program below as main.go here before running these commands.
|
||||
go run -mod=mod . generate "$NH_SCRATCH/response.pb"
|
||||
shasum -a 256 "$NH_SCRATCH/response.pb"
|
||||
cmp "$NH_SCRATCH/response.pb" "$GREPTIMEDB_CHECKOUT/tests/compatibility/wire/native_histogram_remote_read/response.pb"
|
||||
cd "$GREPTIMEDB_CHECKOUT"
|
||||
GREPTIME_NATIVE_HISTOGRAM_REMOTE_READ_OUTPUT="$NH_SCRATCH/greptime-response.snappy" \
|
||||
cargo nextest run -p servers -E 'test(native_histogram_remote_read_wire_fixture)'
|
||||
cd "$NH_SCRATCH"
|
||||
go run -mod=mod . check "$NH_SCRATCH/greptime-response.snappy"
|
||||
```
|
||||
|
||||
The check consumes `storage/remote.FromQueryResult` iterators to exhaustion,
|
||||
checks both iterator and series-set errors, integer/float types, exact integer
|
||||
counts, timestamps, spans, bucket placement, custom bounds, reset hints, and
|
||||
NaN bits. It calls both `AtHistogram` and `AtFloatHistogram` on integer custom
|
||||
samples, including a single finite-bound bucket. Intentional integer-to-float
|
||||
conversion is not expected to retain large-integer precision.
|
||||
|
||||
Verified on 2026-10-26 against the response emitted by the Rust test:
|
||||
`PASS: 9 histogram samples, both iterator accessors, no series or iterator errors`.
|
||||
|
||||
The tested baseline is the exact revision above, not a claimed minimum
|
||||
Prometheus release. This Go check is a manual implementation check; it is not
|
||||
part of the existing Go CI. The Rust regression is the automated guard.
|
||||
|
||||
```go
|
||||
package main
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"math"
|
||||
"os"
|
||||
"reflect"
|
||||
|
||||
"github.com/golang/snappy"
|
||||
"github.com/prometheus/prometheus/prompb"
|
||||
"github.com/prometheus/prometheus/storage/remote"
|
||||
"github.com/prometheus/prometheus/tsdb/chunkenc"
|
||||
)
|
||||
|
||||
func expected() *prompb.ReadResponse {
|
||||
hs := []prompb.Histogram{
|
||||
{Count: &prompb.Histogram_CountInt{CountInt: (1<<53)+13}, ZeroCount: &prompb.Histogram_ZeroCountInt{ZeroCountInt: (1<<53)+1}, Sum: 10, Schema: 1, ZeroThreshold: 0.001, ResetHint: prompb.Histogram_NO,
|
||||
PositiveSpans: []prompb.BucketSpan{{Offset: -2, Length: 2}, {Offset: 3, Length: 1}}, PositiveDeltas: []int64{1,2,-1}, NegativeSpans: []prompb.BucketSpan{{Offset: -3, Length: 1}, {Offset: 2, Length: 2}}, NegativeDeltas: []int64{1,2,-1}},
|
||||
{Count: &prompb.Histogram_CountFloat{CountFloat: 6}, ZeroCount: &prompb.Histogram_ZeroCountFloat{ZeroCountFloat: 0.5}, Sum: 20, Schema: 2, ZeroThreshold: 0.002, ResetHint: prompb.Histogram_GAUGE,
|
||||
PositiveSpans: []prompb.BucketSpan{{Offset: 3, Length: 2}}, PositiveCounts: []float64{2,3.5}},
|
||||
{Count: &prompb.Histogram_CountInt{CountInt: 1}, ZeroCount: &prompb.Histogram_ZeroCountInt{}, Sum: 1, Schema: -53, CustomValues: []float64{1,2}, PositiveSpans: []prompb.BucketSpan{{Length: 1}}, PositiveDeltas: []int64{1}},
|
||||
{Count: &prompb.Histogram_CountFloat{CountFloat: 3.5}, ZeroCount: &prompb.Histogram_ZeroCountFloat{}, Sum: 4, Schema: -53, CustomValues: []float64{1,2}, PositiveSpans: []prompb.BucketSpan{{Length: 2}}, PositiveCounts: []float64{1,2.5}, ResetHint: prompb.Histogram_GAUGE},
|
||||
{Count: &prompb.Histogram_CountInt{}, ZeroCount: &prompb.Histogram_ZeroCountInt{}},
|
||||
{Count: &prompb.Histogram_CountFloat{}, ZeroCount: &prompb.Histogram_ZeroCountFloat{}},
|
||||
{Count: &prompb.Histogram_CountInt{CountInt: 8}, ZeroCount: &prompb.Histogram_ZeroCountInt{}, Sum: math.Float64frombits(0x7ff8000000000123), PositiveSpans: []prompb.BucketSpan{{Length: 2}}, PositiveDeltas: []int64{3,-1}},
|
||||
{Count: &prompb.Histogram_CountInt{}, ZeroCount: &prompb.Histogram_ZeroCountInt{}, Sum: math.Float64frombits(0x7ff0000000000002)},
|
||||
{Count: &prompb.Histogram_CountInt{CountInt: math.MaxInt64}, ZeroCount: &prompb.Histogram_ZeroCountInt{ZeroCountInt: math.MaxInt64}},
|
||||
}
|
||||
for i := range hs { hs[i].Timestamp = int64(i+1)*1000 }
|
||||
return &prompb.ReadResponse{Results: []*prompb.QueryResult{{Timeseries: []*prompb.TimeSeries{{Labels: []prompb.Label{{Name:"__name__", Value:"native_histogram_remote_read"}}, Histograms:hs}}}}}
|
||||
}
|
||||
|
||||
func check(ok bool, args ...any) { if !ok { panic(fmt.Sprint(args...)) } }
|
||||
func main() {
|
||||
want := expected()
|
||||
if os.Args[1] == "generate" {
|
||||
data, err := want.Marshal(); if err != nil { panic(err) }
|
||||
if err := os.WriteFile(os.Args[2], data, 0644); err != nil { panic(err) }
|
||||
fmt.Printf("generated %d bytes\n", len(data)); return
|
||||
}
|
||||
data, err := os.ReadFile(os.Args[2]); if err != nil { panic(err) }
|
||||
data, err = snappy.Decode(nil, data); if err != nil { panic(err) }
|
||||
var got prompb.ReadResponse
|
||||
if err = got.Unmarshal(data); err != nil { panic(err) }
|
||||
check(len(got.Results)==1 && len(got.Results[0].Timeseries)==1, "result/series count")
|
||||
hs := got.Results[0].Timeseries[0].Histograms
|
||||
expectedHs := want.Results[0].Timeseries[0].Histograms
|
||||
check(len(hs)==len(expectedHs), "histogram count")
|
||||
for i := range hs {
|
||||
g,w := hs[i], expectedHs[i]
|
||||
check(math.Float64bits(g.Sum)==math.Float64bits(w.Sum), "sum bits ", i)
|
||||
g.Sum,w.Sum=0,0
|
||||
check(reflect.DeepEqual(g,w), "protobuf fields ", i, ": ", g, " expected ", w)
|
||||
}
|
||||
ss := remote.FromQueryResult(true, got.Results[0])
|
||||
samples,series := 0,0
|
||||
for ss.Next() {
|
||||
series++
|
||||
it := ss.At().Iterator(nil)
|
||||
for typ := it.Next(); typ != chunkenc.ValNone; typ=it.Next() {
|
||||
w := expectedHs[samples]
|
||||
if w.IsFloatHistogram() {
|
||||
check(typ==chunkenc.ValFloatHistogram, "float type ", samples)
|
||||
t,h:=it.AtFloatHistogram(nil)
|
||||
check(t==w.Timestamp && h.Count==w.GetCountFloat() && h.ZeroCount==w.GetZeroCountFloat(), "float count/time ", samples)
|
||||
check(reflect.DeepEqual(h.PositiveBuckets,w.PositiveCounts), "float buckets ", samples)
|
||||
check(reflect.DeepEqual(h.CustomValues,w.CustomValues), "float bounds ", samples)
|
||||
check(math.Float64bits(h.Sum)==math.Float64bits(w.Sum), "float sum ", samples)
|
||||
check(int32(h.CounterResetHint)==int32(w.ResetHint), "float reset ", samples)
|
||||
} else {
|
||||
check(typ==chunkenc.ValHistogram, "integer type ", samples)
|
||||
t,h:=it.AtHistogram(nil)
|
||||
check(t==w.Timestamp && h.Count==w.GetCountInt() && h.ZeroCount==w.GetZeroCountInt(), "integer count/time ", samples)
|
||||
check(reflect.DeepEqual(h.PositiveBuckets,w.PositiveDeltas) && reflect.DeepEqual(h.NegativeBuckets,w.NegativeDeltas), "integer deltas ", samples)
|
||||
check(reflect.DeepEqual(h.CustomValues,w.CustomValues), "integer bounds ", samples)
|
||||
check(math.Float64bits(h.Sum)==math.Float64bits(w.Sum), "integer sum ", samples)
|
||||
check(int32(h.CounterResetHint)==int32(w.ResetHint), "integer reset ", samples)
|
||||
_,f:=it.AtFloatHistogram(nil)
|
||||
check(reflect.DeepEqual(f.CustomValues,w.CustomValues), "converted bounds ", samples)
|
||||
if w.Schema == -53 {
|
||||
buckets:=f.PositiveBucketIterator()
|
||||
check(buckets.Next(), "custom bucket missing")
|
||||
b:=buckets.At()
|
||||
check(b.Lower==math.Inf(-1) && b.Upper==1 && b.Count==1, "custom bucket placement: ",b)
|
||||
check(!buckets.Next(), "extra custom bucket")
|
||||
}
|
||||
}
|
||||
_,f:=it.AtFloatHistogram(nil)
|
||||
check(f.Schema==w.Schema && f.ZeroThreshold==w.ZeroThreshold, "schema/threshold ",samples)
|
||||
check(len(f.PositiveSpans)==len(w.PositiveSpans) && len(f.NegativeSpans)==len(w.NegativeSpans), "span counts ",samples)
|
||||
for i,s:=range f.PositiveSpans { check(s.Offset==w.PositiveSpans[i].Offset && s.Length==w.PositiveSpans[i].Length,"positive spans") }
|
||||
for i,s:=range f.NegativeSpans { check(s.Offset==w.NegativeSpans[i].Offset && s.Length==w.NegativeSpans[i].Length,"negative spans") }
|
||||
samples++
|
||||
}
|
||||
check(it.Err()==nil,"iterator error: ",it.Err())
|
||||
}
|
||||
check(ss.Err()==nil,"series error: ",ss.Err())
|
||||
check(series==1 && samples==9,"series/sample count ",series,"/",samples)
|
||||
fmt.Println("PASS: 9 histogram samples, both iterator accessors, no series or iterator errors")
|
||||
}
|
||||
```
|
||||
Binary file not shown.
Reference in New Issue
Block a user