diff --git a/config/config.md b/config/config.md index 1b18565cc0..91cb4fe766 100644 --- a/config/config.md +++ b/config/config.md @@ -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. | diff --git a/config/frontend.example.toml b/config/frontend.example.toml index ec1f9c89b4..3635a63f06 100644 --- a/config/frontend.example.toml +++ b/config/frontend.example.toml @@ -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 diff --git a/config/standalone.example.toml b/config/standalone.example.toml index 5ada605b92..72ba992cce 100644 --- a/config/standalone.example.toml +++ b/config/standalone.example.toml @@ -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 diff --git a/docs/rfcs/2026-08-04-native-histograms.md b/docs/rfcs/2026-08-04-native-histograms.md index 1e7692bf51..c714b35351 100644 --- a/docs/rfcs/2026-08-04-native-histograms.md +++ b/docs/rfcs/2026-08-04-native-histograms.md @@ -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. diff --git a/src/cmd/src/standalone.rs b/src/cmd/src/standalone.rs index 2ffcc5e729..2b5bd5e8ce 100644 --- a/src/cmd/src/standalone.rs +++ b/src/cmd/src/standalone.rs @@ -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] diff --git a/src/cmd/tests/load_config_test.rs b/src/cmd/tests/load_config_test.rs index fd32680f4c..70594d0653 100644 --- a/src/cmd/tests/load_config_test.rs +++ b/src/cmd/tests/load_config_test.rs @@ -153,6 +153,12 @@ fn test_load_frontend_example_config() { let options = GreptimeOptions::::load_layered_options(example_config.to_str(), "") .unwrap(); + assert!( + !options + .component + .otlp + .experimental_enable_exponential_histogram + ); let expected = GreptimeOptions:: { component: FrontendOptions { default_timezone: Some("UTC".to_string()), @@ -329,6 +335,12 @@ fn test_load_standalone_example_config() { let options = GreptimeOptions::::load_layered_options(example_config.to_str(), "") .unwrap(); + assert!( + !options + .component + .otlp + .experimental_enable_exponential_histogram + ); let expected = GreptimeOptions:: { 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::::load_layered_options(config.path().to_str(), "") + .unwrap(); + assert!( + frontend + .component + .otlp + .experimental_enable_exponential_histogram + ); + + let standalone = + GreptimeOptions::::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(); diff --git a/src/frontend/src/frontend.rs b/src/frontend/src/frontend.rs index b8fea7b604..18cedc0545 100644 --- a/src/frontend/src/frontend.rs +++ b/src/frontend/src/frontend.rs @@ -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] diff --git a/src/frontend/src/instance/otlp.rs b/src/frontend/src/instance/otlp.rs index 459a05321c..0808619853 100644 --- a/src/frontend/src/instance/otlp.rs +++ b/src/frontend/src/instance/otlp.rs @@ -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 { + ) -> ServerResult { self.plugins .get::() .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)] diff --git a/src/frontend/src/server.rs b/src/frontend/src/server.rs index 8fdebd8683..f0dc226ba8 100644 --- a/src/frontend/src/server.rs +++ b/src/frontend/src/server.rs @@ -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 { diff --git a/src/frontend/src/service_config/otlp.rs b/src/frontend/src/service_config/otlp.rs index 3b01f42289..5dcab2c8ce 100644 --- a/src/frontend/src/service_config/otlp.rs +++ b/src/frontend/src/service_config/otlp.rs @@ -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::(&serialized).unwrap(), options); } } diff --git a/src/operator/src/insert.rs b/src/operator/src/insert.rs index fabc1b3778..ead517ab80 100644 --- a/src/operator/src/insert.rs +++ b/src/operator/src/insert.rs @@ -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, ) diff --git a/src/servers/src/http.rs b/src/servers/src/http.rs index bf7478eacf..c0f6af8184 100644 --- a/src/servers/src/http.rs +++ b/src/servers/src/http.rs @@ -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( otlp_handler: OpenTelemetryProtocolHandlerRef, with_metric_engine: bool, + experimental_enable_exponential_histogram: bool, ) -> Router { 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, }) } diff --git a/src/servers/src/http/otlp.rs b/src/servers/src/http/otlp.rs index 6540f6d4c4..d04022cdf8 100644 --- a/src/servers/src/http/otlp.rs +++ b/src/servers/src/http/otlp.rs @@ -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>) -> 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>, bytes: Bytes, -) -> Result> { +) -> Result { 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 { 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 IntoResponse for OtlpResponse { fn into_response(self) -> axum::response::Response { let mut header_map = write_cost_header_map(self.write_cost); diff --git a/src/servers/src/lib.rs b/src/servers/src/lib.rs index 2a3796aab2..79fd392a1a 100644 --- a/src/servers/src/lib.rs +++ b/src/servers/src/lib.rs @@ -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; diff --git a/src/servers/src/native_histogram.rs b/src/servers/src/native_histogram.rs new file mode 100644 index 0000000000..f1dae52699 --- /dev/null +++ b/src/servers/src/native_histogram.rs @@ -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 = std::result::Result; + +/// Returns the canonical column schema for a native histogram value. +pub(crate) fn native_histogram_column_schema() -> Result { + 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 { + 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> { + 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(¤t_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 { + 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) -> Value { + Value { value_data } +} + +fn list_value(values: impl IntoIterator) -> Value { + pb_value(ValueData::ListValue(ListValue { + items: values.into_iter().map(pb_value).collect(), + })) +} + +fn i32_list_value(values: impl IntoIterator) -> Value { + list_value(values.into_iter().map(ValueData::I32Value)) +} + +fn i32_span_lengths(name: &str, spans: &[BucketSpan]) -> Result> { + 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) -> Value { + list_value(values.into_iter().map(ValueData::I64Value)) +} + +fn f64_list_value(values: impl IntoIterator) -> Value { + list_value(values.into_iter().map(ValueData::F64Value)) +} + +fn bucket_counts_from_deltas(deltas: &[i64]) -> Result> { + 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")); + } +} diff --git a/src/servers/src/otel_arrow.rs b/src/servers/src/otel_arrow.rs index f905b39f5b..320ef87daf 100644 --- a/src/servers/src/otel_arrow.rs +++ b/src/servers/src/otel_arrow.rs @@ -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 { handler: T, @@ -42,6 +50,31 @@ impl OtelArrowServiceHandler { } } +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 { type ArrowMetricsStream = futures::channel::mpsc::Receiver>; @@ -55,6 +88,11 @@ impl ArrowMetricsService for OtelArrowServiceHandler 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); + } +} diff --git a/src/servers/src/otlp/metrics.rs b/src/servers/src/otlp/metrics.rs index 6d94b3a139..332fa46f1d 100644 --- a/src/servers/src/otlp/metrics.rs +++ b/src/servers/src/otlp/metrics.rs @@ -13,9 +13,16 @@ // limitations under the License. use ahash::HashSet; -use api::v1::{RowInsertRequests, Value}; +use api::greptime_proto::io::prometheus::write::v2::histogram::{ + Count as PromCount, ResetHint, ZeroCount as PromZeroCount, +}; +use api::greptime_proto::io::prometheus::write::v2::{BucketSpan, Histogram as PromHistogram}; +use api::v1::value::ValueData; +use api::v1::{RowInsertRequests, SemanticType, Value}; use common_grpc::precision::Precision; +use common_query::native_histogram::native_histogram_value_type; use common_query::prelude::{GREPTIME_COUNT, greptime_timestamp, greptime_value}; +use common_query::prometheus::PROMETHEUS_STALE_NAN_BITS; use lazy_static::lazy_static; use otel_arrow_rust::proto::opentelemetry::collector::metrics::v1::ExportMetricsServiceRequest; use otel_arrow_rust::proto::opentelemetry::common::v1::{AnyValue, KeyValue, any_value}; @@ -26,8 +33,10 @@ use table::requests::{ SEMANTIC_METRIC_TEMPORALITY, SEMANTIC_METRIC_TYPE, SEMANTIC_METRIC_UNIT, }; -use crate::error::Result; +use crate::error::{self, Result}; +use crate::native_histogram::{encode_native_histogram, native_histogram_column_schema}; use crate::otlp::trace::{KEY_SERVICE_INSTANCE_ID, KEY_SERVICE_NAME}; +use crate::query_handler::MetricsIngestOutcome; use crate::row_writer::{self, MultiTableData, TableData}; pub use crate::semantic::SemanticIndex; use crate::semantic::{ @@ -82,6 +91,9 @@ lazy_static! { const OTEL_SCOPE_NAME: &str = "name"; const OTEL_SCOPE_VERSION: &str = "version"; const OTEL_SCOPE_SCHEMA_URL: &str = "schema_url"; +const MIN_EXPONENTIAL_HISTOGRAM_SCALE: i32 = -4; +const MAX_EXPONENTIAL_HISTOGRAM_SCALE: i32 = 8; +const MAX_REJECTION_MESSAGE_BYTES: usize = 512; /// Convert OpenTelemetry metrics to GreptimeDB insert requests /// @@ -94,9 +106,15 @@ const OTEL_SCOPE_SCHEMA_URL: &str = "schema_url"; pub fn to_grpc_insert_requests( request: ExportMetricsServiceRequest, metric_ctx: &mut OtlpMetricCtx, -) -> Result<(RowInsertRequests, usize, SemanticIndex)> { +) -> Result<( + RowInsertRequests, + usize, + SemanticIndex, + MetricsIngestOutcome, +)> { let mut table_writer = MultiTableData::default(); let mut semantic_index = SemanticIndex::default(); + let mut outcome = MetricsIngestOutcome::default(); for resource in &request.resource_metrics { let resource_attrs = resource.resource.as_ref().map(|r| { @@ -123,13 +141,46 @@ pub fn to_grpc_insert_requests( scope_attrs.as_ref(), metric_ctx, &mut semantic_index, + &mut outcome, )?; } } } let (requests, rows) = table_writer.into_row_insert_requests(); - Ok((requests, rows, semantic_index)) + validate_sample_kinds(&requests)?; + Ok((requests, rows, semantic_index, outcome)) +} + +fn validate_sample_kinds(requests: &RowInsertRequests) -> Result<()> { + for request in &requests.inserts { + let Some(rows) = &request.rows else { + continue; + }; + let field_count = rows + .schema + .iter() + .filter(|column| column.semantic_type == SemanticType::Field as i32) + .count(); + let has_native_histogram = rows.schema.iter().any(|column| { + column.semantic_type == SemanticType::Field as i32 + && api::helper::is_column_type_value_eq( + column.datatype, + column.datatype_extension.clone(), + native_histogram_value_type(), + ) + }); + if has_native_histogram && field_count != 1 { + return Err(error::InvalidParameterSnafu { + reason: format!( + "OTLP metric `{}` cannot mix native histogram and float sample fields", + request.table_name + ), + } + .build()); + } + } + Ok(()) } /// The tables a metric emits and their per-table `metric.type`. Histogram fans @@ -159,8 +210,10 @@ fn emitted_semantic_tables( (format!("{base}{COUNT_TABLE_SUFFIX}"), METRIC_TYPE_COUNTER), (format!("{base}{SUM_TABLE_SUFFIX}"), METRIC_TYPE_COUNTER), ], - // ExponentialHistogram is a no-op today; Init never reaches encoding. - MetricType::ExponentialHistogram | MetricType::Init => vec![], + MetricType::ExponentialHistogram => { + vec![(base.to_string(), METRIC_TYPE_HISTOGRAM)] + } + MetricType::Init => vec![], } } @@ -170,6 +223,7 @@ fn temporality_value(data: &metric::Data) -> Option<&'static str> { let raw = match data { metric::Data::Sum(sum) => sum.aggregation_temporality, metric::Data::Histogram(hist) => hist.aggregation_temporality, + metric::Data::ExponentialHistogram(hist) => hist.aggregation_temporality, _ => return None, }; match AggregationTemporality::try_from(raw) { @@ -311,6 +365,7 @@ fn encode_metrics( scope_attrs: Option<&Vec>, metric_ctx: &OtlpMetricCtx, semantic_index: &mut SemanticIndex, + outcome: &mut MetricsIngestOutcome, ) -> Result<()> { let name = if metric_ctx.is_legacy { legacy_normalize_otlp_name(&metric.name) @@ -322,12 +377,7 @@ fn encode_metrics( ) }; - // Stamp semantic metadata against the same table name(s) the data is written - // to below. `unit` is captured here (it is otherwise discarded by the row - // encoders) along with the declared type/temporality. - record_metric_semantics(semantic_index, metric, &name, metric_ctx); - - if let Some(data) = &metric.data { + let emitted = if let Some(data) = &metric.data { match data { metric::Data::Gauge(gauge) => { encode_gauge( @@ -338,6 +388,8 @@ fn encode_metrics( scope_attrs, metric_ctx, )?; + add_accepted_data_points(outcome, gauge.data_points.len())?; + !gauge.data_points.is_empty() } metric::Data::Sum(sum) => { encode_sum( @@ -348,6 +400,8 @@ fn encode_metrics( scope_attrs, metric_ctx, )?; + add_accepted_data_points(outcome, sum.data_points.len())?; + !sum.data_points.is_empty() } metric::Data::Summary(summary) => { encode_summary( @@ -358,6 +412,8 @@ fn encode_metrics( scope_attrs, metric_ctx, )?; + add_accepted_data_points(outcome, summary.data_points.len())?; + !summary.data_points.is_empty() } metric::Data::Histogram(hist) => { encode_histogram( @@ -368,16 +424,384 @@ fn encode_metrics( scope_attrs, metric_ctx, )?; + add_accepted_data_points(outcome, hist.data_points.len())?; + !hist.data_points.is_empty() } - // TODO: Convert OTLP exponential histograms into the canonical native-histogram - // Struct. See docs/rfcs/2026-08-04-native-histograms.md. - metric::Data::ExponentialHistogram(_hist) => {} + metric::Data::ExponentialHistogram(hist) => encode_exponential_histogram( + table_writer, + &name, + hist, + resource_attrs, + scope_attrs, + metric_ctx, + outcome, + )?, } + } else { + false + }; + + if emitted { + // Stamp semantic metadata only after at least one row was accepted. + record_metric_semantics(semantic_index, metric, &name, metric_ctx); } Ok(()) } +fn add_accepted_data_points(outcome: &mut MetricsIngestOutcome, count: usize) -> Result<()> { + let count = i64::try_from(count).map_err(|_| { + error::InvalidParameterSnafu { + reason: "OTLP metrics data-point count exceeds i64", + } + .build() + })?; + outcome.accepted_data_points = + outcome + .accepted_data_points + .checked_add(count) + .ok_or_else(|| { + error::InvalidParameterSnafu { + reason: "OTLP accepted data-point count overflows i64", + } + .build() + })?; + Ok(()) +} + +fn reject_data_points( + outcome: &mut MetricsIngestOutcome, + count: usize, + reason: impl FnOnce() -> String, +) -> Result<()> { + if count == 0 { + return Ok(()); + } + let count = i64::try_from(count).map_err(|_| { + error::InvalidParameterSnafu { + reason: "OTLP rejected data-point count exceeds i64", + } + .build() + })?; + outcome.rejected_data_points = + outcome + .rejected_data_points + .checked_add(count) + .ok_or_else(|| { + error::InvalidParameterSnafu { + reason: "OTLP rejected data-point count overflows i64", + } + .build() + })?; + append_rejection_message(&mut outcome.error_message, reason); + Ok(()) +} + +fn append_rejection_message(message: &mut Option, reason: impl FnOnce() -> String) { + let message = message.get_or_insert_with(String::new); + let separator = if message.is_empty() { "" } else { "; " }; + let Some(available) = MAX_REJECTION_MESSAGE_BYTES.checked_sub(message.len()) else { + return; + }; + if available <= separator.len() { + return; + } + let reason = reason(); + message.push_str(separator); + + let available = MAX_REJECTION_MESSAGE_BYTES - message.len(); + if reason.len() <= available { + message.push_str(&reason); + return; + } + + const ELLIPSIS: &str = "..."; + let mut end = available.saturating_sub(ELLIPSIS.len()); + while !reason.is_char_boundary(end) { + end -= 1; + } + message.push_str(&reason[..end]); + if available >= ELLIPSIS.len() { + message.push_str(ELLIPSIS); + } +} + +fn encode_exponential_histogram( + table_writer: &mut MultiTableData, + name: &str, + histogram: &ExponentialHistogram, + resource_attrs: Option<&Vec>, + scope_attrs: Option<&Vec>, + metric_ctx: &OtlpMetricCtx, + outcome: &mut MetricsIngestOutcome, +) -> Result { + if !metric_ctx.experimental_enable_exponential_histogram { + reject_data_points(outcome, histogram.data_points.len(), || { + format!( + "metric `{name}` uses OTLP exponential histograms; set otlp.experimental_enable_exponential_histogram = true to enable ingestion" + ) + })?; + return Ok(false); + } + + match AggregationTemporality::try_from(histogram.aggregation_temporality) { + Ok(AggregationTemporality::Cumulative) => {} + Ok(AggregationTemporality::Delta) => { + reject_data_points(outcome, histogram.data_points.len(), || { + format!( + "metric `{name}` uses delta OTLP exponential histograms; only cumulative temporality is supported" + ) + })?; + return Ok(false); + } + _ => { + reject_data_points(outcome, histogram.data_points.len(), || { + format!( + "metric `{name}` has unspecified OTLP exponential histogram temporality; cumulative temporality is required" + ) + })?; + return Ok(false); + } + } + + let column_schema = native_histogram_column_schema().map_err(|error| { + error::InternalSnafu { + err_msg: error.to_string(), + } + .build() + })?; + let mut emitted = false; + for (index, data_point) in histogram.data_points.iter().enumerate() { + let (value, timestamp_nanos) = match exponential_histogram_value(data_point) { + Ok(value) => value, + Err(reason) => { + reject_data_points(outcome, 1, || { + format!("metric `{name}` data point {index}: {reason}") + })?; + continue; + } + }; + + let table = table_writer.get_or_default_table_data( + name, + APPROXIMATE_COLUMN_COUNT, + histogram.data_points.len(), + ); + let mut row = table.alloc_one_row(); + write_tags_and_timestamp( + table, + &mut row, + resource_attrs, + scope_attrs, + Some(data_point.attributes.as_ref()), + timestamp_nanos, + metric_ctx, + )?; + row_writer::write_by_schema( + table, + std::iter::once((column_schema.clone(), Some(value))), + &mut row, + )?; + table.add_row(row); + add_accepted_data_points(outcome, 1)?; + emitted = true; + } + + Ok(emitted) +} + +fn exponential_histogram_value( + data_point: &ExponentialHistogramDataPoint, +) -> std::result::Result<(ValueData, i64), String> { + if data_point.start_time_unix_nano > data_point.time_unix_nano { + return Err(format!( + "start_time_unix_nano {} exceeds time_unix_nano {}", + data_point.start_time_unix_nano, data_point.time_unix_nano + )); + } + + let timestamp_nanos = i64::try_from(data_point.time_unix_nano) + .map_err(|_| format!("time_unix_nano {} overflows i64", data_point.time_unix_nano))?; + let timestamp = timestamp_millis(data_point.time_unix_nano, "time_unix_nano")?; + let start_timestamp = + timestamp_millis(data_point.start_time_unix_nano, "start_time_unix_nano")?; + + let no_recorded_value = data_point.flags & DataPointFlags::NoRecordedValueMask as u32 != 0; + let histogram = if no_recorded_value { + PromHistogram { + sum: f64::from_bits(PROMETHEUS_STALE_NAN_BITS), + schema: 0, + zero_threshold: 0.0, + reset_hint: ResetHint::Unspecified as i32, + timestamp, + start_timestamp, + count: Some(PromCount::CountInt(0)), + zero_count: Some(PromZeroCount::ZeroCountInt(0)), + ..Default::default() + } + } else { + if data_point.scale < MIN_EXPONENTIAL_HISTOGRAM_SCALE { + return Err(format!( + "scale {} is unsupported; minimum supported scale is {}", + data_point.scale, MIN_EXPONENTIAL_HISTOGRAM_SCALE + )); + } + if !data_point.zero_threshold.is_finite() || data_point.zero_threshold < 0.0 { + return Err(format!( + "zero_threshold {} must be finite and non-negative", + data_point.zero_threshold + )); + } + + let downscale_shift = if data_point.scale > MAX_EXPONENTIAL_HISTOGRAM_SCALE { + let shift = data_point + .scale + .checked_sub(MAX_EXPONENTIAL_HISTOGRAM_SCALE) + .ok_or_else(|| "downscale shift overflows i32".to_string())?; + u32::try_from(shift).map_err(|_| "downscale shift exceeds u32".to_string())? + } else { + 0 + }; + let (positive_spans, positive_deltas, positive_count) = + convert_bucket_range("positive", data_point.positive.as_ref(), downscale_shift)?; + let (negative_spans, negative_deltas, negative_count) = + convert_bucket_range("negative", data_point.negative.as_ref(), downscale_shift)?; + let bucket_count = data_point + .zero_count + .checked_add(positive_count) + .and_then(|count| count.checked_add(negative_count)) + .ok_or_else(|| "bucket observation total overflows u64".to_string())?; + if bucket_count != data_point.count { + return Err(format!( + "buckets contain {bucket_count} observations, declared count is {}", + data_point.count + )); + } + + i64::try_from(data_point.count) + .map_err(|_| format!("count {} overflows i64", data_point.count))?; + i64::try_from(data_point.zero_count) + .map_err(|_| format!("zero_count {} overflows i64", data_point.zero_count))?; + + let sum = match data_point.sum { + Some(sum) if sum.is_nan() => f64::NAN, + Some(sum) => sum, + None => f64::NAN, + }; + PromHistogram { + sum, + schema: data_point.scale.min(MAX_EXPONENTIAL_HISTOGRAM_SCALE), + zero_threshold: data_point.zero_threshold, + negative_spans, + negative_deltas, + positive_spans, + positive_deltas, + reset_hint: ResetHint::Unspecified as i32, + timestamp, + start_timestamp, + count: Some(PromCount::CountInt(data_point.count)), + zero_count: Some(PromZeroCount::ZeroCountInt(data_point.zero_count)), + ..Default::default() + } + }; + + encode_native_histogram(&histogram) + .map(|value| (value, timestamp_nanos)) + .map_err(|error| format!("OTLP exponential histogram cannot be encoded: {error}")) +} + +fn timestamp_millis(timestamp_nanos: u64, name: &str) -> std::result::Result { + i64::try_from(timestamp_nanos / 1_000_000) + .map_err(|_| format!("{name} {timestamp_nanos} milliseconds overflow i64")) +} + +fn convert_bucket_range( + name: &str, + buckets: Option<&exponential_histogram_data_point::Buckets>, + downscale_shift: u32, +) -> std::result::Result<(Vec, Vec, u64), String> { + let Some(buckets) = buckets else { + return Ok((Vec::new(), Vec::new(), 0)); + }; + if buckets.bucket_counts.is_empty() { + return Ok((Vec::new(), Vec::new(), 0)); + } + + let mut merged = Vec::<(i32, u64)>::with_capacity(buckets.bucket_counts.len()); + let mut total = 0u64; + for (position, count) in buckets.bucket_counts.iter().copied().enumerate() { + let position = + i32::try_from(position).map_err(|_| format!("{name} bucket length exceeds i32"))?; + let source_index = buckets.offset.checked_add(position).ok_or_else(|| { + format!( + "{name} bucket index overflows i32 at offset {} and position {position}", + buckets.offset + ) + })?; + let target_index = downscale_bucket_index(source_index, downscale_shift)? + .checked_add(1) + .ok_or_else(|| format!("{name} shifted bucket index overflows i32"))?; + if let Some((last_index, last_count)) = merged.last_mut() { + if *last_index == target_index { + *last_count = last_count + .checked_add(count) + .ok_or_else(|| format!("{name} merged bucket count overflows u64"))?; + total = total + .checked_add(count) + .ok_or_else(|| format!("{name} bucket count total overflows u64"))?; + continue; + } + let next_index = last_index + .checked_add(1) + .ok_or_else(|| format!("{name} target bucket index overflows i32"))?; + if target_index != next_index { + return Err(format!( + "{name} bucket indexes are not contiguous after downscaling" + )); + } + } + total = total + .checked_add(count) + .ok_or_else(|| format!("{name} bucket count total overflows u64"))?; + merged.push((target_index, count)); + } + + let length = u32::try_from(merged.len()) + .map_err(|_| format!("{name} bucket span length exceeds u32"))?; + let span = BucketSpan { + offset: merged[0].0, + length, + }; + let mut deltas = Vec::with_capacity(merged.len()); + let mut previous = 0i64; + for (_, count) in merged { + let count = i64::try_from(count) + .map_err(|_| format!("{name} bucket count {count} overflows i64"))?; + let delta = count + .checked_sub(previous) + .ok_or_else(|| format!("{name} bucket delta overflows i64"))?; + deltas.push(delta); + previous = count; + } + + Ok((vec![span], deltas, total)) +} + +fn downscale_bucket_index(index: i32, downscale_shift: u32) -> std::result::Result { + if downscale_shift == 0 { + return Ok(index); + } + if downscale_shift >= i32::BITS { + return Ok(if index < 0 { -1 } else { 0 }); + } + + let divisor = 1i64 + .checked_shl(downscale_shift) + .ok_or_else(|| format!("downscale shift {downscale_shift} is invalid"))?; + i32::try_from(i64::from(index).div_euclid(divisor)) + .map_err(|_| format!("downscaled bucket index {index} overflows i32")) +} + #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum AttributeType { Resource, @@ -620,17 +1044,17 @@ fn encode_histogram( let count_table_name = format!("{}{}", normalized_name, COUNT_TABLE_SUFFIX); let data_points_len = hist.data_points.len(); - // Note that the row and columns number here is approximate - let mut bucket_table = TableData::new(APPROXIMATE_COLUMN_COUNT, data_points_len * 3); - let mut sum_table = TableData::new(APPROXIMATE_COLUMN_COUNT, data_points_len); - let mut count_table = TableData::new(APPROXIMATE_COLUMN_COUNT, data_points_len); - for data_point in &hist.data_points { + let bucket_table = table_writer.get_or_default_table_data( + &bucket_table_name, + APPROXIMATE_COLUMN_COUNT, + data_points_len * 3, + ); let mut accumulated_count = 0; for (idx, count) in data_point.bucket_counts.iter().enumerate() { let mut bucket_row = bucket_table.alloc_one_row(); write_tags_and_timestamp( - &mut bucket_table, + bucket_table, &mut bucket_row, resource_attrs, scope_attrs, @@ -641,7 +1065,7 @@ fn encode_histogram( if let Some(upper_bounds) = data_point.explicit_bounds.get(idx) { row_writer::write_tag( - &mut bucket_table, + bucket_table, HISTOGRAM_LE_COLUMN, upper_bounds, &mut bucket_row, @@ -649,7 +1073,7 @@ fn encode_histogram( } else if idx == data_point.explicit_bounds.len() { // The last bucket row_writer::write_tag( - &mut bucket_table, + bucket_table, HISTOGRAM_LE_COLUMN, f64::INFINITY, &mut bucket_row, @@ -658,7 +1082,7 @@ fn encode_histogram( accumulated_count += count; row_writer::write_f64( - &mut bucket_table, + bucket_table, greptime_value(), accumulated_count as f64, &mut bucket_row, @@ -668,9 +1092,14 @@ fn encode_histogram( } if let Some(sum) = data_point.sum { + let sum_table = table_writer.get_or_default_table_data( + &sum_table_name, + APPROXIMATE_COLUMN_COUNT, + data_points_len, + ); let mut sum_row = sum_table.alloc_one_row(); write_tags_and_timestamp( - &mut sum_table, + sum_table, &mut sum_row, resource_attrs, scope_attrs, @@ -679,13 +1108,18 @@ fn encode_histogram( metric_ctx, )?; - row_writer::write_f64(&mut sum_table, greptime_value(), sum, &mut sum_row)?; + row_writer::write_f64(sum_table, greptime_value(), sum, &mut sum_row)?; sum_table.add_row(sum_row); } + let count_table = table_writer.get_or_default_table_data( + &count_table_name, + APPROXIMATE_COLUMN_COUNT, + data_points_len, + ); let mut count_row = count_table.alloc_one_row(); write_tags_and_timestamp( - &mut count_table, + count_table, &mut count_row, resource_attrs, scope_attrs, @@ -695,7 +1129,7 @@ fn encode_histogram( )?; row_writer::write_f64( - &mut count_table, + count_table, greptime_value(), data_point.count as f64, &mut count_row, @@ -703,10 +1137,6 @@ fn encode_histogram( count_table.add_row(count_row); } - table_writer.add_table_data(bucket_table_name, bucket_table); - table_writer.add_table_data(sum_table_name, sum_table); - table_writer.add_table_data(count_table_name, count_table); - Ok(()) } @@ -851,6 +1281,7 @@ mod tests { use otel_arrow_rust::proto::opentelemetry::metrics::v1::{ AggregationTemporality, HistogramDataPoint, NumberDataPoint, SummaryDataPoint, }; + use otel_arrow_rust::proto::opentelemetry::resource::v1::Resource; use super::*; @@ -1361,4 +1792,560 @@ mod tests { ); } } + + fn exponential_buckets( + offset: i32, + bucket_counts: Vec, + ) -> exponential_histogram_data_point::Buckets { + exponential_histogram_data_point::Buckets { + offset, + bucket_counts, + } + } + + fn exponential_point() -> ExponentialHistogramDataPoint { + ExponentialHistogramDataPoint { + start_time_unix_nano: 1_000_000, + time_unix_nano: 2_000_000, + count: 28, + sum: None, + scale: 9, + zero_count: 7, + positive: Some(exponential_buckets(-3, vec![1, 2, 3, 4, 5, 6])), + zero_threshold: 0.0, + ..Default::default() + } + } + + fn native_field(value: &ValueData, name: &str) -> Option { + let ValueData::StructValue(value) = value else { + panic!("expected native histogram Struct value"); + }; + let index = common_query::native_histogram::NATIVE_HISTOGRAM_FIELD_NAMES + .iter() + .position(|field| *field == name) + .unwrap(); + value.items[index].value_data.clone() + } + + fn i32_list(value: Option) -> Vec { + let Some(ValueData::ListValue(value)) = value else { + panic!("expected i32 list"); + }; + value + .items + .into_iter() + .map(|item| match item.value_data { + Some(ValueData::I32Value(value)) => value, + _ => panic!("expected i32 value"), + }) + .collect() + } + + fn i64_list(value: Option) -> Vec { + let Some(ValueData::ListValue(value)) = value else { + panic!("expected i64 list"); + }; + value + .items + .into_iter() + .map(|item| match item.value_data { + Some(ValueData::I64Value(value)) => value, + _ => panic!("expected i64 value"), + }) + .collect() + } + + #[test] + fn test_downscale_bucket_index_uses_signed_floor_division() { + assert_eq!(downscale_bucket_index(-3, 1).unwrap(), -2); + assert_eq!(downscale_bucket_index(-2, 1).unwrap(), -1); + assert_eq!(downscale_bucket_index(-1, 1).unwrap(), -1); + assert_eq!(downscale_bucket_index(0, 1).unwrap(), 0); + assert_eq!(downscale_bucket_index(1, 1).unwrap(), 0); + assert_eq!(downscale_bucket_index(2, 1).unwrap(), 1); + assert_eq!(downscale_bucket_index(i32::MIN, 32).unwrap(), -1); + assert_eq!(downscale_bucket_index(i32::MAX, 32).unwrap(), 0); + } + + #[test] + fn test_convert_bucket_range_downscales_before_prometheus_shift() { + let buckets = exponential_buckets(-3, vec![1, 2, 3, 4, 5, 6]); + let (spans, deltas, total) = convert_bucket_range("positive", Some(&buckets), 1).unwrap(); + + assert_eq!( + spans, + vec![BucketSpan { + offset: -1, + length: 4 + }] + ); + assert_eq!(deltas, vec![1, 4, 4, -3]); + assert_eq!(total, 21); + } + + #[test] + fn test_exponential_histogram_value_uses_integer_family() { + use common_query::native_histogram::{ + COUNT_F64_FIELD, COUNT_I64_FIELD, POSITIVE_BUCKETS_I64_FIELD, + POSITIVE_SPAN_LENGTHS_FIELD, POSITIVE_SPAN_OFFSETS_FIELD, SCHEMA_FIELD, SUM_FIELD, + ZERO_COUNT_I64_FIELD, + }; + + let (value, timestamp_nanos) = exponential_histogram_value(&exponential_point()).unwrap(); + + assert_eq!(timestamp_nanos, 2_000_000); + assert_eq!( + native_field(&value, SCHEMA_FIELD), + Some(ValueData::I32Value(8)) + ); + assert_eq!( + native_field(&value, COUNT_I64_FIELD), + Some(ValueData::I64Value(28)) + ); + assert_eq!( + native_field(&value, ZERO_COUNT_I64_FIELD), + Some(ValueData::I64Value(7)) + ); + assert_eq!(native_field(&value, COUNT_F64_FIELD), None); + assert_eq!( + i32_list(native_field(&value, POSITIVE_SPAN_OFFSETS_FIELD)), + vec![-1] + ); + assert_eq!( + i32_list(native_field(&value, POSITIVE_SPAN_LENGTHS_FIELD)), + vec![4] + ); + assert_eq!( + i64_list(native_field(&value, POSITIVE_BUCKETS_I64_FIELD)), + vec![1, 5, 9, 6] + ); + let Some(ValueData::F64Value(sum)) = native_field(&value, SUM_FIELD) else { + panic!("expected histogram sum"); + }; + assert_eq!(sum.to_bits(), f64::NAN.to_bits()); + assert_ne!(sum.to_bits(), PROMETHEUS_STALE_NAN_BITS); + } + + #[test] + fn test_no_recorded_value_ignores_other_value_fields() { + use common_query::native_histogram::{ + COUNT_I64_FIELD, POSITIVE_BUCKETS_I64_FIELD, SCHEMA_FIELD, SUM_FIELD, + ZERO_COUNT_I64_FIELD, ZERO_THRESHOLD_FIELD, + }; + + let point = ExponentialHistogramDataPoint { + start_time_unix_nano: 1_000_000, + time_unix_nano: 2_000_000, + count: u64::MAX, + sum: Some(1.0), + scale: i32::MIN, + zero_count: u64::MAX, + positive: Some(exponential_buckets(i32::MAX, vec![u64::MAX])), + flags: DataPointFlags::NoRecordedValueMask as u32, + zero_threshold: f64::NAN, + ..Default::default() + }; + let (value, _) = exponential_histogram_value(&point).unwrap(); + + assert_eq!( + native_field(&value, SCHEMA_FIELD), + Some(ValueData::I32Value(0)) + ); + assert_eq!( + native_field(&value, ZERO_THRESHOLD_FIELD), + Some(ValueData::F64Value(0.0)) + ); + assert_eq!( + native_field(&value, COUNT_I64_FIELD), + Some(ValueData::I64Value(0)) + ); + assert_eq!( + native_field(&value, ZERO_COUNT_I64_FIELD), + Some(ValueData::I64Value(0)) + ); + assert!(i64_list(native_field(&value, POSITIVE_BUCKETS_I64_FIELD)).is_empty()); + let Some(ValueData::F64Value(sum)) = native_field(&value, SUM_FIELD) else { + panic!("expected histogram sum"); + }; + assert_eq!(sum.to_bits(), PROMETHEUS_STALE_NAN_BITS); + } + + #[test] + fn test_non_flag_nan_sum_is_normalized() { + use common_query::native_histogram::SUM_FIELD; + + let point = ExponentialHistogramDataPoint { + sum: Some(f64::from_bits(PROMETHEUS_STALE_NAN_BITS)), + ..Default::default() + }; + let (value, _) = exponential_histogram_value(&point).unwrap(); + let Some(ValueData::F64Value(sum)) = native_field(&value, SUM_FIELD) else { + panic!("expected histogram sum"); + }; + assert_eq!(sum.to_bits(), f64::NAN.to_bits()); + assert_ne!(sum.to_bits(), PROMETHEUS_STALE_NAN_BITS); + } + + #[test] + fn test_exponential_histogram_rejects_invalid_values() { + let mut cases = Vec::new(); + + let mut point = exponential_point(); + point.scale = -5; + cases.push((point, "scale -5 is unsupported")); + + let mut point = exponential_point(); + point.zero_threshold = f64::INFINITY; + cases.push((point, "must be finite and non-negative")); + + let mut point = exponential_point(); + point.start_time_unix_nano = point.time_unix_nano + 1; + cases.push((point, "start_time_unix_nano")); + + let mut point = exponential_point(); + point.count = 27; + cases.push((point, "declared count is 27")); + + let point = ExponentialHistogramDataPoint { + count: u64::MAX, + zero_count: u64::MAX, + ..Default::default() + }; + cases.push((point, "count 18446744073709551615 overflows i64")); + + let point = ExponentialHistogramDataPoint { + count: 1, + scale: 8, + positive: Some(exponential_buckets(i32::MAX, vec![1])), + ..Default::default() + }; + cases.push((point, "shifted bucket index overflows i32")); + + for (point, expected) in cases { + let error = exponential_histogram_value(&point).unwrap_err(); + assert!( + error.contains(expected), + "expected {expected:?}, got {error}" + ); + } + + let buckets = exponential_buckets(0, vec![u64::MAX, 1]); + let error = convert_bucket_range("positive", Some(&buckets), 1).unwrap_err(); + assert!( + error.contains("merged bucket count overflows u64"), + "{error}" + ); + + let buckets = exponential_buckets(i32::MAX, vec![1, 1]); + let error = convert_bucket_range("positive", Some(&buckets), 1).unwrap_err(); + assert!(error.contains("bucket index overflows i32"), "{error}"); + } + + fn metrics_request(metrics: Vec) -> ExportMetricsServiceRequest { + ExportMetricsServiceRequest { + resource_metrics: vec![ResourceMetrics { + scope_metrics: vec![ScopeMetrics { + metrics, + ..Default::default() + }], + ..Default::default() + }], + } + } + + fn exponential_metric( + name: impl Into, + data_points: Vec, + temporality: AggregationTemporality, + ) -> Metric { + Metric { + name: name.into(), + data: Some(metric::Data::ExponentialHistogram(ExponentialHistogram { + data_points, + aggregation_temporality: temporality as i32, + })), + ..Default::default() + } + } + + fn histogram_metric(name: impl Into) -> Metric { + Metric { + name: name.into(), + data: Some(metric::Data::Histogram(Histogram { + data_points: vec![HistogramDataPoint { + start_time_unix_nano: 1_000_000, + time_unix_nano: 2_000_000, + count: 1, + sum: Some(1.0), + bucket_counts: vec![1], + ..Default::default() + }], + aggregation_temporality: AggregationTemporality::Cumulative as i32, + })), + ..Default::default() + } + } + + #[test] + fn test_exponential_histogram_gate_and_partial_outcome() { + let request = metrics_request(vec![ + Metric { + name: "temperature".to_string(), + data: Some(metric::Data::Gauge(Gauge { + data_points: vec![NumberDataPoint::default()], + })), + ..Default::default() + }, + exponential_metric( + "latency", + vec![exponential_point()], + AggregationTemporality::Cumulative, + ), + ]); + let (requests, _, semantic_index, outcome) = + to_grpc_insert_requests(request, &mut OtlpMetricCtx::default()).unwrap(); + + assert_eq!(outcome.accepted_data_points, 1); + assert_eq!(outcome.rejected_data_points, 1); + assert!( + outcome + .error_message + .as_deref() + .unwrap() + .contains("otlp.experimental_enable_exponential_histogram") + ); + assert_eq!(requests.inserts.len(), 1); + assert_eq!(requests.inserts[0].table_name, "temperature"); + let semantics = decode(&semantic_index); + assert!(semantics.contains_key("temperature")); + assert!(!semantics.contains_key("latency")); + + let empty = metrics_request(vec![exponential_metric( + "empty", + vec![], + AggregationTemporality::Cumulative, + )]); + let (_, _, _, outcome) = + to_grpc_insert_requests(empty, &mut OtlpMetricCtx::default()).unwrap(); + assert_eq!(outcome.rejected_data_points, 0); + assert_eq!(outcome.error_message, None); + } + + #[test] + fn test_exponential_histogram_cannot_share_table_with_scalar_metric() { + let request = metrics_request(vec![ + Metric { + name: "latency".to_string(), + data: Some(metric::Data::Gauge(Gauge { + data_points: vec![NumberDataPoint { + value: Some(number_data_point::Value::AsDouble(1.0)), + ..Default::default() + }], + })), + ..Default::default() + }, + exponential_metric( + "latency", + vec![exponential_point()], + AggregationTemporality::Cumulative, + ), + ]); + let mut ctx = OtlpMetricCtx { + experimental_enable_exponential_histogram: true, + ..Default::default() + }; + + let error = to_grpc_insert_requests(request, &mut ctx).unwrap_err(); + assert!( + error + .to_string() + .contains("cannot mix native histogram and float sample fields") + ); + } + + #[test] + fn test_histogram_cannot_replace_exponential_histogram_table() { + let request = metrics_request(vec![ + exponential_metric( + "latency_bucket", + vec![exponential_point()], + AggregationTemporality::Cumulative, + ), + histogram_metric("latency"), + ]); + let mut ctx = OtlpMetricCtx { + experimental_enable_exponential_histogram: true, + ..Default::default() + }; + + let error = to_grpc_insert_requests(request, &mut ctx).unwrap_err(); + assert!( + error + .to_string() + .contains("cannot mix native histogram and float sample fields") + ); + } + + #[test] + fn test_histograms_with_same_name_across_resources_are_merged() { + let metric = histogram_metric("latency"); + let request = ExportMetricsServiceRequest { + resource_metrics: ["service-a", "service-b"] + .into_iter() + .map(|service| ResourceMetrics { + resource: Some(Resource { + attributes: vec![keyvalue("service.name", service)], + ..Default::default() + }), + scope_metrics: vec![ScopeMetrics { + metrics: vec![metric.clone()], + ..Default::default() + }], + ..Default::default() + }) + .collect(), + }; + + let (requests, rows, _, outcome) = + to_grpc_insert_requests(request, &mut OtlpMetricCtx::default()).unwrap(); + + assert_eq!(outcome.accepted_data_points, 2); + assert_eq!(rows, 6); + assert_eq!(requests.inserts.len(), 3); + for request in requests.inserts { + assert_eq!( + request.rows.unwrap().rows.len(), + 2, + "{}", + request.table_name + ); + } + } + + #[test] + fn test_exponential_histogram_rejects_temporality_before_stale_point() { + let stale = ExponentialHistogramDataPoint { + flags: DataPointFlags::NoRecordedValueMask as u32, + ..Default::default() + }; + for temporality in [ + AggregationTemporality::Delta, + AggregationTemporality::Unspecified, + ] { + let request = metrics_request(vec![exponential_metric( + "latency", + vec![stale.clone()], + temporality, + )]); + let mut ctx = OtlpMetricCtx { + experimental_enable_exponential_histogram: true, + ..Default::default() + }; + let (requests, rows, semantic_index, outcome) = + to_grpc_insert_requests(request, &mut ctx).unwrap(); + + assert_eq!(outcome.accepted_data_points, 0); + assert_eq!(outcome.rejected_data_points, 1); + assert_eq!(rows, 0); + assert!(requests.inserts.is_empty()); + assert!(semantic_index.is_empty()); + } + } + + #[test] + fn test_exponential_histogram_legacy_and_new_modes_share_struct() { + use common_query::prelude::greptime_native_histogram; + + let mut point = exponential_point(); + point.sum = Some(42.0); + let request = metrics_request(vec![exponential_metric( + "request.duration", + vec![point], + AggregationTemporality::Cumulative, + )]); + let mut new_ctx = OtlpMetricCtx { + experimental_enable_exponential_histogram: true, + ..Default::default() + }; + let (new_requests, _, _, _) = + to_grpc_insert_requests(request.clone(), &mut new_ctx).unwrap(); + let mut legacy_ctx = OtlpMetricCtx { + experimental_enable_exponential_histogram: true, + is_legacy: true, + ..Default::default() + }; + let (legacy_requests, _, _, _) = to_grpc_insert_requests(request, &mut legacy_ctx).unwrap(); + + let new_insert = &new_requests.inserts[0]; + let legacy_insert = &legacy_requests.inserts[0]; + assert_eq!(new_insert.table_name, "request_duration"); + assert_eq!(legacy_insert.table_name, "request_duration"); + let new_rows = new_insert.rows.as_ref().unwrap(); + let legacy_rows = legacy_insert.rows.as_ref().unwrap(); + let field = greptime_native_histogram(); + let new_histogram = new_rows.rows[0].values[new_rows + .schema + .iter() + .position(|column| column.column_name == field) + .unwrap()] + .clone(); + let legacy_histogram = legacy_rows.rows[0].values[legacy_rows + .schema + .iter() + .position(|column| column.column_name == field) + .unwrap()] + .clone(); + assert_eq!(new_histogram, legacy_histogram); + assert!(matches!( + new_rows.rows[0].values[new_rows + .schema + .iter() + .position(|column| column.column_name == greptime_timestamp()) + .unwrap()] + .value_data, + Some(ValueData::TimestampMillisecondValue(2)) + )); + assert!(matches!( + legacy_rows.rows[0].values[legacy_rows + .schema + .iter() + .position(|column| column.column_name == greptime_timestamp()) + .unwrap()] + .value_data, + Some(ValueData::TimestampNanosecondValue(2_000_000)) + )); + } + + #[test] + fn test_rejection_message_is_bounded() { + let request = metrics_request(vec![exponential_metric( + "x".repeat(1_000), + vec![ExponentialHistogramDataPoint::default()], + AggregationTemporality::Cumulative, + )]); + let (_, _, _, outcome) = + to_grpc_insert_requests(request, &mut OtlpMetricCtx::default()).unwrap(); + + assert_eq!(outcome.rejected_data_points, 1); + assert!(outcome.error_message.unwrap().len() <= MAX_REJECTION_MESSAGE_BYTES); + } + + #[test] + fn test_rejection_message_reason_is_lazy_after_cap() { + let mut outcome = MetricsIngestOutcome { + error_message: Some("x".repeat(MAX_REJECTION_MESSAGE_BYTES)), + ..Default::default() + }; + let mut reason_built = false; + + reject_data_points(&mut outcome, 1, || { + reason_built = true; + "unused".to_string() + }) + .unwrap(); + + assert_eq!(outcome.rejected_data_points, 1); + assert!(!reason_built); + } } diff --git a/src/servers/src/prom_remote_write/v2.rs b/src/servers/src/prom_remote_write/v2.rs index 1bcec98d5f..846ef3ee6b 100644 --- a/src/servers/src/prom_remote_write/v2.rs +++ b/src/servers/src/prom_remote_write/v2.rs @@ -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, ) -> 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, -) -> 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 { - 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> { - 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(¤t_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 { - 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) -> Value { - Value { value_data } -} - -fn list_value(values: impl IntoIterator) -> Value { - pb_value(ValueData::ListValue(ListValue { - items: values.into_iter().map(pb_value).collect(), - })) -} - -fn i32_list_value(values: impl IntoIterator) -> Value { - list_value(values.into_iter().map(ValueData::I32Value)) -} - -fn i32_span_lengths(name: &str, spans: &[BucketSpan]) -> Result> { - 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) -> Value { - list_value(values.into_iter().map(ValueData::I64Value)) -} - -fn f64_list_value(values: impl IntoIterator) -> Value { - list_value(values.into_iter().map(ValueData::F64Value)) -} - -fn bucket_counts_from_deltas(deltas: &[i64]) -> Result> { - 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( diff --git a/src/servers/src/query_handler.rs b/src/servers/src/query_handler.rs index 96f412ac1e..35ec1cc2ff 100644 --- a/src/servers/src/query_handler.rs +++ b/src/servers/src/query_handler.rs @@ -70,6 +70,15 @@ pub struct TraceIngestOutcome { pub error_message: Option, } +/// 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, +} + #[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; + ) -> Result; /// Handling opentelemetry traces request async fn traces( diff --git a/src/session/src/protocol_ctx.rs b/src/session/src/protocol_ctx.rs index 6e84761c2d..537dec5c41 100644 --- a/src/session/src/protocol_ctx.rs +++ b/src/session/src/protocol_ctx.rs @@ -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, 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, diff --git a/src/standalone/src/options.rs b/src/standalone/src/options.rs index bedf9eebc7..4f31ac2ebd 100644 --- a/src/standalone/src/options.rs +++ b/src/standalone/src/options.rs @@ -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, diff --git a/tests-integration/src/test_util.rs b/tests-integration/src/test_util.rs index 30b9a07348..cb958a84e5 100644 --- a/tests-integration/src/test_util.rs +++ b/tests-integration/src/test_util.rs @@ -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, http_opts: Option, memory_limiter: Option, + 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()); diff --git a/tests-integration/tests/grpc.rs b/tests-integration/tests/grpc.rs index 23862870db..9d0e13dddc 100644 --- a/tests-integration/tests/grpc.rs +++ b/tests-integration/tests/grpc.rs @@ -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 { + 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)) } diff --git a/tests-integration/tests/http.rs b/tests-integration/tests/http.rs index f89d75e9ef..2a66c4353e 100644 --- a/tests-integration/tests/http.rs +++ b/tests-integration/tests/http.rs @@ -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;