[PoC] Add AttributeFn to metric Stream

sdk/metric/example_test.go

@@ -8,9 +8,11 @@ import (
 	"fmt"
 	"log"
 	"regexp"
+	"strings"
 
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/baggage"
 	"go.opentelemetry.io/otel/sdk/instrumentation"
 	"go.opentelemetry.io/otel/sdk/metric"
 	"go.opentelemetry.io/otel/sdk/metric/exemplar"
@@ -219,6 +221,71 @@ func ExampleNewView_attributeFilter() {
 	)
 }
 
+func ExampleNewView_addBaggageAttributes() {
+	// Create a view that adds my custom baggage attribute to all measurements
+	// of the "latency" instrument from the "slo" instrumentation library.
+	view := metric.NewView(
+		metric.Instrument{
+			Name:  "latency",
+			Scope: instrumentation.Scope{Name: "slo"},
+		},
+		metric.Stream{
+			AttributeFn: func(ctx context.Context, a attribute.Set) attribute.Set {
+				b := baggage.FromContext(ctx)
+				const key = "my.custom.baggage.attribute"
+				m := b.Member(key)
+				if m.Key() != key {
+					// Return the original set if baggage member not set.
+					return a
+				}
+				return attribute.NewSet(append(
+					[]attribute.KeyValue{attribute.String(key, m.Value())},
+					a.ToSlice()...,
+				)...)
+			},
+		},
+	)
+
+	// The created view can then be registered with the OpenTelemetry metric
+	// SDK using the WithView option.
+	_ = metric.NewMeterProvider(
+		metric.WithView(view),
+	)
+}
+
+func ExampleNewView_reduceCardinality() {
+	// Replaces know paths with low-cardinality template values.
+	routeAttr := attribute.String("http.route", "/api/v1/users/{user_id}")
+	view := metric.NewView(
+		metric.Instrument{
+			Name:  "latency",
+			Scope: instrumentation.Scope{Name: "http"},
+		},
+		metric.Stream{
+			AttributeFn: func(_ context.Context, a attribute.Set) attribute.Set {
+				v, ok := a.Value("http.route")
+				if !ok {
+					return a
+				}
+				if strings.HasPrefix(v.AsString(), "/api/v1/users/") {
+					// Replace the user ID with a low-cardinality value.
+					return attribute.NewSet(append(
+						[]attribute.KeyValue{routeAttr},
+						a.ToSlice()...,
+					)...)
+				}
+				return a
+			},
+		},
+	)
+
+	// The created view can then be registered with the OpenTelemetry metric
+	// SDK using the WithView option.
+	_ = metric.NewMeterProvider(
+		metric.WithView(view),
+	)
+}
+
 func ExampleNewView_exponentialHistogram() {
 	// Create a view that makes the "latency" instrument from the "http"
 	// instrumentation library to be reported as an exponential histogram.

sdk/metric/instrument.go

@@ -137,6 +137,10 @@ type Stream struct {
 	Unit string
 	// Aggregation the stream uses for an instrument.
 	Aggregation Aggregation
+	// AttributeFn is a function that updates attributes recorded made in the
+	// measurement stream. This is used to add or modify attributes before
+	// aggregation and filtering.
+	AttributeFn func(context.Context, attribute.Set) attribute.Set
 	// AttributeFilter is an attribute Filter applied to the attributes
 	// recorded for an instrument's measurement. If the filter returns false
 	// the attribute will not be recorded, otherwise, if it returns true, it

sdk/metric/internal/aggregate/aggregate.go

@@ -33,6 +33,9 @@ type Builder[N int64 | float64] struct {
 	// Filter is the attribute filter the aggregate function will use on the
 	// input of measurements.
 	Filter attribute.Filter
+	// Annotater returns a new attribute set for the measurement
+	// attributes.
+	Annotater func(context.Context, attribute.Set) attribute.Set
 	// ReservoirFunc is the factory function used by aggregate functions to
 	// create new exemplar reservoirs for a new seen attribute set.
 	//
@@ -57,6 +60,18 @@ func (b Builder[N]) resFunc() func(attribute.Set) FilteredExemplarReservoir[N] {
 	return dropReservoir
 }
 
+func (b Builder[N]) annotate(m Measure[N]) Measure[N] {
+	if b.Annotater != nil {
+		fn := b.Annotater // Copy to make it immutable after assignment.
+		return func(ctx context.Context, n N, a attribute.Set) {
+			m(ctx, n, fn(ctx, a))
+		}
+	}
+	return func(ctx context.Context, n N, a attribute.Set) {
+		m(ctx, n, a)
+	}
+}
+
 type fltrMeasure[N int64 | float64] func(ctx context.Context, value N, fltrAttr attribute.Set, droppedAttr []attribute.KeyValue)
 
 func (b Builder[N]) filter(f fltrMeasure[N]) Measure[N] {
@@ -77,9 +92,9 @@ func (b Builder[N]) LastValue() (Measure[N], ComputeAggregation) {
 	lv := newLastValue[N](b.AggregationLimit, b.resFunc())
 	switch b.Temporality {
 	case metricdata.DeltaTemporality:
-		return b.filter(lv.measure), lv.delta
+		return b.annotate(b.filter(lv.measure)), lv.delta
 	default:
-		return b.filter(lv.measure), lv.cumulative
+		return b.annotate(b.filter(lv.measure)), lv.cumulative
 	}
 }
 
@@ -90,9 +105,9 @@ func (b Builder[N]) PrecomputedLastValue() (Measure[N], ComputeAggregation) {
 	lv := newPrecomputedLastValue[N](b.AggregationLimit, b.resFunc())
 	switch b.Temporality {
 	case metricdata.DeltaTemporality:
-		return b.filter(lv.measure), lv.delta
+		return b.annotate(b.filter(lv.measure)), lv.delta
 	default:
-		return b.filter(lv.measure), lv.cumulative
+		return b.annotate(b.filter(lv.measure)), lv.cumulative
 	}
 }
 
@@ -102,9 +117,9 @@ func (b Builder[N]) PrecomputedSum(monotonic bool) (Measure[N], ComputeAggregati
 	s := newPrecomputedSum[N](monotonic, b.AggregationLimit, b.resFunc())
 	switch b.Temporality {
 	case metricdata.DeltaTemporality:
-		return b.filter(s.measure), s.delta
+		return b.annotate(b.filter(s.measure)), s.delta
 	default:
-		return b.filter(s.measure), s.cumulative
+		return b.annotate(b.filter(s.measure)), s.cumulative
 	}
 }
 
@@ -113,9 +128,9 @@ func (b Builder[N]) Sum(monotonic bool) (Measure[N], ComputeAggregation) {
 	s := newSum[N](monotonic, b.AggregationLimit, b.resFunc())
 	switch b.Temporality {
 	case metricdata.DeltaTemporality:
-		return b.filter(s.measure), s.delta
+		return b.annotate(b.filter(s.measure)), s.delta
 	default:
-		return b.filter(s.measure), s.cumulative
+		return b.annotate(b.filter(s.measure)), s.cumulative
 	}
 }
 
@@ -128,9 +143,9 @@ func (b Builder[N]) ExplicitBucketHistogram(
 	h := newHistogram[N](boundaries, noMinMax, noSum, b.AggregationLimit, b.resFunc())
 	switch b.Temporality {
 	case metricdata.DeltaTemporality:
-		return b.filter(h.measure), h.delta
+		return b.annotate(b.filter(h.measure)), h.delta
 	default:
-		return b.filter(h.measure), h.cumulative
+		return b.annotate(b.filter(h.measure)), h.cumulative
 	}
 }
 
@@ -143,9 +158,9 @@ func (b Builder[N]) ExponentialBucketHistogram(
 	h := newExponentialHistogram[N](maxSize, maxScale, noMinMax, noSum, b.AggregationLimit, b.resFunc())
 	switch b.Temporality {
 	case metricdata.DeltaTemporality:
-		return b.filter(h.measure), h.delta
+		return b.annotate(b.filter(h.measure)), h.delta
 	default:
-		return b.filter(h.measure), h.cumulative
+		return b.annotate(b.filter(h.measure)), h.cumulative
 	}
 }
 

sdk/metric/pipeline.go

@@ -385,6 +385,7 @@ func (i *inserter[N]) cachedAggregator(
 				i.pipeline.exemplarFilter,
 			),
 		}
+		b.Annotater = stream.AttributeFn
 		b.Filter = stream.AttributeFilter
 		// A value less than or equal to zero will disable the aggregation
 		// limits for the builder (an all the created aggregates).