Skip to content

fix: scope shared aggregation state to the owning stream (#2300) #2314

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
gillesbergerp wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix: scope shared aggregation state to the owning stream (#2300) #2314

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions metrics_sdk/lib/opentelemetry/sdk/metrics/aggregation.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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'
Expand Down
24 changes: 13 additions & 11 deletions metrics_sdk/lib/opentelemetry/sdk/metrics/aggregation/explicit_bucket_histogram.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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.
Expand All @@ -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)

@kaylareopelle kaylareopelle Sep 9, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good catch with @aggregation_temporality.temporality!

hdp
Comment thread
gillesbergerp marked this conversation as resolved.
end
data_points.clear
Expand All @@ -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
Expand All @@ -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

Expand Down Expand Up @@ -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
Expand All @@ -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
Expand Down
Loading