diff --git a/src/servers/benches/otlp_metrics.rs b/src/servers/benches/otlp_metrics.rs index eccf7f4f167..130537696f9 100644 --- a/src/servers/benches/otlp_metrics.rs +++ b/src/servers/benches/otlp_metrics.rs @@ -12,16 +12,242 @@ // See the License for the specific language governing permissions and // limitations under the License. +//! Measures OTLP request-to-row conversion, excluding request cloning and storage. +//! Save a baseline with `cargo bench -p servers --bench otlp_metrics -- --save-baseline before`, +//! then compare with `cargo bench -p servers --bench otlp_metrics -- --baseline before`. + #[path = "otlp_metrics/fixtures.rs"] mod fixtures; use std::hint::black_box; use std::time::Duration; +use api::v1::value::ValueData; +use common_catalog::consts::SEMANTIC_GRAPH_WINDOW_NANOS; +use common_query::prelude::{greptime_timestamp, set_default_prefix}; use criterion::{BatchSize, BenchmarkId, Criterion, Throughput, criterion_group, criterion_main}; +use otel_arrow_rust::proto::opentelemetry::collector::metrics::v1::ExportMetricsServiceRequest; +use otel_arrow_rust::proto::opentelemetry::common::v1::{AnyValue, KeyValue, any_value}; +use otel_arrow_rust::proto::opentelemetry::metrics::v1::{ + AggregationTemporality, ExponentialHistogram, ExponentialHistogramDataPoint, Gauge, Histogram, + HistogramDataPoint, Metric, NumberDataPoint, ResourceMetrics, ScopeMetrics, + exponential_histogram_data_point::Buckets, metric, number_data_point, +}; +use otel_arrow_rust::proto::opentelemetry::resource::v1::Resource; use servers::otlp::metrics::to_grpc_insert_requests; use session::protocol_ctx::OtlpMetricCtx; +const POINTS: usize = 128; + +fn attribute(key: &str, value: &str) -> KeyValue { + KeyValue { + key: key.to_string(), + value: Some(AnyValue { + value: Some(any_value::Value::StringValue(value.to_string())), + }), + } +} + +fn timestamp(index: usize) -> u64 { + ((index % 4 + 1) as i64 * SEMANTIC_GRAPH_WINDOW_NANOS) as u64 + index as u64 * 1_000_000 +} + +fn point_attributes() -> Vec { + vec![ + attribute("http.request.method", "GET"), + attribute("route", "/api"), + ] +} + +fn request(data: metric::Data) -> ExportMetricsServiceRequest { + ExportMetricsServiceRequest { + resource_metrics: vec![ResourceMetrics { + resource: Some(Resource { + attributes: vec![ + attribute("service.name", "api"), + attribute("service.namespace", "shop"), + attribute("service.instance.id", "instance-1"), + attribute("host.id", "host-1"), + attribute("k8s.pod.name", "api-1"), + ], + ..Default::default() + }), + scope_metrics: vec![ScopeMetrics { + metrics: vec![Metric { + name: "request_duration".to_string(), + data: Some(data), + ..Default::default() + }], + ..Default::default() + }], + ..Default::default() + }], + } +} + +fn exponential(bucket_counts: Vec, scale: i32, reject: bool) -> metric::Data { + let count = 2 + 2 * bucket_counts.iter().sum::(); + metric::Data::ExponentialHistogram(ExponentialHistogram { + aggregation_temporality: AggregationTemporality::Cumulative as i32, + data_points: (0..POINTS) + .map(|index| ExponentialHistogramDataPoint { + attributes: point_attributes(), + start_time_unix_nano: 1, + time_unix_nano: timestamp(index), + count: count + u64::from(reject && index % 4 == 3), + sum: Some(100.0), + scale, + zero_count: 2, + positive: Some(Buckets { + offset: -128, + bucket_counts: bucket_counts.clone(), + }), + negative: Some(Buckets { + offset: -128, + bucket_counts: bucket_counts.clone(), + }), + zero_threshold: 1e-6, + ..Default::default() + }) + .collect(), + }) +} + +fn bench_otlp_metrics(c: &mut Criterion) { + set_default_prefix(None).unwrap(); + let gauge = metric::Data::Gauge(Gauge { + data_points: (0..POINTS) + .map(|index| NumberDataPoint { + attributes: point_attributes(), + time_unix_nano: timestamp(index), + value: Some(number_data_point::Value::AsDouble(1.0)), + ..Default::default() + }) + .collect(), + }); + let histogram = metric::Data::Histogram(Histogram { + aggregation_temporality: AggregationTemporality::Delta as i32, + data_points: (0..POINTS) + .map(|index| HistogramDataPoint { + attributes: point_attributes(), + start_time_unix_nano: 1, + time_unix_nano: timestamp(index), + count: 17, + sum: Some(100.0), + bucket_counts: vec![1; 17], + explicit_bounds: (0..16).map(|index| index as f64).collect(), + ..Default::default() + }) + .collect(), + }); + let workloads = [ + ("gauge", gauge, POINTS, POINTS), + ("classic_16", histogram, POINTS, POINTS * 19), + ( + "exponential_16", + exponential(vec![1; 16], 0, false), + POINTS, + POINTS, + ), + ( + "exponential_256", + exponential(vec![1; 256], 0, false), + POINTS, + POINTS, + ), + ( + "exponential_sparse_256", + exponential((0..256).map(|i| u64::from(i % 16 == 0)).collect(), 0, false), + POINTS, + POINTS, + ), + ( + "exponential_downscale_256", + exponential(vec![1; 256], 10, false), + POINTS, + POINTS, + ), + ( + "exponential_reject_quarter_256", + exponential(vec![1; 256], 0, true), + POINTS * 3 / 4, + POINTS * 3 / 4, + ), + ]; + + let mut group = c.benchmark_group("otlp_metrics"); + group.sample_size(30); + group.warm_up_time(Duration::from_secs(1)); + group.measurement_time(Duration::from_secs(3)); + group.throughput(Throughput::Elements(POINTS as u64)); + for (name, data, accepted, rows) in workloads { + let request = request(data); + for resource_info in [false, true] { + let ctx = OtlpMetricCtx { + resource_info, + ..Default::default() + }; + let conversion = to_grpc_insert_requests(request.clone(), &mut ctx.clone()).unwrap(); + assert_eq!( + conversion.outcome.accepted_data_points, accepted as i64, + "{name}" + ); + assert_eq!( + conversion.outcome.rejected_data_points, + (POINTS - accepted) as i64, + "{name}" + ); + assert_eq!(conversion.rows, rows, "{name}"); + assert_eq!(conversion.resource_info.is_some(), resource_info, "{name}"); + if let Some(descriptor) = conversion.resource_info { + let descriptor_rows = descriptor.inserts[0].rows.as_ref().unwrap(); + let ts_index = descriptor_rows + .schema + .iter() + .position(|column| column.column_name == greptime_timestamp()) + .unwrap(); + let expected = (0..4) + .filter(|window| accepted == POINTS || *window != 3) + .map(|window| { + Some(ValueData::TimestampMillisecondValue( + (timestamp(POINTS - 4 + window) / 1_000_000) as i64, + )) + }) + .collect::>(); + let actual = descriptor_rows + .rows + .iter() + .map(|row| row.values[ts_index].value_data.clone()) + .collect::>(); + assert_eq!(actual, expected, "{name}"); + } + group.bench_with_input( + BenchmarkId::new( + name, + if resource_info { + "descriptor_on" + } else { + "descriptor_off" + }, + ), + &request, + |b, request| { + b.iter_batched( + || (request.clone(), ctx.clone()), + |(request, mut ctx)| { + black_box( + to_grpc_insert_requests(black_box(request), &mut ctx).unwrap(), + ) + }, + BatchSize::LargeInput, + ); + }, + ); + } + } + group.finish(); +} + fn bench_conversion(c: &mut Criterion) { let fixtures = fixtures::fixtures(); for (group_name, delta) in [ @@ -79,6 +305,6 @@ fn bench_conversion(c: &mut Criterion) { criterion_group! { name = benches; config = Criterion::default().sample_size(20).warm_up_time(Duration::from_secs(1)).measurement_time(Duration::from_secs(3)); - targets = bench_conversion + targets = bench_conversion, bench_otlp_metrics } criterion_main!(benches); diff --git a/src/servers/src/otlp/metrics.rs b/src/servers/src/otlp/metrics.rs index 27462512e64..c1226d6ed6f 100644 --- a/src/servers/src/otlp/metrics.rs +++ b/src/servers/src/otlp/metrics.rs @@ -1591,6 +1591,95 @@ mod tests { } } + #[test] + fn test_conversion_describes_only_accepted_exponential_histogram_windows() { + set_default_prefix(None).unwrap(); + let window = common_catalog::consts::SEMANTIC_GRAPH_WINDOW_NANOS as u64; + let mut valid = exponential_point(); + valid.time_unix_nano = window + 1_000_000; + + let mut invalid_count = valid.clone(); + invalid_count.time_unix_nano += 8_000_000; + invalid_count.count += 1; + let mut invalid_scale = valid.clone(); + invalid_scale.time_unix_nano += window; + invalid_scale.scale = -5; + // OTLP bucket conversion succeeds, but the shared native encoder rejects this index. + let invalid_native = ExponentialHistogramDataPoint { + time_unix_nano: 3 * window + 1_000_000, + count: 1, + positive: Some(exponential_buckets(2000, vec![1])), + ..Default::default() + }; + assert!( + exponential_histogram_value(&invalid_native, AggregationTemporality::Cumulative) + .unwrap_err() + .contains("out of range") + ); + let stale = ExponentialHistogramDataPoint { + time_unix_nano: 4 * window + 1_000_000, + count: u64::MAX, + scale: i32::MIN, + flags: DataPointFlags::NoRecordedValueMask as u32, + ..Default::default() + }; + let invalid_time = ExponentialHistogramDataPoint { + time_unix_nano: 5 * window + 1_000_000, + start_time_unix_nano: 5 * window + 2_000_000, + ..stale.clone() + }; + let rejected = vec![invalid_count, invalid_scale, invalid_native, invalid_time]; + let mut mixed = rejected.clone(); + mixed.extend([valid, stale]); + for (points, expected) in [ + (rejected, vec![]), + ( + mixed, + vec![ + (window / 1_000_000 + 1) as i64, + (4 * window / 1_000_000 + 1) as i64, + ], + ), + ] { + let mut request = metrics_request(vec![exponential_metric( + "latency", + points, + AggregationTemporality::Cumulative, + )]); + request.resource_metrics[0].resource = Some(Resource { + attributes: vec![keyvalue("service.name", "api")], + ..Default::default() + }); + let conversion = to_grpc_insert_requests(request, &mut descriptor_ctx()).unwrap(); + assert_eq!(conversion.outcome.rejected_data_points, 4); + assert_eq!( + conversion.outcome.accepted_data_points, + expected.len() as i64 + ); + assert_eq!(conversion.rows, expected.len()); + assert_eq!(conversion.resource_info.is_some(), !expected.is_empty()); + let actual = conversion + .resource_info + .map(|descriptor| { + let rows = descriptor.inserts[0].rows.as_ref().unwrap(); + let ts_index = rows + .schema + .iter() + .position(|column| column.column_name == greptime_timestamp()) + .unwrap(); + rows.rows + .iter() + .map(|row| match row.values[ts_index].value_data { + Some(ValueData::TimestampMillisecondValue(ts)) => ts, + _ => panic!("expected descriptor timestamp"), + }) + .collect::>() + }) + .unwrap_or_default(); + assert_eq!(actual, expected); + } + } + #[test] fn test_conversion_skips_descriptor_on_metric_name_collision() { set_default_prefix(None).unwrap(); diff --git a/src/servers/src/otlp/metrics/resource_info.rs b/src/servers/src/otlp/metrics/resource_info.rs index c7ad72c3e47..e9c88cf4bc5 100644 --- a/src/servers/src/otlp/metrics/resource_info.rs +++ b/src/servers/src/otlp/metrics/resource_info.rs @@ -217,6 +217,15 @@ mod tests { .all(|(k, _)| k != "os.type" && k != "service.instance.id") ); + let mut reordered = attrs.clone(); + reordered.reverse(); + data.observe_times(&reordered, [200, 75]); + assert_eq!(data.rows.len(), 1); + assert_eq!( + data.rows.values().next().unwrap().values().next(), + Some(&200) + ); + data.observe_times(&[kv("host.id", "h-2")], [10]); assert_eq!(data.rows.len(), 2); @@ -225,6 +234,14 @@ mod tests { assert!(empty.into_row_insert_requests().unwrap().is_none()); } + #[test] + fn observe_without_accepted_points_does_not_emit_rows() { + let mut data = ResourceInfoData::default(); + data.observe_times(&[kv("service.name", "api")], []); + assert!(data.is_empty()); + assert!(data.into_row_insert_requests().unwrap().is_none()); + } + /// Earlier windows would keep their metric rows but lose their entities. #[test] fn observe_keeps_one_row_per_graph_window() {