Combine precomputed values of filtered attribute sets

CHANGELOG.md

@@ -14,6 +14,10 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
   This `Registration` can be used to unregister callbacks. (#3522)
 - Add `Producer` interface and `Reader.RegisterProducer(Producer)` to `go.opentelemetry.io/otel/sdk/metric` to enable external metric Producers. (#3524)
 
+### Fixed
+
+- Asynchronous instruments that use sum aggregators and attribute filters correctly add values from equivalent attribute sets that have been filtered. (#3439, #3549)
+
 ### Changed
 
 - The `InstrumentProvider` from `go.opentelemetry.io/otel/sdk/metric/asyncint64` is removed.

sdk/metric/internal/filter.go

@@ -21,6 +21,24 @@ import (
 	"go.opentelemetry.io/otel/sdk/metric/metricdata"
 )
 
+type filterAgg[N int64 | float64] interface {
+	Aggregator[N]
+
+	// filtered records values for attributes that have been filtered.
+	filtered(N, attribute.Set)
+}
+
+// NewFilter wraps an Aggregator with an attribute filtering function.
+func NewFilter[N int64 | float64](agg Aggregator[N], fn attribute.Filter) Aggregator[N] {
+	if fn == nil {
+		return agg
+	}
+	if fa, ok := agg.(filterAgg[N]); ok {
+		return newPrecomputedFilter(fa, fn)
+	}
+	return newFilter(agg, fn)
+}
+
 // filter is an aggregator that applies attribute filter when Aggregating. filters
 // do not have any backing memory, and must be constructed with a backing Aggregator.
 type filter[N int64 | float64] struct {
@@ -31,15 +49,11 @@ type filter[N int64 | float64] struct {
 	seen map[attribute.Set]attribute.Set
 }
 
-// NewFilter wraps an Aggregator with an attribute filtering function.
-func NewFilter[N int64 | float64](agg Aggregator[N], fn attribute.Filter) Aggregator[N] {
-	if fn == nil {
-		return agg
-	}
+func newFilter[N int64 | float64](agg Aggregator[N], fn attribute.Filter) *filter[N] {
 	return &filter[N]{
 		filter:     fn,
 		aggregator: agg,
-		seen:       map[attribute.Set]attribute.Set{},
+		seen:       make(map[attribute.Set]attribute.Set),
 	}
 }
 
@@ -62,3 +76,44 @@ 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 precomputed Aggregations. The precomputed 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 filterAgg[N]
+
+	sync.Mutex
+	seen map[attribute.Set]attribute.Set
+}
+
+func newPrecomputedFilter[N int64 | float64](agg filterAgg[N], fn attribute.Filter) *precomputedFilter[N] {
+	return &precomputedFilter[N]{
+		filter:     fn,
+		aggregator: agg,
+		seen:       make(map[attribute.Set]attribute.Set),
+	}
+}
+func (f *precomputedFilter[N]) Aggregate(measurement N, attr attribute.Set) {
+	// TODO (#3006): drop stale attributes from seen.
+	f.Lock()
+	defer f.Unlock()
+	fAttr, ok := f.seen[attr]
+	if !ok {
+		fAttr, _ = attr.Filter(f.filter)
+		f.seen[attr] = fAttr
+	}
+	if fAttr.Equals(&attr) {
+		// No filtering done.
+		f.aggregator.Aggregate(measurement, fAttr)
+	} else {
+		f.aggregator.filtered(measurement, fAttr)
+	}
+}
+
+func (f *precomputedFilter[N]) Aggregation() metricdata.Aggregation {
+	return f.aggregator.Aggregation()
+}

sdk/metric/internal/filter_test.go

@@ -15,6 +15,8 @@
 package internal // import "go.opentelemetry.io/otel/sdk/metric/internal"
 
 import (
+	"fmt"
+	"strings"
 	"sync"
 	"testing"
 
@@ -194,3 +196,90 @@ 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]) filtered(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/sum.go

@@ -22,7 +22,7 @@ import (
 	"go.opentelemetry.io/otel/sdk/metric/metricdata"
 )
 
-// valueMap is the storage for all sums.
+// valueMap is the storage for sums.
 type valueMap[N int64 | float64] struct {
 	sync.Mutex
 	values map[attribute.Set]N
@@ -32,12 +32,6 @@ func newValueMap[N int64 | float64]() *valueMap[N] {
 	return &valueMap[N]{values: make(map[attribute.Set]N)}
 }
 
-func (s *valueMap[N]) set(value N, attr attribute.Set) { // nolint: unused  // This is indeed used.
-	s.Lock()
-	s.values[attr] = value
-	s.Unlock()
-}
-
 func (s *valueMap[N]) Aggregate(value N, attr attribute.Set) {
 	s.Lock()
 	s.values[attr] += value
@@ -164,6 +158,43 @@ func (s *cumulativeSum[N]) Aggregation() metricdata.Aggregation {
 	return out
 }
 
+type precomputedValue[N int64 | float64] struct {
+	// measured is the value directly measured.
+	measured N
+	// filtered is the sum of values from spatially aggregations.
+	filtered N
+}
+
+// valueMap 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 as a cumulative sum for attr.
+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()
+}
+
+// filtered records value with spatially re-aggregated attrs.
+func (s *precomputedMap[N]) filtered(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
 // measurements as their pre-computed arithmetic sum. Each sum is scoped by
 // attributes and the aggregation cycle the measurements were made in.
@@ -176,56 +207,52 @@ func (s *cumulativeSum[N]) Aggregation() metricdata.Aggregation {
 // is up to the caller to ensure this is accurate.
 func NewPrecomputedDeltaSum[N int64 | float64](monotonic bool) Aggregator[N] {
 	return &precomputedDeltaSum[N]{
-		recorded:  make(map[attribute.Set]N),
-		reported:  make(map[attribute.Set]N),
-		monotonic: monotonic,
-		start:     now(),
+		precomputedMap: newPrecomputedMap[N](),
+		reported:       make(map[attribute.Set]N),
+		monotonic:      monotonic,
+		start:          now(),
 	}
 }
 
 // precomputedDeltaSum summarizes a set of measurements recorded over all
 // aggregation cycles as the delta arithmetic sum.
 type precomputedDeltaSum[N int64 | float64] struct {
-	sync.Mutex
-	recorded map[attribute.Set]N
+	*precomputedMap[N]
+
 	reported map[attribute.Set]N
 
 	monotonic bool
 	start     time.Time
 }
 
-// Aggregate records value as a cumulative sum for attr.
-func (s *precomputedDeltaSum[N]) Aggregate(value N, attr attribute.Set) {
-	s.Lock()
-	s.recorded[attr] = value
-	s.Unlock()
-}
-
 func (s *precomputedDeltaSum[N]) Aggregation() metricdata.Aggregation {
 	s.Lock()
 	defer s.Unlock()
 
-	if len(s.recorded) == 0 {
+	if len(s.values) == 0 {
 		return nil
 	}
 
 	t := now()
 	out := metricdata.Sum[N]{
 		Temporality: metricdata.DeltaTemporality,
 		IsMonotonic: s.monotonic,
-		DataPoints:  make([]metricdata.DataPoint[N], 0, len(s.recorded)),
+		DataPoints:  make([]metricdata.DataPoint[N], 0, len(s.values)),
 	}
-	for attr, recorded := range s.recorded {
-		value := recorded - s.reported[attr]
+	for attr, value := range s.values {
+		v := value.measured + value.filtered
+		delta := v - s.reported[attr]
 		out.DataPoints = append(out.DataPoints, metricdata.DataPoint[N]{
 			Attributes: attr,
 			StartTime:  s.start,
 			Time:       t,
-			Value:      value,
+			Value:      delta,
 		})
-		if value != 0 {
-			s.reported[attr] = recorded
+		if delta != 0 {
+			s.reported[attr] = v
 		}
+		value.filtered = N(0)
+		s.values[attr] = value
 		// TODO (#3006): This will use an unbounded amount of memory if there
 		// are unbounded number of attribute sets being aggregated. Attribute
 		// sets that become "stale" need to be forgotten so this will not
@@ -247,16 +274,49 @@ func (s *precomputedDeltaSum[N]) Aggregation() metricdata.Aggregation {
 // The output Aggregation will report recorded values as cumulative
 // temporality. It is up to the caller to ensure this is accurate.
 func NewPrecomputedCumulativeSum[N int64 | float64](monotonic bool) Aggregator[N] {
-	return &precomputedSum[N]{newCumulativeSum[N](monotonic)}
+	return &precomputedCumulativeSum[N]{
+		precomputedMap: newPrecomputedMap[N](),
+		monotonic:      monotonic,
+		start:          now(),
+	}
 }
 
-// precomputedSum summarizes a set of measurements recorded over all
+// precomputedCumulativeSum summarizes a set of measurements recorded over all
 // aggregation cycles directly as the cumulative arithmetic sum.
-type precomputedSum[N int64 | float64] struct {
-	*cumulativeSum[N]
+type precomputedCumulativeSum[N int64 | float64] struct {
+	*precomputedMap[N]
+
+	monotonic bool
+	start     time.Time
 }
 
-// Aggregate records value as a cumulative sum for attr.
-func (s *precomputedSum[N]) Aggregate(value N, attr attribute.Set) {
-	s.set(value, attr)
+func (s *precomputedCumulativeSum[N]) Aggregation() metricdata.Aggregation {
+	s.Lock()
+	defer s.Unlock()
+
+	if len(s.values) == 0 {
+		return nil
+	}
+
+	t := now()
+	out := metricdata.Sum[N]{
+		Temporality: metricdata.CumulativeTemporality,
+		IsMonotonic: s.monotonic,
+		DataPoints:  make([]metricdata.DataPoint[N], 0, len(s.values)),
+	}
+	for attr, value := range s.values {
+		out.DataPoints = append(out.DataPoints, metricdata.DataPoint[N]{
+			Attributes: attr,
+			StartTime:  s.start,
+			Time:       t,
+			Value:      value.measured + value.filtered,
+		})
+		value.filtered = N(0)
+		s.values[attr] = value
+		// TODO (#3006): This will use an unbounded amount of memory if there
+		// are unbounded number of attribute sets being aggregated. Attribute
+		// sets that become "stale" need to be forgotten so this will not
+		// overload the system.
+	}
+	return out
 }