Metric SDK: Sum duplicate async observations regardless of filtering

CHANGELOG.md

@@ -31,6 +31,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
   The `AttributeKeys` fields allows users to specify an allow-list of attributes allowed to be recorded for a view.
   This change is made to ensure compatibility with the OpenTelemetry specification. (#4288)
 - If an attribute set is omitted from an async callback, the previous value will no longer be exported. (#4290)
+- If an attribute set is Observed multiple times in an async callback, the values will be summed instead of the last observation winning. (#4289)
 - Allow the explicit bucket histogram aggregation to be used for the up-down counter, observable counter, observable up-down counter, and observable gauge in the `go.opentelemetry.io/otel/sdk/metric` package. (#4332)
 - Restrict `Meter`s in `go.opentelemetry.io/otel/sdk/metric` to only register and collect instruments it created. (#4333)
 

sdk/metric/internal/aggregate/aggregator.go

@@ -38,22 +38,3 @@ type aggregator[N int64 | float64] interface {
 	// measurements made and ends an aggregation cycle.
 	Aggregation() metricdata.Aggregation
 }
-
-// precomputeAggregator is an Aggregator that receives values to aggregate that
-// have been pre-computed by the caller.
-type precomputeAggregator[N int64 | float64] interface {
-	// The Aggregate method of the embedded Aggregator is used to record
-	// pre-computed measurements, scoped by attributes that have not been
-	// filtered by an attribute filter.
-	aggregator[N]
-
-	// aggregateFiltered records measurements scoped by attributes that have
-	// been filtered by an attribute filter.
-	//
-	// Pre-computed measurements of filtered attributes need to be recorded
-	// separate from those that haven't been filtered so they can be added to
-	// the non-filtered pre-computed measurements in a collection cycle and
-	// then resets after the cycle (the non-filtered pre-computed measurements
-	// are not reset).
-	aggregateFiltered(N, attribute.Set)
-}

sdk/metric/internal/aggregate/filter.go

@@ -27,9 +27,6 @@ func newFilter[N int64 | float64](agg aggregator[N], fn attribute.Filter) aggreg
 	if fn == nil {
 		return agg
 	}
-	if fa, ok := agg.(precomputeAggregator[N]); ok {
-		return newPrecomputedFilter(fa, fn)
-	}
 	return &filter[N]{
 		filter:     fn,
 		aggregator: agg,
@@ -59,43 +56,3 @@ func (f *filter[N]) Aggregate(measurement N, attr attribute.Set) {
 func (f *filter[N]) Aggregation() metricdata.Aggregation {
 	return f.aggregator.Aggregation()
 }
-
-// precomputedFilter is an aggregator that applies attribute filter when
-// Aggregating for pre-computed Aggregations. The pre-computed Aggregations
-// need to operate normally when no attribute filtering is done (for sums this
-// means setting the value), but when attribute filtering is done it needs to
-// be added to any set value.
-type precomputedFilter[N int64 | float64] struct {
-	filter     attribute.Filter
-	aggregator precomputeAggregator[N]
-}
-
-// newPrecomputedFilter returns a precomputedFilter Aggregator that wraps agg
-// with the attribute filter fn.
-//
-// This should not be used to wrap a non-pre-computed Aggregator. Use a
-// precomputedFilter instead.
-func newPrecomputedFilter[N int64 | float64](agg precomputeAggregator[N], fn attribute.Filter) *precomputedFilter[N] {
-	return &precomputedFilter[N]{
-		filter:     fn,
-		aggregator: agg,
-	}
-}
-
-// Aggregate records the measurement, scoped by attr, and aggregates it
-// into an aggregation.
-func (f *precomputedFilter[N]) Aggregate(measurement N, attr attribute.Set) {
-	fAttr, _ := attr.Filter(f.filter)
-	if fAttr.Equals(&attr) {
-		// No filtering done.
-		f.aggregator.Aggregate(measurement, fAttr)
-	} else {
-		f.aggregator.aggregateFiltered(measurement, fAttr)
-	}
-}
-
-// Aggregation returns an Aggregation, for all the aggregated
-// measurements made and ends an aggregation cycle.
-func (f *precomputedFilter[N]) Aggregation() metricdata.Aggregation {
-	return f.aggregator.Aggregation()
-}

sdk/metric/internal/aggregate/filter_test.go

@@ -15,8 +15,6 @@
 package aggregate // import "go.opentelemetry.io/otel/sdk/metric/internal/aggregate"
 
 import (
-	"fmt"
-	"strings"
 	"sync"
 	"testing"
 
@@ -196,90 +194,3 @@ func TestFilterConcurrent(t *testing.T) {
 		testFilterConcurrent[float64](t)
 	})
 }
-
-func TestPrecomputedFilter(t *testing.T) {
-	t.Run("Int64", testPrecomputedFilter[int64]())
-	t.Run("Float64", testPrecomputedFilter[float64]())
-}
-
-func testPrecomputedFilter[N int64 | float64]() func(t *testing.T) {
-	return func(t *testing.T) {
-		agg := newTestFilterAgg[N]()
-		f := newFilter[N](agg, testAttributeFilter)
-		require.IsType(t, &precomputedFilter[N]{}, f)
-
-		var (
-			powerLevel = attribute.Int("power-level", 9000)
-			user       = attribute.String("user", "Alice")
-			admin      = attribute.Bool("admin", true)
-		)
-		a := attribute.NewSet(powerLevel)
-		key := a
-		f.Aggregate(1, a)
-		assert.Equal(t, N(1), agg.values[key].measured, str(a))
-		assert.Equal(t, N(0), agg.values[key].filtered, str(a))
-
-		a = attribute.NewSet(powerLevel, user)
-		f.Aggregate(2, a)
-		assert.Equal(t, N(1), agg.values[key].measured, str(a))
-		assert.Equal(t, N(2), agg.values[key].filtered, str(a))
-
-		a = attribute.NewSet(powerLevel, user, admin)
-		f.Aggregate(3, a)
-		assert.Equal(t, N(1), agg.values[key].measured, str(a))
-		assert.Equal(t, N(5), agg.values[key].filtered, str(a))
-
-		a = attribute.NewSet(powerLevel)
-		f.Aggregate(2, a)
-		assert.Equal(t, N(2), agg.values[key].measured, str(a))
-		assert.Equal(t, N(5), agg.values[key].filtered, str(a))
-
-		a = attribute.NewSet(user)
-		f.Aggregate(3, a)
-		assert.Equal(t, N(2), agg.values[key].measured, str(a))
-		assert.Equal(t, N(5), agg.values[key].filtered, str(a))
-		assert.Equal(t, N(3), agg.values[*attribute.EmptySet()].filtered, str(a))
-
-		_ = f.Aggregation()
-		assert.Equal(t, 1, agg.aggregationN, "failed to propagate Aggregation")
-	}
-}
-
-func str(a attribute.Set) string {
-	iter := a.Iter()
-	out := make([]string, 0, iter.Len())
-	for iter.Next() {
-		kv := iter.Attribute()
-		out = append(out, fmt.Sprintf("%s:%#v", kv.Key, kv.Value.AsInterface()))
-	}
-	return strings.Join(out, ",")
-}
-
-type testFilterAgg[N int64 | float64] struct {
-	values       map[attribute.Set]precomputedValue[N]
-	aggregationN int
-}
-
-func newTestFilterAgg[N int64 | float64]() *testFilterAgg[N] {
-	return &testFilterAgg[N]{
-		values: make(map[attribute.Set]precomputedValue[N]),
-	}
-}
-
-func (a *testFilterAgg[N]) Aggregate(val N, attr attribute.Set) {
-	v := a.values[attr]
-	v.measured = val
-	a.values[attr] = v
-}
-
-// nolint: unused  // Used to agg filtered.
-func (a *testFilterAgg[N]) aggregateFiltered(val N, attr attribute.Set) {
-	v := a.values[attr]
-	v.filtered += val
-	a.values[attr] = v
-}
-
-func (a *testFilterAgg[N]) Aggregation() metricdata.Aggregation {
-	a.aggregationN++
-	return nil
-}

sdk/metric/internal/aggregate/sum.go

@@ -150,63 +150,6 @@ func (s *cumulativeSum[N]) Aggregation() metricdata.Aggregation {
 	return out
 }
 
-// precomputedValue is the recorded measurement value for a set of attributes.
-type precomputedValue[N int64 | float64] struct {
-	// measured is the last value measured for a set of attributes that were
-	// not filtered.
-	measured N
-	// filtered is the sum of values from measurements that had their
-	// attributes filtered.
-	filtered N
-}
-
-// precomputedMap is the storage for precomputed sums.
-type precomputedMap[N int64 | float64] struct {
-	sync.Mutex
-	values map[attribute.Set]precomputedValue[N]
-}
-
-func newPrecomputedMap[N int64 | float64]() *precomputedMap[N] {
-	return &precomputedMap[N]{
-		values: make(map[attribute.Set]precomputedValue[N]),
-	}
-}
-
-// Aggregate records value with the unfiltered attributes attr.
-//
-// If a previous measurement was made for the same attribute set:
-//
-//   - If that measurement's attributes were not filtered, this value overwrite
-//     that value.
-//   - If that measurement's attributes were filtered, this value will be
-//     recorded along side that value.
-func (s *precomputedMap[N]) Aggregate(value N, attr attribute.Set) {
-	s.Lock()
-	v := s.values[attr]
-	v.measured = value
-	s.values[attr] = v
-	s.Unlock()
-}
-
-// aggregateFiltered records value with the filtered attributes attr.
-//
-// If a previous measurement was made for the same attribute set:
-//
-//   - If that measurement's attributes were not filtered, this value will be
-//     recorded along side that value.
-//   - If that measurement's attributes were filtered, this value will be
-//     added to it.
-//
-// This method should not be used if attr have not been reduced by an attribute
-// filter.
-func (s *precomputedMap[N]) aggregateFiltered(value N, attr attribute.Set) { // nolint: unused  // Used to agg filtered.
-	s.Lock()
-	v := s.values[attr]
-	v.filtered += value
-	s.values[attr] = v
-	s.Unlock()
-}
-
 // newPrecomputedDeltaSum returns an Aggregator that summarizes a set of
 // pre-computed sums. Each sum is scoped by attributes and the aggregation
 // cycle the measurements were made in.
@@ -218,17 +161,17 @@ func (s *precomputedMap[N]) aggregateFiltered(value N, attr attribute.Set) { //
 // The output Aggregation will report recorded values as delta temporality.
 func newPrecomputedDeltaSum[N int64 | float64](monotonic bool) aggregator[N] {
 	return &precomputedDeltaSum[N]{
-		precomputedMap: newPrecomputedMap[N](),
-		reported:       make(map[attribute.Set]N),
-		monotonic:      monotonic,
-		start:          now(),
+		valueMap:  newValueMap[N](),
+		reported:  make(map[attribute.Set]N),
+		monotonic: monotonic,
+		start:     now(),
 	}
 }
 
 // precomputedDeltaSum summarizes a set of pre-computed sums recorded over all
 // aggregation cycles as the delta of these sums.
 type precomputedDeltaSum[N int64 | float64] struct {
-	*precomputedMap[N]
+	*valueMap[N]
 
 	reported map[attribute.Set]N
 
@@ -263,15 +206,14 @@ func (s *precomputedDeltaSum[N]) Aggregation() metricdata.Aggregation {
 		DataPoints:  make([]metricdata.DataPoint[N], 0, len(s.values)),
 	}
 	for attr, value := range s.values {
-		v := value.measured + value.filtered
-		delta := v - s.reported[attr]
+		delta := value - s.reported[attr]
 		out.DataPoints = append(out.DataPoints, metricdata.DataPoint[N]{
 			Attributes: attr,
 			StartTime:  s.start,
 			Time:       t,
 			Value:      delta,
 		})
-		newReported[attr] = v
+		newReported[attr] = value
 		// Unused attribute sets do not report.
 		delete(s.values, attr)
 	}
@@ -294,15 +236,15 @@ func (s *precomputedDeltaSum[N]) Aggregation() metricdata.Aggregation {
 // temporality.
 func newPrecomputedCumulativeSum[N int64 | float64](monotonic bool) aggregator[N] {
 	return &precomputedCumulativeSum[N]{
-		precomputedMap: newPrecomputedMap[N](),
-		monotonic:      monotonic,
-		start:          now(),
+		valueMap:  newValueMap[N](),
+		monotonic: monotonic,
+		start:     now(),
 	}
 }
 
 // precomputedCumulativeSum directly records and reports a set of pre-computed sums.
 type precomputedCumulativeSum[N int64 | float64] struct {
-	*precomputedMap[N]
+	*valueMap[N]
 
 	monotonic bool
 	start     time.Time
@@ -337,7 +279,7 @@ func (s *precomputedCumulativeSum[N]) Aggregation() metricdata.Aggregation {
 			Attributes: attr,
 			StartTime:  s.start,
 			Time:       t,
-			Value:      value.measured + value.filtered,
+			Value:      value,
 		})
 		// Unused attribute sets do not report.
 		delete(s.values, attr)