Combine precomputed values of filtered attribute sets

CHANGELOG.md

@@ -18,6 +18,10 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 - The deprecated `go.opentelemetry.io/otel/sdk/metric/view` package is removed. (#3520)
 
+### Fixed
+
+- Asynchronous instruments that use sum aggregators and attribute filters correctly add values from equivalent attribute sets that have been filtered. (#3439, #3549)
+
 ### Changed
 
 - Global error handler uses an atomic value instead of a mutex. (#3543)

sdk/metric/internal/filter.go

@@ -21,25 +21,43 @@ import (
 	"go.opentelemetry.io/otel/sdk/metric/metricdata"
 )
 
+type filteredSet struct {
+	filtered bool
+	attrs    attribute.Set
+}
+
 // 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 {
 	filter     attribute.Filter
 	aggregator Aggregator[N]
 
+	// Used to aggreagte if an aggregator aggregates values differently for
+	// spatically reaggregated attributes.
+	filtered func(N, attribute.Set)
+
 	sync.Mutex
-	seen map[attribute.Set]attribute.Set
+	seen map[attribute.Set]filteredSet
 }
 
 // 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
 	}
+	af, ok := agg.(interface{ aggregateFiltered(N, attribute.Set) })
+	if ok {
+		return &filter[N]{
+			filter:     fn,
+			aggregator: agg,
+			filtered:   af.aggregateFiltered,
+			seen:       make(map[attribute.Set]filteredSet),
+		}
+	}
 	return &filter[N]{
 		filter:     fn,
 		aggregator: agg,
-		seen:       map[attribute.Set]attribute.Set{},
+		seen:       make(map[attribute.Set]filteredSet),
 	}
 }
 
@@ -51,10 +69,15 @@ func (f *filter[N]) Aggregate(measurement N, attr attribute.Set) {
 	defer f.Unlock()
 	fAttr, ok := f.seen[attr]
 	if !ok {
-		fAttr, _ = attr.Filter(f.filter)
+		a, na := attr.Filter(f.filter)
+		fAttr = filteredSet{filtered: len(na) != 0, attrs: a}
 		f.seen[attr] = fAttr
 	}
-	f.aggregator.Aggregate(measurement, fAttr)
+	if fAttr.filtered && f.filtered != nil {
+		f.filtered(measurement, fAttr.attrs)
+	} else {
+		f.aggregator.Aggregate(measurement, fAttr.attrs)
+	}
 }
 
 // Aggregation returns an Aggregation, for all the aggregated

sdk/metric/internal/sum.go

@@ -201,6 +201,13 @@ func (s *precomputedDeltaSum[N]) Aggregate(value N, attr attribute.Set) {
 	s.Unlock()
 }
 
+// aggregateFiltered records value with spatially re-aggregated attrs.
+func (s *precomputedDeltaSum[N]) aggregateFiltered(value N, attr attribute.Set) { // nolint: unused  // used to filter.
+	s.Lock()
+	s.recorded[attr] += value
+	s.Unlock()
+}
+
 func (s *precomputedDeltaSum[N]) Aggregation() metricdata.Aggregation {
 	s.Lock()
 	defer s.Unlock()
@@ -260,3 +267,8 @@ type precomputedSum[N int64 | float64] struct {
 func (s *precomputedSum[N]) Aggregate(value N, attr attribute.Set) {
 	s.set(value, attr)
 }
+
+// aggregateFiltered records value with spatially re-aggregated attrs.
+func (s *precomputedSum[N]) aggregateFiltered(value N, attr attribute.Set) { // nolint: unused  // used to filter.
+	s.valueMap.Aggregate(value, attr)
+}

sdk/metric/internal/sum_test.go

@@ -18,6 +18,7 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/sdk/metric/metricdata"
@@ -163,6 +164,29 @@ func TestDeltaSumReset(t *testing.T) {
 	t.Run("Float64", testDeltaSumReset[float64])
 }
 
+func TestAggregateFiltered(t *testing.T) {
+	t.Run("PreComputedDelta", testAggregateFiltered(NewPrecomputedDeltaSum[int64](false)))
+	t.Run("PreComputedCumulativeSum", testAggregateFiltered(NewPrecomputedCumulativeSum[int64](false)))
+}
+
+type af interface{ aggregateFiltered(int64, attribute.Set) }
+
+func testAggregateFiltered[N int64 | float64](a Aggregator[N]) func(*testing.T) {
+	attrs := attribute.NewSet(attribute.String("key", "val"))
+	return func(t *testing.T) {
+		a.Aggregate(1, attrs)
+
+		require.Implements(t, (*af)(nil), a)
+		a.(af).aggregateFiltered(1, attrs)
+
+		agg := a.Aggregation()
+		require.IsType(t, agg, metricdata.Sum[int64]{})
+		sum := agg.(metricdata.Sum[int64])
+		require.Len(t, sum.DataPoints, 1)
+		assert.Equal(t, N(2), sum.DataPoints[0].Value)
+	}
+}
+
 func TestEmptySumNilAggregation(t *testing.T) {
 	assert.Nil(t, NewCumulativeSum[int64](true).Aggregation())
 	assert.Nil(t, NewCumulativeSum[int64](false).Aggregation())

sdk/metric/meter_test.go

@@ -624,6 +624,11 @@ func TestRegisterCallbackDropAggregations(t *testing.T) {
 }
 
 func TestAttributeFilter(t *testing.T) {
+	t.Run("Delta", testAttributeFilter(metricdata.DeltaTemporality))
+	t.Run("Cumulative", testAttributeFilter(metricdata.CumulativeTemporality))
+}
+
+func testAttributeFilter(temporality metricdata.Temporality) func(*testing.T) {
 	one := 1.0
 	two := 2.0
 	testcases := []struct {
@@ -650,10 +655,10 @@ func TestAttributeFilter(t *testing.T) {
 					DataPoints: []metricdata.DataPoint[float64]{
 						{
 							Attributes: attribute.NewSet(attribute.String("foo", "bar")),
-							Value:      2.0, // TODO (#3439): This should be 3.0.
+							Value:      3.0,
 						},
 					},
-					Temporality: metricdata.CumulativeTemporality,
+					Temporality: temporality,
 					IsMonotonic: true,
 				},
 			},
@@ -677,10 +682,10 @@ func TestAttributeFilter(t *testing.T) {
 					DataPoints: []metricdata.DataPoint[float64]{
 						{
 							Attributes: attribute.NewSet(attribute.String("foo", "bar")),
-							Value:      2.0, // TODO (#3439): This should be 3.0.
+							Value:      3.0,
 						},
 					},
-					Temporality: metricdata.CumulativeTemporality,
+					Temporality: temporality,
 					IsMonotonic: false,
 				},
 			},
@@ -729,10 +734,10 @@ func TestAttributeFilter(t *testing.T) {
 					DataPoints: []metricdata.DataPoint[int64]{
 						{
 							Attributes: attribute.NewSet(attribute.String("foo", "bar")),
-							Value:      20, // TODO (#3439): This should be 30.
+							Value:      30,
 						},
 					},
-					Temporality: metricdata.CumulativeTemporality,
+					Temporality: temporality,
 					IsMonotonic: true,
 				},
 			},
@@ -756,10 +761,10 @@ func TestAttributeFilter(t *testing.T) {
 					DataPoints: []metricdata.DataPoint[int64]{
 						{
 							Attributes: attribute.NewSet(attribute.String("foo", "bar")),
-							Value:      20, // TODO (#3439): This should be 30.
+							Value:      30,
 						},
 					},
-					Temporality: metricdata.CumulativeTemporality,
+					Temporality: temporality,
 					IsMonotonic: false,
 				},
 			},
@@ -810,7 +815,7 @@ func TestAttributeFilter(t *testing.T) {
 							Value:      3.0,
 						},
 					},
-					Temporality: metricdata.CumulativeTemporality,
+					Temporality: temporality,
 					IsMonotonic: true,
 				},
 			},
@@ -836,7 +841,7 @@ func TestAttributeFilter(t *testing.T) {
 							Value:      3.0,
 						},
 					},
-					Temporality: metricdata.CumulativeTemporality,
+					Temporality: temporality,
 					IsMonotonic: false,
 				},
 			},
@@ -867,7 +872,7 @@ func TestAttributeFilter(t *testing.T) {
 							Sum:          3.0,
 						},
 					},
-					Temporality: metricdata.CumulativeTemporality,
+					Temporality: temporality,
 				},
 			},
 		},
@@ -892,7 +897,7 @@ func TestAttributeFilter(t *testing.T) {
 							Value:      30,
 						},
 					},
-					Temporality: metricdata.CumulativeTemporality,
+					Temporality: temporality,
 					IsMonotonic: true,
 				},
 			},
@@ -918,7 +923,7 @@ func TestAttributeFilter(t *testing.T) {
 							Value:      30,
 						},
 					},
-					Temporality: metricdata.CumulativeTemporality,
+					Temporality: temporality,
 					IsMonotonic: false,
 				},
 			},
@@ -949,34 +954,38 @@ func TestAttributeFilter(t *testing.T) {
 							Sum:          3.0,
 						},
 					},
-					Temporality: metricdata.CumulativeTemporality,
+					Temporality: temporality,
 				},
 			},
 		},
 	}
 
-	for _, tt := range testcases {
-		t.Run(tt.name, func(t *testing.T) {
-			rdr := NewManualReader()
-			mtr := NewMeterProvider(
-				WithReader(rdr),
-				WithView(NewView(
-					Instrument{Name: "*"},
-					Stream{AttributeFilter: func(kv attribute.KeyValue) bool {
-						return kv.Key == attribute.Key("foo")
-					}},
-				)),
-			).Meter("TestAttributeFilter")
-			require.NoError(t, tt.register(t, mtr))
-
-			m, err := rdr.Collect(context.Background())
-			assert.NoError(t, err)
+	return func(t *testing.T) {
+		for _, tt := range testcases {
+			t.Run(tt.name, func(t *testing.T) {
+				rdr := NewManualReader(WithTemporalitySelector(func(InstrumentKind) metricdata.Temporality {
+					return temporality
+				}))
+				mtr := NewMeterProvider(
+					WithReader(rdr),
+					WithView(NewView(
+						Instrument{Name: "*"},
+						Stream{AttributeFilter: func(kv attribute.KeyValue) bool {
+							return kv.Key == attribute.Key("foo")
+						}},
+					)),
+				).Meter("TestAttributeFilter")
+				require.NoError(t, tt.register(t, mtr))
+
+				m, err := rdr.Collect(context.Background())
+				assert.NoError(t, err)
 
-			require.Len(t, m.ScopeMetrics, 1)
-			require.Len(t, m.ScopeMetrics[0].Metrics, 1)
+				require.Len(t, m.ScopeMetrics, 1)
+				require.Len(t, m.ScopeMetrics[0].Metrics, 1)
 
-			metricdatatest.AssertEqual(t, tt.wantMetric, m.ScopeMetrics[0].Metrics[0], metricdatatest.IgnoreTimestamp())
-		})
+				metricdatatest.AssertEqual(t, tt.wantMetric, m.ScopeMetrics[0].Metrics[0], metricdatatest.IgnoreTimestamp())
+			})
+		}
 	}
 }