Simplify the last-value aggregate

sdk/metric/internal/aggregate/aggregate.go

@@ -42,6 +42,17 @@ type Builder[N int64 | float64] struct {
 	Filter attribute.Filter
 }
 
+func (b Builder[N]) filter(f Measure[N]) Measure[N] {
+	if b.Filter != nil {
+		fltr := b.Filter // Copy to make it immutable after assignment.
+		return func(ctx context.Context, n N, a attribute.Set) {
+			fAttr, _ := a.Filter(fltr)
+			f(ctx, n, fAttr)
+		}
+	}
+	return f
+}
+
 func (b Builder[N]) input(agg aggregator[N]) Measure[N] {
 	if b.Filter != nil {
 		agg = newFilter[N](agg, b.Filter)
@@ -59,11 +70,13 @@ func (b Builder[N]) LastValue() (Measure[N], ComputeAggregation) {
 	// a last-value aggregate.
 	lv := newLastValue[N]()
 
-	return b.input(lv), func(dest *metricdata.Aggregation) int {
-		// TODO (#4220): optimize memory reuse here.
-		*dest = lv.Aggregation()
-
+	return b.filter(lv.measure), func(dest *metricdata.Aggregation) int {
+		// Ignore if dest is not a metricdata.Gauge. The chance for memory
+		// reuse of the DataPoints is missed (better luck next time).
 		gData, _ := (*dest).(metricdata.Gauge[N])
+		lv.computeAggregation(&gData.DataPoints)
+		*dest = gData
+
 		return len(gData.DataPoints)
 	}
 }
@@ -125,3 +138,12 @@ func (b Builder[N]) ExplicitBucketHistogram(cfg aggregation.ExplicitBucketHistog
 		return len(hData.DataPoints)
 	}
 }
+
+// reset ensures s has capacity and sets it length. If the capacity of s too
+// small, a new slice is returned with the specified capacity and length.
+func reset[T any](s []T, length, capacity int) []T {
+	if cap(s) < capacity {
+		return make([]T, length, capacity)
+	}
+	return s[:length]
+}

sdk/metric/internal/aggregate/aggregate_test.go

@@ -0,0 +1,144 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package aggregate // import "go.opentelemetry.io/otel/sdk/metric/internal/aggregate"
+
+import (
+	"context"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata/metricdatatest"
+)
+
+var (
+	keyUser    = "user"
+	userAlice  = attribute.String(keyUser, "Alice")
+	userBob    = attribute.String(keyUser, "Bob")
+	adminTrue  = attribute.Bool("admin", true)
+	adminFalse = attribute.Bool("admin", false)
+
+	alice = attribute.NewSet(userAlice, adminTrue)
+	bob   = attribute.NewSet(userBob, adminFalse)
+
+	// Filtered.
+	attrFltr = func(kv attribute.KeyValue) bool {
+		return kv.Key == attribute.Key(keyUser)
+	}
+	fltrAlice = attribute.NewSet(userAlice)
+	fltrBob   = attribute.NewSet(userBob)
+
+	// Sat Jan 01 2000 00:00:00 GMT+0000.
+	staticTime    = time.Unix(946684800, 0)
+	staticNowFunc = func() time.Time { return staticTime }
+	// Pass to t.Cleanup to override the now function with staticNowFunc and
+	// revert once the test completes. E.g. t.Cleanup(mockTime(now)).
+	mockTime = func(orig func() time.Time) (cleanup func()) {
+		now = staticNowFunc
+		return func() { now = orig }
+	}
+)
+
+type arg[N int64 | float64] struct {
+	ctx   context.Context
+	value N
+	attr  attribute.Set
+}
+
+type output struct {
+	n   int
+	agg metricdata.Aggregation
+}
+
+type teststep[N int64 | float64] struct {
+	input  []arg[N]
+	expect output
+}
+
+func test[N int64 | float64](meas Measure[N], comp ComputeAggregation, steps []teststep[N]) func(*testing.T) {
+	return func(t *testing.T) {
+		t.Helper()
+
+		got := new(metricdata.Aggregation)
+		for i, step := range steps {
+			for _, args := range step.input {
+				meas(args.ctx, args.value, args.attr)
+			}
+
+			t.Logf("step: %d", i)
+			assert.Equal(t, step.expect.n, comp(got), "incorrect data size")
+			metricdatatest.AssertAggregationsEqual(t, step.expect.agg, *got)
+		}
+	}
+}
+
+func benchmarkAggregate[N int64 | float64](factory func() (Measure[N], ComputeAggregation)) func(*testing.B) {
+	counts := []int{1, 10, 100}
+	return func(b *testing.B) {
+		for _, n := range counts {
+			b.Run(strconv.Itoa(n), func(b *testing.B) {
+				benchmarkAggregateN(b, factory, n)
+			})
+		}
+	}
+}
+
+var bmarkRes metricdata.Aggregation
+
+func benchmarkAggregateN[N int64 | float64](b *testing.B, factory func() (Measure[N], ComputeAggregation), count int) {
+	ctx := context.Background()
+	attrs := make([]attribute.Set, count)
+	for i := range attrs {
+		attrs[i] = attribute.NewSet(attribute.Int("value", i))
+	}
+
+	b.Run("Measure", func(b *testing.B) {
+		got := &bmarkRes
+		meas, comp := factory()
+		b.ReportAllocs()
+		b.ResetTimer()
+
+		for n := 0; n < b.N; n++ {
+			for _, attr := range attrs {
+				meas(ctx, 1, attr)
+			}
+		}
+
+		comp(got)
+	})
+
+	b.Run("ComputeAggregation", func(b *testing.B) {
+		comps := make([]ComputeAggregation, b.N)
+		for n := range comps {
+			meas, comp := factory()
+			for _, attr := range attrs {
+				meas(ctx, 1, attr)
+			}
+			comps[n] = comp
+		}
+
+		got := &bmarkRes
+		b.ReportAllocs()
+		b.ResetTimer()
+
+		for n := 0; n < b.N; n++ {
+			comps[n](got)
+		}
+	})
+}

sdk/metric/internal/aggregate/aggregator_test.go

@@ -18,7 +18,6 @@ import (
 	"strconv"
 	"sync"
 	"testing"
-	"time"
 
 	"github.com/stretchr/testify/assert"
 
@@ -33,21 +32,7 @@ const (
 	defaultCycles       = 3
 )
 
-var (
-	alice = attribute.NewSet(attribute.String("user", "alice"), attribute.Bool("admin", true))
-	bob   = attribute.NewSet(attribute.String("user", "bob"), attribute.Bool("admin", false))
-	carol = attribute.NewSet(attribute.String("user", "carol"), attribute.Bool("admin", false))
-
-	// Sat Jan 01 2000 00:00:00 GMT+0000.
-	staticTime    = time.Unix(946684800, 0)
-	staticNowFunc = func() time.Time { return staticTime }
-	// Pass to t.Cleanup to override the now function with staticNowFunc and
-	// revert once the test completes. E.g. t.Cleanup(mockTime(now)).
-	mockTime = func(orig func() time.Time) (cleanup func()) {
-		now = staticNowFunc
-		return func() { now = orig }
-	}
-)
+var carol = attribute.NewSet(attribute.String("user", "carol"), attribute.Bool("admin", false))
 
 func monoIncr[N int64 | float64]() setMap[N] {
 	return setMap[N]{alice: 1, bob: 10, carol: 2}

sdk/metric/internal/aggregate/lastvalue.go

@@ -15,6 +15,7 @@
 package aggregate // import "go.opentelemetry.io/otel/sdk/metric/internal/aggregate"
 
 import (
+	"context"
 	"sync"
 	"time"
 
@@ -28,47 +29,40 @@ type datapoint[N int64 | float64] struct {
 	value     N
 }
 
+func newLastValue[N int64 | float64]() *lastValue[N] {
+	return &lastValue[N]{values: make(map[attribute.Set]datapoint[N])}
+}
+
 // lastValue summarizes a set of measurements as the last one made.
 type lastValue[N int64 | float64] struct {
 	sync.Mutex
 
 	values map[attribute.Set]datapoint[N]
 }
 
-// newLastValue returns an Aggregator that summarizes a set of measurements as
-// the last one made.
-func newLastValue[N int64 | float64]() aggregator[N] {
-	return &lastValue[N]{values: make(map[attribute.Set]datapoint[N])}
-}
-
-func (s *lastValue[N]) Aggregate(value N, attr attribute.Set) {
+func (s *lastValue[N]) measure(ctx context.Context, value N, attr attribute.Set) {
 	d := datapoint[N]{timestamp: now(), value: value}
 	s.Lock()
 	s.values[attr] = d
 	s.Unlock()
 }
 
-func (s *lastValue[N]) Aggregation() metricdata.Aggregation {
+func (s *lastValue[N]) computeAggregation(dest *[]metricdata.DataPoint[N]) {
 	s.Lock()
 	defer s.Unlock()
 
-	if len(s.values) == 0 {
-		return nil
-	}
+	n := len(s.values)
+	*dest = reset(*dest, n, n)
 
-	gauge := metricdata.Gauge[N]{
-		DataPoints: make([]metricdata.DataPoint[N], 0, len(s.values)),
-	}
+	var i int
 	for a, v := range s.values {
-		gauge.DataPoints = append(gauge.DataPoints, metricdata.DataPoint[N]{
-			Attributes: a,
-			// The event time is the only meaningful timestamp, StartTime is
-			// ignored.
-			Time:  v.timestamp,
-			Value: v.value,
-		})
+		(*dest)[i].Attributes = a
+		// The event time is the only meaningful timestamp, StartTime is
+		// ignored.
+		(*dest)[i].Time = v.timestamp
+		(*dest)[i].Value = v.value
 		// Do not report stale values.
 		delete(s.values, a)
+		i++
 	}
-	return gauge
 }