From fd74b9239e6abc31e989389db0958cf3cb71c0c8 Mon Sep 17 00:00:00 2001 From: shuiyisong <113876041+shuiyisong@users.noreply.github.com> Date: Wed, 4 Nov 2026 10:09:57 +0000 Subject: [PATCH] feat(otlp): store validated metric exemplars (#9923) * feat(otlp): persist metric exemplars OTLP metric conversion now emits exemplars through the shared storage contract. It validates timestamps, IDs, values and label budgets, and associates histogram exemplars only with emitted buckets that contain their values. Invalid observations are dropped and counted while valid points and siblings remain ingestible. Frontend ingestion authorizes the emitted destinations and retains confirmed completions across metric and exemplar writes, including logical-table batching. Includes conversion and schema-failure isolation tests; tests that also query the new Prometheus endpoint land in the final layer. Signed-off-by: shuiyisong * feat(otlp): bind exemplars to source metric identities Signed-off-by: shuiyisong * fix(otlp): bound exemplar memory and preserve series Signed-off-by: shuiyisong * fix(otlp): skip exemplars without metric engine Signed-off-by: shuiyisong * fix(otlp): skip exemplars for ordinary tables Signed-off-by: shuiyisong * refactor(exemplar)!: prefix the shared table name Signed-off-by: shuiyisong * fix(otlp): avoid waiting with body permits held Signed-off-by: shuiyisong --------- Signed-off-by: shuiyisong --- src/common/catalog/src/consts.rs | 2 + src/common/query/src/prelude.rs | 2 - src/frontend/AGENTS.md | 10 +- src/frontend/src/instance/otlp.rs | 201 +++- src/frontend/src/server.rs | 1 + src/operator/src/insert.rs | 14 +- src/operator/src/insert/otlp.rs | 9 +- src/servers/src/exemplar.rs | 10 +- src/servers/src/grpc/memory_limit.rs | 6 + src/servers/src/http/memory_limit.rs | 7 +- src/servers/src/http/otlp.rs | 14 +- src/servers/src/http/prom_store.rs | 2 +- src/servers/src/metrics.rs | 2 +- src/servers/src/otel_arrow.rs | 15 +- src/servers/src/otlp/metrics.rs | 453 ++++++++- src/servers/src/otlp/metrics/exemplars.rs | 311 ++++++ src/servers/src/otlp/metrics/prepare.rs | 38 + src/servers/src/postgres.rs | 5 + src/servers/src/postgres/copy_in/mod.rs | 6 +- src/servers/src/postgres/server.rs | 10 + src/servers/src/prom_remote_write/decode.rs | 2 +- src/servers/src/prom_remote_write/v2.rs | 4 +- src/servers/src/query_handler.rs | 5 +- src/servers/src/request_memory_limiter.rs | 2 +- tests-integration/src/instance.rs | 2 +- tests-integration/src/otlp.rs | 994 ++++++++++++++++++- tests-integration/src/tests/instance_test.rs | 52 +- tests-integration/tests/grpc.rs | 100 +- 28 files changed, 2191 insertions(+), 88 deletions(-) create mode 100644 src/servers/src/otlp/metrics/exemplars.rs diff --git a/src/common/catalog/src/consts.rs b/src/common/catalog/src/consts.rs index 27680491ff0..cf2b9363ea8 100644 --- a/src/common/catalog/src/consts.rs +++ b/src/common/catalog/src/consts.rs @@ -224,6 +224,8 @@ pub const SPAN_KIND_SERVER: &str = "SPAN_KIND_SERVER"; pub const SPAN_STATUS_ERROR: &str = "STATUS_CODE_ERROR"; pub const TRACE_TABLE_NAME: &str = "opentelemetry_traces"; pub const TRACE_TABLE_NAME_SESSION_KEY: &str = "trace_table_name"; +/// Shared exemplar destination within each source metric's schema. +pub const METRICS_EXEMPLARS_TABLE: &str = concatcp!(DEFAULT_CATALOG_NAME, "_metrics_exemplars"); // ---- End of special table and fields ---- /// Generate the trace services table name from the trace table name by adding `_services` suffix. diff --git a/src/common/query/src/prelude.rs b/src/common/query/src/prelude.rs index 8bb486a2d30..8d950d77096 100644 --- a/src/common/query/src/prelude.rs +++ b/src/common/query/src/prelude.rs @@ -91,8 +91,6 @@ pub fn greptime_native_histogram() -> &'static str { const GREPTIME_TIMESTAMP: &str = "greptime_timestamp"; /// Default value column name constant for backward compatibility. const GREPTIME_VALUE: &str = "greptime_value"; -/// Shared exemplar destination within each source metric's schema. -pub const METRICS_EXEMPLARS_TABLE: &str = "metrics_exemplars"; /// Query-visible source metric name stored in the exemplar table. pub const EXEMPLAR_METRIC_COLUMN: &str = "greptime_metric"; /// JSON2 object containing exemplar-owned labels. diff --git a/src/frontend/AGENTS.md b/src/frontend/AGENTS.md index 152cbbe6bee..2c8c16b4f84 100644 --- a/src/frontend/AGENTS.md +++ b/src/frontend/AGENTS.md @@ -74,8 +74,14 @@ including caches owned by pipeline and protocol handlers. These tables are ingestion-managed; manual mutation requires restarting serving frontends. -- **OTLP metrics** (`instance/otlp.rs`): authorize all emitted destinations before - schema preflight can reject points or expose destination schemas. +- **OTLP metrics** (`instance/otlp.rs`): authorize source tables before schema + preflight can reject points or expose their schemas. Skip and count exemplars + when metric-engine ingestion is disabled or the source is an ordinary table, + retaining metric rows. Authorize retained exemplar writes before accessing + the exemplar destination or reserving expansion memory. + Reserve exemplar expansion with the transport's memory limiter before building + rows; retain the permit through `WriteProgress` until writes complete. + Expansion must not wait while the request holds transport memory permits. - **Logical-table batching** (`instance/logical_batcher.rs`): `Services` initializes one shared batcher for opted-in HTTP Prom and OTLP metric-engine diff --git a/src/frontend/src/instance/otlp.rs b/src/frontend/src/instance/otlp.rs index 4ed79e7fb41..0c03d82e4e4 100644 --- a/src/frontend/src/instance/otlp.rs +++ b/src/frontend/src/instance/otlp.rs @@ -20,16 +20,18 @@ use std::collections::HashSet; use std::sync::Arc; use api::v1::RowInsertRequests; +use api::v1::value::ValueData; use async_trait::async_trait; use auth::{ OTLP_WRITE, PermissionChecker, PermissionCheckerRef, PermissionReq, PermissionTableTarget, PermissionTableTargets, }; use client::Output; -use common_catalog::consts::{trace_operations_table_name, trace_services_table_name}; +use common_catalog::consts::{ + METRICS_EXEMPLARS_TABLE, trace_operations_table_name, trace_services_table_name, +}; use common_error::ext::BoxedError; -use common_query::prelude::GREPTIME_PHYSICAL_TABLE; -use common_query::{OutputData, OutputMeta}; +use common_query::prelude::{EXEMPLAR_METRIC_COLUMN, GREPTIME_PHYSICAL_TABLE}; use common_telemetry::{tracing, warn}; use opentelemetry_proto::tonic::collector::logs::v1::ExportLogsServiceRequest; use opentelemetry_proto::tonic::collector::trace::v1::ExportTraceServiceRequest; @@ -44,9 +46,11 @@ use servers::otlp::trace::span::TraceSpanGroup; use servers::query_handler::{ MetricsIngestOutcome, OpenTelemetryProtocolHandler, PipelineHandlerRef, TraceIngestOutcome, }; +use servers::request_memory_limiter::ServerMemoryLimiter; use session::context::QueryContextRef; use session::protocol_ctx::OtlpMetricCtx; -use snafu::ResultExt; +use session::write_progress::WriteProgress; +use snafu::{IntoError, ResultExt}; use table::requests::{ OTLP_METRIC_COMPAT_KEY, OTLP_METRIC_COMPAT_PROM, SEMANTIC_PER_TABLE_INDEX_KEY, SEMANTIC_SIGNAL_TYPE, SEMANTIC_SOURCE, SIGNAL_TYPE_LOG, SIGNAL_TYPE_METRIC, @@ -96,6 +100,7 @@ impl OpenTelemetryProtocolHandler for Instance { &self, request: ExportMetricsServiceRequest, ctx: QueryContextRef, + memory_limiter: Option<(ServerMemoryLimiter, u64)>, ) -> ServerResult { self.plugins .get::() @@ -117,7 +122,7 @@ impl OpenTelemetryProtocolHandler for Instance { metric_ctx.resource_info = self.otlp_resource_info; let mut prepared = otlp::metrics::prepare_metrics(request, &mut metric_ctx)?; - // Check all destinations before preflight can expose schemas or discard points. + // Authorize source tables before preflight can expose schemas or discard points. let targets = { let catalog = ctx.current_catalog(); let schema = ctx.current_schema(); @@ -127,7 +132,7 @@ impl OpenTelemetryProtocolHandler for Instance { .iter() .filter(|request| table_names.insert(request.table_name.as_str())) .map(|request| PermissionTableTarget::new(catalog, &schema, &request.table_name)) - .collect() + .collect::>() }; self.check_table_permission( &ctx, @@ -152,6 +157,7 @@ impl OpenTelemetryProtocolHandler for Instance { let otlp::metrics::MetricsConversion { requests, rows, + mut exemplars, semantic_index, resource_info, mut outcome, @@ -167,7 +173,73 @@ impl OpenTelemetryProtocolHandler for Instance { return Ok(outcome); } - let ctx = admit_write(rows as u64, &ctx) + // The engine option only controls new tables; existing ordinary sources cannot bind exemplars. + let before = exemplars.len(); + exemplars.retain(|exemplar| { + preflight + .physical_tables + .contains_key(exemplar.metric.as_ref()) + }); + servers::metrics::EXEMPLARS_DROPPED_TOTAL + .with_label_values(&["otlp"]) + .inc_by((before - exemplars.len()) as u64); + if !exemplars.is_empty() { + self.check_table_permission( + &ctx, + PermissionReq::Action(OTLP_WRITE), + PermissionTableTargets::resolved(vec![PermissionTableTarget::new( + ctx.current_catalog(), + ctx.current_schema(), + METRICS_EXEMPLARS_TABLE, + )]), + ) + .context(AuthSnafu)?; + } + + let ctx = if !exemplars.is_empty() + && let Some((limiter, held_bytes)) = memory_limiter + { + let bytes = + servers::exemplar::estimated_row_bytes(std::iter::once(exemplars.as_slice())) + .and_then(|bytes| u64::try_from(bytes).ok()) + .ok_or_else(|| { + error::MemoryLimitExceededSnafu.into_error( + common_memory_manager::Error::MemoryLimitExceeded { + requested_bytes: u64::MAX, + limit_bytes: limiter.limit_bytes(), + }, + ) + })?; + limiter.check_request_fits(bytes, held_bytes)?; + // Holding body permits while waiting for expansion can stall concurrent requests. + let reservation = Arc::new(if held_bytes > 0 { + limiter.try_acquire(bytes).ok_or_else(|| { + error::MemoryLimitExceededSnafu.into_error( + common_memory_manager::Error::MemoryLimitExceeded { + requested_bytes: bytes, + limit_bytes: limiter.limit_bytes(), + }, + ) + })? + } else { + limiter.acquire(bytes).await? + }); + let mut ctx = ctx.fork(); + // Queued and detached writes retain the reservation after caller cancellation. + ctx.set_write_progress(Arc::new(WriteProgress::with_reservation(reservation))); + Arc::new(ctx) + } else { + ctx + }; + let exemplar_rows = exemplars.len(); + let (exemplars, _) = servers::exemplar::into_rows(exemplars)?; + let total_rows = rows.checked_add(exemplar_rows).ok_or_else(|| { + error::InvalidOtlpMetricInputSnafu { + reason: "combined metric and exemplar row count overflow", + } + .build() + })?; + let ctx = admit_write(total_rows as u64, &ctx) .await .map_err(BoxedError::new) .context(error::ExecuteGrpcQuerySnafu)?; @@ -187,10 +259,107 @@ impl OpenTelemetryProtocolHandler for Instance { Arc::new(c) }; - let output = self - .handle_otlp_metric_row_inserts(requests, ctx.clone(), &metric_ctx) - .await?; - outcome.write_cost = output.meta.cost; + if exemplar_rows > 0 { + let confirmed = Arc::new(ctx.with_confirmed_writes()); + let source_names = exemplars + .inserts + .iter() + .filter_map(|request| request.rows.as_ref()) + .flat_map(|rows| { + let index = rows + .schema + .iter() + .position(|c| c.column_name == EXEMPLAR_METRIC_COLUMN); + rows.rows.iter().filter_map(move |row| { + match &row.values.get(index?)?.value_data { + Some(ValueData::StringValue(name)) => Some(name.as_str()), + _ => None, + } + }) + }) + .collect::>(); + let (source_requests, other): (Vec<_>, Vec<_>) = requests + .inserts + .into_iter() + .partition(|request| source_names.contains(request.table_name.as_str())); + let prepared = self + .inserter + .prepare_metric_row_inserts( + RowInsertRequests { + inserts: source_requests, + }, + confirmed.clone(), + &self.statement_executor, + ctx.extension(PHYSICAL_TABLE_PARAM) + .unwrap_or(GREPTIME_PHYSICAL_TABLE) + .to_string(), + ) + .await + .map_err(BoxedError::new) + .context(error::ExecuteGrpcQuerySnafu); + let source_ready = prepared.is_ok(); + let sources = prepared + .as_ref() + .map(|request| request.source_tables().cloned().collect::>()) + .unwrap_or_default(); + let (metric_output, other_output, exemplar_output) = tokio::join!( + async { + self.inserter + .handle_prepared_metric_inserts(prepared?, confirmed.clone()) + .await + .map_err(BoxedError::new) + .context(error::ExecuteGrpcQuerySnafu) + }, + async { + if other.is_empty() { + Ok(Output::new_with_affected_rows(0)) + } else { + self.handle_otlp_metric_row_inserts( + RowInsertRequests { inserts: other }, + confirmed.clone(), + &metric_ctx, + ) + .await + } + }, + async { + // Failed source preparation must not bind exemplars to a later replacement table. + if !source_ready { + return Ok(Output::new_with_affected_rows(0)); + } + self.inserter + .handle_exemplar_inserts( + exemplars, + confirmed.clone(), + &self.statement_executor, + &sources, + ) + .await + .map_err(BoxedError::new) + .context(error::ExecuteGrpcQuerySnafu) + }, + ); + let mut progress = common_query::WriteProgress::default(); + for output in [metric_output, other_output, exemplar_output] { + match output { + Ok(output) => { + outcome.write_cost += output.meta.cost; + progress.merge(output.write_progress().await); + } + Err(error) => { + progress.merge(common_query::WriteProgress::failed(BoxedError::new(error))) + } + } + } + progress + .into_result() + .context(error::ExecuteGrpcQuerySnafu)?; + } else { + let output = self + .handle_otlp_metric_row_inserts(requests, ctx.clone(), &metric_ctx) + .await?; + outcome.write_cost = output.meta.cost; + } // Derived enrichment follows the accepted metric submission, which may // still be queued in asynchronous mode. Failures remain warning-only @@ -369,8 +538,8 @@ impl Instance { // OTLP tables have one sample field in both the ordinary and physical paths. if let Some(batcher) = batcher { - let (rows, cost) = batcher - .submit_with(requests, ctx.clone(), |mut requests| { + let (mut output, cost) = batcher + .submit_output_with(requests, ctx.clone(), |mut requests| { let ctx = ctx.clone(); async move { Inserter::meter_row_inserts(&mut requests, &ctx) @@ -380,10 +549,8 @@ impl Instance { } }) .await?; - Ok(Output::new( - OutputData::AffectedRows(rows as usize), - OutputMeta::new_with_cost(cost as _), - )) + output.meta.cost = cost as _; + Ok(output) } else if !metric_ctx.with_metric_engine { self.handle_row_inserts(requests, ctx.clone(), false, true) .await diff --git a/src/frontend/src/server.rs b/src/frontend/src/server.rs index 2b79bddb3cd..4cfb2c22f15 100644 --- a/src/frontend/src/server.rs +++ b/src/frontend/src/server.rs @@ -432,6 +432,7 @@ where ) .with_influxdb_handler(instance.clone()) .with_otlp_handler(instance.clone(), with_metric_engine) + .with_memory_limiter(self.server_memory_limiter.clone()) .with_batching_enabled(postgres_batching), ) as Box; diff --git a/src/operator/src/insert.rs b/src/operator/src/insert.rs index 6e921577b7c..c0792ee56e3 100644 --- a/src/operator/src/insert.rs +++ b/src/operator/src/insert.rs @@ -39,9 +39,10 @@ use catalog::CatalogManagerRef; use client::{OutputData, OutputMeta}; use common_base::readable_size::ReadableSize; use common_catalog::consts::{ - DEFAULT_PRIVATE_SCHEMA_NAME, PARENT_SPAN_ID_COLUMN, SERVICE_NAME_COLUMN, SPAN_ID_COLUMN, - TIMESTAMP_END_COLUMN, TRACE_ID_COLUMN, TRACE_TABLE_NAME, TRACE_TABLE_NAME_SESSION_KEY, - default_engine, is_ddl_reserved_table, trace_operations_table_name, trace_services_table_name, + DEFAULT_PRIVATE_SCHEMA_NAME, METRICS_EXEMPLARS_TABLE, PARENT_SPAN_ID_COLUMN, + SERVICE_NAME_COLUMN, SPAN_ID_COLUMN, TIMESTAMP_END_COLUMN, TRACE_ID_COLUMN, TRACE_TABLE_NAME, + TRACE_TABLE_NAME_SESSION_KEY, default_engine, is_ddl_reserved_table, + trace_operations_table_name, trace_services_table_name, }; use common_event_recorder::DEFAULT_EVENTS_TABLE_NAME; use common_frontend::slow_query_event::SLOW_QUERY_TABLE_NAME; @@ -54,8 +55,7 @@ use common_meta::rpc::ddl::TriggerReason; use common_query::Output; use common_query::native_histogram::{is_native_histogram_value_type, native_histogram_value_type}; use common_query::prelude::{ - EXEMPLAR_LABELS_COLUMN, EXEMPLAR_METRIC_COLUMN, METRICS_EXEMPLARS_TABLE, greptime_timestamp, - greptime_value, + EXEMPLAR_LABELS_COLUMN, EXEMPLAR_METRIC_COLUMN, greptime_timestamp, greptime_value, }; use common_telemetry::tracing_context::TracingContext; use common_telemetry::{debug, error, warn}; @@ -2208,7 +2208,9 @@ pub fn validate_exemplar_table(table: &TableInfo) -> Result<()> { .get(store_api::metric_engine_consts::PRIMARY_KEY_ENCODING) .is_some_and(|v| v.eq_ignore_ascii_case("sparse")), InvalidInsertRequestSnafu { - reason: "metrics_exemplars requires Mito, append_mode=true and primary_key_encoding=sparse" + reason: format!( + "{METRICS_EXEMPLARS_TABLE} requires Mito, append_mode=true and primary_key_encoding=sparse" + ) } ); let columns = meta.schema.column_schemas(); diff --git a/src/operator/src/insert/otlp.rs b/src/operator/src/insert/otlp.rs index 80ead7070c5..9405480d3a2 100644 --- a/src/operator/src/insert/otlp.rs +++ b/src/operator/src/insert/otlp.rs @@ -30,7 +30,9 @@ use common_time::timestamp::TimeUnit; use datatypes::schema::ColumnSchema as DataColumn; use session::context::QueryContextRef; use snafu::prelude::*; -use store_api::metric_engine_consts::{LOGICAL_TABLE_METADATA_KEY, PHYSICAL_TABLE_METADATA_KEY}; +use store_api::metric_engine_consts::{ + LOGICAL_TABLE_METADATA_KEY, METRIC_ENGINE_NAME, PHYSICAL_TABLE_METADATA_KEY, +}; use table::TableRef; use crate::error::{self, InvalidInsertRequestSnafu, Result}; @@ -43,6 +45,7 @@ use crate::req_convert::insert::{ImpureDefaultFiller, value_matches_type}; pub struct MetricInsertPreflight { /// Rejections in batch order, then row order. pub rejections: Vec>>, + /// Backing tables for existing or newly created logical metric-engine destinations. pub physical_tables: std::collections::HashMap, } @@ -91,7 +94,7 @@ impl Inserter { tables.insert(request.table_name.clone(), table); } let backing = match tables[&request.table_name].as_ref() { - Some(table) => table + Some(table) if table.table_info().meta.engine == METRIC_ENGINE_NAME => table .table_info() .meta .options @@ -99,7 +102,7 @@ impl Inserter { .get(LOGICAL_TABLE_METADATA_KEY) .cloned(), None if metric_engine => Some(physical_table.to_string()), - None => None, + _ => None, }; if let Some(backing) = backing { physical_tables.insert(request.table_name.clone(), backing); diff --git a/src/servers/src/exemplar.rs b/src/servers/src/exemplar.rs index 40599f14ab0..cd20b796739 100644 --- a/src/servers/src/exemplar.rs +++ b/src/servers/src/exemplar.rs @@ -19,10 +19,10 @@ use std::sync::Arc; use api::v1::value::ValueData; use api::v1::{ColumnDataType, Row, RowInsertRequests, Value}; +use common_catalog::consts::METRICS_EXEMPLARS_TABLE; use common_grpc::precision::Precision; use common_query::prelude::{ - EXEMPLAR_LABELS_COLUMN, EXEMPLAR_METRIC_COLUMN, METRICS_EXEMPLARS_TABLE, greptime_timestamp, - greptime_value, + EXEMPLAR_LABELS_COLUMN, EXEMPLAR_METRIC_COLUMN, greptime_timestamp, greptime_value, }; use datatypes::json; use metric_engine::row_modifier::TsidGenerator; @@ -50,8 +50,10 @@ pub struct Exemplar { pub type ExemplarRequests = BTreeMap, Vec>; /// Bounds row capacity after each database's metrics share one exemplar schema. -pub(crate) fn estimated_row_bytes(requests: &ExemplarRequests) -> Option { - requests.values().try_fold(0usize, |total, exemplars| { +pub fn estimated_row_bytes<'a>( + requests: impl IntoIterator, +) -> Option { + requests.into_iter().try_fold(0usize, |total, exemplars| { let mut tags = BTreeSet::new(); let mut payload = 0usize; // Shared series data is copied into each output row. diff --git a/src/servers/src/grpc/memory_limit.rs b/src/servers/src/grpc/memory_limit.rs index c9c62e88dd1..c69d0ba6374 100644 --- a/src/servers/src/grpc/memory_limit.rs +++ b/src/servers/src/grpc/memory_limit.rs @@ -63,6 +63,12 @@ pub(crate) struct PreDecodeMemoryReservation { _guard: Arc>, } +impl PreDecodeMemoryReservation { + pub(crate) fn reserved_bytes(&self) -> u64 { + self._guard.granted_bytes() + } +} + #[derive(Clone)] pub struct MemoryLimiterExtensionLayer { limiter: ServerMemoryLimiter, diff --git a/src/servers/src/http/memory_limit.rs b/src/servers/src/http/memory_limit.rs index 75eda067223..255920974fd 100644 --- a/src/servers/src/http/memory_limit.rs +++ b/src/servers/src/http/memory_limit.rs @@ -155,7 +155,12 @@ pub async fn decoded_body_accounting_middleware( .and_then(|v| v.parse::().ok()) .unwrap_or(0); - let accounting = BodyMemoryAccounting::default(); + // Share the wire-byte ledger so handlers see all permits held by this request. + let accounting = req + .extensions() + .get::() + .cloned() + .unwrap_or_default(); // Retain the permits for the whole request, like the global middleware. let retained_accounting = accounting.clone(); let (mut parts, body) = req.into_parts(); diff --git a/src/servers/src/http/otlp.rs b/src/servers/src/http/otlp.rs index fc2e6e6aabd..6e18502a9aa 100644 --- a/src/servers/src/http/otlp.rs +++ b/src/servers/src/http/otlp.rs @@ -45,10 +45,12 @@ use crate::http::extractor::{ LogTableName, OtlpMetricOptions, PipelineInfo, SelectInfoWrapper, TraceTableName, }; use crate::http::header::{CONTENT_TYPE_PROTOBUF, write_cost_header_map}; +use crate::http::memory_limit::BodyMemoryAccounting; use crate::metrics::METRIC_HTTP_OPENTELEMETRY_LOGS_ELAPSED; use crate::query_handler::{ MetricsIngestOutcome, OpenTelemetryProtocolHandlerRef, PipelineHandler, TraceIngestOutcome, }; +use crate::request_memory_limiter::ServerMemoryLimiter; #[derive(Clone, prost::Message)] pub struct GoogleRpcStatus { @@ -87,6 +89,7 @@ pub async fn metrics( Extension(mut query_ctx): Extension, http_opts: OtlpMetricOptions, content_type: Option>, + extensions: http::Extensions, bytes: Bytes, ) -> Result { if is_json_content_type(content_type.as_ref().map(|h| &h.0)) { @@ -122,7 +125,16 @@ pub async fn metrics( })); let query_ctx = Arc::new(query_ctx); - match handler.metrics(request, query_ctx).await { + let memory_limiter = extensions + .get::() + .cloned() + .map(|limiter| { + let held_bytes = extensions + .get::() + .map_or(0, BodyMemoryAccounting::reserved_bytes); + (limiter, held_bytes) + }); + match handler.metrics(request, query_ctx, memory_limiter).await { Ok(outcome) => { if outcome.accepted_data_points == 0 && outcome.rejected_data_points > 0 { Ok(OtlpMetricsResponse::Failure(outcome)) diff --git a/src/servers/src/http/prom_store.rs b/src/servers/src/http/prom_store.rs index eb7f0898411..7ccaffd641d 100644 --- a/src/servers/src/http/prom_store.rs +++ b/src/servers/src/http/prom_store.rs @@ -235,7 +235,7 @@ async fn remote_write_v1( query_ctx } else { // Reserve for row expansion before copying shared metric names and series labels. - let bytes = crate::exemplar::estimated_row_bytes(&exemplars) + let bytes = crate::exemplar::estimated_row_bytes(exemplars.values().map(Vec::as_slice)) .and_then(|bytes| u64::try_from(bytes).ok()) .ok_or_else(|| { error::MemoryLimitExceededSnafu.into_error( diff --git a/src/servers/src/metrics.rs b/src/servers/src/metrics.rs index ef5f30ba0b5..cd254559e0f 100644 --- a/src/servers/src/metrics.rs +++ b/src/servers/src/metrics.rs @@ -130,7 +130,7 @@ lazy_static! { ).unwrap(); pub static ref EXEMPLARS_DROPPED_TOTAL: IntCounterVec = register_int_counter_vec!( "greptime_servers_exemplars_dropped_total", - "Exemplars dropped during input validation", + "Exemplars dropped during ingestion", &["protocol"] ).unwrap(); /// The native histograms count of Prometheus remote write per protocol version. diff --git a/src/servers/src/otel_arrow.rs b/src/servers/src/otel_arrow.rs index 490ae4b1c58..fd75c92b5d1 100644 --- a/src/servers/src/otel_arrow.rs +++ b/src/servers/src/otel_arrow.rs @@ -33,7 +33,9 @@ use tonic::{Request, Response, Status, Streaming}; use crate::error; use crate::grpc::context_auth; +use crate::grpc::memory_limit::PreDecodeMemoryReservation; use crate::query_handler::{MetricsIngestOutcome, OpenTelemetryProtocolHandlerRef}; +use crate::request_memory_limiter::ServerMemoryLimiter; const EXPONENTIAL_HISTOGRAM_UNSUPPORTED: &str = "OTel Arrow exponential histograms are unsupported because the Arrow wire format omits zero_threshold"; @@ -120,6 +122,8 @@ impl ArrowMetricsService for OtelArrowServiceHandler().cloned(); + let decode_reservation = extensions.get::().cloned(); // handles incoming requests common_runtime::spawn_global(async move { @@ -159,7 +163,16 @@ impl ArrowMetricsService for OtelArrowServiceHandler outcome, Err(error::Error::InvalidOtlpMetricInput { reason }) => { let _ = sender diff --git a/src/servers/src/otlp/metrics.rs b/src/servers/src/otlp/metrics.rs index c1226d6ed6f..b4f2128812c 100644 --- a/src/servers/src/otlp/metrics.rs +++ b/src/servers/src/otlp/metrics.rs @@ -51,6 +51,7 @@ use crate::semantic::{ METRIC_TYPE_UPDOWN_COUNTER, }; +mod exemplars; mod prepare; mod resource_info; @@ -110,6 +111,8 @@ const MAX_REJECTION_MESSAGE_BYTES: usize = 512; /// Result of converting one OTLP metrics request. #[derive(Debug)] pub struct MetricsConversion { + /// Kept normalized until the frontend reserves memory for row expansion. + pub exemplars: Vec, pub requests: RowInsertRequests, /// Row count of `requests`; the resource descriptor is not counted. pub rows: usize, @@ -391,6 +394,7 @@ fn process_scope_attrs(scope: &ScopeMetrics, metric_ctx: &OtlpMetricCtx) -> Opti }) } +#[allow(clippy::too_many_arguments)] fn encode_metrics( table_writer: &mut MultiTableData, metric: &Metric, @@ -399,6 +403,7 @@ fn encode_metrics( scope_attrs: Option<&Vec>, metric_ctx: &OtlpMetricCtx, outcome: &mut MetricsIngestOutcome, + exemplars: &mut Vec, ) -> Result<()> { let name = &metadata.name; @@ -412,6 +417,7 @@ fn encode_metrics( resource_attrs, scope_attrs, metric_ctx, + exemplars, )?; add_accepted_data_points(outcome, gauge.data_points.len())?; } @@ -423,6 +429,7 @@ fn encode_metrics( resource_attrs, scope_attrs, metric_ctx, + exemplars, )?; add_accepted_data_points(outcome, sum.data_points.len())?; } @@ -445,6 +452,7 @@ fn encode_metrics( scope_attrs, metric_ctx, outcome, + exemplars, )?, metric::Data::ExponentialHistogram(hist) => encode_exponential_histogram( table_writer, @@ -454,6 +462,7 @@ fn encode_metrics( scope_attrs, metric_ctx, outcome, + exemplars, )?, } } @@ -538,6 +547,7 @@ fn append_rejection_message(message: &mut Option, reason: impl FnOnce() } } +#[allow(clippy::too_many_arguments)] fn encode_exponential_histogram( table_writer: &mut MultiTableData, metadata: &MetricMetadata, @@ -546,6 +556,7 @@ fn encode_exponential_histogram( scope_attrs: Option<&Vec>, metric_ctx: &OtlpMetricCtx, outcome: &mut MetricsIngestOutcome, + exemplars: &mut Vec, ) -> Result<()> { let name = &metadata.name; let temporality = match exponential_histogram_gate(histogram) { @@ -606,6 +617,13 @@ fn encode_exponential_histogram( std::iter::once((column_schema.clone(), Some(value))), &mut row, )?; + exemplars::append( + exemplars, + &data_point.exemplars, + name, + table.num_rows(), + metric_ctx, + ); table.add_row(row); add_accepted_data_points(outcome, 1)?; } @@ -898,11 +916,15 @@ fn write_attributes( ) } }; - if key == OTLP_AGGREGATION_TEMPORALITY_LABEL { + // OTLP attribute keys may include __name__, but Prometheus metric identity comes from + // the OTLP metric name. Reject collisions after translation and scope prefixing. + // https://opentelemetry.io/docs/specs/otel/compatibility/prometheus_and_openmetrics/#metric-metadata-1 + if key == OTLP_AGGREGATION_TEMPORALITY_LABEL || key == crate::prom_store::METRIC_NAME_LABEL + { return Err(error::InvalidOtlpMetricInputSnafu { reason: format!( "OTLP attribute `{}` resolves to reserved label `{}`", - attr.key, OTLP_AGGREGATION_TEMPORALITY_LABEL + attr.key, key ), } .build()); @@ -1026,6 +1048,7 @@ fn encode_gauge( resource_attrs: Option<&Vec>, scope_attrs: Option<&Vec>, metric_ctx: &OtlpMetricCtx, + exemplars: &mut Vec, ) -> Result<()> { let table = table_writer.get_or_default_table_data( name, @@ -1046,6 +1069,13 @@ fn encode_gauge( )?; write_data_point_value(table, &mut row, greptime_value(), &data_point.value)?; + exemplars::append( + exemplars, + &data_point.exemplars, + name, + table.num_rows(), + metric_ctx, + ); table.add_row(row); } @@ -1061,6 +1091,7 @@ fn encode_sum( resource_attrs: Option<&Vec>, scope_attrs: Option<&Vec>, metric_ctx: &OtlpMetricCtx, + exemplars: &mut Vec, ) -> Result<()> { let is_delta = matches!( AggregationTemporality::try_from(sum.aggregation_temporality), @@ -1094,6 +1125,13 @@ fn encode_sum( } else { write_data_point_value(table, &mut row, greptime_value(), &data_point.value)?; } + exemplars::append( + exemplars, + &data_point.exemplars, + name, + table.num_rows(), + metric_ctx, + ); table.add_row(row); } @@ -1115,6 +1153,7 @@ const HISTOGRAM_LE_COLUMN: &str = "le"; /// quantile functions on this table. /// Delta points retain their temporality tag and interval values. Bucket counts /// are prefix-summed within each point, never accumulated across timestamps. +#[allow(clippy::too_many_arguments)] fn encode_histogram( table_writer: &mut MultiTableData, name: &str, @@ -1123,6 +1162,7 @@ fn encode_histogram( scope_attrs: Option<&Vec>, metric_ctx: &OtlpMetricCtx, outcome: &mut MetricsIngestOutcome, + exemplars: &mut Vec, ) -> Result<()> { let normalized_name = name; @@ -1185,7 +1225,28 @@ fn encode_histogram( } values }; - for (bound, value) in bucket_values { + let mut by_bucket = vec![Vec::new(); bucket_values.len()]; + for exemplar in &data_point.exemplars { + let value = match exemplars::value(exemplar) { + Ok(value) => value, + Err(_) => { + crate::metrics::EXEMPLARS_DROPPED_TOTAL + .with_label_values(&["otlp"]) + .inc(); + continue; + } + }; + let Some(bucket) = bucket_values.iter().position(|(bound, _)| { + value <= *bound || (value.is_nan() && *bound == f64::INFINITY) + }) else { + crate::metrics::EXEMPLARS_DROPPED_TOTAL + .with_label_values(&["otlp"]) + .inc(); + continue; + }; + by_bucket[bucket].push(exemplar); + } + for ((bound, value), bucket_exemplars) in bucket_values.into_iter().zip(by_bucket) { let mut bucket_row = bucket_table.alloc_one_row(); write_tags_and_timestamp( bucket_table, @@ -1197,9 +1258,17 @@ fn encode_histogram( metric_ctx, )?; write_temporality_tag(bucket_table, &mut bucket_row, is_delta)?; + // Preserve existing OTLP series identities, including the "inf" bucket. row_writer::write_tag(bucket_table, HISTOGRAM_LE_COLUMN, bound, &mut bucket_row)?; row_writer::write_f64(bucket_table, greptime_value(), value, &mut bucket_row)?; + exemplars::append( + exemplars, + bucket_exemplars, + &bucket_table_name, + bucket_table.num_rows(), + metric_ctx, + ); bucket_table.add_row(bucket_row); } @@ -1411,14 +1480,15 @@ fn encode_summary( mod tests { use api::v1::ColumnDataType; use common_query::prelude::set_default_prefix; - use otel_arrow_rust::proto::opentelemetry::common::v1::AnyValue; use otel_arrow_rust::proto::opentelemetry::common::v1::any_value::Value as Val; + use otel_arrow_rust::proto::opentelemetry::common::v1::{AnyValue, InstrumentationScope}; use otel_arrow_rust::proto::opentelemetry::metrics::v1::number_data_point::Value; use otel_arrow_rust::proto::opentelemetry::metrics::v1::summary_data_point::ValueAtQuantile; use otel_arrow_rust::proto::opentelemetry::metrics::v1::{ AggregationTemporality, HistogramDataPoint, NumberDataPoint, SummaryDataPoint, }; use otel_arrow_rust::proto::opentelemetry::resource::v1::Resource; + use session::protocol_ctx::OtlpMetricTranslationStrategy; use super::*; @@ -1536,8 +1606,19 @@ mod tests { } #[test] - fn test_descriptors_exclude_points_rejected_by_destination_preflight() { + fn test_descriptors_and_exemplars_exclude_points_rejected_by_destination_preflight() { let mut request = gauge_request(vec![keyvalue("service.name", "api")], "accepted"); + let Some(metric::Data::Gauge(gauge)) = request.resource_metrics[0].scope_metrics[0].metrics + [0] + .data + .as_mut() else { + unreachable!() + }; + gauge.data_points[0].exemplars.push(Exemplar { + time_unix_nano: 1, + value: Some(exemplar::Value::AsDouble(0.5)), + ..Default::default() + }); let mut rejected = request.resource_metrics[0].clone(); let metric = &mut rejected.scope_metrics[0].metrics[0]; metric.name = "rejected".into(); @@ -1547,7 +1628,28 @@ mod tests { gauge.data_points[0].time_unix_nano = 2 * common_catalog::consts::SEMANTIC_GRAPH_WINDOW_NANOS as u64; request.resource_metrics.push(rejected); - let prepared = prepare_metrics(request, &mut descriptor_ctx()).unwrap(); + let mut prepared = prepare_metrics( + request, + &mut OtlpMetricCtx { + with_metric_engine: true, + ..descriptor_ctx() + }, + ) + .unwrap(); + for request in &mut prepared.requests { + let rows = request.rows.as_mut().unwrap(); + rows.schema.push(ColumnSchema { + column_name: "defaulted_host".into(), + datatype: api::v1::ColumnDataType::String as i32, + semantic_type: SemanticType::Tag as i32, + ..Default::default() + }); + for row in &mut rows.rows { + row.values.push(api::v1::Value { + value_data: Some(ValueData::StringValue("unknown".into())), + }); + } + } let conversion = prepared .finish( vec![None, Some("incompatible schema".into())], @@ -1561,6 +1663,22 @@ mod tests { ), (1, 1) ); + assert_eq!(conversion.exemplars.len(), 1); + assert_eq!( + conversion.exemplars[0].series_labels["defaulted_host"], + "unknown" + ); + let (exemplars, _) = crate::exemplar::into_rows(conversion.exemplars).unwrap(); + let exemplar_rows = exemplars.inserts[0].rows.as_ref().unwrap(); + let metric = exemplar_rows + .schema + .iter() + .position(|c| c.column_name == common_query::prelude::EXEMPLAR_METRIC_COLUMN) + .unwrap(); + assert_eq!( + exemplar_rows.rows[0].values[metric].value_data, + Some(ValueData::StringValue("accepted".into())) + ); let descriptors = conversion.resource_info.unwrap(); let rows = descriptors.inserts[0].rows.as_ref().unwrap(); assert_eq!(rows.rows.len(), 1); @@ -1750,6 +1868,85 @@ mod tests { assert_eq!(attr_value(&attrs, "instance").as_deref(), Some("inst-1")); } + #[test] + fn test_reject_metric_name_attribute() { + set_default_prefix(None).unwrap(); + for (strategy, key) in [ + ( + OtlpMetricTranslationStrategy::UnderscoreEscapingWithSuffixes, + "__name__", + ), + ( + OtlpMetricTranslationStrategy::UnderscoreEscapingWithoutSuffixes, + "__name__", + ), + ( + OtlpMetricTranslationStrategy::NoUtf8EscapingWithSuffixes, + "__name__", + ), + (OtlpMetricTranslationStrategy::NoTranslation, "__name__"), + ( + OtlpMetricTranslationStrategy::UnderscoreEscapingWithSuffixes, + "..name..", + ), + ] { + for source in [ + AttributeType::Resource, + AttributeType::Scope, + AttributeType::DataPoint, + ] { + let mut request = gauge_request(vec![], "source_metric"); + let resource = &mut request.resource_metrics[0]; + let scope = &mut resource.scope_metrics[0]; + let Some(metric::Data::Gauge(gauge)) = scope.metrics[0].data.as_mut() else { + unreachable!(); + }; + gauge.data_points[0].exemplars.push(Exemplar { + value: Some(exemplar::Value::AsDouble(0.25)), + time_unix_nano: 1_000_000, + ..Default::default() + }); + let attributes = vec![keyvalue(key, "wrong_metric")]; + match source { + AttributeType::Resource => { + resource.resource.as_mut().unwrap().attributes = attributes; + } + AttributeType::Scope => { + scope.scope = Some(InstrumentationScope { + attributes, + ..Default::default() + }); + } + AttributeType::DataPoint => gauge.data_points[0].attributes = attributes, + } + let mut ctx = OtlpMetricCtx { + with_metric_engine: true, + metric_translation_strategy: strategy, + promote_all_resource_attrs: true, + promote_scope_attrs: true, + ..Default::default() + }; + let result = to_grpc_insert_requests(request, &mut ctx); + if source == AttributeType::Scope { + let conversion = result.unwrap(); + assert_eq!(1, conversion.exemplars.len()); + assert!( + column_names(&conversion.requests, "source_metric") + .contains(&"otel_scope___name__".to_string()) + ); + } else { + let conversion = result.unwrap(); + assert_eq!(conversion.outcome.accepted_data_points, 0); + assert_eq!(conversion.outcome.rejected_data_points, 1); + assert_eq!(conversion.exemplars.len(), 0); + assert!(conversion.outcome.error_message.unwrap().contains(&format!( + "OTLP attribute `{key}` resolves to reserved label `__name__`" + ))); + } + } + } + } + #[test] fn test_boolean_attributes_keep_series_distinct() { set_default_prefix(None).unwrap(); @@ -1834,6 +2031,7 @@ mod tests { Some(&vec![]), Some(&vec![keyvalue("scope", "otel")]), &OtlpMetricCtx::default(), + &mut Vec::new(), ) .unwrap(); @@ -1884,6 +2082,7 @@ mod tests { Some(&vec![]), Some(&vec![keyvalue("scope", "otel")]), &OtlpMetricCtx::default(), + &mut Vec::new(), ) .unwrap(); @@ -2020,6 +2219,7 @@ mod tests { Some(&vec![keyvalue("scope", "otel")]), &OtlpMetricCtx::default(), &mut outcome, + &mut Vec::new(), ) .unwrap(); @@ -2643,6 +2843,247 @@ mod tests { } } + #[test] + fn test_invalid_exemplars_preserve_metric_payload() { + use otel_arrow_rust::proto::opentelemetry::metrics::v1::{Exemplar, exemplar}; + + let valid = Exemplar { + time_unix_nano: 1, + value: Some(exemplar::Value::AsDouble(0.5)), + ..Default::default() + }; + for invalid in [ + Exemplar::default(), + Exemplar { + trace_id: vec![1], + ..valid.clone() + }, + Exemplar { + span_id: vec![1], + ..valid.clone() + }, + Exemplar { + time_unix_nano: u64::MAX, + ..valid.clone() + }, + Exemplar { + filtered_attributes: vec![keyvalue("http.url", &"x".repeat(128))], + ..valid.clone() + }, + Exemplar { + filtered_attributes: vec![keyvalue("a", "1"), keyvalue("a", "2")], + ..valid.clone() + }, + ] { + let exemplars = vec![invalid.clone(), valid.clone(), invalid, valid.clone()]; + let point = NumberDataPoint { + time_unix_nano: 1, + value: Some(Value::AsDouble(1.0)), + exemplars: exemplars.clone(), + ..Default::default() + }; + let metrics = vec![ + Metric { + name: "gauge".into(), + data: Some(metric::Data::Gauge(Gauge { + data_points: vec![point.clone()], + })), + ..Default::default() + }, + Metric { + name: "sum".into(), + data: Some(metric::Data::Sum(Sum { + data_points: vec![point], + aggregation_temporality: AggregationTemporality::Cumulative as i32, + ..Default::default() + })), + ..Default::default() + }, + exponential_metric( + "histogram", + vec![ExponentialHistogramDataPoint { + exemplars, + ..exponential_point() + }], + AggregationTemporality::Cumulative, + ), + ]; + let conversion = to_grpc_insert_requests( + metrics_request(metrics), + &mut OtlpMetricCtx { + with_metric_engine: true, + ..Default::default() + }, + ) + .unwrap(); + assert_eq!(conversion.rows, 3); + assert_eq!(conversion.outcome.accepted_data_points, 3); + assert_eq!(conversion.outcome.rejected_data_points, 0); + assert_eq!(conversion.exemplars.len(), 6); + } + } + + #[test] + fn test_histogram_exemplars_follow_emitted_buckets() { + use common_query::prelude::EXEMPLAR_METRIC_COLUMN; + use otel_arrow_rust::proto::opentelemetry::metrics::v1::{Exemplar, exemplar}; + use prost::Message; + + for mode in ["delta", "empty", "stale", "missing_infinity"] { + let mut point = HistogramDataPoint { + time_unix_nano: 100, + count: 4, + explicit_bounds: vec![1.0, 2.0], + bucket_counts: vec![1, 1, 2], + exemplars: [5.0, 1.0, 0.5, f64::NEG_INFINITY, f64::INFINITY, f64::NAN] + .into_iter() + .enumerate() + .map(|(i, value)| Exemplar { + time_unix_nano: i as u64 + 1, + value: Some(exemplar::Value::AsDouble(value)), + ..Default::default() + }) + .collect(), + ..Default::default() + }; + point.exemplars.extend([ + Exemplar::default(), + Exemplar { + trace_id: vec![1], + value: Some(exemplar::Value::AsDouble(1.0)), + ..Default::default() + }, + ]); + match mode { + "empty" => { + point.explicit_bounds.clear(); + point.bucket_counts.clear(); + } + "stale" => point.flags = DataPointFlags::NoRecordedValueMask as u32, + "missing_infinity" => { + // Cumulative conversion tolerates a missing infinity bucket. + point.explicit_bounds = vec![1.0]; + point.bucket_counts = vec![1]; + point.count = 1; + } + _ => {} + } + let mut histogram = Histogram { + data_points: vec![point], + aggregation_temporality: if mode == "missing_infinity" { + AggregationTemporality::Cumulative + } else { + AggregationTemporality::Delta + } as i32, + }; + if mode != "missing_infinity" { + histogram.data_points.push(HistogramDataPoint { + explicit_bounds: vec![1.0], + bucket_counts: vec![4], + count: 4, + exemplars: vec![Exemplar::default()], + ..Default::default() + }); + } + let mut request = gauge_request(vec![], "gauge"); + request.resource_metrics[0].scope_metrics[0] + .metrics + .push(Metric { + name: "histogram".into(), + data: Some(metric::Data::Histogram(histogram.clone())), + ..Default::default() + }); + let ctx = OtlpMetricCtx { + with_metric_engine: true, + ..Default::default() + }; + let mut conversion = + to_grpc_insert_requests(request.clone(), &mut ctx.clone()).unwrap(); + assert_eq!(conversion.outcome.accepted_data_points, 2, "{mode}"); + assert_eq!( + conversion.outcome.rejected_data_points, + i64::from(mode != "missing_infinity"), + "{mode}" + ); + + // Exemplars must not change either the histogram or the independent gauge payload. + for point in &mut histogram.data_points { + point.exemplars.clear(); + } + request.resource_metrics[0].scope_metrics[0].metrics[1].data = + Some(metric::Data::Histogram(histogram)); + let mut baseline = to_grpc_insert_requests(request, &mut ctx.clone()).unwrap(); + conversion + .requests + .inserts + .sort_by(|a, b| a.table_name.cmp(&b.table_name)); + baseline + .requests + .inserts + .sort_by(|a, b| a.table_name.cmp(&b.table_name)); + // Compare wire bytes so stale NaN values compare by their bits. + assert_eq!( + conversion.requests.encode_to_vec(), + baseline.requests.encode_to_vec(), + "{mode}" + ); + + let expected = match mode { + "empty" => vec![ + (1, "inf"), + (2, "inf"), + (3, "inf"), + (4, "inf"), + (5, "inf"), + (6, "inf"), + ], + "missing_infinity" => vec![(2, "1"), (3, "1"), (4, "1")], + _ => vec![ + (1, "inf"), + (2, "1"), + (3, "1"), + (4, "1"), + (5, "inf"), + (6, "inf"), + ], + }; + assert_eq!(conversion.exemplars.len(), expected.len(), "{mode}"); + let (exemplars, _) = crate::exemplar::into_rows(conversion.exemplars).unwrap(); + let rows = exemplars.inserts[0].rows.as_ref().unwrap(); + let index = |name| { + rows.schema + .iter() + .position(|c| c.column_name == name) + .unwrap() + }; + let mut actual = Vec::new(); + for row in &rows.rows { + assert_eq!( + row.values[index(EXEMPLAR_METRIC_COLUMN)].value_data, + Some(ValueData::StringValue("histogram_bucket".into())) + ); + let Some(ValueData::TimestampNanosecondValue(timestamp)) = + row.values[index(greptime_timestamp())].value_data + else { + panic!("missing exemplar timestamp") + }; + let Some(ValueData::StringValue(ref bound)) = row.values[index("le")].value_data + else { + panic!("missing bucket label") + }; + actual.push((timestamp, bound.as_str())); + if timestamp == 3 { + assert_eq!( + row.values[index(greptime_value())].value_data, + Some(ValueData::F64Value(0.5)) + ); + } + } + actual.sort_unstable(); + assert_eq!(actual, expected, "{mode}"); + } + } + fn exponential_metric( name: impl Into, data_points: Vec, diff --git a/src/servers/src/otlp/metrics/exemplars.rs b/src/servers/src/otlp/metrics/exemplars.rs new file mode 100644 index 00000000000..8cb8d6e1714 --- /dev/null +++ b/src/servers/src/otlp/metrics/exemplars.rs @@ -0,0 +1,311 @@ +// 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 std::collections::{BTreeMap, HashSet}; +use std::sync::Arc; + +use api::v1::{ColumnSchema, SemanticType, Value, value::ValueData}; +use otel_arrow_rust::proto::opentelemetry::metrics::v1::{Exemplar, exemplar}; +use session::protocol_ctx::OtlpMetricCtx; +use snafu::{OptionExt, ensure}; + +use crate::error::{InvalidOtlpMetricInputSnafu, Result}; +use crate::exemplar::{self as normalized, Exemplar as Observation}; +use crate::otlp::metrics::{scalar_value_string, translate_label_name}; + +/// Observations whose inherited labels are bound after source-row normalization. +pub(crate) struct RowExemplars { + pub(crate) metric: Arc, + /// Starts as an emitted table's row offset; preparation maps it into `row_locations`. + pub(crate) row: usize, + observations: Vec, +} + +impl RowExemplars { + /// Drops observations whose inherited tags cannot be represented by the exemplar schema. + pub(crate) fn finish(mut self, schema: &[ColumnSchema], row: &[Value]) -> Vec { + let labels = normalized::labels( + schema.iter().zip(row).filter_map(|(column, value)| { + if column.semantic_type != SemanticType::Tag as i32 { + return None; + } + match &value.value_data { + Some(ValueData::StringValue(value)) => { + Some((column.column_name.clone(), value.clone())) + } + _ => None, + } + }), + false, + |name| !name.is_empty(), + ); + let Ok(labels) = labels else { + crate::metrics::EXEMPLARS_DROPPED_TOTAL + .with_label_values(&["otlp"]) + .inc_by(self.observations.len() as u64); + return Vec::new(); + }; + let labels = Arc::new(labels); + for observation in &mut self.observations { + observation.series_labels = labels.clone(); + } + self.observations + } +} + +pub(super) fn value(exemplar: &Exemplar) -> Result { + match exemplar + .value + .as_ref() + .context(InvalidOtlpMetricInputSnafu { + reason: "exemplar is missing its value", + })? { + exemplar::Value::AsDouble(value) => Ok(*value), + exemplar::Value::AsInt(value) => Ok(*value as f64), + } +} + +/// Collects exemplars for metric-engine ingestion, retaining each emitted row's identity +/// until destination defaults have been filled. +pub(crate) fn append<'a>( + output: &mut Vec, + exemplars: impl IntoIterator, + metric: &str, + row: usize, + ctx: &OtlpMetricCtx, +) { + let mut exemplars = exemplars.into_iter().peekable(); + if exemplars.peek().is_none() { + return; + } + if !ctx.with_metric_engine { + crate::metrics::EXEMPLARS_DROPPED_TOTAL + .with_label_values(&["otlp"]) + .inc_by(exemplars.count() as u64); + return; + } + let validate_name = |name: &[u8]| !name.is_empty(); + let series_labels = Arc::new(BTreeMap::new()); + let metric: Arc = metric.into(); + let mut pending = RowExemplars { + metric: metric.clone(), + row, + observations: Vec::new(), + }; + for exemplar in exemplars { + let observation: Result<_> = (|| { + let mut names = HashSet::new(); + let mut attributes = Vec::new(); + for attribute in &exemplar.filtered_attributes { + ensure!( + names.insert(&attribute.key), + InvalidOtlpMetricInputSnafu { + reason: format!("duplicate filtered exemplar attribute: {}", attribute.key) + } + ); + if let Some(value) = scalar_value_string(attribute.value.as_ref()) { + let name = + translate_label_name(&attribute.key, ctx.metric_translation_strategy); + attributes.push((name, value)); + } + } + let mut labels = normalized::labels(attributes, true, validate_name)?; + for (name, bytes, length) in [ + ("trace_id", &exemplar.trace_id, 16), + ("span_id", &exemplar.span_id, 8), + ] { + if !bytes.is_empty() { + ensure!( + bytes.len() == length, + InvalidOtlpMetricInputSnafu { + reason: format!("exemplar {name} must contain {length} bytes") + } + ); + labels.insert(name.to_string(), hex::encode(bytes)); + } + } + let labels = normalized::labels(labels, true, validate_name)?; + let timestamp = i64::try_from(exemplar.time_unix_nano).map_err(|_| { + InvalidOtlpMetricInputSnafu { + reason: "exemplar timestamp exceeds signed nanosecond range", + } + .build() + })?; + Ok(Observation { + metric: metric.clone(), + series_labels: series_labels.clone(), + timestamp, + value: value(exemplar)?, + labels, + }) + })(); + match observation { + Ok(observation) => pending.observations.push(observation), + Err(_) => crate::metrics::EXEMPLARS_DROPPED_TOTAL + .with_label_values(&["otlp"]) + .inc(), + } + } + if !pending.observations.is_empty() { + output.push(pending); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::otlp::metrics::string_key_value; + use crate::row_writer; + use crate::row_writer::TableData; + + #[test] + fn test_exemplars_require_metric_engine() { + let exemplar = Exemplar { + time_unix_nano: 1, + value: Some(exemplar::Value::AsDouble(0.5)), + ..Default::default() + }; + let dropped = crate::metrics::EXEMPLARS_DROPPED_TOTAL.with_label_values(&["otlp"]); + for enabled in [false, true] { + let before = dropped.get(); + let mut output = Vec::new(); + append( + &mut output, + [&exemplar], + "latency", + 0, + &OtlpMetricCtx { + with_metric_engine: enabled, + ..Default::default() + }, + ); + assert_eq!(output.len(), usize::from(enabled)); + assert_eq!(dropped.get() - before, u64::from(!enabled)); + } + } + + #[test] + fn test_exemplar_labels_follow_otlp_translation_strategy() { + use otel_arrow_rust::proto::opentelemetry::metrics::v1::{ + Gauge, NumberDataPoint, number_data_point, + }; + use session::protocol_ctx::OtlpMetricTranslationStrategy; + + use crate::otlp::metrics::encode_gauge; + use crate::row_writer::MultiTableData; + + for strategy in [ + OtlpMetricTranslationStrategy::NoTranslation, + OtlpMetricTranslationStrategy::NoUtf8EscapingWithSuffixes, + OtlpMetricTranslationStrategy::UnderscoreEscapingWithSuffixes, + OtlpMetricTranslationStrategy::UnderscoreEscapingWithoutSuffixes, + ] { + for name in ["request.method", "请求"] { + let gauge = Gauge { + data_points: vec![NumberDataPoint { + attributes: vec![string_key_value(name, "GET".into())], + time_unix_nano: 1, + value: Some(number_data_point::Value::AsDouble(1.0)), + exemplars: vec![Exemplar { + filtered_attributes: vec![string_key_value(name, "POST".into())], + time_unix_nano: 1, + value: Some(exemplar::Value::AsDouble(0.5)), + ..Default::default() + }], + ..Default::default() + }], + }; + let mut tables = MultiTableData::default(); + let mut output = Vec::new(); + encode_gauge( + &mut tables, + "requests", + &gauge, + None, + None, + &OtlpMetricCtx { + with_metric_engine: true, + metric_translation_strategy: strategy, + ..Default::default() + }, + &mut output, + ) + .unwrap(); + let (requests, rows) = tables.into_row_insert_requests(); + assert_eq!(rows, 1); + let rows = requests.inserts[0].rows.as_ref().unwrap(); + let output = output + .into_iter() + .flat_map(|pending| { + let row = pending.row; + pending.finish(&rows.schema, &rows.rows[row].values) + }) + .collect::>(); + let name = translate_label_name(name, strategy); + assert_eq!(output.len(), 1); + assert_eq!(output[0].series_labels[&name], "GET"); + assert_eq!(output[0].labels[&name], "POST"); + assert!( + requests.inserts[0] + .rows + .as_ref() + .unwrap() + .schema + .iter() + .any(|column| column.column_name == name) + ); + } + } + } + + #[test] + fn own_time_labels_and_dedicated_id_precedence() { + let mut table = TableData::new(2, 1); + let mut row = table.alloc_one_row(); + row_writer::write_tag(&mut table, "job", "api", &mut row).unwrap(); + let exemplar = Exemplar { + filtered_attributes: vec![ + string_key_value("trace_id", "fallback".into()), + string_key_value("span_id", "fallback_span".into()), + string_key_value("request.id", "r1".into()), + ], + time_unix_nano: 1720000000123456789, + trace_id: vec![0xab; 16], + span_id: vec![0xcd; 8], + value: Some(exemplar::Value::AsDouble(0.32)), + }; + let ctx = OtlpMetricCtx { + with_metric_engine: true, + ..Default::default() + }; + let mut output = Vec::new(); + append(&mut output, [&exemplar], "latency", table.num_rows(), &ctx); + let first = output.pop().unwrap().finish(table.columns(), &row); + assert_eq!(first[0].timestamp, 1720000000123456789); + assert_eq!(first[0].series_labels["job"], "api"); + assert_eq!(first[0].labels["trace_id"], "ab".repeat(16)); + assert_eq!(first[0].labels["span_id"], "cd".repeat(8)); + let mut without_id = exemplar.clone(); + without_id.trace_id.clear(); + append( + &mut output, + [&without_id], + "latency", + table.num_rows(), + &ctx, + ); + let second = output.pop().unwrap().finish(table.columns(), &row); + assert_eq!(second[0].labels["trace_id"], "fallback"); + } +} diff --git a/src/servers/src/otlp/metrics/prepare.rs b/src/servers/src/otlp/metrics/prepare.rs index 6254cf0120c..ef6a731269d 100644 --- a/src/servers/src/otlp/metrics/prepare.rs +++ b/src/servers/src/otlp/metrics/prepare.rs @@ -31,6 +31,7 @@ use table::requests::{ use crate::error::{self, Result}; use crate::metrics::OTLP_EXPONENTIAL_HISTOGRAM_REJECTED_DATA_POINTS; +use crate::otlp::metrics::exemplars::RowExemplars; use crate::otlp::metrics::{ MetricMetadata, MetricsConversion, OTEL_RESOURCE_INFO_TABLE_NAME, ResourceInfoData, add_accepted_data_points, encode_metrics, from_metric_type, has_no_recorded_value, @@ -56,6 +57,7 @@ struct PreparedPoint { resource: usize, timestamp: i64, rows: Range, + exemplars: Vec, } #[derive(Default)] @@ -213,6 +215,7 @@ impl PreparedMetrics { let invalid_interval = !matches!(metric.data, Some(metric::Data::Gauge(_))) && start > timestamp; let mut outcome = MetricsIngestOutcome::default(); + let mut exemplars = Vec::new(); let encoded = if !exponential && (timestamp > i64::MAX as u64 || invalid_interval) { Err(error::InvalidOtlpMetricInputSnafu { reason: "invalid OTLP data-point timestamp or interval", @@ -227,6 +230,7 @@ impl PreparedMetrics { scope_attrs, ctx, &mut outcome, + &mut exemplars, ) }; if let Err(err) = encoded { @@ -259,6 +263,16 @@ impl PreparedMetrics { } let first_row = self.row_locations.len(); for (name, schema, rows) in writer.tables.drain_rows() { + for pending in exemplars + .iter_mut() + .filter(|pending| pending.metric.as_ref() == name) + { + pending.row = self.row_locations.len().checked_add(pending.row).context( + error::InternalSnafu { + err_msg: "exemplar source row offset overflow", + }, + )?; + } let batch = match writer.batches.get(name) { Some(&(columns, batch)) if columns == schema.len() => batch, _ => { @@ -313,6 +327,7 @@ impl PreparedMetrics { resource, timestamp: timestamp as i64, rows: first_row..self.row_locations.len(), + exemplars, }); Ok(true) } @@ -412,6 +427,28 @@ impl PreparedMetrics { rejections[index] = Some("incompatible metric sample kinds or column definitions".to_string()); } + // Bind before survivor merging moves batch schemas and row values. + let mut exemplars = Vec::new(); + for (point, rejection) in self.points.iter_mut().zip(&rejections) { + if rejection.is_some() { + continue; + } + for pending in std::mem::take(&mut point.exemplars) { + let &(batch, row) = + self.row_locations + .get(pending.row) + .context(error::InternalSnafu { + err_msg: "missing exemplar source row", + })?; + let rows = self.requests[batch] + .rows + .as_ref() + .context(error::InternalSnafu { + err_msg: "missing prepared batch", + })?; + exemplars.extend(pending.finish(&rows.schema, &rows.rows[row].values)); + } + } let reuse_batches = rejections.iter().all(Option::is_none) && { let mut tables = HashSet::new(); self.requests.len() <= 1 || self.requests.iter().all(|r| tables.insert(&r.table_name)) @@ -539,6 +576,7 @@ impl PreparedMetrics { ); } Ok(MetricsConversion { + exemplars, requests, rows, semantic_index, diff --git a/src/servers/src/postgres.rs b/src/servers/src/postgres.rs index 1d46ab9c010..e9032bc79bd 100644 --- a/src/servers/src/postgres.rs +++ b/src/servers/src/postgres.rs @@ -47,6 +47,7 @@ use crate::query_handler::sql::ServerSqlQueryHandlerRef; use crate::query_handler::{ CopyInHandlerRef, InfluxdbLineProtocolHandlerRef, OpenTelemetryProtocolHandlerRef, }; +use crate::request_memory_limiter::ServerMemoryLimiter; pub(crate) struct GreptimeDBStartupParameters { version: String, @@ -94,6 +95,7 @@ pub struct PostgresServerHandlerInner { /// HTTP wiring passes from its options. otlp_handler: Option, otlp_with_metric_engine: bool, + memory_limiter: ServerMemoryLimiter, force_tls: bool, auth_driver: AuthDriver, UserInfoRef>, /// SCRAM selected; failures get the uniform metric label. @@ -118,6 +120,8 @@ pub(crate) struct MakePostgresServerHandler { otlp_handler: Option, #[builder(default)] otlp_with_metric_engine: bool, + #[builder(default)] + memory_limiter: ServerMemoryLimiter, user_provider: Option, #[builder(default = "Arc::new(GreptimeDBStartupParameters::new())")] param_provider: Arc, @@ -171,6 +175,7 @@ impl MakePostgresServerHandler { influxdb_handler: self.influxdb_handler.clone(), otlp_handler: self.otlp_handler.clone(), otlp_with_metric_engine: self.otlp_with_metric_engine, + memory_limiter: self.memory_limiter.clone(), force_tls: self.force_tls, auth_driver: AuthDriver::new(self.param_provider.clone()) .with_pid_secret_key_generator(Arc::new(ZeroPidSecretKeyGenerator)), diff --git a/src/servers/src/postgres/copy_in/mod.rs b/src/servers/src/postgres/copy_in/mod.rs index 064b65b8bb5..d63b50367ce 100644 --- a/src/servers/src/postgres/copy_in/mod.rs +++ b/src/servers/src/postgres/copy_in/mod.rs @@ -721,7 +721,11 @@ impl PostgresServerHandlerInner { let request = ExportMetricsServiceRequest::decode(bytes.as_slice()) .map_err(|e| bad_copy_data(format!("invalid OTLP payload: {e}")))?; let outcome = otlp_handler - .metrics(request, Arc::new(ctx)) + .metrics( + request, + Arc::new(ctx), + Some((self.memory_limiter.clone(), 0)), + ) .await .map_err(convert_err)?; // Same failure semantics as the HTTP route: a payload diff --git a/src/servers/src/postgres/server.rs b/src/servers/src/postgres/server.rs index 8bdbce65a9c..d12382a0506 100644 --- a/src/servers/src/postgres/server.rs +++ b/src/servers/src/postgres/server.rs @@ -33,6 +33,7 @@ use crate::query_handler::CopyInHandlerRef; use crate::query_handler::InfluxdbLineProtocolHandlerRef; use crate::query_handler::OpenTelemetryProtocolHandlerRef; use crate::query_handler::sql::ServerSqlQueryHandlerRef; +use crate::request_memory_limiter::ServerMemoryLimiter; use crate::server::{AbortableStream, BaseTcpServer, Server}; use crate::tls::ReloadableTlsServerConfig; @@ -45,6 +46,7 @@ pub struct PostgresServer { influxdb_handler: Option, otlp_handler: Option, otlp_with_metric_engine: bool, + memory_limiter: ServerMemoryLimiter, bind_addr: Option, process_manager: Option, } @@ -85,6 +87,7 @@ impl PostgresServer { influxdb_handler: None, otlp_handler: None, otlp_with_metric_engine: false, + memory_limiter: ServerMemoryLimiter::default(), bind_addr: None, process_manager, } @@ -96,6 +99,12 @@ impl PostgresServer { self } + /// Sets the shared memory limiter used for OTLP exemplar expansion. + pub fn with_memory_limiter(mut self, limiter: ServerMemoryLimiter) -> Self { + self.memory_limiter = limiter; + self + } + /// Wires the influxdb line-protocol handler used by table-less /// `COPY FROM STDIN WITH (FORMAT influxdb)` flushes — the same /// handler the HTTP influxdb write endpoints use, so a flushed batch @@ -137,6 +146,7 @@ impl PostgresServer { influxdb_handler: self.influxdb_handler.clone(), otlp_handler: self.otlp_handler.clone(), otlp_with_metric_engine: self.otlp_with_metric_engine, + memory_limiter: self.memory_limiter.clone(), user_provider: make_handler.user_provider.clone(), param_provider: make_handler.param_provider.clone(), force_tls: make_handler.force_tls, diff --git a/src/servers/src/prom_remote_write/decode.rs b/src/servers/src/prom_remote_write/decode.rs index 580fc5c57a3..7ab90a0ff7b 100644 --- a/src/servers/src/prom_remote_write/decode.rs +++ b/src/servers/src/prom_remote_write/decode.rs @@ -596,7 +596,7 @@ mod tests { ); } assert!( - crate::exemplar::estimated_row_bytes(&requests).unwrap() + crate::exemplar::estimated_row_bytes(requests.values().map(Vec::as_slice)).unwrap() >= count * (metric.len() + label.len()) ); } diff --git a/src/servers/src/prom_remote_write/v2.rs b/src/servers/src/prom_remote_write/v2.rs index a8e11fc65e6..6f82247b727 100644 --- a/src/servers/src/prom_remote_write/v2.rs +++ b/src/servers/src/prom_remote_write/v2.rs @@ -561,7 +561,9 @@ fn convert_remote_write_v2( checked_total(histogram_count_total, counts.histograms, "histogram")?; } - budget.reserve(crate::exemplar::estimated_row_bytes(&exemplars))?; + budget.reserve(crate::exemplar::estimated_row_bytes( + exemplars.values().map(Vec::as_slice), + ))?; Ok(RemoteWriteV2WriteRequests { samples: into_context_req(sample_tables), diff --git a/src/servers/src/query_handler.rs b/src/servers/src/query_handler.rs index c0d1d7c9b0b..62aad96c873 100644 --- a/src/servers/src/query_handler.rs +++ b/src/servers/src/query_handler.rs @@ -56,6 +56,7 @@ use crate::error::Result; use crate::http::jaeger::QueryTraceParams; use crate::influxdb::InfluxdbRequest; use crate::opentsdb::codec::DataPoint; +use crate::request_memory_limiter::ServerMemoryLimiter; pub type OpentsdbProtocolHandlerRef = Arc; pub type InfluxdbLineProtocolHandlerRef = Arc; pub type PromStoreProtocolHandlerRef = Arc; @@ -200,11 +201,13 @@ pub trait PromStoreProtocolHandler { #[async_trait] pub trait OpenTelemetryProtocolHandler: PipelineHandler { - /// Handling opentelemetry metrics request + /// Handles metrics, reserving row expansion against the transport's limiter. + /// The accompanying byte count includes permits already held by this request. async fn metrics( &self, request: ExportMetricsServiceRequest, ctx: QueryContextRef, + memory_limiter: Option<(ServerMemoryLimiter, u64)>, ) -> Result; /// Handling opentelemetry traces request diff --git a/src/servers/src/request_memory_limiter.rs b/src/servers/src/request_memory_limiter.rs index e2be0a08151..4996d40b69a 100644 --- a/src/servers/src/request_memory_limiter.rs +++ b/src/servers/src/request_memory_limiter.rs @@ -70,7 +70,7 @@ impl ServerMemoryLimiter { } /// Rejects requests whose additional bytes cannot fit alongside their own held permits. - pub(crate) fn check_request_fits(&self, bytes: u64, held_bytes: u64) -> Result<()> { + pub fn check_request_fits(&self, bytes: u64, held_bytes: u64) -> Result<()> { let requested_bytes = held_bytes.checked_add(bytes); let limit_bytes = self.limit_bytes(); match requested_bytes { diff --git a/tests-integration/src/instance.rs b/tests-integration/src/instance.rs index 140151d594f..f0fc8845755 100644 --- a/tests-integration/src/instance.rs +++ b/tests-integration/src/instance.rs @@ -778,7 +778,7 @@ mod tests { let errors = frontend::OTLP_RESOURCE_INFO_WRITE_ERRORS.get(); let outcome = - OpenTelemetryProtocolHandler::metrics(&*instance, request, QueryContext::arc()) + OpenTelemetryProtocolHandler::metrics(&*instance, request, QueryContext::arc(), None) .await .unwrap(); assert_eq!( diff --git a/tests-integration/src/otlp.rs b/tests-integration/src/otlp.rs index 98cc66bcafb..47e169e54ec 100644 --- a/tests-integration/src/otlp.rs +++ b/tests-integration/src/otlp.rs @@ -21,6 +21,7 @@ mod test { use api::region::RegionResponse; use api::v1::region::{RegionRequest, region_request}; use client::{DEFAULT_CATALOG_NAME, OutputData}; + use common_catalog::consts::METRICS_EXEMPLARS_TABLE; use common_meta::cache::LayeredCacheRegistryBuilder; use common_meta::node_manager::{ Datanode, DatanodeManager, DatanodeRef, FlownodeManager, FlownodeRef, NodeManagerRef, @@ -1533,6 +1534,7 @@ WITH( .metrics( build_sum_request("custom", AggregationTemporality::Cumulative, &[(60, 7)]), ctx.clone(), + None, ) .await .unwrap(); @@ -1564,7 +1566,8 @@ WITH( AggregationTemporality::Cumulative, &[(60, 1)] ), - Arc::new(fixed) + Arc::new(fixed), + None ) .await .unwrap() @@ -1575,6 +1578,878 @@ WITH( assert_eq!(results[..3], results[3..]); } + fn exemplar_metric_request(count: usize) -> ExportMetricsServiceRequest { + use otel_arrow_rust::proto::opentelemetry::metrics::v1::{Exemplar, exemplar}; + + metric_request( + (0..count) + .map(|i| Metric { + name: format!("exemplar_memory_{i}"), + data: Some(metric::Data::Gauge(Gauge { + data_points: vec![NumberDataPoint { + attributes: vec![keyvalue(&format!("tag_{i}"), "value")], + time_unix_nano: 60_000_000_000, + value: Some(Value::AsDouble(1.0)), + exemplars: vec![Exemplar { + time_unix_nano: 60_000_000_000, + value: Some(exemplar::Value::AsDouble(1.0)), + ..Default::default() + }], + ..Default::default() + }], + })), + ..Default::default() + }) + .collect(), + ) + } + + #[tokio::test(flavor = "multi_thread")] + async fn test_exemplar_otlp_existing_ordinary_tables() { + use common_memory_manager::OnExhaustedPolicy; + use servers::request_memory_limiter::ServerMemoryLimiter; + + for (batching, count) in [(false, 1), (true, 1), (false, 64), (true, 64)] { + let standalone = GreptimeDbStandaloneBuilder::new("exemplar_otlp_ordinary") + .with_logical_batcher(BatcherOptions { + protocols: vec![BatchingProtocol::Otlp], + pending_rows_flush_interval: Duration::from_millis(5), + ..Default::default() + }) + .build() + .await; + let instance = standalone.fe_instance(); + let request = exemplar_metric_request(count); + let mut seed = request.clone(); + for metric in &mut seed.resource_metrics[0].scope_metrics[0].metrics { + let Some(metric::Data::Gauge(gauge)) = &mut metric.data else { + unreachable!() + }; + gauge.data_points[0].exemplars.clear(); + } + // Ordinary tables can already exist when metric-engine ingestion is enabled. + instance + .metrics(seed, QueryContext::arc(), None) + .await + .unwrap(); + let mut ctx = QueryContext::with(DEFAULT_CATALOG_NAME, "public"); + ctx.set_logical_batching_enabled(batching); + ctx.set_protocol_ctx(ProtocolCtx::OtlpMetric(OtlpMetricCtx { + with_metric_engine: true, + ..Default::default() + })); + let ctx = Arc::new(ctx); + let limiter = ServerMemoryLimiter::new(64 * 1024, OnExhaustedPolicy::Fail); + let dropped = servers::metrics::EXEMPLARS_DROPPED_TOTAL.with_label_values(&["otlp"]); + for round in 0..3 { + let mut request = request.clone(); + let metrics = &mut request.resource_metrics[0].scope_metrics[0].metrics; + if round > 0 { + let mut sibling = metrics[0].clone(); + sibling.name = "logical_sibling".into(); + metrics.push(sibling); + } + for metric in metrics { + let Some(metric::Data::Gauge(gauge)) = &mut metric.data else { + unreachable!() + }; + gauge.data_points[0].time_unix_nano += (round + 1) * 60_000_000_000; + } + let before = dropped.get(); + let outcome = instance + .metrics(request, ctx.clone(), Some((limiter.clone(), 0))) + .await + .unwrap(); + assert_eq!( + outcome.accepted_data_points, + count as i64 + i64::from(round > 0) + ); + assert_eq!(outcome.rejected_data_points, 0); + assert_eq!(dropped.get() - before, count as u64); + assert_eq!(limiter.used_bytes(), 0); + if round == 0 { + assert_trace_v2_query(instance, &format!("SELECT COUNT(*)=0 FROM information_schema.tables WHERE table_name='{METRICS_EXEMPLARS_TABLE}'"), ctx.clone()).await.unwrap(); + } + } + for sql in [ + "SELECT COUNT(*)=4 FROM exemplar_memory_0", + "SELECT COUNT(*)=2 FROM logical_sibling", + &format!( + "SELECT COUNT(*)=2 AND MIN(greptime_metric)='logical_sibling' FROM {METRICS_EXEMPLARS_TABLE}" + ), + ] { + assert_trace_v2_query(instance, sql, ctx.clone()) + .await + .unwrap(); + } + } + } + + #[tokio::test(flavor = "multi_thread")] + async fn test_exemplar_otlp_without_metric_engine() { + use axum::body::Body; + use common_memory_manager::OnExhaustedPolicy; + use futures::SinkExt; + use http::{Request, StatusCode}; + use prost::Message; + use servers::http::HttpServerBuilder; + use servers::postgres::PostgresServer; + use servers::request_memory_limiter::ServerMemoryLimiter; + use servers::server::Server; + use servers::tls::ReloadableTlsServerConfig; + use tower::ServiceExt; + + let standalone = GreptimeDbStandaloneBuilder::new("exemplar_otlp_disabled") + .build() + .await; + let instance = standalone.fe_instance(); + // The body fits, but expanding the discarded exemplars would exceed the quota. + let limiter = ServerMemoryLimiter::new(64 * 1024, OnExhaustedPolicy::Fail); + let server = HttpServerBuilder::new(Default::default()) + .with_memory_limiter(limiter.clone()) + .with_otlp_handler(instance.clone(), false) + .build(); + let app = server.build(server.make_app()).unwrap(); + let request = exemplar_metric_request(64); + let response = app + .oneshot( + Request::post("/v1/otlp/v1/metrics") + .header("content-type", "application/x-protobuf") + .body(Body::from(request.encode_to_vec())) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + assert_eq!(limiter.used_bytes(), 0); + + let mut server = PostgresServer::new( + instance.clone(), + instance.clone(), + false, + Arc::new(ReloadableTlsServerConfig::try_new(Default::default()).unwrap()), + 0, + common_runtime::global_runtime(), + None, + None, + ) + .with_otlp_handler(instance.clone(), false) + .with_memory_limiter(limiter.clone()); + server.start("127.0.0.1:0".parse().unwrap()).await.unwrap(); + let (client, connection) = tokio_postgres::connect( + &format!( + "host=127.0.0.1 port={} dbname=public", + server.bind_addr().unwrap().port() + ), + tokio_postgres::NoTls, + ) + .await + .unwrap(); + let connection = tokio::spawn(connection); + let mut sink = Box::pin( + client + .copy_in("COPY FROM STDIN WITH (FORMAT otlp)") + .await + .unwrap(), + ); + let mut request = request; + for metric in &mut request.resource_metrics[0].scope_metrics[0].metrics { + let Some(metric::Data::Gauge(gauge)) = &mut metric.data else { + unreachable!() + }; + gauge.data_points[0].time_unix_nano += 1_000_000_000; + } + sink.as_mut() + .send(bytes::Bytes::from(request.encode_to_vec())) + .await + .unwrap(); + assert_eq!(sink.as_mut().finish().await.unwrap(), 64); + assert_eq!(limiter.used_bytes(), 0); + drop(sink); + drop(client); + connection.await.unwrap().unwrap(); + server.shutdown().await.unwrap(); + + let ctx = Arc::new(QueryContext::with(DEFAULT_CATALOG_NAME, "public")); + assert_trace_v2_query( + instance, + "SELECT COUNT(*)=2 FROM exemplar_memory_0", + ctx.clone(), + ) + .await + .unwrap(); + assert_trace_v2_query( + instance, + &format!("SELECT COUNT(*)=0 FROM information_schema.tables WHERE table_name='{METRICS_EXEMPLARS_TABLE}'"), + ctx, + ) + .await + .unwrap(); + } + + #[tokio::test(flavor = "multi_thread")] + async fn test_exemplar_otlp_memory_limit() { + use std::io::Write; + + use axum::body::Body; + use common_memory_manager::OnExhaustedPolicy; + use flate2::Compression; + use flate2::write::GzEncoder; + use futures::SinkExt; + use http::{Request, StatusCode}; + use prost::Message; + use servers::http::HttpServerBuilder; + use servers::postgres::PostgresServer; + use servers::request_memory_limiter::ServerMemoryLimiter; + use servers::server::Server; + use servers::tls::ReloadableTlsServerConfig; + use tower::ServiceExt; + + let standalone = GreptimeDbStandaloneBuilder::new("exemplar_otlp_memory") + .build() + .await; + let instance = standalone.fe_instance(); + for (count, compressed) in [(64, false), (64, true), (1, true)] { + let request = exemplar_metric_request(count); + let limit = if count == 1 { + let conversion = servers::otlp::metrics::to_grpc_insert_requests( + request.clone(), + &mut OtlpMetricCtx { + with_metric_engine: true, + ..Default::default() + }, + ) + .unwrap(); + let bytes = servers::exemplar::estimated_row_bytes(std::iter::once( + conversion.exemplars.as_slice(), + )) + .unwrap() as u64; + // Fits the exemplar rows plus either body buffer, but not both. + bytes.div_ceil(1024) * 1024 + 1024 + } else { + 64 * 1024 + }; + let limiter = ServerMemoryLimiter::new( + limit, + OnExhaustedPolicy::Wait { + timeout: Duration::from_secs(30), + }, + ); + let server = HttpServerBuilder::new(Default::default()) + .with_memory_limiter(limiter.clone()) + .with_otlp_handler(instance.clone(), true) + .build(); + let app = server.build(server.make_app()).unwrap(); + let body = request.encode_to_vec(); + assert!((body.len() as u64) < limit); + let mut request = Request::post("/v1/otlp/v1/metrics") + .header("content-type", "application/x-protobuf"); + let body = if compressed { + request = request.header("content-encoding", "gzip"); + let mut encoder = GzEncoder::new(Vec::new(), Compression::default()); + encoder.write_all(&body).unwrap(); + encoder.finish().unwrap() + } else { + body.clone() + }; + let response = tokio::time::timeout( + Duration::from_secs(5), + app.oneshot(request.body(Body::from(body)).unwrap()), + ) + .await + .expect("requests must not wait for their own body permits") + .unwrap(); + assert_eq!(response.status(), StatusCode::TOO_MANY_REQUESTS); + assert_eq!(limiter.used_bytes(), 0); + } + let limiter = ServerMemoryLimiter::new(64 * 1024, OnExhaustedPolicy::Fail); + let mut server = PostgresServer::new( + instance.clone(), + instance.clone(), + false, + Arc::new(ReloadableTlsServerConfig::try_new(Default::default()).unwrap()), + 0, + common_runtime::global_runtime(), + None, + None, + ) + .with_otlp_handler(instance.clone(), true) + .with_memory_limiter(limiter.clone()); + server.start("127.0.0.1:0".parse().unwrap()).await.unwrap(); + let (client, connection) = tokio_postgres::connect( + &format!( + "host=127.0.0.1 port={} dbname=public", + server.bind_addr().unwrap().port() + ), + tokio_postgres::NoTls, + ) + .await + .unwrap(); + let connection = tokio::spawn(connection); + let mut sink = Box::pin( + client + .copy_in("COPY FROM STDIN WITH (FORMAT otlp)") + .await + .unwrap(), + ); + sink.as_mut() + .send(bytes::Bytes::from( + exemplar_metric_request(64).encode_to_vec(), + )) + .await + .unwrap(); + let error = sink.as_mut().finish().await.unwrap_err(); + let error = error.as_db_error().unwrap(); + assert_eq!(error.code().code(), "54000", "{error}"); + assert!( + error.message().starts_with("Memory limit exceeded"), + "{error}" + ); + assert_eq!(limiter.used_bytes(), 0); + drop(sink); + drop(client); + connection.await.unwrap().unwrap(); + server.shutdown().await.unwrap(); + + for table in ["exemplar_memory_0", METRICS_EXEMPLARS_TABLE] { + assert!( + instance + .catalog_manager() + .table(DEFAULT_CATALOG_NAME, "public", table, None) + .await + .unwrap() + .is_none() + ); + } + } + + #[tokio::test(flavor = "multi_thread")] + async fn test_exemplar_otlp_memory_wait_and_lifetime() { + use common_memory_manager::OnExhaustedPolicy; + use servers::request_memory_limiter::ServerMemoryLimiter; + + let standalone = GreptimeDbStandaloneBuilder::new("exemplar_otlp_memory_lifetime") + .build() + .await; + let source = standalone.fe_instance(); + let gate = Arc::new(Mutex::new(None)); + let cache_registry = Arc::new( + cache::with_default_composite_cache_registry( + LayeredCacheRegistryBuilder::default().add_cache_registry( + cache::build_fundamental_cache_registry(standalone.kv_backend.clone()), + ), + ) + .unwrap() + .build(), + ); + let instance = Arc::new( + FrontendBuilder::new( + standalone.opts.frontend_options(), + standalone.kv_backend.clone(), + cache_registry, + source.catalog_manager().clone(), + Arc::new(GatedTraceNodeManager { + inner: source.node_manager().clone(), + gate: gate.clone(), + }), + source.procedure_executor().clone(), + Arc::new(catalog::process_manager::ProcessManager::new( + "exemplar_otlp_memory_lifetime".into(), + None, + )), + ) + .try_build() + .await + .unwrap(), + ); + let mut ctx = QueryContext::with(DEFAULT_CATALOG_NAME, "public"); + ctx.set_protocol_ctx(ProtocolCtx::OtlpMetric(OtlpMetricCtx { + with_metric_engine: true, + ..Default::default() + })); + let ctx = Arc::new(ctx); + let request = exemplar_metric_request(1); + instance + .metrics(request.clone(), ctx.clone(), None) + .await + .unwrap(); + let limit = 64 * 1024; + let limiter = ServerMemoryLimiter::new( + limit, + OnExhaustedPolicy::Wait { + timeout: Duration::from_secs(30), + }, + ); + let body = limiter.acquire(limit / 2).await.unwrap(); + let concurrent_body = limiter.acquire(limit / 2).await.unwrap(); + // Body permits cannot be released while their requests wait for expansion. + let error = tokio::time::timeout( + Duration::from_secs(5), + instance.metrics( + request.clone(), + ctx.clone(), + Some((limiter.clone(), body.granted_bytes())), + ), + ) + .await + .expect("requests holding body permits must not wait for expansion") + .unwrap_err(); + assert!(matches!( + error, + servers::error::Error::MemoryLimitExceeded { + source: common_memory_manager::Error::MemoryLimitExceeded { .. }, + .. + } + )); + assert_eq!(limiter.used_bytes(), limit); + drop(concurrent_body); + assert_eq!( + instance + .metrics( + request.clone(), + ctx.clone(), + Some((limiter.clone(), body.granted_bytes())), + ) + .await + .unwrap() + .accepted_data_points, + 1 + ); + assert_eq!(limiter.used_bytes(), body.granted_bytes()); + drop(body); + assert_eq!(limiter.used_bytes(), 0); + + let limiter = ServerMemoryLimiter::new( + limit, + OnExhaustedPolicy::Wait { + timeout: Duration::from_millis(100), + }, + ); + for release in [true, false] { + let held = limiter.acquire(limit).await.unwrap(); + let write = instance.metrics(request.clone(), ctx.clone(), Some((limiter.clone(), 0))); + tokio::pin!(write); + assert!(futures::poll!(&mut write).is_pending()); + if release { + drop(held); + assert_eq!(write.await.unwrap().accepted_data_points, 1); + } else { + assert!(matches!( + write.await, + Err(servers::error::Error::MemoryLimitExceeded { + source: common_memory_manager::Error::MemoryAcquireTimeout { .. }, + .. + }) + )); + drop(held); + } + assert_eq!(limiter.used_bytes(), 0); + } + + let limiter = ServerMemoryLimiter::new( + limit, + OnExhaustedPolicy::Wait { + timeout: Duration::from_secs(10), + }, + ); + let table = instance + .catalog_manager() + .table( + DEFAULT_CATALOG_NAME, + "public", + METRICS_EXEMPLARS_TABLE, + None, + ) + .await + .unwrap() + .unwrap(); + let region_id = instance + .partition_manager() + .find_physical_table_route(table.table_info().table_id()) + .await + .unwrap() + .region_routes[0] + .region + .id; + for cancel in [false, true] { + let (started_tx, started_rx) = oneshot::channel(); + let (release_tx, release_rx) = oneshot::channel(); + *gate.lock().unwrap() = Some(TraceAuxWriteGate { + region_id, + started: started_tx, + release: release_rx, + }); + let write = tokio::spawn({ + let instance = instance.clone(); + let request = request.clone(); + let ctx = ctx.clone(); + let limiter = limiter.clone(); + async move { instance.metrics(request, ctx, Some((limiter, 0))).await } + }); + tokio::time::timeout(Duration::from_secs(10), started_rx) + .await + .unwrap() + .unwrap(); + let reserved = limiter.used_bytes(); + assert!(reserved > 0); + if cancel { + write.abort(); + assert!(write.await.unwrap_err().is_cancelled()); + assert_eq!( + limiter.used_bytes(), + reserved, + "in-flight rows still own the permit" + ); + release_tx.send(true).unwrap(); + } else { + release_tx.send(false).unwrap(); + assert!(write.await.unwrap().is_err()); + } + // Acquiring the entire pool waits for the detached write to drop its guard. + let released = limiter.acquire(limit).await.unwrap(); + drop(released); + assert_eq!(limiter.used_bytes(), 0); + } + } + + #[tokio::test(flavor = "multi_thread")] + async fn test_exemplar_otlp_inherited_label_collisions() { + use otel_arrow_rust::proto::opentelemetry::metrics::v1::{Exemplar, exemplar}; + + let standalone = GreptimeDbStandaloneBuilder::new("exemplar_otlp_label_collisions") + .build() + .await; + let instance = standalone.fe_instance(); + let mut ctx = QueryContext::with(DEFAULT_CATALOG_NAME, "public"); + ctx.set_protocol_ctx(ProtocolCtx::OtlpMetric(OtlpMetricCtx { + with_metric_engine: true, + ..Default::default() + })); + let ctx = Arc::new(ctx); + let dropped = servers::metrics::EXEMPLARS_DROPPED_TOTAL.with_label_values(&["otlp"]); + let before = dropped.get(); + let observation = Exemplar { + time_unix_nano: 60_000_000_000, + value: Some(exemplar::Value::AsDouble(0.5)), + ..Default::default() + }; + let mut metrics = Vec::new(); + for (i, key) in [ + "greptime_metric", + "exemplar_labels", + "greptime.metric", + "exemplar.labels", + ] + .into_iter() + .enumerate() + { + let mut metric = + gauge_metric(&format!("collision_{i}"), &[(60_000_000_000, 1.0, None)]); + let Some(metric::Data::Gauge(gauge)) = &mut metric.data else { + unreachable!() + }; + gauge.data_points[0].attributes = vec![keyvalue(key, "original")]; + gauge.data_points[0].exemplars = vec![observation.clone(); 2]; + metrics.push(metric); + } + for sql in [ + "CREATE TABLE collision_physical (greptime_timestamp TIMESTAMP TIME INDEX, greptime_value DOUBLE) ENGINE=metric WITH ('physical_metric_table'='')", + "CREATE TABLE collision_default (greptime_timestamp TIMESTAMP TIME INDEX, greptime_value DOUBLE, greptime_metric STRING DEFAULT 'original' PRIMARY KEY) ENGINE=metric WITH ('on_physical_table'='collision_physical')", + ] { + instance.do_query(sql, ctx.clone()).await.remove(0).unwrap(); + } + for name in ["collision_default", "retained"] { + let mut metric = gauge_metric(name, &[(60_000_000_000, 1.0, None)]); + let Some(metric::Data::Gauge(gauge)) = &mut metric.data else { + unreachable!() + }; + gauge.data_points[0].exemplars = vec![observation.clone()]; + metrics.push(metric); + } + let outcome = instance + .metrics(metric_request(metrics), ctx.clone(), None) + .await + .unwrap(); + assert_eq!( + (outcome.accepted_data_points, outcome.rejected_data_points), + (6, 0) + ); + assert_eq!(dropped.get(), before + 9); + for (table, column) in [ + ("collision_0", "greptime_metric"), + ("collision_1", "exemplar_labels"), + ("collision_2", "greptime_metric"), + ("collision_3", "exemplar_labels"), + ("collision_default", "greptime_metric"), + ] { + assert_trace_v2_query( + instance, + &format!("SELECT COUNT(*)=1 FROM {table} WHERE {column}='original'"), + ctx.clone(), + ) + .await + .unwrap(); + } + assert_trace_v2_query( + instance, + &format!("SELECT COUNT(*)=1 AND MIN(greptime_metric)='retained' FROM {METRICS_EXEMPLARS_TABLE}"), + ctx, + ) + .await + .unwrap(); + } + + #[tokio::test(flavor = "multi_thread")] + async fn test_exemplar_otlp_defaulted_source_identity() { + use otel_arrow_rust::proto::opentelemetry::metrics::v1::{Exemplar, exemplar}; + + for batching in [false, true] { + let standalone = GreptimeDbStandaloneBuilder::new("exemplar_otlp_defaulted_identity") + .with_logical_batcher(BatcherOptions { + protocols: vec![BatchingProtocol::Otlp], + pending_rows_flush_interval: Duration::from_millis(5), + ..Default::default() + }) + .build() + .await; + let instance = standalone.fe_instance(); + let mut ctx = QueryContext::with(DEFAULT_CATALOG_NAME, "public"); + ctx.set_logical_batching_enabled(batching); + ctx.set_protocol_ctx(ProtocolCtx::OtlpMetric(OtlpMetricCtx { + with_metric_engine: true, + ..Default::default() + })); + let ctx = Arc::new(ctx); + instance.do_query("CREATE TABLE defaults_physical (greptime_timestamp TIMESTAMP(3) TIME INDEX, greptime_value DOUBLE) ENGINE=metric WITH ('physical_metric_table'='')", ctx.clone()).await.remove(0).unwrap(); + for name in ["defaulted_gauge", "defaulted_histogram_bucket"] { + let (le, le_key) = if name.ends_with("_bucket") { + (", le STRING", ", le") + } else { + ("", "") + }; + instance.do_query(&format!("CREATE TABLE {name} (greptime_timestamp TIMESTAMP(3) TIME INDEX, greptime_value DOUBLE, host STRING DEFAULT 'unknown', empty_tag STRING DEFAULT '', nullable_tag STRING{le}, PRIMARY KEY(host, empty_tag, nullable_tag{le_key})) ENGINE=metric WITH ('on_physical_table'='defaults_physical')"), ctx.clone()).await.remove(0).unwrap(); + } + let mut gauge = gauge_metric( + "defaulted_gauge", + &[ + (60_000_000_000, 1.0, None), + (120_000_000_000, 2.0, Some("known")), + (180_000_000_000, 3.0, None), + ], + ); + let Some(metric::Data::Gauge(data)) = &mut gauge.data else { + unreachable!() + }; + for point in &mut data.data_points { + point.exemplars.push(Exemplar { + time_unix_nano: point.time_unix_nano, + value: Some(exemplar::Value::AsDouble(0.5)), + ..Default::default() + }); + } + let mut request = build_histogram_request( + "defaulted_histogram", + vec![HistogramDataPoint { + time_unix_nano: 240_000_000_000, + count: 3, + sum: Some(5.0), + explicit_bounds: vec![1.0, 2.0], + bucket_counts: vec![1, 1, 1], + exemplars: [0.5, 1.5, 3.0] + .into_iter() + .map(|value| Exemplar { + time_unix_nano: 240_000_000_000, + value: Some(exemplar::Value::AsDouble(value)), + ..Default::default() + }) + .collect(), + ..Default::default() + }], + ); + request.resource_metrics[0].scope_metrics[0] + .metrics + .push(gauge); + let outcome = instance.metrics(request, ctx.clone(), None).await.unwrap(); + assert_eq!( + (outcome.accepted_data_points, outcome.rejected_data_points), + (4, 0) + ); + assert_trace_v2_query( + instance, + "SELECT COUNT(*)=2 FROM defaulted_gauge WHERE host='unknown'", + ctx.clone(), + ) + .await + .unwrap(); + assert_trace_v2_query( + instance, + "SELECT COUNT(*)=3 FROM defaulted_histogram_bucket WHERE host='unknown'", + ctx.clone(), + ) + .await + .unwrap(); + assert_trace_v2_query(instance, &format!("SELECT COUNT(*)=6 FROM {METRICS_EXEMPLARS_TABLE} e JOIN defaults_physical m ON e.__table_id=m.__table_id AND e.__tsid=m.__tsid AND e.greptime_timestamp=m.greptime_timestamp"), ctx.clone()).await.unwrap(); + assert_trace_v2_query( + instance, + &format!("SELECT COUNT(*)=6 FROM {METRICS_EXEMPLARS_TABLE} WHERE empty_tag='' AND host IS NOT NULL"), + ctx, + ) + .await + .unwrap(); + } + } + + #[tokio::test(flavor = "multi_thread")] + async fn test_exemplar_otlp_source_identity() { + use otel_arrow_rust::proto::opentelemetry::metrics::v1::{Exemplar, exemplar}; + + for (with_metric_engine, batching) in [(false, false), (true, false), (true, true)] { + let standalone = GreptimeDbStandaloneBuilder::new("exemplar_otlp_identity") + .with_logical_batcher(BatcherOptions { + protocols: vec![BatchingProtocol::Otlp], + pending_rows_flush_interval: Duration::from_millis(5), + ..Default::default() + }) + .build() + .await; + let instance = standalone.fe_instance(); + let mut ctx = QueryContext::with(DEFAULT_CATALOG_NAME, "public"); + ctx.set_logical_batching_enabled(batching); + ctx.set_protocol_ctx(ProtocolCtx::OtlpMetric(OtlpMetricCtx { + with_metric_engine, + promote_all_resource_attrs: true, + ..Default::default() + })); + let ctx = Arc::new(ctx); + let mut request = build_histogram_request( + "latency", + vec![HistogramDataPoint { + attributes: vec![keyvalue("job", "point"), keyvalue("zone", "")], + time_unix_nano: 10_000_000_000, + count: 3, + sum: Some(5.0), + explicit_bounds: vec![1.0, 2.0], + bucket_counts: vec![1, 1, 1], + exemplars: [1.5, 3.0] + .into_iter() + .map(|value| Exemplar { + filtered_attributes: vec![keyvalue("trace_id", "observation")], + time_unix_nano: 9_123_456_789, + value: Some(exemplar::Value::AsDouble(value)), + ..Default::default() + }) + .collect(), + ..Default::default() + }], + ); + request.resource_metrics[0].resource = Some(Resource { + attributes: vec![keyvalue("job", "resource"), keyvalue("service.name", "api")], + ..Default::default() + }); + let result = instance.metrics(request, ctx.clone(), None).await; + assert_eq!(result.unwrap().accepted_data_points, 1); + if !with_metric_engine { + assert_trace_v2_query( + instance, + "SELECT COUNT(*)=3 FROM latency_bucket", + ctx.clone(), + ) + .await + .unwrap(); + assert_trace_v2_query(instance, &format!("SELECT COUNT(*)=0 FROM information_schema.tables WHERE table_name='{METRICS_EXEMPLARS_TABLE}'"), ctx).await.unwrap(); + continue; + } + assert_trace_v2_query( + instance, + &format!("SELECT COUNT(*)=2 FROM {METRICS_EXEMPLARS_TABLE}"), + ctx.clone(), + ) + .await + .unwrap(); + assert_trace_v2_query(instance, &format!("SELECT COUNT(*)=2 FROM {METRICS_EXEMPLARS_TABLE} e JOIN greptime_physical_table m ON e.__table_id=m.__table_id AND e.__tsid=m.__tsid AND e.le=m.le AND e.job=m.job AND e.zone=m.zone WHERE e.greptime_metric='latency_bucket'"), ctx).await.unwrap(); + } + } + + #[tokio::test(flavor = "multi_thread")] + async fn test_exemplar_otlp_schema_failure_isolation() { + use otel_arrow_rust::proto::opentelemetry::metrics::v1::{Exemplar, exemplar}; + + for metric_fails in [false, true] { + let standalone = GreptimeDbStandaloneBuilder::new("exemplar_otlp_partial") + .build() + .await; + let instance = standalone.fe_instance(); + let mut ctx = QueryContext::with(DEFAULT_CATALOG_NAME, "public"); + ctx.set_protocol_ctx(ProtocolCtx::OtlpMetric(OtlpMetricCtx { + with_metric_engine: true, + ..Default::default() + })); + let ctx = Arc::new(ctx); + let (failed, retained) = if metric_fails { + ("exemplar_partial_total", METRICS_EXEMPLARS_TABLE) + } else { + (METRICS_EXEMPLARS_TABLE, "exemplar_partial_total") + }; + instance + .do_query( + &format!( + "CREATE TABLE {failed} (ts TIMESTAMP TIME INDEX, greptime_value STRING)" + ), + ctx.clone(), + ) + .await + .remove(0) + .unwrap(); + let mut request = build_sum_request( + "exemplar.partial", + AggregationTemporality::Cumulative, + &[(10, 1)], + ); + let Some(metric::Data::Sum(sum)) = + &mut request.resource_metrics[0].scope_metrics[0].metrics[0].data + else { + unreachable!() + }; + sum.data_points[0].exemplars.push(Exemplar { + time_unix_nano: 10_123_456_789, + value: Some(exemplar::Value::AsDouble(0.75)), + ..Default::default() + }); + let result = instance.metrics(request, ctx.clone(), None).await; + if metric_fails { + let outcome = result.unwrap(); + assert_eq!(outcome.accepted_data_points, 0); + assert_eq!(outcome.rejected_data_points, 1); + assert_trace_v2_query( + instance, + &format!( + "SELECT COUNT(*)=0 FROM information_schema.tables \ + WHERE table_name='{METRICS_EXEMPLARS_TABLE}'" + ), + ctx, + ) + .await + .unwrap(); + continue; + } + assert!(result.is_err()); + let output = instance + .do_query(&format!("SELECT COUNT(*) FROM {retained}"), ctx) + .await + .remove(0) + .unwrap(); + let OutputData::Stream(stream) = output.data else { + unreachable!() + }; + let rows = RecordBatches::try_collect(stream) + .await + .unwrap() + .pretty_print() + .unwrap(); + assert!(rows.contains("| 1 |"), "{retained}: {rows}"); + } + } + #[tokio::test(flavor = "multi_thread")] pub async fn test_otlp_on_standalone() { let standalone = GreptimeDbStandaloneBuilder::new("test_standalone_otlp") @@ -1616,6 +2491,7 @@ WITH( .metrics( build_sum_request("fixed.delta", AggregationTemporality::Delta, &[(60, 10)]), ctx.clone(), + None, ) .await .unwrap(); @@ -1661,7 +2537,7 @@ WITH( .is_ok() ); - let resp = instance.metrics(req, ctx.clone()).await; + let resp = instance.metrics(req, ctx.clone(), None).await; assert!(resp.is_ok()); let mut output = instance @@ -1689,7 +2565,7 @@ WITH( ), build_sum_request(metric, AggregationTemporality::Cumulative, &[(180, 30)]), ] { - let result = instance.metrics(request, ctx.clone()).await; + let result = instance.metrics(request, ctx.clone(), None).await; assert!(result.is_ok(), "{metric}: {result:?}"); } @@ -1748,7 +2624,7 @@ WITH( unreachable!() }; sum.data_points[0].flags = DataPointFlags::NoRecordedValueMask as u32; - assert!(instance.metrics(stale, ctx.clone()).await.is_ok()); + assert!(instance.metrics(stale, ctx.clone(), None).await.is_ok()); for (matcher, expected_rows) in [ ("otlp_aggregation_temporality=\"delta\"", 0), @@ -1795,7 +2671,10 @@ WITH( ..Default::default() }], }; - let outcome = instance.metrics(malformed, ctx.clone()).await.unwrap(); + let outcome = instance + .metrics(malformed, ctx.clone(), None) + .await + .unwrap(); assert_eq!(0, outcome.accepted_data_points); assert_eq!(1, outcome.rejected_data_points); assert!( @@ -1857,6 +2736,7 @@ WITH( .metrics( build_histogram_request("raw.delta.histogram", points), ctx.clone(), + None, ) .await .unwrap(); @@ -2144,7 +3024,7 @@ WITH( ..Default::default() }], }; - instance.metrics(request, ctx.clone()).await.unwrap(); + instance.metrics(request, ctx.clone(), None).await.unwrap(); let mut output = instance .do_query("SELECT greptime_timestamp FROM my_gauge", ctx.clone()) @@ -2378,7 +3258,7 @@ WITH( ..Default::default() }); instance - .metrics(warm.clone(), Arc::new(ctx.clone())) + .metrics(warm.clone(), Arc::new(ctx.clone()), None) .await .unwrap(); let region = async |name: &str| { @@ -2414,7 +3294,7 @@ WITH( let write = tokio::spawn({ let instance = instance.clone(); let ctx = Arc::new(ctx.clone()); - async move { instance.metrics(request, ctx).await } + async move { instance.metrics(request, ctx, None).await } }); tokio::time::timeout(Duration::from_secs(10), started_rx) .await @@ -2527,7 +3407,12 @@ WITH( let auth::PermissionTableTargets::Resolved(targets) = targets else { panic!("metric destinations must be resolved"); }; - let denied = targets.iter().any(|t| t.table == "restricted_metric"); + let denied = targets.iter().any(|t| { + matches!( + t.table.as_str(), + "restricted_metric" | METRICS_EXEMPLARS_TABLE + ) + }); self.calls .lock() .unwrap() @@ -2575,7 +3460,7 @@ WITH( expected_targets.push("allowed_metric"); } let err = instance - .metrics(metric_request(metrics), ctx.clone()) + .metrics(metric_request(metrics), ctx.clone(), None) .await .unwrap_err(); assert_eq!(err.status_code(), StatusCode::PermissionDenied); @@ -2586,13 +3471,72 @@ WITH( ); } } + let mut metric = gauge_metric("allowed_metric", &[(60_000_000_000, 7.0, None)]); + let Some(metric::Data::Gauge(gauge)) = metric.data.as_mut() else { + unreachable!() + }; + gauge.data_points[0].exemplars.push( + otel_arrow_rust::proto::opentelemetry::metrics::v1::Exemplar { + time_unix_nano: 1, + value: Some( + otel_arrow_rust::proto::opentelemetry::metrics::v1::exemplar::Value::AsDouble( + 0.5, + ), + ), + ..Default::default() + }, + ); + let mut metric_ctx = ctx.fork(); + metric_ctx.set_protocol_ctx(ProtocolCtx::OtlpMetric(OtlpMetricCtx { + with_metric_engine: true, + ..Default::default() + })); + let metric_ctx = Arc::new(metric_ctx); + let request = metric_request(vec![metric]); + let err = instance + .metrics(request.clone(), metric_ctx.clone(), None) + .await + .unwrap_err(); + assert_eq!(err.status_code(), StatusCode::PermissionDenied); + assert_eq!( + std::mem::take(&mut *checker.calls.lock().unwrap()), + vec![vec!["allowed_metric"], vec![METRICS_EXEMPLARS_TABLE]] + ); assert_trace_v2_query( instance, "SELECT COUNT(*)=0 FROM information_schema.tables WHERE table_name='allowed_metric'", - ctx, + ctx.clone(), ) .await .unwrap(); + assert_eq!( + instance + .metrics(request.clone(), ctx.clone(), None) + .await + .unwrap() + .accepted_data_points, + 1 + ); + assert_eq!( + std::mem::take(&mut *checker.calls.lock().unwrap()), + vec![vec!["allowed_metric"]] + ); + // Enabling metric-engine ingestion leaves this ordinary destination unchanged. + assert_eq!( + instance + .metrics(request, metric_ctx, None) + .await + .unwrap() + .accepted_data_points, + 1 + ); + assert_eq!( + std::mem::take(&mut *checker.calls.lock().unwrap()), + vec![vec!["allowed_metric"]] + ); + assert_trace_v2_query(instance, "SELECT COUNT(*)=1 FROM allowed_metric", ctx) + .await + .unwrap(); } #[tokio::test(flavor = "multi_thread")] @@ -2614,6 +3558,7 @@ WITH( ], )]), ctx.clone(), + None, ) .await .unwrap(); @@ -2662,6 +3607,7 @@ WITH( gauge_metric("independent", &[(60_000_000_000, 7.0, None)]), ]), ctx.clone(), + None, ) .await .unwrap(); @@ -2732,7 +3678,7 @@ WITH( request.resource_metrics[0].scope_metrics[0] .metrics .push(native_metric("bad", -5)); - let outcome = instance.metrics(request, ctx.clone()).await.unwrap(); + let outcome = instance.metrics(request, ctx.clone(), None).await.unwrap(); assert_eq!( (outcome.accepted_data_points, outcome.rejected_data_points), (1, 1), @@ -2778,7 +3724,10 @@ WITH( "invalid_backing", &[(120_000_000_000, 1.0, None)], )); - let outcome = instance.metrics(request, Arc::new(selected)).await.unwrap(); + let outcome = instance + .metrics(request, Arc::new(selected), None) + .await + .unwrap(); assert_eq!( (outcome.accepted_data_points, outcome.rejected_data_points), (1, 1) @@ -2823,7 +3772,7 @@ WITH( }) .collect(); let outcome = instance - .metrics(metric_request(vec![metric]), ctx.clone()) + .metrics(metric_request(vec![metric]), ctx.clone(), None) .await .unwrap(); assert_eq!( @@ -2970,6 +3919,7 @@ WITH( .collect(), ), Arc::new(ctx.clone()), + None, ) .await .unwrap(); @@ -3049,7 +3999,7 @@ WITH( metrics.reverse(); } let outcome = instance - .metrics(metric_request(metrics.clone()), ctx.clone()) + .metrics(metric_request(metrics.clone()), ctx.clone(), None) .await .unwrap(); assert_eq!( @@ -3068,6 +4018,7 @@ WITH( gauge_metric("independent", &[(120_000_000_000, 9.0, None)]), ]), ctx.clone(), + None, ) .await .unwrap(); @@ -3085,7 +4036,7 @@ WITH( .unwrap() .clone(); let outcome = instance - .metrics(metric_request(vec![compatible]), ctx.clone()) + .metrics(metric_request(vec![compatible]), ctx.clone(), None) .await .unwrap(); assert_eq!( @@ -3102,7 +4053,7 @@ WITH( .name = "independent".into(); } let outcome = instance - .metrics(metric_request(metrics), ctx.clone()) + .metrics(metric_request(metrics), ctx.clone(), None) .await .unwrap(); assert_eq!( @@ -3146,6 +4097,7 @@ WITH( .collect(), ), ctx.clone(), + None, ) .await .unwrap(); @@ -3241,7 +4193,7 @@ WITH( 60_000_000_000 + if precision == 9 { 1 } else { 1 + round as u64 }; } let outcome = instance - .metrics(metric_request(vec![metric]), ctx.clone()) + .metrics(metric_request(vec![metric]), ctx.clone(), None) .await .unwrap(); assert_eq!( @@ -3314,7 +4266,7 @@ WITH( data.data_points[0].time_unix_nano += if precision == 6 { 1_000 } else { 2 }; } instance - .metrics(metric_request(vec![metric]), ctx.clone()) + .metrics(metric_request(vec![metric]), ctx.clone(), None) .await .unwrap(); assert_trace_v2_query(instance, "SELECT COUNT(*)=2 FROM interval", ctx.clone()) @@ -3347,6 +4299,7 @@ WITH( gauge_metric("independent", &[(60_000_000_000, 7.0, None)]), ]), ctx.clone(), + None, ) .await .unwrap(); @@ -3359,6 +4312,7 @@ WITH( gauge_metric("independent", &[(120_000_000_000, 9.0, None)]), ]), Arc::new(fixed), + None, ) .await .unwrap(); @@ -3393,7 +4347,7 @@ WITH( metrics.reverse(); } let outcome = instance - .metrics(metric_request(metrics), ctx.clone()) + .metrics(metric_request(metrics), ctx.clone(), None) .await .unwrap(); assert_eq!( diff --git a/tests-integration/src/tests/instance_test.rs b/tests-integration/src/tests/instance_test.rs index 86bd1409c1a..0615b684f6b 100644 --- a/tests-integration/src/tests/instance_test.rs +++ b/tests-integration/src/tests/instance_test.rs @@ -16,7 +16,7 @@ use std::env; use std::sync::Arc; use client::{DEFAULT_SCHEMA_NAME, OutputData}; -use common_catalog::consts::DEFAULT_CATALOG_NAME; +use common_catalog::consts::{DEFAULT_CATALOG_NAME, METRICS_EXEMPLARS_TABLE}; use common_error::ext::{ErrorExt, RetryHint}; use common_error::status_code::StatusCode; use common_query::Output; @@ -4560,7 +4560,7 @@ async fn test_exemplar_preserves_physical_table_routing(#[case] batching: bool) let table = frontend .catalog_manager() .table( - "greptime", + DEFAULT_CATALOG_NAME, "public", &format!("routed_{version}_{suffix}"), None, @@ -4574,7 +4574,7 @@ async fn test_exemplar_preserves_physical_table_routing(#[case] batching: bool) ); } let output = execute_sql(frontend, &format!( - "SELECT count(*) FROM metrics_exemplars e JOIN physical_a m ON e.__table_id = m.__table_id AND e.__tsid = m.__tsid WHERE e.greptime_metric = 'routed_{version}_a'" + "SELECT count(*) FROM {METRICS_EXEMPLARS_TABLE} e JOIN physical_a m ON e.__table_id = m.__table_id AND e.__tsid = m.__tsid WHERE e.greptime_metric = 'routed_{version}_a'" )).await; check_output_stream( output.data, @@ -4583,7 +4583,12 @@ async fn test_exemplar_preserves_physical_table_routing(#[case] batching: bool) .await; let table = frontend .catalog_manager() - .table("greptime", "public", "metrics_exemplars", None) + .table( + DEFAULT_CATALOG_NAME, + "public", + METRICS_EXEMPLARS_TABLE, + None, + ) .await .unwrap() .unwrap(); @@ -4776,7 +4781,7 @@ async fn test_exemplar_precreated_table_options( execute_sql(frontend, "CREATE TABLE source_physical (ts TIMESTAMP TIME INDEX, val DOUBLE) ENGINE=metric WITH ('physical_metric_table'='')").await; execute_sql(frontend, "CREATE TABLE metric_source (job STRING PRIMARY KEY, ts TIMESTAMP TIME INDEX, val DOUBLE) ENGINE=metric WITH ('on_physical_table'='source_physical')").await; execute_sql(frontend, &format!( - "CREATE TABLE metrics_exemplars (__table_id INT UNSIGNED NOT NULL, __tsid BIGINT UNSIGNED NOT NULL, greptime_metric STRING, greptime_timestamp TIMESTAMP({precision}) TIME INDEX, greptime_value DOUBLE, exemplar_labels JSON2, PRIMARY KEY (__table_id, __tsid, greptime_metric)) ENGINE=mito WITH ('append_mode'='{append_mode}', 'primary_key_encoding'='{encoding}', 'compaction.type'='twcs')" + "CREATE TABLE {METRICS_EXEMPLARS_TABLE} (__table_id INT UNSIGNED NOT NULL, __tsid BIGINT UNSIGNED NOT NULL, greptime_metric STRING, greptime_timestamp TIMESTAMP({precision}) TIME INDEX, greptime_value DOUBLE, exemplar_labels JSON2, PRIMARY KEY (__table_id, __tsid, greptime_metric)) ENGINE=mito WITH ('append_mode'='{append_mode}', 'primary_key_encoding'='{encoding}', 'compaction.type'='twcs')" )).await; let (request, _) = into_rows(vec![Exemplar { metric: "metric_source".into(), @@ -4795,15 +4800,20 @@ async fn test_exemplar_precreated_table_options( let error = result.unwrap_err(); assert_eq!(error.status_code(), StatusCode::InvalidArguments); assert!( - error.output_msg().contains( - "metrics_exemplars requires Mito, append_mode=true and primary_key_encoding=sparse" - ), + error.output_msg().contains(&format!( + "{METRICS_EXEMPLARS_TABLE} requires Mito, append_mode=true and primary_key_encoding=sparse" + )), "{error:?}" ); } let table = frontend .catalog_manager() - .table("greptime", "public", "metrics_exemplars", None) + .table( + DEFAULT_CATALOG_NAME, + "public", + METRICS_EXEMPLARS_TABLE, + None, + ) .await .unwrap() .unwrap(); @@ -4815,10 +4825,10 @@ async fn test_exemplar_precreated_table_options( let sql = if accepted { let timestamp = 1_234_567_890 / 10_i64.pow(9 - precision); format!( - "SELECT count(*) FROM metrics_exemplars WHERE job = 'api' AND greptime_value = 2.0 AND CAST(greptime_timestamp AS BIGINT) = {timestamp}" + "SELECT count(*) FROM {METRICS_EXEMPLARS_TABLE} WHERE job = 'api' AND greptime_value = 2.0 AND CAST(greptime_timestamp AS BIGINT) = {timestamp}" ) } else { - "SELECT count(*) FROM metrics_exemplars".into() + format!("SELECT count(*) FROM {METRICS_EXEMPLARS_TABLE}") }; let output = execute_sql(frontend, &sql).await; check_output_stream( @@ -4875,7 +4885,12 @@ async fn test_exemplar_source_identity_policy() { assert!( frontend .catalog_manager() - .table("greptime", "public", "metrics_exemplars", None) + .table( + DEFAULT_CATALOG_NAME, + "public", + METRICS_EXEMPLARS_TABLE, + None + ) .await .unwrap() .is_none() @@ -4885,7 +4900,7 @@ async fn test_exemplar_source_identity_policy() { execute_sql(frontend, create).await; let source = frontend .catalog_manager() - .table("greptime", "public", "metric_source", None) + .table(DEFAULT_CATALOG_NAME, "public", "metric_source", None) .await .unwrap() .unwrap() @@ -4895,7 +4910,7 @@ async fn test_exemplar_source_identity_policy() { execute_sql(frontend, create).await; let replacement = frontend .catalog_manager() - .table("greptime", "public", "metric_source", None) + .table(DEFAULT_CATALOG_NAME, "public", "metric_source", None) .await .unwrap() .unwrap() @@ -4915,7 +4930,12 @@ async fn test_exemplar_source_identity_policy() { .unwrap(); let table = frontend .catalog_manager() - .table("greptime", "public", "metrics_exemplars", None) + .table( + DEFAULT_CATALOG_NAME, + "public", + METRICS_EXEMPLARS_TABLE, + None, + ) .await .unwrap() .unwrap(); @@ -4940,7 +4960,7 @@ async fn test_exemplar_source_identity_policy() { assert_eq!(info.meta.primary_key_indices, [0, 1, 2, 3]); let output = execute_sql( frontend, - "SELECT __table_id FROM metrics_exemplars ORDER BY __table_id", + &format!("SELECT __table_id FROM {METRICS_EXEMPLARS_TABLE} ORDER BY __table_id"), ) .await; let OutputData::Stream(stream) = output.data else { diff --git a/tests-integration/tests/grpc.rs b/tests-integration/tests/grpc.rs index 818a940d76b..e7a718f3e3c 100644 --- a/tests-integration/tests/grpc.rs +++ b/tests-integration/tests/grpc.rs @@ -35,7 +35,7 @@ use auth::user_provider_from_option; use base64::prelude::{BASE64_STANDARD, Engine as _}; use catalog::process_manager::{ProcessManager, ProcessManagerRef}; use client::{Client, DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, Database, OutputData}; -use common_catalog::consts::{DEFAULT_PRIVATE_SCHEMA_NAME, MITO_ENGINE}; +use common_catalog::consts::{DEFAULT_PRIVATE_SCHEMA_NAME, METRICS_EXEMPLARS_TABLE, MITO_ENGINE}; use common_event_recorder::DEFAULT_EVENTS_TABLE_NAME; use common_frontend::slow_query_event::SLOW_QUERY_TABLE_NAME; use common_grpc::channel_manager::ClientTlsOption; @@ -133,6 +133,7 @@ macro_rules! grpc_tests { test_otel_arrow_auth, test_otel_arrow_delta_histogram, test_otel_arrow_exponential_histogram, + test_otel_arrow_exemplars_without_metric_engine, test_insert_and_select, test_dbname, test_grpc_message_size_ok, @@ -1241,6 +1242,103 @@ pub async fn test_otel_arrow_delta_histogram(store_type: StorageType) { let _ = server.shutdown().await; } +pub async fn test_otel_arrow_exemplars_without_metric_engine(store_type: StorageType) { + use datatypes::arrow::array::BinaryArray; + use otel_arrow_rust::Consumer; + use otel_arrow_rust::proto::opentelemetry::metrics::v1::metric; + + let (_instance, server) = setup_grpc_server( + store_type, + "test_otel_arrow_exemplars_without_metric_engine", + ) + .await; + let addr = server.bind_addr().unwrap().to_string(); + let db = Database::new_with_dbname( + format!("{DEFAULT_CATALOG_NAME}-{DEFAULT_SCHEMA_NAME}"), + Client::with_urls(vec![addr.clone()]), + ); + let mut client = ArrowMetricsServiceClient::connect(format!("http://{addr}")) + .await + .unwrap(); + let mut batch = gauge_arrow_batch(20, false); + let exemplars = ArrowRecordBatch::try_from_iter(vec![ + ( + arrow_consts::PARENT_ID, + Arc::new(UInt32Array::from(vec![0_u32])) as ArrayRef, + ), + ( + arrow_consts::TIME_UNIX_NANO, + Arc::new(TimestampNanosecondArray::from(vec![2_000_000_000])) as ArrayRef, + ), + ( + arrow_consts::DOUBLE_VALUE, + Arc::new(Float64Array::from(vec![0.5])) as ArrayRef, + ), + ( + arrow_consts::SPAN_ID, + Arc::new(BinaryArray::from_vec(vec![&[1_u8; 8]])) as ArrayRef, + ), + ( + arrow_consts::TRACE_ID, + Arc::new(BinaryArray::from_vec(vec![&[2_u8; 16]])) as ArrayRef, + ), + ]) + .unwrap(); + batch.arrow_payloads.push(ArrowPayload { + schema_id: "exemplars-20".into(), + r#type: ArrowPayloadType::NumberDpExemplars as i32, + record: serialize_arrow_record_batch(&exemplars), + }); + let decoded = Consumer::default() + .consume_metrics_batches(&mut batch.clone()) + .unwrap(); + let Some(metric::Data::Gauge(gauge)) = + &decoded.resource_metrics[0].scope_metrics[0].metrics[0].data + else { + panic!("expected a gauge") + }; + assert_eq!(gauge.data_points[0].exemplars.len(), 1); + + let request = Request::new(futures::stream::iter([ + batch, + delta_histogram_arrow_batch(21, &[(3, &[1, 2], &[1.0])]), + ])); + let mut response = client.arrow_metrics(request).await.unwrap().into_inner(); + for batch_id in [20, 21] { + let status = response.message().await.unwrap().unwrap(); + assert_eq!(status.batch_id, batch_id); + assert_eq!(status.status_code, ArrowStatusCode::Ok as i32); + assert!(status.status_message.is_empty()); + } + assert!(response.message().await.unwrap().is_none()); + for (sql, count) in [ + ("SELECT COUNT(*) AS c FROM otel_arrow_gauge", 1), + ( + "SELECT COUNT(*) AS c FROM otel_arrow_delta_histogram_bucket", + 2, + ), + ( + &format!( + "SELECT COUNT(*) AS c FROM information_schema.tables WHERE table_name='{METRICS_EXEMPLARS_TABLE}'" + ), + 0, + ), + ] { + let OutputData::Stream(stream) = db.sql(sql).await.unwrap().data else { + panic!("expected a stream") + }; + assert_eq!( + RecordBatches::try_collect(stream) + .await + .unwrap() + .pretty_print() + .unwrap(), + format!("+---+\n| c |\n+---+\n| {count} |\n+---+"), + ); + } + server.shutdown().await.unwrap(); +} + 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;