From d8a96020d9a45617b0403d65785bf3c0a3104bb3 Mon Sep 17 00:00:00 2001 From: Paul Gillesberger Date: Wed, 26 Aug 2026 19:58:20 +0200 Subject: [PATCH] fix: scope shared aggregation state to the owning stream (#2300) RegisteredView shares one aggregation instance across every stream a wildcard/regex view matches, but ExponentialBucketHistogram keyed its scale/bucket caches (@mappings, @previous_*) and every aggregation's @exemplar_reservoir_storage by attributes alone. Two instruments sharing a view and an attribute set therefore corrupted each other's scale (ratcheting toward MIN_SCALE) and leaked exemplars. Key that state by stream (data_points object_id) as well as attributes. --- .../opentelemetry/sdk/metrics/aggregation.rb | 1 + .../aggregation/explicit_bucket_histogram.rb | 24 +-- .../exponential_bucket_histogram.rb | 162 ++++++++++-------- .../sdk/metrics/aggregation/last_value.rb | 20 ++- .../aggregation/stream_scoped_storage.rb | 26 +++ .../sdk/metrics/aggregation/sum.rb | 26 +-- .../explicit_bucket_histogram_test.rb | 16 ++ .../exponential_bucket_histogram_test.rb | 91 ++++++++++ .../metrics/aggregation/last_value_test.rb | 16 ++ .../sdk/metrics/aggregation/sum_test.rb | 16 ++ .../exemplar/exemplar_integration_test.rb | 26 +-- 11 files changed, 307 insertions(+), 117 deletions(-) create mode 100644 metrics_sdk/lib/opentelemetry/sdk/metrics/aggregation/stream_scoped_storage.rb diff --git a/metrics_sdk/lib/opentelemetry/sdk/metrics/aggregation.rb b/metrics_sdk/lib/opentelemetry/sdk/metrics/aggregation.rb index 1698fe3656..4c03ef15a1 100644 --- a/metrics_sdk/lib/opentelemetry/sdk/metrics/aggregation.rb +++ b/metrics_sdk/lib/opentelemetry/sdk/metrics/aggregation.rb @@ -16,6 +16,7 @@ module Aggregation end require 'opentelemetry/sdk/metrics/aggregation/aggregation_temporality' +require 'opentelemetry/sdk/metrics/aggregation/stream_scoped_storage' require 'opentelemetry/sdk/metrics/aggregation/number_data_point' require 'opentelemetry/sdk/metrics/aggregation/histogram_data_point' require 'opentelemetry/sdk/metrics/aggregation/explicit_bucket_histogram' diff --git a/metrics_sdk/lib/opentelemetry/sdk/metrics/aggregation/explicit_bucket_histogram.rb b/metrics_sdk/lib/opentelemetry/sdk/metrics/aggregation/explicit_bucket_histogram.rb index c76532e29b..de2175f5b0 100644 --- a/metrics_sdk/lib/opentelemetry/sdk/metrics/aggregation/explicit_bucket_histogram.rb +++ b/metrics_sdk/lib/opentelemetry/sdk/metrics/aggregation/explicit_bucket_histogram.rb @@ -11,6 +11,8 @@ module Aggregation # Contains the implementation of the ExplicitBucketHistogram aggregation # https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/metrics/sdk.md#explicit-bucket-histogram-aggregation class ExplicitBucketHistogram # rubocop:disable Metrics/ClassLength + include StreamScopedStorage + OVERFLOW_ATTRIBUTE_SET = { 'otel.metric.overflow' => true }.freeze attr_reader :exemplar_reservoir @@ -31,7 +33,7 @@ def initialize( @boundaries = boundaries && !boundaries.empty? ? boundaries.sort : nil @record_min_max = record_min_max @exemplar_reservoir = exemplar_reservoir || Metrics::Exemplar::AlignedHistogramBucketExemplarReservoir.new(boundaries: @boundaries) - @exemplar_reservoir_storage = {} + @exemplar_reservoir_storage = new_stream_storage end # Returns the current histogram data points, clearing them for delta temporality. @@ -41,8 +43,8 @@ def collect(start_time, end_time, data_points) hdps = data_points.values.map! do |hdp| hdp.start_time_unix_nano = start_time hdp.time_unix_nano = end_time - reservoir = @exemplar_reservoir_storage[hdp.attributes] - hdp.exemplars = reservoir&.collect(attributes: hdp.attributes, aggregation_temporality: @aggregation_temporality) + reservoir = @exemplar_reservoir_storage[data_points][hdp.attributes] + hdp.exemplars = reservoir&.collect(attributes: hdp.attributes, aggregation_temporality: @aggregation_temporality.temporality) hdp end data_points.clear @@ -52,8 +54,8 @@ def collect(start_time, end_time, data_points) data_points.values.map! do |hdp| hdp.start_time_unix_nano ||= start_time # Start time of a data point is from the first observation. hdp.time_unix_nano = end_time - reservoir = @exemplar_reservoir_storage[hdp.attributes] - hdp.exemplars = reservoir&.collect(attributes: hdp.attributes, aggregation_temporality: @aggregation_temporality) + reservoir = @exemplar_reservoir_storage[data_points][hdp.attributes] + hdp.exemplars = reservoir&.collect(attributes: hdp.attributes, aggregation_temporality: @aggregation_temporality.temporality) hdp = hdp.dup hdp.bucket_counts = hdp.bucket_counts.dup hdp @@ -71,7 +73,7 @@ def update(amount, attributes, data_points, cardinality_limit, exemplar_offer: f create_new_data_point(attributes, data_points) end - update_histogram_data_point(hdp, amount, exemplar_offer: exemplar_offer) + update_histogram_data_point(hdp, amount, data_points, exemplar_offer: exemplar_offer) nil end @@ -103,8 +105,8 @@ def create_new_data_point(attributes, data_points) ) end - def update_histogram_data_point(hdp, amount, exemplar_offer: false) - reservior_update(hdp.attributes, amount, exemplar_offer) + def update_histogram_data_point(hdp, amount, stream_key, exemplar_offer: false) + reservior_update(hdp.attributes, amount, exemplar_offer, stream_key) if @record_min_max hdp.max = amount if amount > hdp.max @@ -119,12 +121,12 @@ def update_histogram_data_point(hdp, amount, exemplar_offer: false) hdp.bucket_counts[bucket_index] += 1 end - def reservior_update(attributes, amount, exemplar_offer) - reservoir = @exemplar_reservoir_storage[attributes] + def reservior_update(attributes, amount, exemplar_offer, stream_key) + reservoir = @exemplar_reservoir_storage[stream_key][attributes] unless reservoir reservoir = @exemplar_reservoir.dup reservoir.reset - @exemplar_reservoir_storage[attributes] = reservoir + @exemplar_reservoir_storage[stream_key][attributes] = reservoir end return unless exemplar_offer diff --git a/metrics_sdk/lib/opentelemetry/sdk/metrics/aggregation/exponential_bucket_histogram.rb b/metrics_sdk/lib/opentelemetry/sdk/metrics/aggregation/exponential_bucket_histogram.rb index c9debc003a..8ebcb7e555 100644 --- a/metrics_sdk/lib/opentelemetry/sdk/metrics/aggregation/exponential_bucket_histogram.rb +++ b/metrics_sdk/lib/opentelemetry/sdk/metrics/aggregation/exponential_bucket_histogram.rb @@ -17,6 +17,8 @@ module Metrics module Aggregation # Contains the implementation of the {https://opentelemetry.io/docs/specs/otel/metrics/data-model/#exponentialhistogram ExponentialBucketHistogram} aggregation class ExponentialBucketHistogram # rubocop:disable Metrics/ClassLength + include StreamScopedStorage + OVERFLOW_ATTRIBUTE_SET = { 'otel.metric.overflow' => true }.freeze # relate to min max scale: https://opentelemetry.io/docs/specs/otel/metrics/sdk/#support-a-minimum-and-maximum-scale @@ -54,45 +56,57 @@ def initialize( @scale = validate_scale(max_scale) @exemplar_reservoir = exemplar_reservoir || DEFAULT_RESERVOIR - @exemplar_reservoir_storage = {} + @exemplar_reservoir_storage = new_stream_storage @mapping = new_mapping(@scale) # Previous state for cumulative aggregation - @previous_positive = {} # nil - @previous_negative = {} # nil - @previous_min = {} # Float::INFINITY - @previous_max = {} # -Float::INFINITY - @previous_sum = {} # 0 - @previous_count = {} # 0 - @previous_zero_count = {} # 0 - @previous_scale = {} # nil - - # Cache mappings per attribute set - @mappings = {} - @previous_mappings = {} + @previous_positive = new_stream_storage # nil + @previous_negative = new_stream_storage # nil + @previous_min = new_stream_storage # Float::INFINITY + @previous_max = new_stream_storage # -Float::INFINITY + @previous_sum = new_stream_storage # 0 + @previous_count = new_stream_storage # 0 + @previous_zero_count = new_stream_storage # 0 + @previous_scale = new_stream_storage # nil + + @mappings = new_stream_storage + @previous_mappings = new_stream_storage end # when aggregation temporality is cumulative, merge and downscale will happen. # rubocop:disable-next Metrics/MethodLength def collect(start_time, end_time, data_points) + stream_key = data_points + stream_exemplar_reservoir_storage = @exemplar_reservoir_storage[stream_key] + if @aggregation_temporality.delta? # Set timestamps and 'move' data point values to result. hdps = data_points.values.map! do |hdp| hdp.start_time_unix_nano = start_time hdp.time_unix_nano = end_time - reservoir = @exemplar_reservoir_storage[hdp.attributes] - hdp.exemplars = reservoir&.collect(attributes: hdp.attributes, aggregation_temporality: @aggregation_temporality) + reservoir = stream_exemplar_reservoir_storage[hdp.attributes] + hdp.exemplars = reservoir&.collect(attributes: hdp.attributes, aggregation_temporality: @aggregation_temporality.temporality) hdp end data_points.clear - @mappings.clear + @mappings.delete(stream_key) hdps else # CUMULATIVE temporality - merge current data_points to previous data_points # and only keep the merged data_points in @previous_* merged_data_points = {} + stream_previous_positive = @previous_positive[stream_key] + stream_previous_negative = @previous_negative[stream_key] + stream_previous_min = @previous_min[stream_key] + stream_previous_max = @previous_max[stream_key] + stream_previous_sum = @previous_sum[stream_key] + stream_previous_count = @previous_count[stream_key] + stream_previous_zero_count = @previous_zero_count[stream_key] + stream_previous_scale = @previous_scale[stream_key] + stream_mappings = @mappings[stream_key] + stream_previous_mappings = @previous_mappings[stream_key] # this will slow down the operation especially if large amount of data_points present # but it should be fine since with cumulative, the data_points are merged into previous_* and not kept in data_points @@ -108,25 +122,25 @@ def collect(start_time, end_time, data_points) current_scale = hdp.scale # Setup previous positive, negative bucket and scale based on three different cases - @previous_positive[attributes] = current_positive.copy_empty if @previous_positive[attributes].nil? - @previous_negative[attributes] = current_negative.copy_empty if @previous_negative[attributes].nil? - @previous_scale[attributes] = current_scale if @previous_scale[attributes].nil? + stream_previous_positive[attributes] = current_positive.copy_empty if stream_previous_positive[attributes].nil? + stream_previous_negative[attributes] = current_negative.copy_empty if stream_previous_negative[attributes].nil? + stream_previous_scale[attributes] = current_scale if stream_previous_scale[attributes].nil? # Determine minimum scale for merging - min_scale = [@previous_scale[attributes], current_scale].min + min_scale = [stream_previous_scale[attributes], current_scale].min # Calculate ranges for positive and negative buckets low_positive, high_positive = get_low_high_previous_current( - @previous_positive[attributes], + stream_previous_positive[attributes], current_positive, - @previous_scale[attributes], + stream_previous_scale[attributes], current_scale, min_scale ) low_negative, high_negative = get_low_high_previous_current( - @previous_negative[attributes], + stream_previous_negative[attributes], current_negative, - @previous_scale[attributes], + stream_previous_scale[attributes], current_scale, min_scale ) @@ -138,79 +152,79 @@ def collect(start_time, end_time, data_points) ].min # Downscale previous buckets if necessary - downscale_change = @previous_scale[attributes] - min_scale - downscale(downscale_change, @previous_positive[attributes], @previous_negative[attributes]) + downscale_change = stream_previous_scale[attributes] - min_scale + downscale(downscale_change, stream_previous_positive[attributes], stream_previous_negative[attributes]) # Merge current buckets into previous buckets (kind like update); it's always :cumulative - merge_buckets(@previous_positive[attributes], current_positive, current_scale, min_scale, @aggregation_temporality) - merge_buckets(@previous_negative[attributes], current_negative, current_scale, min_scale, @aggregation_temporality) + merge_buckets(stream_previous_positive[attributes], current_positive, current_scale, min_scale, @aggregation_temporality) + merge_buckets(stream_previous_negative[attributes], current_negative, current_scale, min_scale, @aggregation_temporality) # initialize min, max, sum, count, zero_count for first time - @previous_min[attributes] = Float::INFINITY if @previous_min[attributes].nil? - @previous_max[attributes] = -Float::INFINITY if @previous_max[attributes].nil? - @previous_sum[attributes] = 0 if @previous_sum[attributes].nil? - @previous_count[attributes] = 0 if @previous_count[attributes].nil? - @previous_zero_count[attributes] = 0 if @previous_zero_count[attributes].nil? + stream_previous_min[attributes] = Float::INFINITY if stream_previous_min[attributes].nil? + stream_previous_max[attributes] = -Float::INFINITY if stream_previous_max[attributes].nil? + stream_previous_sum[attributes] = 0 if stream_previous_sum[attributes].nil? + stream_previous_count[attributes] = 0 if stream_previous_count[attributes].nil? + stream_previous_zero_count[attributes] = 0 if stream_previous_zero_count[attributes].nil? # Update aggregated values - @previous_min[attributes] = [@previous_min[attributes], current_min].min - @previous_max[attributes] = [@previous_max[attributes], current_max].max - @previous_sum[attributes] += current_sum - @previous_count[attributes] += current_count - @previous_zero_count[attributes] += current_zero_count - @previous_scale[attributes] = min_scale + stream_previous_min[attributes] = [stream_previous_min[attributes], current_min].min + stream_previous_max[attributes] = [stream_previous_max[attributes], current_max].max + stream_previous_sum[attributes] += current_sum + stream_previous_count[attributes] += current_count + stream_previous_zero_count[attributes] += current_zero_count + stream_previous_scale[attributes] = min_scale # Create merged data point - reservoir = @exemplar_reservoir_storage[attributes] + reservoir = stream_exemplar_reservoir_storage[attributes] merged_hdp = ExponentialHistogramDataPoint.new( attributes, start_time, end_time, - @previous_count[attributes], - @previous_sum[attributes], - @previous_scale[attributes], - @previous_zero_count[attributes], - @previous_positive[attributes].dup, - @previous_negative[attributes].dup, + stream_previous_count[attributes], + stream_previous_sum[attributes], + stream_previous_scale[attributes], + stream_previous_zero_count[attributes], + stream_previous_positive[attributes].dup, + stream_previous_negative[attributes].dup, 0, # flags - reservoir&.collect(attributes: attributes, aggregation_temporality: @aggregation_temporality), # exemplars - @previous_min[attributes], - @previous_max[attributes], + reservoir&.collect(attributes: attributes, aggregation_temporality: @aggregation_temporality.temporality), # exemplars + stream_previous_min[attributes], + stream_previous_max[attributes], @zero_threshold ) merged_data_points[attributes] = merged_hdp - @previous_mappings[attributes] = @mappings[attributes] if @mappings[attributes] # Preserve mapping for next collection + stream_previous_mappings[attributes] = stream_mappings[attributes] if stream_mappings[attributes] # Preserve mapping for next collection end # when you have no local_data_points, the loop from cumulative aggregation will not run # so return last merged data points if exists - if data_points.empty? && !@previous_positive.empty? - @previous_positive.each_key do |attributes| - reservoir = @exemplar_reservoir_storage[attributes] + if data_points.empty? && !stream_previous_positive.empty? + stream_previous_positive.each_key do |attributes| + reservoir = stream_exemplar_reservoir_storage[attributes] merged_hdp = ExponentialHistogramDataPoint.new( attributes, start_time, end_time, - @previous_count[attributes], - @previous_sum[attributes], - @previous_scale[attributes], - @previous_zero_count[attributes], - @previous_positive[attributes].dup, - @previous_negative[attributes].dup, + stream_previous_count[attributes], + stream_previous_sum[attributes], + stream_previous_scale[attributes], + stream_previous_zero_count[attributes], + stream_previous_positive[attributes].dup, + stream_previous_negative[attributes].dup, 0, # flags - reservoir&.collect(attributes: attributes, aggregation_temporality: @aggregation_temporality), # exemplars - @previous_min[attributes], - @previous_max[attributes], + reservoir&.collect(attributes: attributes, aggregation_temporality: @aggregation_temporality.temporality), # exemplars + stream_previous_min[attributes], + stream_previous_max[attributes], @zero_threshold ) merged_data_points[attributes] = merged_hdp end end - # Swap current with previous mappings for next cycle - @mappings = @previous_mappings - @previous_mappings = {} + # Swap current with previous mappings for next cycle, for this stream only + @mappings[stream_key] = stream_previous_mappings + @previous_mappings[stream_key] = {} # clear data_points since the data is merged into previous_* already; # otherwise we will have duplicated data_points in the next collect @@ -229,7 +243,7 @@ def update(amount, attributes, data_points, cardinality_limit, exemplar_offer: f create_new_data_point(attributes, data_points) end - update_histogram_data_point(hdp, attributes, amount, exemplar_offer: exemplar_offer) + update_histogram_data_point(hdp, attributes, amount, data_points, exemplar_offer: exemplar_offer) nil end @@ -265,8 +279,8 @@ def create_new_data_point(attributes, data_points) end # rubocop:disable-next Metrics/CyclomaticComplexity,Metrics/MethodLength - def update_histogram_data_point(hdp, attributes, amount, exemplar_offer: false) - reservior_update(attributes, amount, exemplar_offer) + def update_histogram_data_point(hdp, attributes, amount, stream_key, exemplar_offer: false) + reservior_update(attributes, amount, exemplar_offer, stream_key) if @record_min_max hdp.max = amount if amount > hdp.max @@ -289,11 +303,11 @@ def update_histogram_data_point(hdp, attributes, amount, exemplar_offer: false) # Reset scale to max_scale if transitioning from all-zeros to first non-zero value if buckets.counts == [0] && hdp.scale == 0 && hdp.count > hdp.zero_count hdp.scale = @scale - @mappings.delete(attributes) # Clear any cached mapping + @mappings[stream_key].delete(attributes) # Clear any cached mapping end # Get or create mapping for this attribute set - mapping = @mappings[attributes] ||= new_mapping(hdp.scale) + mapping = @mappings[stream_key][attributes] ||= new_mapping(hdp.scale) bucket_index = mapping.map_to_index(amount) rescaling_needed = false @@ -320,7 +334,7 @@ def update_histogram_data_point(hdp, attributes, amount, exemplar_offer: false) downscale(scale_change, hdp.positive, hdp.negative) new_scale = mapping.scale - scale_change mapping = new_mapping(new_scale) - @mappings[attributes] = mapping # Update cache + @mappings[stream_key][attributes] = mapping # Update cache bucket_index = mapping.map_to_index(amount) OpenTelemetry.logger.debug "Rescaled with new scale #{new_scale} from #{low} and #{high}; bucket_index is updated to #{bucket_index}" @@ -352,12 +366,12 @@ def grow_buckets(span, buckets) buckets.grow(span + 1, @size) end - def reservior_update(attributes, amount, exemplar_offer) - reservoir = @exemplar_reservoir_storage[attributes] + def reservior_update(attributes, amount, exemplar_offer, stream_key) + reservoir = @exemplar_reservoir_storage[stream_key][attributes] unless reservoir reservoir = @exemplar_reservoir.dup reservoir.reset - @exemplar_reservoir_storage[attributes] = reservoir + @exemplar_reservoir_storage[stream_key][attributes] = reservoir end return unless exemplar_offer diff --git a/metrics_sdk/lib/opentelemetry/sdk/metrics/aggregation/last_value.rb b/metrics_sdk/lib/opentelemetry/sdk/metrics/aggregation/last_value.rb index 4fe7431218..f8cb86f9e1 100644 --- a/metrics_sdk/lib/opentelemetry/sdk/metrics/aggregation/last_value.rb +++ b/metrics_sdk/lib/opentelemetry/sdk/metrics/aggregation/last_value.rb @@ -10,6 +10,8 @@ module Metrics module Aggregation # Contains the implementation of the LastValue aggregation class LastValue + include StreamScopedStorage + OVERFLOW_ATTRIBUTE_SET = { 'otel.metric.overflow' => true }.freeze attr_reader :exemplar_reservoir @@ -19,15 +21,17 @@ class LastValue def initialize(exemplar_reservoir: nil) @exemplar_reservoir = exemplar_reservoir || DEFAULT_RESERVOIR - @exemplar_reservoir_storage = {} + @exemplar_reservoir_storage = new_stream_storage end # Returns the current data points, clearing them for delta temporality. def collect(start_time, end_time, data_points) + stream_exemplar_reservoir_storage = @exemplar_reservoir_storage[data_points] + ndps = data_points.values.map! do |ndp| ndp.start_time_unix_nano = start_time ndp.time_unix_nano = end_time - reservoir = @exemplar_reservoir_storage[ndp.attributes] + reservoir = stream_exemplar_reservoir_storage[ndp.attributes] ndp.exemplars = reservoir&.collect(attributes: ndp.attributes, aggregation_temporality: :delta) ndp end @@ -46,7 +50,7 @@ def update(increment, attributes, data_points, cardinality_limit, exemplar_offer create_new_data_point(attributes, data_points) end - update_number_data_point(ndp, increment, exemplar_offer: exemplar_offer) + update_number_data_point(ndp, increment, data_points, exemplar_offer: exemplar_offer) nil end @@ -63,17 +67,17 @@ def create_new_data_point(attributes, data_points) ) end - def update_number_data_point(ndp, increment, exemplar_offer: false) + def update_number_data_point(ndp, increment, stream_key, exemplar_offer: false) ndp.value = increment - reservior_update(ndp.attributes, increment, exemplar_offer) + reservior_update(ndp.attributes, increment, exemplar_offer, stream_key) end - def reservior_update(attributes, increment, exemplar_offer) - reservoir = @exemplar_reservoir_storage[attributes] + def reservior_update(attributes, increment, exemplar_offer, stream_key) + reservoir = @exemplar_reservoir_storage[stream_key][attributes] unless reservoir reservoir = @exemplar_reservoir.dup reservoir.reset - @exemplar_reservoir_storage[attributes] = reservoir + @exemplar_reservoir_storage[stream_key][attributes] = reservoir end return unless exemplar_offer diff --git a/metrics_sdk/lib/opentelemetry/sdk/metrics/aggregation/stream_scoped_storage.rb b/metrics_sdk/lib/opentelemetry/sdk/metrics/aggregation/stream_scoped_storage.rb new file mode 100644 index 0000000000..e98fcd0310 --- /dev/null +++ b/metrics_sdk/lib/opentelemetry/sdk/metrics/aggregation/stream_scoped_storage.rb @@ -0,0 +1,26 @@ +# frozen_string_literal: true + +# Copyright The OpenTelemetry Authors +# +# SPDX-License-Identifier: Apache-2.0 + +module OpenTelemetry + module SDK + module Metrics + module Aggregation + # Builds the nested storage aggregations use to keep per-stream state + # apart, since a wildcard or regex view shares one aggregation instance + # across every stream it matches. + module StreamScopedStorage + private + + # Streams are keyed by their +data_points+ hash, which is mutable, so + # the outer hash must compare by identity rather than by value. + def new_stream_storage + Hash.new { |h, k| h[k] = {} }.compare_by_identity + end + end + end + end + end +end diff --git a/metrics_sdk/lib/opentelemetry/sdk/metrics/aggregation/sum.rb b/metrics_sdk/lib/opentelemetry/sdk/metrics/aggregation/sum.rb index da918d2db8..ddac46e013 100644 --- a/metrics_sdk/lib/opentelemetry/sdk/metrics/aggregation/sum.rb +++ b/metrics_sdk/lib/opentelemetry/sdk/metrics/aggregation/sum.rb @@ -10,6 +10,8 @@ module Metrics module Aggregation # Contains the implementation of the Sum aggregation class Sum + include StreamScopedStorage + OVERFLOW_ATTRIBUTE_SET = { 'otel.metric.overflow' => true }.freeze attr_reader :exemplar_reservoir @@ -24,18 +26,20 @@ def initialize(aggregation_temporality: ENV.fetch('OTEL_EXPORTER_OTLP_METRICS_TE @aggregation_temporality = AggregationTemporality.determine_temporality(aggregation_temporality: aggregation_temporality, instrument_kind: instrument_kind, default: :cumulative) @monotonic = monotonic @exemplar_reservoir = exemplar_reservoir || DEFAULT_RESERVOIR - @exemplar_reservoir_storage = {} + @exemplar_reservoir_storage = new_stream_storage end # Returns the current sum data points, clearing them for delta temporality. def collect(start_time, end_time, data_points) + stream_exemplar_reservoir_storage = @exemplar_reservoir_storage[data_points] + if @aggregation_temporality.delta? # Set timestamps and 'move' data point values to result. ndps = data_points.values.map! do |ndp| ndp.start_time_unix_nano = start_time ndp.time_unix_nano = end_time - reservoir = @exemplar_reservoir_storage[ndp.attributes] - ndp.exemplars = reservoir&.collect(attributes: ndp.attributes, aggregation_temporality: @aggregation_temporality) + reservoir = stream_exemplar_reservoir_storage[ndp.attributes] + ndp.exemplars = reservoir&.collect(attributes: ndp.attributes, aggregation_temporality: @aggregation_temporality.temporality) ndp end data_points.clear @@ -45,8 +49,8 @@ def collect(start_time, end_time, data_points) data_points.values.map! do |ndp| ndp.start_time_unix_nano ||= start_time # Start time of a data point is from the first observation. ndp.time_unix_nano = end_time - reservoir = @exemplar_reservoir_storage[ndp.attributes] - ndp.exemplars = reservoir&.collect(attributes: ndp.attributes, aggregation_temporality: @aggregation_temporality) + reservoir = stream_exemplar_reservoir_storage[ndp.attributes] + ndp.exemplars = reservoir&.collect(attributes: ndp.attributes, aggregation_temporality: @aggregation_temporality.temporality) ndp.dup end end @@ -65,7 +69,7 @@ def update(increment, attributes, data_points, cardinality_limit, exemplar_offer create_new_data_point(attributes, data_points) end - update_number_data_point(ndp, increment, exemplar_offer: exemplar_offer) + update_number_data_point(ndp, increment, data_points, exemplar_offer: exemplar_offer) nil end @@ -92,17 +96,17 @@ def create_new_data_point(attributes, data_points) ) end - def update_number_data_point(ndp, increment, exemplar_offer: false) - reservior_update(ndp.attributes, increment, exemplar_offer) + def update_number_data_point(ndp, increment, stream_key, exemplar_offer: false) + reservior_update(ndp.attributes, increment, exemplar_offer, stream_key) ndp.value += increment end - def reservior_update(attributes, increment, exemplar_offer) - reservoir = @exemplar_reservoir_storage[attributes] + def reservior_update(attributes, increment, exemplar_offer, stream_key) + reservoir = @exemplar_reservoir_storage[stream_key][attributes] unless reservoir reservoir = @exemplar_reservoir.dup reservoir.reset - @exemplar_reservoir_storage[attributes] = reservoir + @exemplar_reservoir_storage[stream_key][attributes] = reservoir end return unless exemplar_offer diff --git a/metrics_sdk/test/opentelemetry/sdk/metrics/aggregation/explicit_bucket_histogram_test.rb b/metrics_sdk/test/opentelemetry/sdk/metrics/aggregation/explicit_bucket_histogram_test.rb index fa116a8d37..090645def0 100644 --- a/metrics_sdk/test/opentelemetry/sdk/metrics/aggregation/explicit_bucket_histogram_test.rb +++ b/metrics_sdk/test/opentelemetry/sdk/metrics/aggregation/explicit_bucket_histogram_test.rb @@ -416,4 +416,20 @@ end end end + + # https://github.com/open-telemetry/opentelemetry-ruby/issues/2300 + it 'keeps exemplar reservoirs independent per stream when streams share one aggregation instance' do + attrs = { 'outcome' => 'success' } + stream_a_dps = {} + stream_b_dps = {} + + ebh.update(1, attrs, stream_a_dps, cardinality_limit, exemplar_offer: true) + ebh.update(2, attrs, stream_b_dps, cardinality_limit, exemplar_offer: true) + + a_hdp = ebh.collect(start_time, end_time, stream_a_dps).first + b_hdp = ebh.collect(start_time, end_time, stream_b_dps).first + + _(a_hdp.exemplars.map(&:value)).must_equal([1]) + _(b_hdp.exemplars.map(&:value)).must_equal([2]) + end end diff --git a/metrics_sdk/test/opentelemetry/sdk/metrics/aggregation/exponential_bucket_histogram_test.rb b/metrics_sdk/test/opentelemetry/sdk/metrics/aggregation/exponential_bucket_histogram_test.rb index 43e931dd6a..ecbee2944e 100644 --- a/metrics_sdk/test/opentelemetry/sdk/metrics/aggregation/exponential_bucket_histogram_test.rb +++ b/metrics_sdk/test/opentelemetry/sdk/metrics/aggregation/exponential_bucket_histogram_test.rb @@ -336,4 +336,95 @@ end end end + + # https://github.com/open-telemetry/opentelemetry-ruby/issues/2300 + describe 'when multiple streams share one aggregation instance via a view' do + it 'keeps delta scale and bucket state independent per stream, matching independent instances' do + attrs = { 'outcome' => 'success' } + duration_values = [0.143741, 0.327218, 0.156199, 0.201044, 0.118377] + batch_size_values = [1.0, 2.0, 1.0, 2.0, 1.0] + body_size_values = [351.0, 2269.0, 640.0, 1810.0, 402.0] + + independent_result = lambda do |values| + agg = OpenTelemetry::SDK::Metrics::Aggregation::ExponentialBucketHistogram.new(aggregation_temporality: :delta, zero_threshold: 0) + dps = {} + values.each { |v| agg.update(v, attrs, dps, cardinality_limit) } + agg.collect(start_time, end_time, dps).first + end + + expected_duration = independent_result.call(duration_values) + expected_batch_size = independent_result.call(batch_size_values) + expected_body_size = independent_result.call(body_size_values) + + shared = OpenTelemetry::SDK::Metrics::Aggregation::ExponentialBucketHistogram.new(aggregation_temporality: :delta, zero_threshold: 0) + duration_dps = {} + batch_size_dps = {} + body_size_dps = {} + + duration_values.each_index do |i| + shared.update(duration_values[i], attrs, duration_dps, cardinality_limit) + shared.update(batch_size_values[i], attrs, batch_size_dps, cardinality_limit) + shared.update(body_size_values[i], attrs, body_size_dps, cardinality_limit) + end + + duration_hdp = shared.collect(start_time, end_time, duration_dps).first + batch_size_hdp = shared.collect(start_time, end_time, batch_size_dps).first + body_size_hdp = shared.collect(start_time, end_time, body_size_dps).first + + _(duration_hdp.scale).must_equal(expected_duration.scale) + _(duration_hdp.count).must_equal(expected_duration.count) + _(duration_hdp.positive.counts).must_equal(expected_duration.positive.counts) + + _(batch_size_hdp.scale).must_equal(expected_batch_size.scale) + _(batch_size_hdp.count).must_equal(expected_batch_size.count) + _(batch_size_hdp.positive.counts).must_equal(expected_batch_size.positive.counts) + + _(body_size_hdp.scale).must_equal(expected_body_size.scale) + _(body_size_hdp.count).must_equal(expected_body_size.count) + _(body_size_hdp.positive.counts).must_equal(expected_body_size.positive.counts) + end + + it 'keeps cumulative previous-state independent per stream across collect cycles' do + attrs = { 'outcome' => 'success' } + shared = OpenTelemetry::SDK::Metrics::Aggregation::ExponentialBucketHistogram.new(aggregation_temporality: :cumulative, zero_threshold: 0) + stream_a_dps = {} + stream_b_dps = {} + + shared.update(1.0, attrs, stream_a_dps, cardinality_limit) + shared.update(1000.0, attrs, stream_b_dps, cardinality_limit) + + a_first = shared.collect(start_time, end_time, stream_a_dps).first + b_first = shared.collect(start_time, end_time, stream_b_dps).first + + _(a_first.count).must_equal(1) + _(b_first.count).must_equal(1) + + shared.update(2.0, attrs, stream_a_dps, cardinality_limit) + shared.update(2000.0, attrs, stream_b_dps, cardinality_limit) + + a_second = shared.collect(start_time, end_time, stream_a_dps).first + b_second = shared.collect(start_time, end_time, stream_b_dps).first + + _(a_second.count).must_equal(2) + _(a_second.positive.counts.sum).must_equal(2) + _(b_second.count).must_equal(2) + _(b_second.positive.counts.sum).must_equal(2) + end + + it 'keeps exemplar reservoirs independent per stream' do + attrs = { 'outcome' => 'success' } + shared = OpenTelemetry::SDK::Metrics::Aggregation::ExponentialBucketHistogram.new(aggregation_temporality: :delta, zero_threshold: 0) + stream_a_dps = {} + stream_b_dps = {} + + shared.update(1.0, attrs, stream_a_dps, cardinality_limit, exemplar_offer: true) + shared.update(2.0, attrs, stream_b_dps, cardinality_limit, exemplar_offer: true) + + a_hdp = shared.collect(start_time, end_time, stream_a_dps).first + b_hdp = shared.collect(start_time, end_time, stream_b_dps).first + + _(a_hdp.exemplars.map(&:value)).must_equal([1.0]) + _(b_hdp.exemplars.map(&:value)).must_equal([2.0]) + end + end end diff --git a/metrics_sdk/test/opentelemetry/sdk/metrics/aggregation/last_value_test.rb b/metrics_sdk/test/opentelemetry/sdk/metrics/aggregation/last_value_test.rb index abf1ca4ac1..3bf5070134 100644 --- a/metrics_sdk/test/opentelemetry/sdk/metrics/aggregation/last_value_test.rb +++ b/metrics_sdk/test/opentelemetry/sdk/metrics/aggregation/last_value_test.rb @@ -91,4 +91,20 @@ end end end + + # https://github.com/open-telemetry/opentelemetry-ruby/issues/2300 + it 'keeps exemplar reservoirs independent per stream when streams share one aggregation instance' do + attrs = { 'outcome' => 'success' } + stream_a_dps = {} + stream_b_dps = {} + + last_value_aggregation.update(1, attrs, stream_a_dps, cardinality_limit, exemplar_offer: true) + last_value_aggregation.update(2, attrs, stream_b_dps, cardinality_limit, exemplar_offer: true) + + a_ndp = last_value_aggregation.collect(start_time, end_time, stream_a_dps).first + b_ndp = last_value_aggregation.collect(start_time, end_time, stream_b_dps).first + + _(a_ndp.exemplars.map(&:value)).must_equal([1]) + _(b_ndp.exemplars.map(&:value)).must_equal([2]) + end end diff --git a/metrics_sdk/test/opentelemetry/sdk/metrics/aggregation/sum_test.rb b/metrics_sdk/test/opentelemetry/sdk/metrics/aggregation/sum_test.rb index 3813d71187..2d86c3dccd 100644 --- a/metrics_sdk/test/opentelemetry/sdk/metrics/aggregation/sum_test.rb +++ b/metrics_sdk/test/opentelemetry/sdk/metrics/aggregation/sum_test.rb @@ -204,4 +204,20 @@ end end end + + # https://github.com/open-telemetry/opentelemetry-ruby/issues/2300 + it 'keeps exemplar reservoirs independent per stream when streams share one aggregation instance' do + attrs = { 'outcome' => 'success' } + stream_a_dps = {} + stream_b_dps = {} + + sum_aggregation.update(1, attrs, stream_a_dps, cardinality_limit, exemplar_offer: true) + sum_aggregation.update(2, attrs, stream_b_dps, cardinality_limit, exemplar_offer: true) + + a_ndp = sum_aggregation.collect(start_time, end_time, stream_a_dps).first + b_ndp = sum_aggregation.collect(start_time, end_time, stream_b_dps).first + + _(a_ndp.exemplars.map(&:value)).must_equal([1]) + _(b_ndp.exemplars.map(&:value)).must_equal([2]) + end end diff --git a/metrics_sdk/test/opentelemetry/sdk/metrics/exemplar/exemplar_integration_test.rb b/metrics_sdk/test/opentelemetry/sdk/metrics/exemplar/exemplar_integration_test.rb index 197f20f253..f4f828831e 100644 --- a/metrics_sdk/test/opentelemetry/sdk/metrics/exemplar/exemplar_integration_test.rb +++ b/metrics_sdk/test/opentelemetry/sdk/metrics/exemplar/exemplar_integration_test.rb @@ -254,20 +254,20 @@ _(exponential_histogram.description).must_equal('data size') _(exponential_histogram.data_points.size).must_equal 2 - # First data point: {} attributes, count=1, sum=100, 2 exemplars + # First data point: {} attributes, count=1, sum=100, 1 exemplar _(exponential_histogram.data_points[0].attributes).must_equal({}) _(exponential_histogram.data_points[0].count).must_equal 1 _(exponential_histogram.data_points[0].sum).must_equal 100 _(exponential_histogram.data_points[0].scale).must_equal 20 - _(exponential_histogram.data_points[0].exemplars.size).must_equal 2 + _(exponential_histogram.data_points[0].exemplars.size).must_equal 1 - # Second data point: {'type' => 'upload'} attributes, count=2, sum=600, 4 exemplars + # Second data point: {'type' => 'upload'} attributes, count=2, sum=600, 2 exemplars _(exponential_histogram.data_points[1].attributes).must_equal('type' => 'upload') _(exponential_histogram.data_points[1].count).must_equal 2 _(exponential_histogram.data_points[1].sum).must_equal 600 _(exponential_histogram.data_points[1].scale).must_equal 7 - _(exponential_histogram.data_points[1].exemplars.size).must_equal 4 - _(exponential_histogram.data_points[1].exemplars.map(&:value).sort).must_equal [200, 200, 400, 400] + _(exponential_histogram.data_points[1].exemplars.size).must_equal 2 + _(exponential_histogram.data_points[1].exemplars.map(&:value).sort).must_equal [200, 400] end it 'emits counter metrics with exemplars and customized reservoir' do @@ -351,13 +351,13 @@ sum200 = sum_metric.data_points.find { |dp| dp.attributes['status'] == '200' } _(sum200.value).must_equal 30 # 10 + 20 - _(sum200.exemplars.size).must_equal 4 - _(sum200.exemplars.map(&:value)).must_equal [10, 10, 20, 20] + _(sum200.exemplars.size).must_equal 2 + _(sum200.exemplars.map(&:value)).must_equal [10, 20] sum500 = sum_metric.data_points.find { |dp| dp.attributes['status'] == '500' } _(sum500.value).must_equal 3 - _(sum500.exemplars.size).must_equal 2 - _(sum500.exemplars.map(&:value)).must_equal [3, 3] + _(sum500.exemplars.size).must_equal 1 + _(sum500.exemplars.map(&:value)).must_equal [3] # Second metric (LastValue aggregation): status='200' value=20, status='500' value=3 lastvalue_metric = metrics_with_exemplars.find { |m| m.aggregation_temporality.nil? } @@ -365,13 +365,13 @@ lastvalue200 = lastvalue_metric.data_points.find { |dp| dp.attributes['status'] == '200' } _(lastvalue200.value).must_equal 20 # Last value - _(lastvalue200.exemplars.size).must_equal 4 - _(lastvalue200.exemplars.map(&:value)).must_equal [10, 10, 20, 20] + _(lastvalue200.exemplars.size).must_equal 2 + _(lastvalue200.exemplars.map(&:value)).must_equal [10, 20] lastvalue500 = lastvalue_metric.data_points.find { |dp| dp.attributes['status'] == '500' } _(lastvalue500.value).must_equal 3 - _(lastvalue500.exemplars.size).must_equal 2 - _(lastvalue500.exemplars.map(&:value)).must_equal [3, 3] + _(lastvalue500.exemplars.size).must_equal 1 + _(lastvalue500.exemplars.map(&:value)).must_equal [3] # Verify all exemplars are properly formed metrics_with_exemplars.each do |metric|