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;