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

[gillesbergerp]

Fixes #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.

[Copilot]

Pull request overview

This PR addresses issue #2300 by ensuring aggregation-internal “shared state” (exponential histogram scale/mapping caches and exemplar reservoir storage) is scoped to the owning metric stream when a single aggregation instance is reused across multiple streams via wildcard/regex views.

Changes:

• Key exemplar reservoir storage by stream + attributes to prevent cross-stream exemplar contamination when views share an aggregation instance.
• Key ExponentialBucketHistogram per-stream caches (@mappings, @previous_*) by stream + attributes to prevent downscale/mapping corruption across instruments.
• Update and add tests to assert per-stream independence for exemplars and exponential histogram bucket/scale behavior.

Reviewed changes

Copilot reviewed 9 out of 9 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
metrics_sdk/lib/opentelemetry/sdk/metrics/aggregation/sum.rb	Scope exemplar reservoir storage per stream (then attributes) when views share an aggregation instance.
metrics_sdk/lib/opentelemetry/sdk/metrics/aggregation/last_value.rb	Scope exemplar reservoir storage per stream (then attributes) when views share an aggregation instance.
metrics_sdk/lib/opentelemetry/sdk/metrics/aggregation/explicit_bucket_histogram.rb	Scope exemplar reservoir storage per stream (then attributes) when views share an aggregation instance.
metrics_sdk/lib/opentelemetry/sdk/metrics/aggregation/exponential_bucket_histogram.rb	Scope exponential histogram per-stream caches and exemplar storage to avoid cross-stream scale/mapping corruption.
metrics_sdk/test/opentelemetry/sdk/metrics/exemplar/exemplar_integration_test.rb	Adjust exemplar expectations to reflect per-stream isolation (no cross-stream duplication).
metrics_sdk/test/opentelemetry/sdk/metrics/aggregation/sum_test.rb	Add regression test ensuring exemplar reservoirs remain independent per stream with a shared aggregation instance.
metrics_sdk/test/opentelemetry/sdk/metrics/aggregation/last_value_test.rb	Add regression test ensuring exemplar reservoirs remain independent per stream with a shared aggregation instance.
metrics_sdk/test/opentelemetry/sdk/metrics/aggregation/explicit_bucket_histogram_test.rb	Add regression test ensuring exemplar reservoirs remain independent per stream with a shared aggregation instance.
metrics_sdk/test/opentelemetry/sdk/metrics/aggregation/exponential_bucket_histogram_test.rb	Add regression tests for per-stream independence of delta scale/mappings, cumulative previous-state, and exemplars.

Suppressed comments (4)

metrics_sdk/lib/opentelemetry/sdk/metrics/aggregation/sum.rb:53

• Same issue in the cumulative path: aggregation_temporality should be a symbol (@aggregation_temporality.temporality) so delta reservoirs can reset correctly when reused, and to keep the API consistent with ExemplarReservoir#collect.

              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 = stream_exemplar_reservoir_storage[ndp.attributes]
                ndp.exemplars = reservoir&.collect(attributes: ndp.attributes, aggregation_temporality: @aggregation_temporality)
                ndp.dup

metrics_sdk/lib/opentelemetry/sdk/metrics/aggregation/explicit_bucket_histogram.rb:60

• Same issue in the cumulative path: pass a symbol temporality to reservoir.collect (e.g. @aggregation_temporality.temporality) so delta reservoirs reset correctly and the call matches ExemplarReservoir#collect’s API.

              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 = stream_exemplar_reservoir_storage[hdp.attributes]
                hdp.exemplars = reservoir&.collect(attributes: hdp.attributes, aggregation_temporality: @aggregation_temporality)
                hdp = hdp.dup

metrics_sdk/lib/opentelemetry/sdk/metrics/aggregation/exponential_bucket_histogram.rb:193

• In the cumulative path, pass a symbol temporality into reservoir.collect (e.g. @aggregation_temporality.temporality). The current call passes an AggregationTemporality instance, which means delta-capable reservoirs will never reset when they should.

                  stream_previous_positive[attributes].dup,
                  stream_previous_negative[attributes].dup,
                  0, # flags
                  reservoir&.collect(attributes: attributes, aggregation_temporality: @aggregation_temporality), # exemplars
                  stream_previous_min[attributes],
                  stream_previous_max[attributes],
                  @zero_threshold

metrics_sdk/lib/opentelemetry/sdk/metrics/aggregation/exponential_bucket_histogram.rb:218

• Same issue when returning previously-merged cumulative datapoints: aggregation_temporality should be a symbol (e.g. @aggregation_temporality.temporality) so ExemplarReservoir#collect can correctly apply its delta reset behavior.

                    stream_previous_positive[attributes].dup,
                    stream_previous_negative[attributes].dup,
                    0, # flags
                    reservoir&.collect(attributes: attributes, aggregation_temporality: @aggregation_temporality), # exemplars
                    stream_previous_min[attributes],
                    stream_previous_max[attributes],

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[xuan-cao-swi]

Thanks, LGTM. I will get other maintainers to take a look as well.

[kaylareopelle]

@gillesbergerp, could you take a look at the merge conflicts on this PR?

[kaylareopelle]

I think this looks great and solves the bug report well. Thank you for your contribution, @gillesbergerp! One small refactoring suggestion, but other than that and the merge conflicts, this looks good to me.

[kaylareopelle]

Good catch with @aggregation_temporality.temporality!

[kaylareopelle]

What do you think about putting Hash.new { |h, k| h[k] = {} }.compare_by_identity into a helper method that can be shared amongst the aggregations?

[gillesbergerp]

Good idea. I introduced a StreamScopedStorage module. Let me know if you prefer a different home for the helper