diff --git a/src/mito-codec/src/row_converter/sparse.rs b/src/mito-codec/src/row_converter/sparse.rs index c70293886f8..df26dfb9709 100644 --- a/src/mito-codec/src/row_converter/sparse.rs +++ b/src/mito-codec/src/row_converter/sparse.rs @@ -301,13 +301,33 @@ impl SparsePrimaryKeyCodec { Ok(()) } + /// Exact length of a sparse metric key, including table id and TSID. + pub fn encoded_raw_key_len<'a>( + tag_values: impl IntoIterator, + ) -> Result { + tag_values.into_iter().try_fold(22usize, |length, value| { + value + .len() + .div_ceil(8) + .checked_mul(9) + .and_then(|encoded| encoded.checked_add(6)) + .and_then(|encoded| length.checked_add(encoded)) + .ok_or_else(|| { + InvalidSparsePrimaryKeySnafu { + reason: "encoded primary key length overflows usize", + } + .build() + }) + }) + } + pub fn encode_raw_tag_value<'a, I>(&self, row: I, buffer: &mut Vec) -> Result<()> where I: Iterator, { for (tag_column_id, tag_value) in row { let value_len = tag_value.len(); - buffer.reserve(6 + value_len / 8 * 9); + buffer.reserve(6 + value_len.div_ceil(8) * 9); buffer.put_u32(tag_column_id); buffer.put_u8(1); buffer.put_u8(!tag_value.is_empty() as u8); @@ -630,6 +650,38 @@ mod tests { use super::*; + #[test] + fn raw_key_length_matches_encoding_at_chunk_boundaries() { + let codec = SparsePrimaryKeyCodec::schemaless(); + for lengths in [ + vec![], + vec![0], + vec![1, 7, 8, 9], + vec![15, 16, 17], + vec![0, 128, 257], + ] { + let values: Vec<_> = lengths + .into_iter() + .map(|length| vec![b'x'; length]) + .collect(); + let length = + SparsePrimaryKeyCodec::encoded_raw_key_len(values.iter().map(Vec::as_slice)) + .unwrap(); + let mut encoded = Vec::with_capacity(length); + codec.encode_internal(42, 99, &mut encoded).unwrap(); + codec + .encode_raw_tag_value( + values + .iter() + .enumerate() + .map(|(index, value)| (index as u32 + 10, value.as_slice())), + &mut encoded, + ) + .unwrap(); + assert_eq!(length, encoded.len()); + } + } + fn test_region_metadata() -> RegionMetadataRef { let mut builder = RegionMetadataBuilder::new(RegionId::new(1, 1)); builder diff --git a/src/servers/Cargo.toml b/src/servers/Cargo.toml index 509699d8ba2..fea9a00586a 100644 --- a/src/servers/Cargo.toml +++ b/src/servers/Cargo.toml @@ -181,6 +181,10 @@ common-version.workspace = true name = "bench_prom" harness = false +[[bench]] +name = "bench_otlp_decode" +harness = false + [[bench]] name = "prom_decode" harness = false diff --git a/src/servers/benches/bench_otlp_decode.rs b/src/servers/benches/bench_otlp_decode.rs new file mode 100644 index 00000000000..c0d8838693b --- /dev/null +++ b/src/servers/benches/bench_otlp_decode.rs @@ -0,0 +1,153 @@ +// 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::hint::black_box; +use std::path::PathBuf; +use std::time::Duration; + +use bytes::Bytes; +use criterion::{Criterion, Throughput, criterion_group, criterion_main}; +use otel_arrow_rust::proto::opentelemetry::collector::metrics::v1::ExportMetricsServiceRequest; +use otel_arrow_rust::proto::opentelemetry::metrics::v1::{ + Gauge, Metric, NumberDataPoint, ResourceMetrics, ScopeMetrics, metric, +}; +use prost::Message; + +#[path = "otlp_decode/capacity_hint.rs"] +mod decode; + +fn decode_metrics(c: &mut Criterion) { + let directory = std::env::var_os("OTLP_DECODE_CORPUS") + .expect("set OTLP_DECODE_CORPUS to uncompressed sample_loader OTLP .pb requests"); + let mut paths: Vec = std::fs::read_dir(directory) + .unwrap() + .map(|entry| entry.unwrap().path()) + .filter(|path| path.extension().is_some_and(|extension| extension == "pb")) + .collect(); + paths.sort(); + let corpus: Vec = paths + .into_iter() + .map(|path| std::fs::read(path).unwrap().into()) + .collect(); + assert!(!corpus.is_empty()); + for payload in &corpus { + assert_eq!( + ExportMetricsServiceRequest::decode(payload.clone()).unwrap(), + ExportMetricsServiceRequest::decode(payload.as_ref()).unwrap() + ); + assert_eq!( + ExportMetricsServiceRequest::decode(payload.clone()).unwrap(), + decode::decode_metrics(payload.clone()).unwrap() + ); + } + if let Some(report) = std::env::var_os("OTLP_DECODE_CAPACITY_REPORT") { + let standard: usize = corpus + .iter() + .map(|payload| { + attribute_capacity(&ExportMetricsServiceRequest::decode(payload.clone()).unwrap()) + }) + .sum(); + let hinted: usize = corpus + .iter() + .map(|payload| attribute_capacity(&decode::decode_metrics(payload.clone()).unwrap())) + .sum(); + std::fs::write(report, format!("standard_attribute_capacity_bytes={standard}\nhinted_attribute_capacity_bytes={hinted}\n")).unwrap(); + } + let mut group = c.benchmark_group("otlp_metrics_decode"); + group.sample_size(50); + group.warm_up_time(Duration::from_secs(2)); + group.measurement_time(Duration::from_secs(5)); + group.throughput(Throughput::Bytes( + corpus.iter().map(|payload| payload.len() as u64).sum(), + )); + group.bench_function("bytes", |b| { + b.iter(|| { + for payload in &corpus { + black_box(ExportMetricsServiceRequest::decode(black_box(payload.clone())).unwrap()); + } + }) + }); + group.bench_function("slice", |b| { + b.iter(|| { + for payload in &corpus { + black_box( + ExportMetricsServiceRequest::decode(black_box(payload.as_ref())).unwrap(), + ); + } + }) + }); + group.bench_function("capacity_hint", |b| { + b.iter(|| { + for payload in &corpus { + black_box(decode::decode_metrics(black_box(payload.clone())).unwrap()); + } + }) + }); + group.finish(); + + let small = Bytes::from( + ExportMetricsServiceRequest { + resource_metrics: vec![ResourceMetrics { + scope_metrics: vec![ScopeMetrics { + metrics: vec![Metric { + data: Some(metric::Data::Gauge(Gauge { + data_points: vec![NumberDataPoint::default()], + })), + ..Default::default() + }], + ..Default::default() + }], + ..Default::default() + }], + } + .encode_to_vec(), + ); + let mut group = c.benchmark_group("otlp_small_decode"); + group.measurement_time(Duration::from_secs(3)); + group.bench_function("bytes", |b| { + b.iter(|| black_box(ExportMetricsServiceRequest::decode(black_box(small.clone())).unwrap())) + }); + group.bench_function("capacity_hint", |b| { + b.iter(|| black_box(decode::decode_metrics(black_box(small.clone())).unwrap())) + }); + group.finish(); +} + +fn attribute_capacity(request: &ExportMetricsServiceRequest) -> usize { + request + .resource_metrics + .iter() + .flat_map(|resource| &resource.scope_metrics) + .flat_map(|scope| &scope.metrics) + .map(|metric| { + let points = match &metric.data { + Some(metric::Data::Gauge(gauge)) => &gauge.data_points, + Some(metric::Data::Sum(sum)) => &sum.data_points, + _ => return 0, + }; + points + .iter() + .map(|point| { + point.attributes.capacity() + * std::mem::size_of::< + otel_arrow_rust::proto::opentelemetry::common::v1::KeyValue, + >() + }) + .sum::() + }) + .sum() +} + +criterion_group!(benches, decode_metrics); +criterion_main!(benches); diff --git a/src/servers/benches/otlp_decode/README.md b/src/servers/benches/otlp_decode/README.md new file mode 100644 index 00000000000..c3ffef1b838 --- /dev/null +++ b/src/servers/benches/otlp_decode/README.md @@ -0,0 +1,55 @@ +# OTLP metrics decode experiment + +This experiment uses uncompressed protobuf requests generated by +`metrics-bench-suite/bin/sample_loader` with `configs/samples_1750`, +`--protocol otlp`, `--replica 0`, and `--max-samples 1000`. +Save HTTP request bodies as `.pb` files, then run: + +```sh +OTLP_DECODE_CORPUS=/path/to/corpus \ +OTLP_DECODE_CAPACITY_REPORT=/path/to/capacity.txt \ +cargo bench -p servers --bench bench_otlp_decode +``` + +The harness checks decoded-message equality before timing. Each iteration +decodes and drops the entire corpus; it does not append successive requests +into an uncleared message. Throughput is input protobuf bytes, not database +ingestion throughput. The capacity report measures attribute-vector backing +storage, excluding strings and the rest of the message; it is not process RSS. + +## Candidates + +- **bytes:** current prost decoder over `Bytes`. +- **slice:** prost decoder over `&[u8]`. +- **capacity_hint:** bounded preallocation of scalar data-point attributes. + This follows the enterprise Remote Write decoder's focus on reducing + repeated-container allocations. It still delegates wire validation, + recursion handling and unknown fields to prost. It is benchmark-only. + +## Local results (2026-09-10) + +Release-mode isolated harness, the actual pinned `otel-arrow-rust` message +types, prost/prost-build 0.14.1, bytes 1.11.1; 12 suite requests, each containing +1,000 input samples. The isolated harness builds only benchmark dependencies, +not a release database binary. Criterion: 50 samples, 2 seconds warm-up, +5 seconds measurement for the real corpus. + +| Decoder | Corpus time, final run | Throughput | +| --- | ---: | ---: | +| bytes | 17.568 ms | 334.79 MiB/s | +| slice | 18.596 ms | 316.30 MiB/s | +| capacity hint (32) | 17.437 ms | 337.32 MiB/s | + +An earlier repeat measured 17.637 ms vs 17.327 ms for bytes vs capacity hint. +The observed improvement therefore ranged from approximately 0.7% to 1.8%. +A capacity hint of 16 measured 17.775 ms vs a 17.635 ms baseline. + +The 32-entry hint increased aggregate attribute-vector capacity from +15,980,160 to 21,414,848 bytes (+34%). On a single empty scalar data point, +baseline decoding measured 87.130 ns vs 89.584 ns with the hint (+2.8%). + +**Conclusion:** retain the current production decoder. The small, variable +improvement does not justify additional decoding wrappers and memory overhead. +Keep the candidate here so future alternatives can be compared with the same +corpus. These results do not establish an end-to-end ingestion speedup: the +standalone smoke test's default 3-second batching interval dominates latency. diff --git a/src/servers/benches/otlp_decode/capacity_hint.rs b/src/servers/benches/otlp_decode/capacity_hint.rs new file mode 100644 index 00000000000..258811b8598 --- /dev/null +++ b/src/servers/benches/otlp_decode/capacity_hint.rs @@ -0,0 +1,166 @@ +// 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 bytes::{Buf, BufMut, Bytes}; +use otel_arrow_rust::proto::opentelemetry::collector::metrics::v1::ExportMetricsServiceRequest; +use otel_arrow_rust::proto::opentelemetry::metrics::v1::{ + Gauge, Metric, NumberDataPoint, ResourceMetrics, ScopeMetrics, Sum, metric, +}; +use prost::encoding::{DecodeContext, WireType, decode_varint, message}; +use prost::{DecodeError, Message}; + +/// Decode scalar metrics with bounded attribute capacity hints. Prost still +/// handles field validation, unknown fields, oneofs and recursive messages. +pub fn decode_metrics(bytes: Bytes) -> Result { + RequestDecoder::decode(bytes).map(|request| request.0) +} + +macro_rules! decoder { + ($name:ident, $inner:ty, |$this:ident, $tag:ident, $wire:ident, $buf:ident, $ctx:ident| $merge:block) => { + #[derive(Debug, Default)] + struct $name($inner); + + impl Message for $name { + fn encode_raw(&self, buf: &mut impl BufMut) { + self.0.encode_raw(buf); + } + + fn merge_field( + &mut self, + $tag: u32, + $wire: WireType, + $buf: &mut impl Buf, + $ctx: DecodeContext, + ) -> Result<(), DecodeError> { + let $this = self; + $merge + } + + fn encoded_len(&self) -> usize { + self.0.encoded_len() + } + + fn clear(&mut self) { + self.0.clear(); + } + } + }; +} + +decoder!( + RequestDecoder, + ExportMetricsServiceRequest, + |this, tag, wire, buf, ctx| { + if tag == 1 { + let mut resource = ResourceDecoder::default(); + message::merge(wire, &mut resource, buf, ctx)?; + this.0.resource_metrics.push(resource.0); + Ok(()) + } else { + this.0.merge_field(tag, wire, buf, ctx) + } + } +); + +decoder!( + ResourceDecoder, + ResourceMetrics, + |this, tag, wire, buf, ctx| { + if tag == 2 { + let mut scope = ScopeDecoder::default(); + message::merge(wire, &mut scope, buf, ctx)?; + this.0.scope_metrics.push(scope.0); + Ok(()) + } else { + this.0.merge_field(tag, wire, buf, ctx) + } + } +); + +decoder!(ScopeDecoder, ScopeMetrics, |this, tag, wire, buf, ctx| { + if tag == 2 { + let mut metric = MetricDecoder::default(); + message::merge(wire, &mut metric, buf, ctx)?; + this.0.metrics.push(metric.0); + Ok(()) + } else { + this.0.merge_field(tag, wire, buf, ctx) + } +}); + +decoder!(MetricDecoder, Metric, |this, tag, wire, buf, ctx| { + match tag { + 5 => { + let mut gauge = GaugeDecoder(match this.0.data.take() { + Some(metric::Data::Gauge(gauge)) => gauge, + _ => Gauge::default(), + }); + message::merge(wire, &mut gauge, buf, ctx)?; + this.0.data = Some(metric::Data::Gauge(gauge.0)); + Ok(()) + } + 7 => { + let mut sum = SumDecoder(match this.0.data.take() { + Some(metric::Data::Sum(sum)) => sum, + _ => Sum::default(), + }); + message::merge(wire, &mut sum, buf, ctx)?; + this.0.data = Some(metric::Data::Sum(sum.0)); + Ok(()) + } + _ => this.0.merge_field(tag, wire, buf, ctx), + } +}); + +decoder!(GaugeDecoder, Gauge, |this, tag, wire, buf, ctx| { + if tag == 1 { + this.0 + .data_points + .push(decode_number_point(wire, buf, ctx)?); + Ok(()) + } else { + this.0.merge_field(tag, wire, buf, ctx) + } +}); + +decoder!(SumDecoder, Sum, |this, tag, wire, buf, ctx| { + if tag == 1 { + this.0 + .data_points + .push(decode_number_point(wire, buf, ctx)?); + Ok(()) + } else { + this.0.merge_field(tag, wire, buf, ctx) + } +}); + +fn decode_number_point( + wire: WireType, + buf: &mut impl Buf, + ctx: DecodeContext, +) -> Result { + // Large, label-rich samples otherwise repeatedly grow a Vec while decoding. + // The hint is capped independently of the untrusted length; small points + // retain prost's normal allocation behavior. + let capacity = match decode_varint(&mut buf.chunk()) { + Ok(len) if len >= 256 => 32, + _ => 0, + }; + let mut point = NumberDataPoint { + attributes: Vec::with_capacity(capacity), + ..Default::default() + }; + message::merge(wire, &mut point, buf, ctx)?; + Ok(point) +} diff --git a/src/servers/src/otlp/metrics.rs b/src/servers/src/otlp/metrics.rs index 6b7bbd3fab1..190d32485e2 100644 --- a/src/servers/src/otlp/metrics.rs +++ b/src/servers/src/otlp/metrics.rs @@ -12,6 +12,8 @@ // See the License for the specific language governing permissions and // limitations under the License. +use std::borrow::Cow; + use ahash::HashSet; use api::greptime_proto::io::prometheus::write::v2::histogram::{ Count as PromCount, ResetHint, ZeroCount as PromZeroCount, @@ -125,7 +127,7 @@ pub struct MetricsConversion { /// /// for data structure of OTLP metrics. pub fn to_grpc_insert_requests( - request: ExportMetricsServiceRequest, + mut request: ExportMetricsServiceRequest, metric_ctx: &mut OtlpMetricCtx, ) -> Result { let mut table_writer = MultiTableData::default(); @@ -133,7 +135,7 @@ pub fn to_grpc_insert_requests( let mut outcome = MetricsIngestOutcome::default(); let mut resource_info = ResourceInfoData::default(); - for resource in &request.resource_metrics { + for resource in &mut request.resource_metrics { if metric_ctx.resource_info && !metric_ctx.is_legacy && let Some(r) = resource.resource.as_ref() @@ -147,10 +149,10 @@ pub fn to_grpc_insert_requests( attrs }); - for scope in &resource.scope_metrics { + for scope in &mut resource.scope_metrics { let scope_attrs = process_scope_attrs(scope, metric_ctx); - for metric in &scope.metrics { + for metric in &mut scope.metrics { if metric.data.is_none() { continue; } @@ -458,7 +460,7 @@ fn process_scope_attrs(scope: &ScopeMetrics, metric_ctx: &OtlpMetricCtx) -> Opti fn encode_metrics( table_writer: &mut MultiTableData, - metric: &Metric, + metric: &mut Metric, resource_attrs: Option<&Vec>, scope_attrs: Option<&Vec>, metric_ctx: &OtlpMetricCtx, @@ -475,7 +477,7 @@ fn encode_metrics( ) }; - let emitted = if let Some(data) = &metric.data { + let emitted = if let Some(data) = &mut metric.data { match data { metric::Data::Gauge(gauge) => { encode_gauge( @@ -666,7 +668,7 @@ fn encode_exponential_histogram( &mut row, resource_attrs, scope_attrs, - Some(data_point.attributes.as_ref()), + DataPointAttributes::Borrowed(&data_point.attributes), timestamp_nanos, metric_ctx, )?; @@ -922,6 +924,30 @@ enum AttributeType { Legacy, } +fn attribute_key<'a>( + name: &'a str, + attribute_type: AttributeType, + metric_ctx: &OtlpMetricCtx, +) -> Result> { + let key = match attribute_type { + AttributeType::Resource | AttributeType::DataPoint => { + translate_label_name(name, metric_ctx.metric_translation_strategy) + } + AttributeType::Scope => Cow::Owned(format!( + "otel_scope_{}", + translate_label_name(name, metric_ctx.metric_translation_strategy) + )), + AttributeType::Legacy => Cow::Owned(legacy_normalize_otlp_name(name)), + }; + if key == OTLP_AGGREGATION_TEMPORALITY_LABEL { + return Err(error::InvalidOtlpMetricInputSnafu { + reason: format!("OTLP attribute `{name}` resolves to reserved label `{key}`"), + } + .build()); + } + Ok(key) +} + fn write_attributes( writer: &mut TableData, row: &mut Vec, @@ -939,27 +965,7 @@ fn write_attributes( let Some(value) = scalar_value_string(attr.value.as_ref()) else { continue; }; - let key = match attribute_type { - AttributeType::Resource | AttributeType::DataPoint => { - translate_label_name(&attr.key, metric_ctx.metric_translation_strategy) - } - AttributeType::Scope => { - format!( - "otel_scope_{}", - translate_label_name(&attr.key, metric_ctx.metric_translation_strategy) - ) - } - AttributeType::Legacy => legacy_normalize_otlp_name(&attr.key), - }; - if key == OTLP_AGGREGATION_TEMPORALITY_LABEL { - return Err(error::InvalidOtlpMetricInputSnafu { - reason: format!( - "OTLP attribute `{}` resolves to reserved label `{}`", - attr.key, OTLP_AGGREGATION_TEMPORALITY_LABEL - ), - } - .build()); - } + let key = attribute_key(&attr.key, attribute_type, metric_ctx)?; tags.push((key, value)); } row_writer::write_tags(writer, tags.into_iter(), row)?; @@ -967,6 +973,37 @@ fn write_attributes( Ok(()) } +fn write_owned_attributes( + writer: &mut TableData, + row: &mut Vec, + attrs: Vec, + attribute_type: AttributeType, + metric_ctx: &OtlpMetricCtx, +) -> Result<()> { + let mut tags = Vec::with_capacity(attrs.len()); + for attr in attrs { + let value = match attr.value.and_then(|value| value.value) { + Some(any_value::Value::StringValue(value)) => value, + Some(any_value::Value::IntValue(value)) => value.to_string(), + Some(any_value::Value::DoubleValue(value)) => value.to_string(), + _ => continue, + }; + let key = match attribute_key(&attr.key, attribute_type, metric_ctx)? { + Cow::Borrowed(_) => attr.key, + Cow::Owned(key) => key, + }; + tags.push((key, value)); + } + row_writer::write_tags(writer, tags.into_iter(), row) +} + +enum DataPointAttributes<'a> { + // Histograms and summaries fan out the same attributes into multiple rows. + Borrowed(&'a Vec), + // Gauge/sum points emit one row, so their strings can move into that row. + Owned(Vec), +} + fn write_timestamp( table: &mut TableData, row: &mut Vec, @@ -1036,7 +1073,7 @@ fn write_tags_and_timestamp( row: &mut Vec, resource_attrs: Option<&Vec>, scope_attrs: Option<&Vec>, - data_point_attrs: Option<&Vec>, + data_point_attrs: DataPointAttributes<'_>, timestamp_nanos: i64, metric_ctx: &OtlpMetricCtx, ) -> Result<()> { @@ -1049,13 +1086,6 @@ fn write_tags_and_timestamp( metric_ctx, )?; write_attributes(table, row, scope_attrs, AttributeType::Legacy, metric_ctx)?; - write_attributes( - table, - row, - data_point_attrs, - AttributeType::Legacy, - metric_ctx, - )?; } else { // TODO(shuiyisong): check `__type__` and `__unit__` tags in prometheus write_attributes( @@ -1066,13 +1096,20 @@ fn write_tags_and_timestamp( metric_ctx, )?; write_attributes(table, row, scope_attrs, AttributeType::Scope, metric_ctx)?; - write_attributes( - table, - row, - data_point_attrs, - AttributeType::DataPoint, - metric_ctx, - )?; + } + + let attribute_type = if metric_ctx.is_legacy { + AttributeType::Legacy + } else { + AttributeType::DataPoint + }; + match data_point_attrs { + DataPointAttributes::Borrowed(attrs) => { + write_attributes(table, row, Some(attrs), attribute_type, metric_ctx)? + } + DataPointAttributes::Owned(attrs) => { + write_owned_attributes(table, row, attrs, attribute_type, metric_ctx)? + } } write_timestamp(table, row, timestamp_nanos, metric_ctx.is_legacy)?; @@ -1087,7 +1124,7 @@ fn write_tags_and_timestamp( fn encode_gauge( table_writer: &mut MultiTableData, name: &str, - gauge: &Gauge, + gauge: &mut Gauge, resource_attrs: Option<&Vec>, scope_attrs: Option<&Vec>, metric_ctx: &OtlpMetricCtx, @@ -1098,14 +1135,14 @@ fn encode_gauge( gauge.data_points.len(), ); - for data_point in &gauge.data_points { + for data_point in &mut gauge.data_points { let mut row = table.alloc_one_row(); write_tags_and_timestamp( table, &mut row, resource_attrs, scope_attrs, - Some(data_point.attributes.as_ref()), + DataPointAttributes::Owned(std::mem::take(&mut data_point.attributes)), data_point.time_unix_nano as i64, metric_ctx, )?; @@ -1122,7 +1159,7 @@ fn encode_gauge( fn encode_sum( table_writer: &mut MultiTableData, name: &str, - sum: &Sum, + sum: &mut Sum, resource_attrs: Option<&Vec>, scope_attrs: Option<&Vec>, metric_ctx: &OtlpMetricCtx, @@ -1137,14 +1174,14 @@ fn encode_sum( sum.data_points.len(), ); - for data_point in &sum.data_points { + for data_point in &mut sum.data_points { let mut row = table.alloc_one_row(); write_tags_and_timestamp( table, &mut row, resource_attrs, scope_attrs, - Some(data_point.attributes.as_ref()), + DataPointAttributes::Owned(std::mem::take(&mut data_point.attributes)), data_point.time_unix_nano as i64, metric_ctx, )?; @@ -1256,7 +1293,7 @@ fn encode_histogram( &mut bucket_row, resource_attrs, scope_attrs, - Some(data_point.attributes.as_ref()), + DataPointAttributes::Borrowed(&data_point.attributes), data_point.time_unix_nano as i64, metric_ctx, )?; @@ -1279,7 +1316,7 @@ fn encode_histogram( &mut sum_row, resource_attrs, scope_attrs, - Some(data_point.attributes.as_ref()), + DataPointAttributes::Borrowed(&data_point.attributes), data_point.time_unix_nano as i64, metric_ctx, )?; @@ -1304,7 +1341,7 @@ fn encode_histogram( &mut count_row, resource_attrs, scope_attrs, - Some(data_point.attributes.as_ref()), + DataPointAttributes::Borrowed(&data_point.attributes), data_point.time_unix_nano as i64, metric_ctx, )?; @@ -1401,7 +1438,7 @@ fn encode_summary( &mut row, resource_attrs, scope_attrs, - Some(data_point.attributes.as_ref()), + DataPointAttributes::Borrowed(&data_point.attributes), data_point.time_unix_nano as i64, metric_ctx, )?; @@ -1442,7 +1479,7 @@ fn encode_summary( &mut row, resource_attrs, scope_attrs, - Some(data_point.attributes.as_ref()), + DataPointAttributes::Borrowed(&data_point.attributes), data_point.time_unix_nano as i64, metric_ctx, )?; @@ -1468,7 +1505,7 @@ fn encode_summary( &mut row, resource_attrs, scope_attrs, - Some(data_point.attributes.as_ref()), + DataPointAttributes::Borrowed(&data_point.attributes), data_point.time_unix_nano as i64, metric_ctx, )?; @@ -1495,7 +1532,7 @@ fn encode_summary( &mut row, resource_attrs, scope_attrs, - Some(data_point.attributes.as_ref()), + DataPointAttributes::Borrowed(&data_point.attributes), data_point.time_unix_nano as i64, metric_ctx, )?; @@ -1535,6 +1572,101 @@ mod tests { } } + #[test] + fn owned_scalar_attributes_preserve_borrowed_row_semantics() { + use session::protocol_ctx::OtlpMetricTranslationStrategy::*; + + let resource = vec![keyvalue("shared.key", "resource")]; + let scope = vec![keyvalue("shared.key", "scope")]; + let attributes = vec![ + keyvalue("shared.key", "first"), + keyvalue("shared_key", "normalized collision"), + keyvalue("shared.key", "last"), + keyvalue("é汉", "unicode"), + KeyValue { + key: "int".into(), + value: Some(AnyValue { + value: Some(Val::IntValue(-123)), + }), + }, + KeyValue { + key: "double".into(), + value: Some(AnyValue { + value: Some(Val::DoubleValue(2.5)), + }), + }, + KeyValue { + key: "ignored_bool".into(), + value: Some(AnyValue { + value: Some(Val::BoolValue(true)), + }), + }, + KeyValue { + key: "ignored_missing".into(), + value: None, + }, + ]; + for is_legacy in [false, true] { + for strategy in [ + UnderscoreEscapingWithSuffixes, + UnderscoreEscapingWithoutSuffixes, + NoUtf8EscapingWithSuffixes, + NoTranslation, + ] { + let ctx = OtlpMetricCtx { + is_legacy, + metric_translation_strategy: strategy, + ..Default::default() + }; + let mut borrowed = TableData::new(0, 1); + let mut borrowed_row = borrowed.alloc_one_row(); + write_tags_and_timestamp( + &mut borrowed, + &mut borrowed_row, + Some(&resource), + Some(&scope), + DataPointAttributes::Borrowed(&attributes), + 123_000_000, + &ctx, + ) + .unwrap(); + borrowed.add_row(borrowed_row); + let mut owned = TableData::new(0, 1); + let mut owned_row = owned.alloc_one_row(); + write_tags_and_timestamp( + &mut owned, + &mut owned_row, + Some(&resource), + Some(&scope), + DataPointAttributes::Owned(attributes.clone()), + 123_000_000, + &ctx, + ) + .unwrap(); + owned.add_row(owned_row); + assert_eq!( + borrowed.into_schema_and_rows(), + owned.into_schema_and_rows() + ); + } + } + } + + #[test] + fn owned_scalar_attributes_reject_reserved_label() { + let mut table = TableData::new(0, 1); + let mut row = table.alloc_one_row(); + let error = write_owned_attributes( + &mut table, + &mut row, + vec![keyvalue(OTLP_AGGREGATION_TEMPORALITY_LABEL, "reserved")], + AttributeType::DataPoint, + &OtlpMetricCtx::default(), + ) + .unwrap_err(); + assert!(matches!(error, error::Error::InvalidOtlpMetricInput { .. })); + } + fn descriptor_ctx() -> OtlpMetricCtx { OtlpMetricCtx { resource_info: true, @@ -1716,11 +1848,11 @@ mod tests { ..Default::default() }, ]; - let gauge = Gauge { data_points }; + let mut gauge = Gauge { data_points }; encode_gauge( &mut tables, "datamon", - &gauge, + &mut gauge, Some(&vec![]), Some(&vec![keyvalue("scope", "otel")]), &OtlpMetricCtx::default(), @@ -1763,14 +1895,14 @@ mod tests { ..Default::default() }, ]; - let sum = Sum { + let mut sum = Sum { data_points, ..Default::default() }; encode_sum( &mut tables, "datamon", - &sum, + &mut sum, Some(&vec![]), Some(&vec![keyvalue("scope", "otel")]), &OtlpMetricCtx::default(), diff --git a/src/servers/src/otlp/metrics/translator.rs b/src/servers/src/otlp/metrics/translator.rs index 8b3b45d3274..b65c24ef627 100644 --- a/src/servers/src/otlp/metrics/translator.rs +++ b/src/servers/src/otlp/metrics/translator.rs @@ -12,6 +12,8 @@ // See the License for the specific language governing permissions and // limitations under the License. +use std::borrow::Cow; + use ahash::HashMap; use lazy_static::lazy_static; use otel_arrow_rust::proto::opentelemetry::metrics::v1::Metric; @@ -88,11 +90,11 @@ pub fn translate_metric_name( } } -pub fn translate_label_name(name: &str, strategy: OtlpMetricTranslationStrategy) -> String { +pub fn translate_label_name(name: &str, strategy: OtlpMetricTranslationStrategy) -> Cow<'_, str> { if strategy.should_escape() { normalize_label_name(name) } else { - name.to_string() + Cow::Borrowed(name) } } @@ -269,21 +271,34 @@ pub(crate) fn clean_unit_name(name: &str) -> String { } // See https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/145942706622aba5c276ca47f48df438228bfea4/pkg/translator/prometheus/normalize_label.go#L27 -pub fn normalize_label_name(name: &str) -> String { - if name.is_empty() { - return name.to_string(); - } - - let n = NON_ALPHA_NUM_CHAR.replace_all(name, UNDERSCORE); - if let Some((_, first)) = n.char_indices().next() - && first.is_ascii_digit() +pub fn normalize_label_name(name: &str) -> Cow<'_, str> { + // Most attribute names already use the Prometheus ASCII alphabet. Borrow + // them so an existing column does not allocate a fresh name for every row. + let normalized = if name + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || byte == b'_') { - return format!("key_{}", n); + Cow::Borrowed(name) + } else { + // Match the old Unicode regex: each invalid code point, not each UTF-8 + // byte, becomes one underscore. Consecutive replacements stay distinct. + Cow::Owned( + name.chars() + .map(|ch| if ch.is_ascii_alphanumeric() { ch } else { '_' }) + .collect::(), + ) + }; + if normalized + .as_bytes() + .first() + .is_some_and(u8::is_ascii_digit) + { + return Cow::Owned(format!("key_{normalized}")); } - if n.starts_with(UNDERSCORE) && !n.starts_with(DOUBLE_UNDERSCORE) { - return format!("key{}", n); + if normalized.starts_with(UNDERSCORE) && !normalized.starts_with(DOUBLE_UNDERSCORE) { + return Cow::Owned(format!("key{normalized}")); } - n.to_string() + normalized } /// Normalize otlp instrumentation, metric and attribute names @@ -306,6 +321,33 @@ mod tests { use super::*; + #[test] + fn label_scan_matches_regex_normalization() { + fn reference(name: &str) -> String { + let normalized = NON_ALPHA_NUM_CHAR.replace_all(name, UNDERSCORE); + if normalized.starts_with(|ch: char| ch.is_ascii_digit()) { + format!("key_{normalized}") + } else if normalized.starts_with('_') && !normalized.starts_with("__") { + format!("key{normalized}") + } else { + normalized.into_owned() + } + } + // Covers prefix rules and one replacement per Unicode scalar, including + // adjacent invalid characters and already-normalized underscores. + let alphabet = [ + "", "a", "Z", "0", "_", ".", "-", ":", " ", "\n", "\0", "é", "汉", "💡", "\u{0301}", + ]; + for first in alphabet { + for second in alphabet { + for third in alphabet { + let name = format!("{first}{second}{third}"); + assert_eq!(normalize_label_name(&name), reference(&name), "{name:?}"); + } + } + } + } + #[test] fn test_legacy_normalize_otlp_name() { assert_eq!(