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

feat(otlp): support cumulative exponential histograms (#8900)

Browse Source 
* feat: implement exponential histogram

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

* chore: remove duplicate tests

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

* fix(otlp): enforce exponential histogram ingestion safety

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

* chore: update rfc

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

* fix: test

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

* fix(otlp): remove protocol-coupled histogram checks

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

* perf(otlp): reuse native histogram schema across data points

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

* fix: merge repeated OTLP histogram fragments

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

* fix(otlp): build rejection messages lazily

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

* fix: add doc

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

---------

Signed-off-by: shuiyisong <xixing.sys@gmail.com>
This commit is contained in:
shuiyisong
2026-08-24 12:59:44 +00:00
committed by GitHub
parent 82444635f5
commit 1c5eabcbbf
24 changed files with 2322 additions and 630 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config/config.md
+2
View File
@@ -74,6 +74,7 @@
| `jaeger.enable` | Bool | `true` | Whether to enable Jaeger protocol in HTTP API. |
| `otlp` | -- | -- | OpenTelemetry protocol options. |
| `otlp.enable` | Bool | `true` | Whether to enable OpenTelemetry protocol in HTTP API. |
| `otlp.experimental_enable_exponential_histogram` | Bool | `false` | Experimental: enable cumulative OTLP exponential histogram ingestion. |
| `otlp.trace_ingest_chunk_size` | Integer | `512` | Maximum spans per trace ingest chunk. Set to 0 to disable splitting. |
| `prom_store` | -- | -- | Prometheus remote storage options |
| `prom_store.enable` | Bool | `true` | Whether to enable Prometheus remote write and read in HTTP API. |
@@ -316,6 +317,7 @@
| `jaeger.enable` | Bool | `true` | Whether to enable Jaeger protocol in HTTP API. |
| `otlp` | -- | -- | OpenTelemetry protocol options. |
| `otlp.enable` | Bool | `true` | Whether to enable OpenTelemetry protocol in HTTP API. |
| `otlp.experimental_enable_exponential_histogram` | Bool | `false` | Experimental: enable cumulative OTLP exponential histogram ingestion. |
| `otlp.trace_ingest_chunk_size` | Integer | `512` | Maximum spans per trace ingest chunk. Set to 0 to disable splitting. |
| `prom_store` | -- | -- | Prometheus remote storage options |
| `prom_store.enable` | Bool | `true` | Whether to enable Prometheus remote write and read in HTTP API. |
config/frontend.example.toml
+2
View File
@@ -237,6 +237,8 @@ enable = true
[otlp]
## Whether to enable OpenTelemetry protocol in HTTP API.
enable = true
## Experimental: enable cumulative OTLP exponential histogram ingestion.
experimental_enable_exponential_histogram = false
## Maximum spans per trace ingest chunk. Set to 0 to disable splitting.
trace_ingest_chunk_size = 512
config/standalone.example.toml
+2
View File
@@ -204,6 +204,8 @@ enable = true
[otlp]
## Whether to enable OpenTelemetry protocol in HTTP API.
enable = true
## Experimental: enable cumulative OTLP exponential histogram ingestion.
experimental_enable_exponential_histogram = false
## Maximum spans per trace ingest chunk. Set to 0 to disable splitting.
trace_ingest_chunk_size = 512
docs/rfcs/2026-08-04-native-histograms.md
+82 -27
View File
@@ -7,22 +7,23 @@ Author: codex
# Summary
GreptimeDB stores a Prometheus native histogram in one Struct-valued field and
evaluates it as a first-class PromQL sample. This document records the supported
protocols, the storage invariant, and the compatibility decisions needed to make
that path predictable.
GreptimeDB stores native histograms in one Prometheus-compatible Struct-valued
field and evaluates them as first-class PromQL samples. Prometheus Remote Write
2.0 supplies native histograms directly. OTLP `ExponentialHistogram` is an
ingestion transport: accepted cumulative points are normalized into the same
Struct before persistence and are queried only as native histograms.
Native histograms are experimental. Prometheus Remote Write 2.0 is the only
supported native-histogram ingestion protocol. Remote Write 1.0 histogram
payloads are rejected instead of being acknowledged and dropped.
Native histograms are experimental. Prometheus Remote Write 2.0 and cumulative
OTLP exponential histograms are supported behind separate configuration gates.
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.
# Goals
1. Prevent silent native-histogram data loss at protocol boundaries.
2. Keep one stable persisted representation for integer, float, exponential,
and custom-bucket native histograms.
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
mistaken for support.
@@ -32,20 +33,37 @@ continues to return scalar samples only.
- Supporting native histograms in Remote Write 1.0.
- Returning native histograms through Prometheus Remote Read.
- Persisting Remote Write metric metadata.
- Ingesting OTLP exponential histograms.
- Persisting exemplars.
- Providing an OTLP-specific histogram query surface or reconstructing and
re-exporting the original OTLP point after persistence.
- Removing mixed float/histogram handling from PromQL expressions.
- Propagating PromQL annotations produced on datanodes back to the frontend.
# Data Model and Invariants
Each native histogram is stored in the configured native-histogram field
(`greptime_native_histogram` by default), a Struct whose children preserve the
wire-level schema, zero threshold, sum, reset hint, start timestamp, custom
bounds, spans, and either the integer or float count family.
Integer bucket deltas are converted to absolute integer counts for storage;
float bucket counts are already absolute. No separate sample-kind discriminator
is stored because the populated count family identifies it.
Each accepted native histogram, whether received directly through Remote Write
2.0 or normalized from OTLP `ExponentialHistogram`, is stored in the configured
native-histogram field (`greptime_native_histogram` by default) as the same
canonical Struct. The persistence boundary admits only values accepted by the
shared native-histogram validation.
## Prometheus Remote Write 2.0
Remote Write 2.0 native histograms enter the persistence path in Prometheus
format. The Struct preserves the validated schema, zero threshold, sum, reset
hint, start timestamp, custom bounds, spans, and either the integer or float
count family. Integer bucket deltas are converted to absolute integer counts
for storage; float bucket counts are already absolute. No separate sample-kind
discriminator is stored because the populated count family identifies it.
## OTLP ExponentialHistogram
OTLP `ExponentialHistogram` does not introduce a second persisted format.
Accepted cumulative points are normalized into a valid Prometheus native
histogram and then pass through the same validator and Struct encoder. The
Struct does not retain the raw OTLP point or a source-protocol discriminator,
and queries use only the native-histogram PromQL behavior. Detailed conversion
and rejection rules are listed under [OTLP](#otlp).
Within one resolved catalog, schema, and physical-table routing context, a
metric name has exactly one persisted sample kind:
@@ -78,7 +96,7 @@ that send native histograms must use Remote Write 2.0.
## Remote Write 2.0
Remote Write 2.0 accepts integer and float native histograms while
`http.experimental_enable_prometheus_native_histogram` is enabled. Supported
`prom_store.experimental_enable_prometheus_native_histogram` is enabled. Supported
exponential schemas are `-4` through `8`; schema `-53` represents native
histograms with custom buckets.
@@ -103,11 +121,46 @@ explicit if sampled responses are implemented.
## OTLP
OTLP exponential histograms are not converted in this version. The ingestion
branch intentionally remains deferred until it can map temporality, scale,
reset behavior, attributes, and rejected-point reporting into the canonical
native-histogram path. The code carries an explicit TODO rather than a partial
encoder.
OTLP exponential histograms are accepted when
`otlp.experimental_enable_exponential_histogram` is enabled. The option defaults
to false and applies to OTLP/HTTP. Disabled points are rejected rather than
silently acknowledged. OTel Arrow exponential histograms are rejected because
the current Arrow wire format omits `zero_threshold`; accepting them would
silently change the distribution. Cumulative temporality is required; delta and
unspecified exponential histograms are rejected before their points are
converted. Explicit OTLP histograms keep their existing `_bucket`, `_sum`, and
`_count` representation, including their existing delta behavior.
OTLP scales `-4` through `8` map directly to Prometheus schemas. Higher scales
are downscaled to schema `8`: dense counts that collide are merged before the
OTLP lower-bound index is shifted by one to the Prometheus upper-bound index.
This preserves count mass but irreversibly loses distinctions between source
buckets that merge. Lower scales are rejected. OTLP counts always populate the
integer histogram family. The transmitted non-negative finite zero threshold
and zero count are preserved; the Struct start timestamp is stored in
milliseconds and the reset hint is unknown. Point timestamps and attributes
retain their existing mode-specific conversion rules. Legacy mode retains its
normalized name, attribute rules, and nanosecond row timestamp. Non-legacy mode
retains Prometheus-compatible translation and its millisecond row timestamp.
Both modes write the same canonical Struct.
An absent sum is stored as an ordinary quiet NaN, so the sample remains
selectable while `histogram_sum` is unknown. Any NaN sum on a point without
`NoRecordedValue` is normalized to the same value, even if its payload has the
Prometheus stale-marker bits. Only an exponential-histogram point carrying OTLP
`NoRecordedValue` becomes an empty schema-0 integer histogram with the canonical
Prometheus stale-NaN sum; its attributes and timestamps remain. This
interpretation intentionally does not change gauges, sums, or explicit histograms.
Invalid points are skipped while unrelated valid points continue. Mixed
accepted/rejected OTLP/HTTP requests return partial success; a request with only
rejected points returns `InvalidArgument`. OTel Arrow uses an `OK` batch status
for mixed batches and `INVALID_ARGUMENT` when all points are rejected. Rejection
details are bounded, and metric metadata is emitted only for a metric that
produced an accepted row.
Minimum, maximum, and exemplars are not persisted. Delta accumulation,
zero-run span compaction, and a dedicated rejection metric remain deferred.
# PromQL Compatibility Decisions
@@ -208,20 +261,22 @@ boundary already recorded in
# Testing and Compatibility
Compatibility coverage includes protocol rejection, kind exclusivity,
Compatibility coverage includes protocol rejection, OTLP scale conversion and
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.
This work does not change GreptimeDB's persisted Struct, protobuf dependencies,
or public configuration. Existing native-histogram data remains readable.
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.
# Future Work
- Native-histogram Remote Read, including exact integer round-trips and streamed
chunks with start timestamps.
- OTLP exponential-histogram conversion with partial-success reporting.
- OTLP exponential-histogram delta accumulation.
- Persistent Remote Write metadata and accurate help/unit updates.
- Native-histogram exemplars and exemplar query APIs.
- Start-timestamp overlap annotations.
src/cmd/src/standalone.rs
+3 -1
View File
@@ -1129,7 +1129,9 @@ mod tests {
fn test_toml() {
let opts = StandaloneOptions::default();
let toml_string = toml::to_string(&opts).unwrap();
let _parsed: StandaloneOptions = toml::from_str(&toml_string).unwrap();
assert!(toml_string.contains("experimental_enable_exponential_histogram = false"));
let parsed: StandaloneOptions = toml::from_str(&toml_string).unwrap();
assert_eq!(parsed.otlp, opts.otlp);
}
#[test]
src/cmd/tests/load_config_test.rs
+49
View File
@@ -153,6 +153,12 @@ fn test_load_frontend_example_config() {
let options =
GreptimeOptions::<FrontendOptions>::load_layered_options(example_config.to_str(), "")
.unwrap();
assert!(
!options
.component
.otlp
.experimental_enable_exponential_histogram
);
let expected = GreptimeOptions::<FrontendOptions> {
component: FrontendOptions {
default_timezone: Some("UTC".to_string()),
@@ -329,6 +335,12 @@ fn test_load_standalone_example_config() {
let options =
GreptimeOptions::<StandaloneOptions>::load_layered_options(example_config.to_str(), "")
.unwrap();
assert!(
!options
.component
.otlp
.experimental_enable_exponential_histogram
);
let expected = GreptimeOptions::<StandaloneOptions> {
component: StandaloneOptions {
default_timezone: Some("UTC".to_string()),
@@ -380,6 +392,43 @@ fn test_load_standalone_example_config() {
similar_asserts::assert_eq!(options, expected);
}
#[test]
fn test_load_otlp_exponential_histogram_option() {
let config = tempfile::NamedTempFile::new().unwrap();
std::fs::write(
config.path(),
"[otlp]\nexperimental_enable_exponential_histogram = true\n",
)
.unwrap();
let frontend =
GreptimeOptions::<FrontendOptions>::load_layered_options(config.path().to_str(), "")
.unwrap();
assert!(
frontend
.component
.otlp
.experimental_enable_exponential_histogram
);
let standalone =
GreptimeOptions::<StandaloneOptions>::load_layered_options(config.path().to_str(), "")
.unwrap();
assert!(
standalone
.component
.otlp
.experimental_enable_exponential_histogram
);
assert!(
standalone
.component
.frontend_options()
.otlp
.experimental_enable_exponential_histogram
);
}
#[test]
fn test_load_standalone_user_provider_from_config() {
let config = tempfile::NamedTempFile::new().unwrap();
src/frontend/src/frontend.rs
+3 -1
View File
@@ -218,7 +218,9 @@ mod tests {
fn test_toml() {
let opts = FrontendOptions::default();
let toml_string = toml::to_string(&opts).unwrap();
let _parsed: FrontendOptions = toml::from_str(&toml_string).unwrap();
assert!(toml_string.contains("experimental_enable_exponential_histogram = false"));
let parsed: FrontendOptions = toml::from_str(&toml_string).unwrap();
assert_eq!(parsed.otlp, opts.otlp);
}
#[test]
src/frontend/src/instance/otlp.rs
+22 -9
View File
@@ -27,7 +27,7 @@ use client::Output;
use common_catalog::consts::{trace_operations_table_name, trace_services_table_name};
use common_error::ext::BoxedError;
use common_query::prelude::GREPTIME_PHYSICAL_TABLE;
use common_telemetry::tracing;
use common_telemetry::{tracing, warn};
use opentelemetry_proto::tonic::collector::logs::v1::ExportLogsServiceRequest;
use opentelemetry_proto::tonic::collector::trace::v1::ExportTraceServiceRequest;
use otel_arrow_rust::proto::opentelemetry::collector::metrics::v1::ExportMetricsServiceRequest;
@@ -38,7 +38,7 @@ use servers::interceptor::{OpenTelemetryProtocolInterceptor, OpenTelemetryProtoc
use servers::otlp;
use servers::otlp::trace::span::TraceSpanGroup;
use servers::query_handler::{
OpenTelemetryProtocolHandler, PipelineHandlerRef, TraceIngestOutcome,
MetricsIngestOutcome, OpenTelemetryProtocolHandler, PipelineHandlerRef, TraceIngestOutcome,
};
use session::context::QueryContextRef;
use snafu::ResultExt;
@@ -90,7 +90,7 @@ impl OpenTelemetryProtocolHandler for Instance {
&self,
request: ExportMetricsServiceRequest,
ctx: QueryContextRef,
) -> ServerResult<Output> {
) -> ServerResult<MetricsIngestOutcome> {
self.plugins
.get::<PermissionCheckerRef>()
.as_ref()
@@ -120,8 +120,19 @@ impl OpenTelemetryProtocolHandler for Instance {
.unwrap_or_default();
metric_ctx.is_legacy = is_legacy;
let (requests, rows, semantic_index) =
let (requests, rows, semantic_index, mut outcome) =
otlp::metrics::to_grpc_insert_requests(request, &mut metric_ctx)?;
if outcome.rejected_data_points > 0 {
warn!(
"Rejected {} OTLP exponential histogram data points: {}",
outcome.rejected_data_points,
outcome.error_message.as_deref().unwrap_or_default()
);
}
if outcome.accepted_data_points == 0 {
return Ok(outcome);
}
self.check_row_insert_permission(&requests, &ctx, PermissionReq::Action(OTLP_WRITE))
.context(AuthSnafu)?;
self.cache_otlp_legacy(&input_names, &ctx, is_legacy)?;
@@ -143,9 +154,9 @@ impl OpenTelemetryProtocolHandler for Instance {
Arc::new(c)
};
// If the user uses the legacy path, it is by default without metric engine.
if metric_ctx.is_legacy || !metric_ctx.with_metric_engine {
self.handle_row_inserts(requests, ctx, false, false)
// OTLP tables have one sample field in both the legacy and physical paths.
let output = if metric_ctx.is_legacy || !metric_ctx.with_metric_engine {
self.handle_row_inserts(requests, ctx, false, true)
.await
.map_err(BoxedError::new)
.context(error::ExecuteGrpcQuerySnafu)
@@ -154,11 +165,13 @@ impl OpenTelemetryProtocolHandler for Instance {
.extension(PHYSICAL_TABLE_PARAM)
.unwrap_or(GREPTIME_PHYSICAL_TABLE)
.to_string();
self.handle_metric_row_inserts(requests, ctx, physical_table.clone())
self.handle_metric_row_inserts(requests, ctx, physical_table)
.await
.map_err(BoxedError::new)
.context(error::ExecuteGrpcQuerySnafu)
}
}?;
outcome.write_cost = output.meta.cost;
Ok(outcome)
}
#[tracing::instrument(skip_all)]
src/frontend/src/server.rs
+5 -2
View File
@@ -158,8 +158,11 @@ where
}
if opts.otlp.enable {
builder = builder
.with_otlp_handler(self.instance.clone(), opts.prom_store.with_metric_engine);
builder = builder.with_otlp_handler(
self.instance.clone(),
opts.prom_store.with_metric_engine,
opts.otlp.experimental_enable_exponential_histogram,
);
}
if opts.jaeger.enable {
src/frontend/src/service_config/otlp.rs
+13
View File
@@ -20,6 +20,7 @@ const DEFAULT_TRACE_INGEST_CHUNK_SIZE: usize = 512;
#[serde(default)]
pub struct OtlpOptions {
pub enable: bool,
pub experimental_enable_exponential_histogram: bool,
/// Maximum spans per trace ingest chunk. Set to 0 to disable splitting.
pub trace_ingest_chunk_size: usize,
}
@@ -28,6 +29,7 @@ impl Default for OtlpOptions {
fn default() -> Self {
Self {
enable: true,
experimental_enable_exponential_histogram: false,
trace_ingest_chunk_size: DEFAULT_TRACE_INGEST_CHUNK_SIZE,
}
}
@@ -41,10 +43,12 @@ mod tests {
fn test_otlp_options() {
let default = OtlpOptions::default();
assert!(default.enable);
assert!(!default.experimental_enable_exponential_histogram);
assert_eq!(default.trace_ingest_chunk_size, 512);
let options: OtlpOptions = toml::from_str("enable = false").unwrap();
assert!(!options.enable);
assert!(!options.experimental_enable_exponential_histogram);
assert_eq!(
options.trace_ingest_chunk_size,
DEFAULT_TRACE_INGEST_CHUNK_SIZE
@@ -52,6 +56,15 @@ mod tests {
let options: OtlpOptions = toml::from_str("trace_ingest_chunk_size = 0").unwrap();
assert!(options.enable);
assert!(!options.experimental_enable_exponential_histogram);
assert_eq!(options.trace_ingest_chunk_size, 0);
let options: OtlpOptions =
toml::from_str("experimental_enable_exponential_histogram = true").unwrap();
assert!(options.experimental_enable_exponential_histogram);
let serialized = toml::to_string(&options).unwrap();
assert!(serialized.contains("experimental_enable_exponential_histogram = true"));
assert_eq!(toml::from_str::<OtlpOptions>(&serialized).unwrap(), options);
}
}
src/operator/src/insert.rs
+9 -6
View File
@@ -982,8 +982,8 @@ impl Inserter {
/// When `accommodate_existing_schema` is true, it may modify the input `req` to
/// accommodate it with existing schema. See [`create_or_alter_tables_on_demand`](Self::create_or_alter_tables_on_demand)
/// for more details.
/// When `accommodate_existing_schema` is true and `is_single_value` is true, it also consider fields when modifying the
/// input `req`.
/// When `is_single_value` is true, it also rejects native-histogram/float kind changes.
/// When both options are true, it considers fields when modifying the input `req`.
fn get_alter_table_expr_on_demand(
&self,
req: &mut RowInsertRequest,
@@ -1018,8 +1018,7 @@ impl Inserter {
return Ok(None);
};
// If accommodate_existing_schema is true, update request schema for Timestamp/Field columns
if accommodate_existing_schema {
if is_single_value {
let request_is_native_histogram = request_is_native_histogram(request_schema);
let table_is_native_histogram = table_is_native_histogram(table);
ensure!(
@@ -1030,6 +1029,10 @@ impl Inserter {
),
}
);
}
// If accommodate_existing_schema is true, update request schema for Timestamp/Field columns
if accommodate_existing_schema {
let table_schema = table.schema();
// Find timestamp column name
let ts_col_name = table_schema.timestamp_column().map(|c| c.name.clone());
@@ -1638,7 +1641,7 @@ mod tests {
}),
};
let error = inserter
.get_alter_table_expr_on_demand(&mut histogram_req, &table, &ctx, true, true, true)
.get_alter_table_expr_on_demand(&mut histogram_req, &table, &ctx, false, true, true)
.unwrap_err();
assert!(
error
@@ -1666,7 +1669,7 @@ mod tests {
&mut sample_req,
&histogram_table,
&ctx,
true,
false,
true,
true,
)
src/servers/src/http.rs
+8 -1
View File
@@ -724,11 +724,16 @@ impl HttpServerBuilder {
self,
handler: OpenTelemetryProtocolHandlerRef,
with_metric_engine: bool,
experimental_enable_exponential_histogram: bool,
) -> Self {
Self {
router: self.router.nest(
&format!("/{HTTP_API_VERSION}/otlp"),
HttpServer::route_otlp(handler, with_metric_engine),
HttpServer::route_otlp(
handler,
with_metric_engine,
experimental_enable_exponential_histogram,
),
),
..self
}
@@ -1386,6 +1391,7 @@ impl HttpServer {
fn route_otlp<S>(
otlp_handler: OpenTelemetryProtocolHandlerRef,
with_metric_engine: bool,
experimental_enable_exponential_histogram: bool,
) -> Router<S> {
Router::new()
.route("/v1/metrics", routing::post(otlp::metrics))
@@ -1397,6 +1403,7 @@ impl HttpServer {
)
.with_state(OtlpState {
with_metric_engine,
experimental_enable_exponential_histogram,
handler: otlp_handler,
})
}
src/servers/src/http/otlp.rs
+63 -12
View File
@@ -27,7 +27,9 @@ use mime_guess::mime;
use opentelemetry_proto::tonic::collector::logs::v1::{
ExportLogsServiceRequest, ExportLogsServiceResponse,
};
use opentelemetry_proto::tonic::collector::metrics::v1::ExportMetricsServiceResponse;
use opentelemetry_proto::tonic::collector::metrics::v1::{
ExportMetricsPartialSuccess, ExportMetricsServiceResponse,
};
use opentelemetry_proto::tonic::collector::trace::v1::{
ExportTracePartialSuccess, ExportTraceServiceRequest, ExportTraceServiceResponse,
};
@@ -44,7 +46,9 @@ use crate::http::extractor::{
};
use crate::http::header::{CONTENT_TYPE_PROTOBUF, write_cost_header_map};
use crate::metrics::METRIC_HTTP_OPENTELEMETRY_LOGS_ELAPSED;
use crate::query_handler::{OpenTelemetryProtocolHandlerRef, PipelineHandler, TraceIngestOutcome};
use crate::query_handler::{
MetricsIngestOutcome, OpenTelemetryProtocolHandlerRef, PipelineHandler, TraceIngestOutcome,
};
#[derive(Clone, prost::Message)]
pub struct GoogleRpcStatus {
@@ -73,6 +77,7 @@ fn content_type_to_string(content_type: Option<&TypedHeader<ContentType>>) -> St
#[derive(Clone)]
pub struct OtlpState {
pub with_metric_engine: bool,
pub experimental_enable_exponential_histogram: bool,
pub handler: OpenTelemetryProtocolHandlerRef,
}
@@ -84,7 +89,7 @@ pub async fn metrics(
http_opts: OtlpMetricOptions,
content_type: Option<TypedHeader<ContentType>>,
bytes: Bytes,
) -> Result<OtlpResponse<ExportMetricsServiceResponse>> {
) -> Result<OtlpMetricsResponse> {
if is_json_content_type(content_type.as_ref().map(|h| &h.0)) {
return error::UnsupportedJsonContentTypeSnafu {}.fail();
}
@@ -103,6 +108,7 @@ pub async fn metrics(
let OtlpState {
with_metric_engine,
experimental_enable_exponential_histogram,
handler,
} = state;
@@ -111,6 +117,7 @@ pub async fn metrics(
resource_attrs: http_opts.resource_attrs,
promote_scope_attrs: http_opts.promote_scope_attrs,
with_metric_engine,
experimental_enable_exponential_histogram,
// set is_legacy later
is_legacy: false,
metric_type: MetricType::Init,
@@ -118,15 +125,15 @@ pub async fn metrics(
}));
let query_ctx = Arc::new(query_ctx);
handler
.metrics(request, query_ctx)
.await
.map(|o| OtlpResponse {
resp_body: ExportMetricsServiceResponse {
partial_success: None,
},
write_cost: o.meta.cost,
})
handler.metrics(request, query_ctx).await.map(|outcome| {
if outcome.accepted_data_points == 0 && outcome.rejected_data_points > 0 {
OtlpMetricsResponse::Failure(outcome)
} else if outcome.rejected_data_points > 0 {
OtlpMetricsResponse::PartialSuccess(outcome)
} else {
OtlpMetricsResponse::FullSuccess(outcome)
}
})
}
#[axum_macros::debug_handler]
@@ -259,6 +266,50 @@ pub struct OtlpResponse<T: Message> {
write_cost: usize,
}
pub enum OtlpMetricsResponse {
FullSuccess(MetricsIngestOutcome),
PartialSuccess(MetricsIngestOutcome),
Failure(MetricsIngestOutcome),
}
impl IntoResponse for OtlpMetricsResponse {
fn into_response(self) -> axum::response::Response {
match self {
OtlpMetricsResponse::FullSuccess(outcome) => {
let mut header_map = write_cost_header_map(outcome.write_cost);
header_map.insert(header::CONTENT_TYPE, CONTENT_TYPE_PROTOBUF.clone());
let body = ExportMetricsServiceResponse {
partial_success: None,
};
(header_map, body.encode_to_vec()).into_response()
}
OtlpMetricsResponse::PartialSuccess(outcome) => {
let mut header_map = write_cost_header_map(outcome.write_cost);
header_map.insert(header::CONTENT_TYPE, CONTENT_TYPE_PROTOBUF.clone());
let body = ExportMetricsServiceResponse {
partial_success: Some(ExportMetricsPartialSuccess {
rejected_data_points: outcome.rejected_data_points,
error_message: outcome.error_message.unwrap_or_default(),
}),
};
(header_map, body.encode_to_vec()).into_response()
}
OtlpMetricsResponse::Failure(outcome) => {
let status = GoogleRpcStatus {
code: tonic::Code::InvalidArgument as i32,
message: outcome.error_message.unwrap_or_default(),
};
(
StatusCode::BAD_REQUEST,
[(header::CONTENT_TYPE, CONTENT_TYPE_PROTOBUF.as_ref())],
status.encode_to_vec(),
)
.into_response()
}
}
}
}
impl<T: Message> IntoResponse for OtlpResponse<T> {
fn into_response(self) -> axum::response::Response {
let mut header_map = write_cost_header_map(self.write_cost);
src/servers/src/lib.rs
+1
View File
@@ -34,6 +34,7 @@ pub mod interceptor;
pub mod metrics;
pub mod metrics_handler;
pub mod mysql;
pub(crate) mod native_histogram;
pub mod opentsdb;
pub mod otel_arrow;
pub mod otlp;
src/servers/src/native_histogram.rs
+524
View File
@@ -0,0 +1,524 @@
// 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::histogram::{Count, ZeroCount};
use api::greptime_proto::io::prometheus::write::v2::{BucketSpan, Histogram};
use api::helper::ColumnDataTypeWrapper;
use api::v1::value::ValueData;
use api::v1::{ColumnSchema, ListValue, SemanticType, Value};
use common_query::native_histogram::*;
use common_query::prelude::greptime_native_histogram;
use snafu::{Snafu, ensure};
const MAX_NATIVE_HISTOGRAM_SCHEMA: i32 = 8;
const MAX_REDUCIBLE_NATIVE_HISTOGRAM_SCHEMA: i32 = 52;
#[derive(Debug, Snafu)]
#[snafu(display("{message}"))]
pub(crate) struct NativeHistogramError {
message: String,
}
type Result<T> = std::result::Result<T, NativeHistogramError>;
/// Returns the canonical column schema for a native histogram value.
pub(crate) fn native_histogram_column_schema() -> Result<ColumnSchema> {
let (datatype, datatype_extension) =
ColumnDataTypeWrapper::try_from(native_histogram_value_type().clone())
.map_err(|error| NativeHistogramError {
message: format!("native histogram type cannot be encoded: {error}"),
})?
.into_parts();
Ok(ColumnSchema {
column_name: greptime_native_histogram().to_string(),
datatype: datatype as i32,
semantic_type: SemanticType::Field as i32,
datatype_extension,
options: None,
})
}
/// Validates and encodes a Prometheus histogram into the canonical Struct value.
pub(crate) fn encode_native_histogram(histogram: &Histogram) -> Result<ValueData> {
let uses_float_counts = native_histogram_uses_float_counts(histogram)?;
validate_native_histogram(histogram, uses_float_counts)?;
let mut items = Vec::with_capacity(NATIVE_HISTOGRAM_FIELD_NAMES.len());
let positive_span_lengths = i32_span_lengths("positive", &histogram.positive_spans)?;
let negative_span_lengths = i32_span_lengths("negative", &histogram.negative_spans)?;
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)),
i32_list_value(positive_span_lengths),
i32_list_value(histogram.negative_spans.iter().map(|span| span.offset)),
i32_list_value(negative_span_lengths),
]);
if uses_float_counts {
validate_float_native_histogram_counts(histogram)?;
let count = match histogram.count.as_ref() {
Some(Count::CountFloat(count)) => *count,
_ => 0.0,
};
let zero_count = match histogram.zero_count.as_ref() {
Some(ZeroCount::ZeroCountFloat(zero_count)) => *zero_count,
_ => 0.0,
};
items.extend([
null_pb_value(),
null_pb_value(),
i64_list_value(std::iter::empty()),
i64_list_value(std::iter::empty()),
pb_value(ValueData::F64Value(count)),
pb_value(ValueData::F64Value(zero_count)),
f64_list_value(histogram.positive_counts.iter().copied()),
f64_list_value(histogram.negative_counts.iter().copied()),
]);
} else {
let count = match histogram.count.as_ref() {
Some(Count::CountInt(count)) => *count,
_ => 0,
};
let zero_count = match histogram.zero_count.as_ref() {
Some(ZeroCount::ZeroCountInt(zero_count)) => *zero_count,
_ => 0,
};
let positive_buckets = bucket_counts_from_deltas(&histogram.positive_deltas)?;
let negative_buckets = bucket_counts_from_deltas(&histogram.negative_deltas)?;
validate_integer_native_histogram_counts(histogram, &positive_buckets, &negative_buckets)?;
let count = i64::try_from(count).map_err(|_| NativeHistogramError {
message: format!("native histogram integer count {count} overflows i64"),
})?;
let zero_count = i64::try_from(zero_count).map_err(|_| NativeHistogramError {
message: format!("native histogram integer zero_count {zero_count} overflows i64"),
})?;
items.extend([
pb_value(ValueData::I64Value(count)),
pb_value(ValueData::I64Value(zero_count)),
i64_list_value(positive_buckets),
i64_list_value(negative_buckets),
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 validate_native_histogram(histogram: &Histogram, uses_float_counts: bool) -> Result<()> {
let exponential_overflow_index = validate_native_histogram_schema(histogram.schema)?;
validate_native_histogram_custom_values(histogram)?;
if histogram.schema == CUSTOM_BUCKETS_SCHEMA {
ensure!(
histogram.zero_threshold == 0.0 && native_histogram_zero_count_is_zero(histogram),
NativeHistogramSnafu {
message: "custom native histogram must not use a zero bucket"
}
);
ensure!(
histogram.negative_spans.is_empty()
&& histogram.negative_deltas.is_empty()
&& histogram.negative_counts.is_empty(),
NativeHistogramSnafu {
message: "custom native histogram must not use negative buckets"
}
);
}
let (positive_buckets, negative_buckets) = if uses_float_counts {
(
histogram.positive_counts.len(),
histogram.negative_counts.len(),
)
} else {
(
histogram.positive_deltas.len(),
histogram.negative_deltas.len(),
)
};
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()).map_err(|_| NativeHistogramError {
message: "custom native histogram has too many custom_values".to_string(),
})?,
)
};
validate_native_histogram_spans(
"positive",
&histogram.positive_spans,
positive_buckets,
bucket_index_range,
)?;
validate_native_histogram_spans(
"negative",
&histogram.negative_spans,
negative_buckets,
bucket_index_range,
)?;
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_NATIVE_HISTOGRAM_SCHEMA + 1..=MAX_REDUCIBLE_NATIVE_HISTOGRAM_SCHEMA).contains(&schema) {
Err(NativeHistogramError {
message: format!("native histogram schema {schema} must be reduced before ingestion"),
})
} else {
Err(NativeHistogramError {
message: format!("native histogram schema {schema} is unsupported"),
})
}
}
fn validate_native_histogram_custom_values(histogram: &Histogram) -> Result<()> {
if histogram.schema != CUSTOM_BUCKETS_SCHEMA {
ensure!(
histogram.custom_values.is_empty(),
NativeHistogramSnafu {
message: "standard native histogram must not use custom_values"
}
);
return Ok(());
}
for value in &histogram.custom_values {
ensure!(
!value.is_nan() && *value != f64::INFINITY,
NativeHistogramSnafu {
message: "custom native histogram custom_values must not contain +Inf or NaN"
}
);
}
for values in histogram.custom_values.windows(2) {
ensure!(
values[0] < values[1],
NativeHistogramSnafu {
message: "custom native histogram custom_values must be sorted"
}
);
}
Ok(())
}
fn validate_native_histogram_spans(
name: &str,
spans: &[BucketSpan],
bucket_count: usize,
bucket_index_range: (i32, i32),
) -> Result<()> {
let span_len = spans.iter().try_fold(0usize, |sum, span| {
let length = usize::try_from(span.length).map_err(|_| NativeHistogramError {
message: format!("native histogram {name} span length exceeds usize"),
})?;
sum.checked_add(length).ok_or_else(|| NativeHistogramError {
message: format!("native histogram {name} spans overflow"),
})
})?;
ensure!(
span_len == bucket_count,
NativeHistogramSnafu {
message: format!(
"native histogram {name} spans describe {span_len} buckets, found {bucket_count}"
)
}
);
let mut current_index = 0i32;
for (span_index, span) in spans.iter().enumerate() {
ensure!(
span.offset >= 0 || (span_index == 0 && bucket_index_range.0 == i32::MIN),
NativeHistogramSnafu {
message: format!(
"native histogram {name} span {} has negative offset {}",
span_index + 1,
span.offset
)
}
);
current_index = if span_index == 0 {
span.offset
} else {
current_index
.checked_add(span.offset)
.ok_or_else(|| NativeHistogramError {
message: format!("native histogram {name} span index overflows i32"),
})?
};
for _ in 0..span.length {
ensure!(
(bucket_index_range.0..=bucket_index_range.1).contains(&current_index),
NativeHistogramSnafu {
message: format!(
"native histogram {name} bucket index {current_index} is out of range"
)
}
);
current_index = current_index
.checked_add(1)
.ok_or_else(|| NativeHistogramError {
message: format!("native histogram {name} span index overflows i32"),
})?;
}
}
Ok(())
}
fn validate_float_native_histogram_counts(histogram: &Histogram) -> Result<()> {
let count = match histogram.count.as_ref() {
Some(Count::CountFloat(count)) => *count,
_ => 0.0,
};
ensure!(
count >= 0.0 || count.is_nan(),
NativeHistogramSnafu {
message: "native histogram float count must not be negative"
}
);
let zero_count = match histogram.zero_count.as_ref() {
Some(ZeroCount::ZeroCountFloat(zero_count)) => *zero_count,
_ => 0.0,
};
ensure!(
zero_count >= 0.0 || zero_count.is_nan(),
NativeHistogramSnafu {
message: "native histogram float zero_count must not be negative"
}
);
for (name, counts) in [
("positive", &histogram.positive_counts),
("negative", &histogram.negative_counts),
] {
for (index, count) in counts.iter().enumerate() {
ensure!(
*count >= 0.0 || count.is_nan(),
NativeHistogramSnafu {
message: format!(
"native histogram {name} bucket {} count must not be negative",
index + 1
)
}
);
}
}
Ok(())
}
fn validate_integer_native_histogram_counts(
histogram: &Histogram,
positive_buckets: &[i64],
negative_buckets: &[i64],
) -> Result<()> {
let count = match histogram.count.as_ref() {
Some(Count::CountInt(count)) => *count,
_ => 0,
};
let zero_count = match histogram.zero_count.as_ref() {
Some(ZeroCount::ZeroCountInt(zero_count)) => *zero_count,
_ => 0,
};
let bucket_count =
positive_buckets
.iter()
.chain(negative_buckets)
.try_fold(zero_count, |total, bucket| {
let bucket = u64::try_from(*bucket).map_err(|_| NativeHistogramError {
message: "native histogram bucket count is negative".to_string(),
})?;
total
.checked_add(bucket)
.ok_or_else(|| NativeHistogramError {
message: "native histogram bucket total overflows u64".to_string(),
})
})?;
ensure!(
if histogram.sum.is_nan() {
bucket_count <= count
} else {
bucket_count == count
},
NativeHistogramSnafu {
message: format!(
"native histogram has {bucket_count} observations in buckets, count is {count}"
)
}
);
Ok(())
}
fn native_histogram_zero_count_is_zero(histogram: &Histogram) -> bool {
match histogram.zero_count.as_ref() {
Some(ZeroCount::ZeroCountInt(zero_count)) => *zero_count == 0,
Some(ZeroCount::ZeroCountFloat(zero_count)) => *zero_count == 0.0,
None => true,
}
}
fn native_histogram_uses_float_counts(histogram: &Histogram) -> Result<bool> {
let uses_float_count = matches!(histogram.count, Some(Count::CountFloat(_)))
|| matches!(histogram.zero_count, Some(ZeroCount::ZeroCountFloat(_)));
let uses_int_count = matches!(histogram.count, Some(Count::CountInt(_)))
|| matches!(histogram.zero_count, Some(ZeroCount::ZeroCountInt(_)));
let uses_float_buckets =
!histogram.positive_counts.is_empty() || !histogram.negative_counts.is_empty();
let uses_int_buckets =
!histogram.positive_deltas.is_empty() || !histogram.negative_deltas.is_empty();
ensure!(
!matches!(
(&histogram.count, &histogram.zero_count),
(Some(Count::CountInt(_)), Some(ZeroCount::ZeroCountFloat(_)))
| (Some(Count::CountFloat(_)), Some(ZeroCount::ZeroCountInt(_)))
),
NativeHistogramSnafu {
message: "native histogram count and zero_count must use the same integer or float family"
}
);
ensure!(
!(uses_float_buckets && uses_int_buckets),
NativeHistogramSnafu {
message: "native histogram bucket counts must use either integer deltas or float counts"
}
);
ensure!(
!(uses_float_count && uses_int_buckets),
NativeHistogramSnafu {
message: "float native histogram must not use integer bucket deltas"
}
);
ensure!(
!(uses_int_count && uses_float_buckets),
NativeHistogramSnafu {
message: "integer native histogram must not use float bucket counts"
}
);
Ok(uses_float_count || uses_float_buckets)
}
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 i32_span_lengths(name: &str, spans: &[BucketSpan]) -> Result<Vec<i32>> {
spans
.iter()
.map(|span| {
i32::try_from(span.length).map_err(|_| NativeHistogramError {
message: format!(
"native histogram {name} span length {} overflows i32",
span.length
),
})
})
.collect()
}
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)
.ok_or_else(|| NativeHistogramError {
message: "native histogram bucket count overflows i64".to_string(),
})?;
ensure!(
count >= 0,
NativeHistogramSnafu {
message: "native histogram bucket count is negative"
}
);
buckets.push(count);
}
Ok(buckets)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn shared_errors_are_protocol_neutral() {
let error = encode_native_histogram(&Histogram {
schema: 9,
..Default::default()
})
.unwrap_err();
assert_eq!(
error.to_string(),
"native histogram schema 9 must be reduced before ingestion"
);
assert!(!error.to_string().contains("remote write"));
assert!(!error.to_string().contains("OTLP"));
}
}
src/servers/src/otel_arrow.rs
+90 -18
View File
@@ -12,6 +12,8 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use std::sync::Arc;
use auth::UserProviderRef;
use common_error::ext::ErrorExt;
use common_error::status_code::status_to_tonic_code;
@@ -19,14 +21,20 @@ use common_telemetry::error;
use futures::SinkExt;
use otel_arrow_rust::Consumer;
use otel_arrow_rust::proto::opentelemetry::arrow::v1::arrow_metrics_service_server::ArrowMetricsService;
use otel_arrow_rust::proto::opentelemetry::arrow::v1::{BatchArrowRecords, BatchStatus};
use otel_arrow_rust::proto::opentelemetry::arrow::v1::{
BatchArrowRecords, BatchStatus, StatusCode as ArrowStatusCode,
};
use otel_arrow_rust::proto::opentelemetry::metrics::v1::metric;
use session::protocol_ctx::{OtlpMetricCtx, ProtocolCtx};
use tonic::metadata::{Entry, MetadataValue};
use tonic::service::Interceptor;
use tonic::{Request, Response, Status, Streaming};
use crate::error;
use crate::grpc::context_auth;
use crate::query_handler::OpenTelemetryProtocolHandlerRef;
use crate::query_handler::{MetricsIngestOutcome, OpenTelemetryProtocolHandlerRef};
const EXPONENTIAL_HISTOGRAM_UNSUPPORTED: &str = "OTel Arrow exponential histograms are unsupported because the Arrow wire format omits zero_threshold";
pub struct OtelArrowServiceHandler<T> {
handler: T,
@@ -42,6 +50,31 @@ impl<T> OtelArrowServiceHandler<T> {
}
}
fn batch_status(
batch_id: i64,
outcome: MetricsIngestOutcome,
has_exponential_histogram_data_points: bool,
) -> BatchStatus {
let status_code = if outcome.accepted_data_points == 0 && outcome.rejected_data_points > 0 {
ArrowStatusCode::InvalidArgument
} else {
ArrowStatusCode::Ok
};
let status_message = match outcome.error_message {
// Arrow keeps the feature gate off, so these fail before per-point validation.
Some(_) if has_exponential_histogram_data_points => {
EXPONENTIAL_HISTOGRAM_UNSUPPORTED.to_string()
}
Some(message) => message,
None => String::new(),
};
BatchStatus {
batch_id,
status_code: status_code as i32,
status_message,
}
}
#[async_trait::async_trait]
impl ArrowMetricsService for OtelArrowServiceHandler<OpenTelemetryProtocolHandlerRef> {
type ArrowMetricsStream = futures::channel::mpsc::Receiver<Result<BatchStatus, Status>>;
@@ -55,6 +88,11 @@ impl ArrowMetricsService for OtelArrowServiceHandler<OpenTelemetryProtocolHandle
let query_ctx = context_auth::create_query_context_from_grpc_metadata(&headers)?;
context_auth::check_auth(self.user_provider.clone(), &headers, query_ctx.clone()).await?;
let query_ctx = {
let mut ctx = query_ctx.fork();
ctx.set_protocol_ctx(ProtocolCtx::OtlpMetric(OtlpMetricCtx::default()));
Arc::new(ctx)
};
let handler = self.handler.clone();
@@ -73,11 +111,7 @@ impl ArrowMetricsService for OtelArrowServiceHandler<OpenTelemetryProtocolHandle
return;
}
};
let batch_status = BatchStatus {
batch_id: batch.batch_id,
status_code: 0,
status_message: Default::default(),
};
let batch_id = batch.batch_id;
let request = match consumer.consume_metrics_batches(&mut batch).map_err(|e| {
error::HandleOtelArrowRequestSnafu {
err_msg: e.to_string(),
@@ -98,17 +132,33 @@ impl ArrowMetricsService for OtelArrowServiceHandler<OpenTelemetryProtocolHandle
return;
}
};
// use metric engine by default
if let Err(e) = handler.metrics(request, query_ctx.clone()).await {
let _ = sender
.send(Err(Status::new(
status_to_tonic_code(e.status_code()),
e.to_string(),
)))
.await;
error!(e; "Failed to ingest metrics from otel-arrow");
return;
}
let has_exponential_histogram_data_points = request
.resource_metrics
.iter()
.flat_map(|resource| &resource.scope_metrics)
.flat_map(|scope| &scope.metrics)
.any(|item| {
matches!(
item.data.as_ref(),
Some(metric::Data::ExponentialHistogram(histogram))
if !histogram.data_points.is_empty()
)
});
let outcome = match handler.metrics(request, query_ctx.clone()).await {
Ok(outcome) => outcome,
Err(e) => {
let _ = sender
.send(Err(Status::new(
status_to_tonic_code(e.status_code()),
e.to_string(),
)))
.await;
error!(e; "Failed to ingest metrics from otel-arrow");
return;
}
};
let batch_status =
batch_status(batch_id, outcome, has_exponential_histogram_data_points);
let _ = sender.send(Ok(batch_status)).await;
}
});
@@ -131,3 +181,25 @@ impl Interceptor for HeaderInterceptor {
Ok(request)
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn batch_status_explains_arrow_exponential_histogram_limit() {
let status = batch_status(
7,
MetricsIngestOutcome {
rejected_data_points: 1,
error_message: Some("internal OTLP rejection detail".to_string()),
..Default::default()
},
true,
);
assert_eq!(7, status.batch_id);
assert_eq!(ArrowStatusCode::InvalidArgument as i32, status.status_code);
assert_eq!(EXPONENTIAL_HISTOGRAM_UNSUPPORTED, status.status_message);
}
}
src/servers/src/otlp/metrics.rs
+1019 -32
View File
File diff suppressed because it is too large Load Diff
src/servers/src/prom_remote_write/v2.rs
+47 -515
View File
@@ -15,19 +15,23 @@
use std::collections::hash_map::Entry;
use ahash::{HashMap, HashMapExt, HashSet, HashSetExt};
#[cfg(test)]
use api::greptime_proto::io::prometheus::write::v2::BucketSpan;
#[cfg(test)]
use api::greptime_proto::io::prometheus::write::v2::histogram::{Count, ZeroCount};
use api::greptime_proto::io::prometheus::write::v2::{
BucketSpan, Exemplar, Histogram, Metadata, Sample, metadata,
Exemplar, Histogram, Metadata, Sample, metadata,
};
#[cfg(test)]
use api::greptime_proto::io::prometheus::write::v2::{Request, TimeSeries};
use api::helper::ColumnDataTypeWrapper;
#[cfg(test)]
use api::v1::ColumnSchema;
use api::v1::value::ValueData;
use api::v1::{
ColumnDataType, ColumnSchema, ListValue, RowInsertRequest, Rows, SemanticType, Value,
};
use api::v1::{ColumnDataType, RowInsertRequest, Rows, SemanticType, Value};
use bytes::{Buf, Bytes};
use common_grpc::precision::Precision;
use common_query::native_histogram::NATIVE_HISTOGRAM_FIELD;
#[cfg(test)]
use common_query::native_histogram::*;
use common_query::prelude::{greptime_native_histogram, greptime_timestamp, greptime_value};
use pipeline::{ContextOpt, ContextReq};
@@ -42,6 +46,7 @@ use table::requests::{
};
use crate::error::{self, Result};
use crate::native_histogram::{encode_native_histogram, native_histogram_column_schema};
use crate::prom_remote_write::row_builder::PromCtx;
use crate::prom_remote_write::validation::validate_label_name;
use crate::prom_remote_write::{REMOTE_WRITE_V2_VERSION, try_decompress};
@@ -59,8 +64,6 @@ use crate::semantic::{
type PromTags<'a> = Vec<(&'a str, String)>;
type ResolvedSeriesLabels<'a> = (PromCtx, String, PromTags<'a>);
const MAX_REMOTE_WRITE_V2_SCHEMA: i32 = 8;
const MAX_REDUCIBLE_REMOTE_WRITE_V2_SCHEMA: i32 = 52;
const TIME_SERIES_LABELS_REFS_TAG: u32 = 1;
const TIME_SERIES_SAMPLES_TAG: u32 = 2;
const TIME_SERIES_HISTOGRAMS_TAG: u32 = 3;
@@ -597,6 +600,19 @@ fn write_native_histogram<'a>(
histogram: &Histogram,
tags: impl Iterator<Item = (&'a str, String)>,
) -> Result<()> {
let value = encode_native_histogram(histogram).map_err(|error| {
error::InvalidPromRemoteRequestSnafu {
msg: format!("remote write v2 {error}"),
}
.build()
})?;
let column_schema = native_histogram_column_schema().map_err(|error| {
error::InvalidPromRemoteRequestSnafu {
msg: format!("remote write v2 {error}"),
}
.build()
})?;
// 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.
@@ -608,8 +624,11 @@ fn write_native_histogram<'a>(
Precision::Millisecond,
&mut row,
)?;
write_native_histogram_value(table_data, histogram, &mut row)?;
row_writer::write_by_schema(
table_data,
std::iter::once((column_schema, Some(value))),
&mut row,
)?;
row_writer::write_tags(table_data, tags, &mut row)?;
table_data.add_row(row);
@@ -617,512 +636,6 @@ fn write_native_histogram<'a>(
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: greptime_native_histogram().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 uses_float_counts = native_histogram_uses_float_counts(histogram)?;
validate_native_histogram(histogram, uses_float_counts)?;
let mut items = Vec::with_capacity(NATIVE_HISTOGRAM_FIELD_NAMES.len());
let positive_span_lengths = i32_span_lengths("positive", &histogram.positive_spans)?;
let negative_span_lengths = i32_span_lengths("negative", &histogram.negative_spans)?;
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)),
i32_list_value(positive_span_lengths.iter().copied()),
i32_list_value(histogram.negative_spans.iter().map(|span| span.offset)),
i32_list_value(negative_span_lengths.iter().copied()),
]);
if uses_float_counts {
validate_float_native_histogram_counts(histogram)?;
let count = match histogram.count.as_ref() {
Some(Count::CountFloat(count)) => *count,
_ => 0.0,
};
let zero_count = match histogram.zero_count.as_ref() {
Some(ZeroCount::ZeroCountFloat(zero_count)) => *zero_count,
_ => 0.0,
};
items.extend([
null_pb_value(),
null_pb_value(),
i64_list_value(std::iter::empty()),
i64_list_value(std::iter::empty()),
pb_value(ValueData::F64Value(count)),
pb_value(ValueData::F64Value(zero_count)),
f64_list_value(histogram.positive_counts.iter().copied()),
f64_list_value(histogram.negative_counts.iter().copied()),
]);
} else {
let count = match histogram.count.as_ref() {
Some(Count::CountInt(count)) => *count,
_ => 0,
};
let zero_count = match histogram.zero_count.as_ref() {
Some(ZeroCount::ZeroCountInt(zero_count)) => *zero_count,
_ => 0,
};
let positive_buckets = bucket_counts_from_deltas(&histogram.positive_deltas)?;
let negative_buckets = bucket_counts_from_deltas(&histogram.negative_deltas)?;
validate_integer_native_histogram_counts(histogram, &positive_buckets, &negative_buckets)?;
let count = i64::try_from(count)
.ok()
.context(error::InvalidPromRemoteRequestSnafu {
msg: format!(
"remote write v2 native histogram integer count {count} overflows i64"
),
})?;
let zero_count = i64::try_from(zero_count).ok().context(
error::InvalidPromRemoteRequestSnafu {
msg: format!(
"remote write v2 native histogram integer zero_count {zero_count} overflows i64"
),
},
)?;
items.extend([
pb_value(ValueData::I64Value(count)),
pb_value(ValueData::I64Value(zero_count)),
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 validate_native_histogram(histogram: &Histogram, uses_float_counts: bool) -> Result<()> {
let exponential_overflow_index = validate_native_histogram_schema(histogram.schema)?;
validate_native_histogram_custom_values(histogram)?;
if histogram.schema == CUSTOM_BUCKETS_SCHEMA {
ensure!(
histogram.zero_threshold == 0.0 && native_histogram_zero_count_is_zero(histogram),
error::InvalidPromRemoteRequestSnafu {
msg: "remote write v2 custom native histogram must not use a zero bucket"
.to_string(),
}
);
ensure!(
histogram.negative_spans.is_empty()
&& histogram.negative_deltas.is_empty()
&& histogram.negative_counts.is_empty(),
error::InvalidPromRemoteRequestSnafu {
msg: "remote write v2 custom native histogram must not use negative buckets"
.to_string(),
}
);
}
let (positive_buckets, negative_buckets) = if uses_float_counts {
(
histogram.positive_counts.len(),
histogram.negative_counts.len(),
)
} else {
(
histogram.positive_deltas.len(),
histogram.negative_deltas.len(),
)
};
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"
.to_string(),
},
)?,
)
};
validate_native_histogram_spans(
"positive",
&histogram.positive_spans,
positive_buckets,
bucket_index_range,
)?;
validate_native_histogram_spans(
"negative",
&histogram.negative_spans,
negative_buckets,
bucket_index_range,
)?;
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) {
error::InvalidPromRemoteRequestSnafu {
msg: format!(
"remote write v2 native histogram schema {schema} must be reduced before ingestion"
),
}
.fail()
} else {
error::InvalidPromRemoteRequestSnafu {
msg: format!("remote write v2 native histogram schema {schema} is unsupported"),
}
.fail()
}
}
fn validate_native_histogram_custom_values(histogram: &Histogram) -> Result<()> {
if histogram.schema != CUSTOM_BUCKETS_SCHEMA {
ensure!(
histogram.custom_values.is_empty(),
error::InvalidPromRemoteRequestSnafu {
msg: "remote write v2 standard native histogram must not use custom_values"
.to_string(),
}
);
return Ok(());
}
for value in &histogram.custom_values {
ensure!(
!value.is_nan() && *value != f64::INFINITY,
error::InvalidPromRemoteRequestSnafu {
msg: "remote write v2 custom native histogram custom_values must not contain +Inf or NaN"
.to_string(),
}
);
}
for values in histogram.custom_values.windows(2) {
ensure!(
values[0] < values[1],
error::InvalidPromRemoteRequestSnafu {
msg: "remote write v2 custom native histogram custom_values must be sorted"
.to_string(),
}
);
}
Ok(())
}
fn validate_native_histogram_spans(
name: &str,
spans: &[BucketSpan],
bucket_count: usize,
bucket_index_range: (i32, i32),
) -> Result<()> {
let span_len = spans
.iter()
.try_fold(0usize, |sum, span| sum.checked_add(span.length as usize))
.with_context(|| error::InvalidPromRemoteRequestSnafu {
msg: format!("remote write v2 native histogram {name} spans overflow"),
})?;
ensure!(
span_len == bucket_count,
error::InvalidPromRemoteRequestSnafu {
msg: format!(
"remote write v2 native histogram {name} spans describe {span_len} buckets, found {bucket_count}"
),
}
);
let mut current_index = 0i32;
for (span_index, span) in spans.iter().enumerate() {
ensure!(
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 {}",
span_index + 1,
span.offset
),
}
);
current_index = if span_index == 0 {
span.offset
} else {
current_index.checked_add(span.offset).with_context(|| {
error::InvalidPromRemoteRequestSnafu {
msg: format!(
"remote write v2 native histogram {name} span index overflows i32"
),
}
})?
};
for _ in 0..span.length {
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)
.context(error::InvalidPromRemoteRequestSnafu {
msg: format!(
"remote write v2 native histogram {name} span index overflows i32"
),
})?;
}
}
Ok(())
}
fn validate_float_native_histogram_counts(histogram: &Histogram) -> Result<()> {
let count = match histogram.count.as_ref() {
Some(Count::CountFloat(count)) => *count,
_ => 0.0,
};
ensure!(
count >= 0.0 || count.is_nan(),
error::InvalidPromRemoteRequestSnafu {
msg: "remote write v2 native histogram float count must not be negative".to_string(),
}
);
let zero_count = match histogram.zero_count.as_ref() {
Some(ZeroCount::ZeroCountFloat(zero_count)) => *zero_count,
_ => 0.0,
};
ensure!(
zero_count >= 0.0 || zero_count.is_nan(),
error::InvalidPromRemoteRequestSnafu {
msg: "remote write v2 native histogram float zero_count must not be negative"
.to_string(),
}
);
for (name, counts) in [
("positive", &histogram.positive_counts),
("negative", &histogram.negative_counts),
] {
for (index, count) in counts.iter().enumerate() {
ensure!(
*count >= 0.0 || count.is_nan(),
error::InvalidPromRemoteRequestSnafu {
msg: format!(
"remote write v2 native histogram {name} bucket {} count must not be negative",
index + 1
),
}
);
}
}
Ok(())
}
fn validate_integer_native_histogram_counts(
histogram: &Histogram,
positive_buckets: &[i64],
negative_buckets: &[i64],
) -> Result<()> {
let count = match histogram.count.as_ref() {
Some(Count::CountInt(count)) => *count,
_ => 0,
};
let zero_count = match histogram.zero_count.as_ref() {
Some(ZeroCount::ZeroCountInt(zero_count)) => *zero_count,
_ => 0,
};
let bucket_count = positive_buckets
.iter()
.chain(negative_buckets)
.try_fold(zero_count, |total, bucket| {
total.checked_add(*bucket as u64)
})
.with_context(|| error::InvalidPromRemoteRequestSnafu {
msg: "remote write v2 native histogram bucket total overflows u64".to_string(),
})?;
ensure!(
if histogram.sum.is_nan() {
bucket_count <= count
} else {
bucket_count == count
},
error::InvalidPromRemoteRequestSnafu {
msg: format!(
"remote write v2 native histogram has {bucket_count} observations in buckets, count is {count}"
),
}
);
Ok(())
}
fn native_histogram_zero_count_is_zero(histogram: &Histogram) -> bool {
match histogram.zero_count.as_ref() {
Some(ZeroCount::ZeroCountInt(zero_count)) => *zero_count == 0,
Some(ZeroCount::ZeroCountFloat(zero_count)) => *zero_count == 0.0,
None => true,
}
}
fn native_histogram_uses_float_counts(histogram: &Histogram) -> Result<bool> {
let uses_float_count = matches!(histogram.count, Some(Count::CountFloat(_)))
|| matches!(histogram.zero_count, Some(ZeroCount::ZeroCountFloat(_)));
let uses_int_count = matches!(histogram.count, Some(Count::CountInt(_)))
|| matches!(histogram.zero_count, Some(ZeroCount::ZeroCountInt(_)));
let uses_float_buckets =
!histogram.positive_counts.is_empty() || !histogram.negative_counts.is_empty();
let uses_int_buckets =
!histogram.positive_deltas.is_empty() || !histogram.negative_deltas.is_empty();
if matches!(
(&histogram.count, &histogram.zero_count),
(Some(Count::CountInt(_)), Some(ZeroCount::ZeroCountFloat(_)))
| (Some(Count::CountFloat(_)), Some(ZeroCount::ZeroCountInt(_)))
) {
return error::InvalidPromRemoteRequestSnafu {
msg: "remote write v2 native histogram count and zero_count must use the same integer or float family".to_string(),
}
.fail();
}
ensure!(
!(uses_float_buckets && uses_int_buckets),
error::InvalidPromRemoteRequestSnafu {
msg: "remote write v2 native histogram bucket counts must use either integer deltas or float counts".to_string(),
}
);
ensure!(
!(uses_float_count && uses_int_buckets),
error::InvalidPromRemoteRequestSnafu {
msg: "remote write v2 float native histogram must not use integer bucket deltas"
.to_string(),
}
);
ensure!(
!(uses_int_count && uses_float_buckets),
error::InvalidPromRemoteRequestSnafu {
msg: "remote write v2 integer native histogram must not use float bucket counts"
.to_string(),
}
);
Ok(uses_float_count || uses_float_buckets)
}
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 i32_span_lengths(name: &str, spans: &[BucketSpan]) -> Result<Vec<i32>> {
spans
.iter()
.map(|span| {
i32::try_from(span.length)
.ok()
.context(error::InvalidPromRemoteRequestSnafu {
msg: format!(
"remote write v2 native histogram {name} span length {} overflows i32",
span.length
),
})
})
.collect()
}
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)
.with_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 {
@@ -2371,6 +1884,25 @@ mod tests {
);
}
#[test]
fn test_into_context_req_preserves_exponential_zero_threshold() {
for zero_threshold in [-1.0, f64::NAN] {
let ctx_req = decode_test_request(request_with_histogram(Histogram {
zero_threshold,
..Default::default()
}))
.unwrap();
let rows = ctx_req.histograms.all_req().next().unwrap().rows.unwrap();
let Some(ValueData::F64Value(actual)) =
histogram_field_value(&rows, 0, ZERO_THRESHOLD_FIELD)
else {
panic!("expected zero threshold");
};
assert_eq!(zero_threshold.to_bits(), actual.to_bits());
}
}
#[test]
fn test_into_context_req_rejects_internal_histogram_labels() {
let mut request = test_util::request_with_labels_and_samples(
src/servers/src/query_handler.rs
+10 -1
View File
@@ -70,6 +70,15 @@ pub struct TraceIngestOutcome {
pub error_message: Option<String>,
}
/// Result of ingesting one OTLP metrics request or Arrow batch.
#[derive(Debug, Default, Clone)]
pub struct MetricsIngestOutcome {
pub write_cost: usize,
pub accepted_data_points: i64,
pub rejected_data_points: i64,
pub error_message: Option<String>,
}
#[async_trait]
pub trait InfluxdbLineProtocolHandler {
/// A successful request will not return a response.
@@ -133,7 +142,7 @@ pub trait OpenTelemetryProtocolHandler: PipelineHandler {
&self,
request: ExportMetricsServiceRequest,
ctx: QueryContextRef,
) -> Result<Output>;
) -> Result<MetricsIngestOutcome>;
/// Handling opentelemetry traces request
async fn traces(
src/session/src/protocol_ctx.rs
+2
View File
@@ -42,6 +42,7 @@ impl ProtocolCtx {
/// If true, all scope attributes will be promoted to the final table schema.
/// Along with the scope name, scope version and scope schema URL.
/// - `with_metric_engine`
/// - `experimental_enable_exponential_histogram`
/// - `is_legacy`
/// If the user uses OTLP metrics ingestion before v0.16, it uses the old path.
/// So we call this path 'legacy'.
@@ -53,6 +54,7 @@ pub struct OtlpMetricCtx {
pub resource_attrs: HashSet<String>,
pub promote_scope_attrs: bool,
pub with_metric_engine: bool,
pub experimental_enable_exponential_histogram: bool,
pub is_legacy: bool,
pub metric_type: MetricType,
pub metric_translation_strategy: OtlpMetricTranslationStrategy,
src/standalone/src/options.rs
+4 -1
View File
@@ -24,7 +24,7 @@ use file_engine::config::EngineConfig as FileEngineConfig;
use flow::FlowConfig;
use frontend::frontend::FrontendOptions;
use frontend::service_config::{
InfluxdbOptions, JaegerOptions, MysqlOptions, OpentsdbOptions, PostgresOptions,
InfluxdbOptions, JaegerOptions, MysqlOptions, OpentsdbOptions, OtlpOptions, PostgresOptions,
PromStoreOptions,
};
use mito2::config::MitoConfig;
@@ -56,6 +56,7 @@ pub struct StandaloneOptions {
pub opentsdb: OpentsdbOptions,
pub influxdb: InfluxdbOptions,
pub jaeger: JaegerOptions,
pub otlp: OtlpOptions,
pub prom_store: PromStoreOptions,
pub wal: DatanodeWalConfig,
pub storage: StorageConfig,
@@ -94,6 +95,7 @@ impl Default for StandaloneOptions {
opentsdb: OpentsdbOptions::default(),
influxdb: InfluxdbOptions::default(),
jaeger: JaegerOptions::default(),
otlp: OtlpOptions::default(),
prom_store: PromStoreOptions::default(),
wal: DatanodeWalConfig::default(),
storage: StorageConfig::default(),
@@ -153,6 +155,7 @@ impl StandaloneOptions {
opentsdb: cloned_opts.opentsdb,
influxdb: cloned_opts.influxdb,
jaeger: cloned_opts.jaeger,
otlp: cloned_opts.otlp,
prom_store: cloned_opts.prom_store,
meta_client: None,
logging: cloned_opts.logging,
tests-integration/src/test_util.rs
+20 -2
View File
@@ -490,7 +490,7 @@ pub async fn setup_test_http_app_with_frontend_and_slow_query_threshold(
.with_log_ingest_handler(instance.fe_instance().clone(), None, None)
.with_logs_handler(instance.fe_instance().clone())
.with_influxdb_handler(instance.fe_instance().clone())
.with_otlp_handler(instance.fe_instance().clone(), true)
.with_otlp_handler(instance.fe_instance().clone(), true, false)
.with_jaeger_handler(instance.fe_instance().clone())
.with_greptime_config_options(instance.opts.to_toml().unwrap())
.build();
@@ -510,6 +510,18 @@ pub async fn setup_test_http_app_with_frontend_and_user_provider(
user_provider,
None,
None,
false,
)
.await
}
pub async fn setup_test_http_app_with_otlp_exponential_histogram(
store_type: StorageType,
name: &str,
enabled: bool,
) -> (Router, TestGuard) {
setup_test_http_app_with_frontend_and_custom_options(
store_type, name, None, None, None, enabled,
)
.await
}
@@ -520,6 +532,7 @@ pub async fn setup_test_http_app_with_frontend_and_custom_options(
user_provider: Option<UserProviderRef>,
http_opts: Option<HttpOptions>,
memory_limiter: Option<ServerMemoryLimiter>,
experimental_enable_exponential_histogram: bool,
) -> (Router, TestGuard) {
let plugins = Plugins::new();
if let Some(user_provider) = user_provider.clone() {
@@ -541,7 +554,12 @@ pub async fn setup_test_http_app_with_frontend_and_custom_options(
.with_log_ingest_handler(instance.fe_instance().clone(), None, None)
.with_logs_handler(instance.fe_instance().clone())
.with_influxdb_handler(instance.fe_instance().clone())
.with_otlp_handler(instance.fe_instance().clone(), true)
.with_otlp_handler(
instance.fe_instance().clone(),
true,
experimental_enable_exponential_histogram,
)
.with_prometheus_handler(instance.fe_instance().clone())
.with_jaeger_handler(instance.fe_instance().clone())
.with_dashboard_handler(instance.fe_instance().clone())
.with_greptime_config_options(instance.opts.to_toml().unwrap());
tests-integration/tests/grpc.rs
+198 -1
View File
@@ -12,6 +12,8 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use std::sync::Arc;
use api::v1::alter_table_expr::Kind;
use api::v1::promql_request::Promql;
use api::v1::value::ValueData;
@@ -32,8 +34,20 @@ use common_recordbatch::RecordBatches;
use common_runtime::Runtime;
use common_runtime::runtime::{BuilderBuild, RuntimeTrait};
use common_test_util::find_workspace_path;
use otel_arrow_rust::proto::opentelemetry::arrow::v1::BatchArrowRecords;
use datatypes::arrow::array::{
Array, ArrayRef, Float64Array, Int32Array, ListBuilder, StringArray, StructArray,
TimestampNanosecondArray, UInt8Array, UInt16Array, UInt32Array, UInt64Array, UInt64Builder,
};
use datatypes::arrow::datatypes::{DataType, Field};
use datatypes::arrow::ipc::writer::StreamWriter;
use datatypes::arrow::record_batch::RecordBatch as ArrowRecordBatch;
use otel_arrow_rust::otlp::metrics::MetricType as ArrowMetricType;
use otel_arrow_rust::proto::opentelemetry::arrow::v1::arrow_metrics_service_client::ArrowMetricsServiceClient;
use otel_arrow_rust::proto::opentelemetry::arrow::v1::{
ArrowPayload, ArrowPayloadType, BatchArrowRecords, StatusCode as ArrowStatusCode,
};
use otel_arrow_rust::proto::opentelemetry::metrics::v1::AggregationTemporality;
use otel_arrow_rust::schema::consts as arrow_consts;
use servers::grpc::GrpcServerConfig;
use servers::grpc::builder::GrpcServerBuilder;
use servers::http::prometheus::{
@@ -85,6 +99,7 @@ macro_rules! grpc_tests {
test_auto_create_table_with_hints,
test_auto_create_table_disabled_by_config,
test_otel_arrow_auth,
test_otel_arrow_exponential_histogram,
test_insert_and_select,
test_dbname,
test_grpc_message_size_ok,
@@ -378,6 +393,188 @@ pub async fn test_otel_arrow_auth(store_type: StorageType) {
let _ = fe_grpc_server.shutdown().await;
}
// The pinned otel-arrow Producer cannot hash List-typed bucket schemas yet, so
// serialize these test-only record batches directly into the same Arrow stream format.
fn serialize_arrow_record_batch(record_batch: &ArrowRecordBatch) -> Vec<u8> {
let mut bytes = Vec::new();
let mut writer = StreamWriter::try_new(&mut bytes, record_batch.schema_ref()).unwrap();
writer.write(record_batch).unwrap();
writer.finish().unwrap();
drop(writer);
bytes
}
fn exponential_histogram_arrow_batch(batch_id: i64, scales: &[i32]) -> BatchArrowRecords {
let resource = StructArray::from(vec![(
Arc::new(Field::new(arrow_consts::ID, DataType::UInt16, true)),
Arc::new(UInt16Array::from(vec![0_u16])) as ArrayRef,
)]);
let scope = StructArray::from(vec![(
Arc::new(Field::new(arrow_consts::ID, DataType::UInt16, true)),
Arc::new(UInt16Array::from(vec![0_u16])) as ArrayRef,
)]);
let metrics = ArrowRecordBatch::try_from_iter(vec![
(
arrow_consts::ID,
Arc::new(UInt16Array::from(vec![0_u16])) as ArrayRef,
),
(arrow_consts::RESOURCE, Arc::new(resource) as ArrayRef),
(arrow_consts::SCOPE, Arc::new(scope) as ArrayRef),
(
arrow_consts::METRIC_TYPE,
Arc::new(UInt8Array::from(vec![
ArrowMetricType::ExponentialHistogram as u8,
])) as ArrayRef,
),
(
arrow_consts::NAME,
Arc::new(StringArray::from(vec!["otel.arrow.exponential.latency"])) as ArrayRef,
),
(
arrow_consts::AGGREGATION_TEMPORALITY,
Arc::new(Int32Array::from(vec![
AggregationTemporality::Cumulative as i32,
])) as ArrayRef,
),
])
.unwrap();
let point_count = scales.len();
let mut positive_counts = ListBuilder::new(UInt64Builder::new());
let mut negative_counts = ListBuilder::new(UInt64Builder::new());
for _ in scales {
positive_counts.values().append_slice(&[1, 2]);
positive_counts.append(true);
negative_counts.append(true);
}
let positive_counts = positive_counts.finish();
let negative_counts = negative_counts.finish();
let positive = StructArray::from(vec![
(
Arc::new(Field::new(
arrow_consts::EXP_HISTOGRAM_OFFSET,
DataType::Int32,
true,
)),
Arc::new(Int32Array::from(vec![-1; point_count])) as ArrayRef,
),
(
Arc::new(Field::new(
arrow_consts::EXP_HISTOGRAM_BUCKET_COUNTS,
positive_counts.data_type().clone(),
true,
)),
Arc::new(positive_counts) as ArrayRef,
),
]);
let negative = StructArray::from(vec![
(
Arc::new(Field::new(
arrow_consts::EXP_HISTOGRAM_OFFSET,
DataType::Int32,
true,
)),
Arc::new(Int32Array::from(vec![0; point_count])) as ArrayRef,
),
(
Arc::new(Field::new(
arrow_consts::EXP_HISTOGRAM_BUCKET_COUNTS,
negative_counts.data_type().clone(),
true,
)),
Arc::new(negative_counts) as ArrayRef,
),
]);
let data_points = ArrowRecordBatch::try_from_iter(vec![
(
arrow_consts::ID,
Arc::new(UInt32Array::from_iter_values(
(0..point_count).map(|id| u32::try_from(id).unwrap()),
)) as ArrayRef,
),
(
arrow_consts::PARENT_ID,
Arc::new(UInt16Array::from(vec![0_u16; point_count])) as ArrayRef,
),
(
arrow_consts::START_TIME_UNIX_NANO,
Arc::new(TimestampNanosecondArray::from(vec![
1_000_000_000;
point_count
])) as ArrayRef,
),
(
arrow_consts::TIME_UNIX_NANO,
Arc::new(TimestampNanosecondArray::from(vec![
3_000_000_000;
point_count
])) as ArrayRef,
),
(
arrow_consts::HISTOGRAM_COUNT,
Arc::new(UInt64Array::from(vec![4_u64; point_count])) as ArrayRef,
),
(
arrow_consts::HISTOGRAM_SUM,
Arc::new(Float64Array::from(vec![8.0; point_count])) as ArrayRef,
),
(
arrow_consts::EXP_HISTOGRAM_SCALE,
Arc::new(Int32Array::from(scales.to_vec())) as ArrayRef,
),
(
arrow_consts::EXP_HISTOGRAM_ZERO_COUNT,
Arc::new(UInt64Array::from(vec![1_u64; point_count])) as ArrayRef,
),
(
arrow_consts::EXP_HISTOGRAM_POSITIVE,
Arc::new(positive) as ArrayRef,
),
(
arrow_consts::EXP_HISTOGRAM_NEGATIVE,
Arc::new(negative) as ArrayRef,
),
(
arrow_consts::FLAGS,
Arc::new(UInt32Array::from(vec![0_u32; point_count])) as ArrayRef,
),
])
.unwrap();
BatchArrowRecords {
batch_id,
arrow_payloads: vec![
ArrowPayload {
schema_id: format!("metrics-{batch_id}"),
r#type: ArrowPayloadType::UnivariateMetrics as i32,
record: serialize_arrow_record_batch(&metrics),
},
ArrowPayload {
schema_id: format!("exp-histogram-{batch_id}"),
r#type: ArrowPayloadType::ExpHistogramDataPoints as i32,
record: serialize_arrow_record_batch(&data_points),
},
],
headers: vec![],
}
}
pub async fn test_otel_arrow_exponential_histogram(store_type: StorageType) {
let (_instance, server) =
setup_grpc_server(store_type, "test_otel_arrow_exponential_histogram").await;
let addr = server.bind_addr().unwrap().to_string();
let mut client = ArrowMetricsServiceClient::connect(format!("http://{addr}"))
.await
.unwrap();
let batch = exponential_histogram_arrow_batch(0, &[0]);
let request = Request::new(futures::stream::once(async { batch }));
let mut response = client.arrow_metrics(request).await.unwrap().into_inner();
let status = response.message().await.unwrap().unwrap();
assert_eq!(0, status.batch_id);
assert_eq!(ArrowStatusCode::InvalidArgument as i32, status.status_code);
assert!(status.status_message.contains("omits zero_threshold"));
let _ = server.shutdown().await;
}
fn basic_auth(username: &str, password: &str) -> String {
format!("Basic {}", basic_auth_credentials(username, password))
}
tests-integration/tests/http.rs
+144 -1
View File
@@ -50,7 +50,9 @@ use log_query::{AggFunc, Context, Limit, LogExpr, LogQuery, TimeFilter};
use loki_proto::logproto::{EntryAdapter, LabelPairAdapter, PushRequest, StreamAdapter};
use loki_proto::prost_types::Timestamp;
use opentelemetry_proto::tonic::collector::logs::v1::ExportLogsServiceRequest;
use opentelemetry_proto::tonic::collector::metrics::v1::ExportMetricsServiceRequest;
use opentelemetry_proto::tonic::collector::metrics::v1::{
ExportMetricsServiceRequest, ExportMetricsServiceResponse,
};
use opentelemetry_proto::tonic::collector::trace::v1::{
ExportTraceServiceRequest, ExportTraceServiceResponse,
};
@@ -160,6 +162,7 @@ macro_rules! http_tests {
test_pipeline_index_options,
test_otlp_metrics_new,
test_otlp_exponential_histogram,
test_otlp_metric_translation_strategies,
test_otlp_traces_v0,
test_otlp_traces_v1,
@@ -2201,6 +2204,11 @@ default_merge_mode = "last_non_null"
[jaeger]
enable = true
[otlp]
enable = true
experimental_enable_exponential_histogram = false
trace_ingest_chunk_size = 512
[prom_store]
enable = true
with_metric_engine = true
@@ -6709,6 +6717,140 @@ pub async fn test_otlp_metrics_new(store_type: StorageType) {
guard.remove_all().await;
}
pub async fn test_otlp_exponential_histogram(store_type: StorageType) {
use opentelemetry_proto::tonic::metrics::v1::{
AggregationTemporality, ExponentialHistogram, ExponentialHistogramDataPoint, Metric,
ResourceMetrics, ScopeMetrics, exponential_histogram_data_point, metric,
};
use tests_integration::test_util::setup_test_http_app_with_otlp_exponential_histogram;
common_telemetry::init_default_ut_logging();
let req = ExportMetricsServiceRequest {
resource_metrics: vec![ResourceMetrics {
scope_metrics: vec![ScopeMetrics {
metrics: vec![Metric {
name: "otlp.exponential.latency".to_string(),
data: Some(metric::Data::ExponentialHistogram(ExponentialHistogram {
data_points: vec![
ExponentialHistogramDataPoint {
start_time_unix_nano: 1_000_000_000,
time_unix_nano: 3_000_000_000,
count: 4,
sum: Some(8.0),
scale: 0,
zero_count: 1,
positive: Some(exponential_histogram_data_point::Buckets {
offset: -1,
bucket_counts: vec![1, 2],
}),
..Default::default()
},
ExponentialHistogramDataPoint {
start_time_unix_nano: 1_000_000_000,
time_unix_nano: 4_000_000_000,
scale: -5,
..Default::default()
},
],
aggregation_temporality: AggregationTemporality::Cumulative as i32,
})),
..Default::default()
}],
..Default::default()
}],
..Default::default()
}],
};
let body = req.encode_to_vec();
let headers = || {
vec![(
HeaderName::from_static("content-type"),
HeaderValue::from_static("application/x-protobuf"),
)]
};
let (app, mut guard) = setup_test_http_app_with_otlp_exponential_histogram(
store_type,
"test_otlp_exponential_histogram_disabled",
false,
)
.await;
let client = TestClient::new(app).await;
let res = send_req(
&client,
headers(),
"/v1/otlp/v1/metrics",
body.clone(),
false,
)
.await;
assert_eq!(StatusCode::BAD_REQUEST, res.status());
let status = GoogleRpcStatus::decode(res.bytes().await.as_ref()).unwrap();
assert_eq!(3, status.code);
assert!(
status
.message
.contains("otlp.experimental_enable_exponential_histogram")
);
validate_data(
"otlp_exponential_histogram_disabled_no_table",
&client,
"select count(*) from information_schema.tables where table_name = 'otlp_exponential_latency';",
"[[0]]",
)
.await;
guard.remove_all().await;
let (app, mut guard) = setup_test_http_app_with_otlp_exponential_histogram(
store_type,
"test_otlp_exponential_histogram_enabled",
true,
)
.await;
let client = TestClient::new(app).await;
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();
let partial_success = response.partial_success.unwrap();
assert_eq!(1, partial_success.rejected_data_points);
assert!(partial_success.error_message.contains("scale -5"));
validate_data(
"otlp_exponential_histogram_row",
&client,
"select greptime_timestamp, greptime_native_histogram from otlp_exponential_latency;",
"[[3000,{\"count_f64\":null,\"count_i64\":4,\"custom_values\":[],\"negative_buckets_f64\":[],\"negative_buckets_i64\":[],\"negative_span_lengths\":[],\"negative_span_offsets\":[],\"positive_buckets_f64\":[],\"positive_buckets_i64\":[1,2],\"positive_span_lengths\":[2],\"positive_span_offsets\":[0],\"reset_hint\":0,\"schema\":0,\"start_timestamp\":1000,\"sum\":8.0,\"zero_count_f64\":null,\"zero_count_i64\":1,\"zero_threshold\":0.0}]]",
)
.await;
validate_data(
"otlp_exponential_histogram_metadata",
&client,
"select count(*) from information_schema.tables where table_name = 'otlp_exponential_latency' and create_options like '%greptime.semantic.metric.type=histogram%' and create_options like '%greptime.semantic.metric.temporality=cumulative%';",
"[[1]]",
)
.await;
for query in [
"histogram_count(otlp_exponential_latency)",
"histogram_sum(otlp_exponential_latency)",
"histogram_quantile(0.5,otlp_exponential_latency)",
"histogram_count(rate(otlp_exponential_latency[5s]))",
] {
let res = client
.get(&format!(
"/v1/prometheus/api/v1/query?query={}&time=3",
encode(query)
))
.send()
.await;
assert_eq!(StatusCode::OK, res.status(), "query: {query}");
let body = res.text().await;
assert!(body.contains("otlp_exponential_latency"), "{query}: {body}");
}
guard.remove_all().await;
}
pub async fn test_otlp_metric_translation_strategies(store_type: StorageType) {
common_telemetry::init_default_ut_logging();
let (app, mut guard) =
@@ -10646,6 +10788,7 @@ pub async fn test_http_memory_limit(store_type: StorageType) {
None,
Some(http_opts),
Some(memory_limiter),
false,
)
.await;