Combine precomputed values of filtered attribute sets

CHANGELOG.md

@@ -107,6 +107,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 ### Fixed
 
+- Asynchronous instruments that use sum aggregators and attribute filters correctly add values from equivalent attribute sets that have been filtered. (#3439, #3549)
 - The `RegisterCallback` method of the `Meter` from `go.opentelemetry.io/otel/sdk/metric` only registers a callback for instruments created by that meter.
   Trying to register a callback with instruments from a different meter will result in an error being returned. (#3584)
 

sdk/metric/internal/aggregator.go

@@ -22,13 +22,13 @@ import (
 )
 
 // now is used to return the current local time while allowing tests to
-// override the the default time.Now function.
+// override the default time.Now function.
 var now = time.Now
 
 // Aggregator forms an aggregation from a collection of recorded measurements.
 //
-// Aggregators need to be comparable so they can be de-duplicated by the SDK when
-// it creates them for multiple views.
+// Aggregators need to be comparable so they can be de-duplicated by the SDK
+// when it creates them for multiple views.
 type Aggregator[N int64 | float64] interface {
 	// Aggregate records the measurement, scoped by attr, and aggregates it
 	// into an aggregation.
@@ -38,3 +38,22 @@ 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/filter.go

@@ -21,8 +21,26 @@ import (
 	"go.opentelemetry.io/otel/sdk/metric/metricdata"
 )
 
-// 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.
+// NewFilter returns an Aggregator that wraps an agg with an attribute
+// filtering function. Both pre-computed non-pre-computed Aggregators can be
+// passed for agg. An appropriate Aggregator will be returned for the detected
+// type.
+func NewFilter[N int64 | float64](agg Aggregator[N], fn attribute.Filter) Aggregator[N] {
+	if fn == nil {
+		return agg
+	}
+	if fa, ok := agg.(precomputeAggregator[N]); ok {
+		return newPrecomputedFilter(fa, fn)
+	}
+	return newFilter(agg, fn)
+}
+
+// filter wraps an aggregator with an attribute filter. All recorded
+// measurements will have their attributes filtered before they are passed to
+// the underlying aggregator's Aggregate method.
+//
+// This should not be used to wrap a pre-computed Aggregator. Use a
+// precomputedFilter instead.
 type filter[N int64 | float64] struct {
 	filter     attribute.Filter
 	aggregator Aggregator[N]
@@ -31,15 +49,16 @@ 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
-	}
+// newFilter returns an filter Aggregator that wraps agg with the attribute
+// filter fn.
+//
+// This should not be used to wrap a pre-computed Aggregator. Use a
+// precomputedFilter instead.
+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 +81,54 @@ 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]
+
+	sync.Mutex
+	seen map[attribute.Set]attribute.Set
+}
+
+// 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,
+		seen:       make(map[attribute.Set]attribute.Set),
+	}
+}
+
+// Aggregate records the measurement, scoped by attr, and aggregates it
+// into an aggregation.
+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.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/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]) 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
+}