Add metrics Enricher API

CHANGELOG.md

@@ -13,6 +13,8 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 - `EventOption` and the related `NewEventConfig` function are added to the `go.opentelemetry.io/otel` package to configure Span events. (#1254)
 - A `TextMapPropagator` and associated `TextMapCarrier` are added to the `go.opentelemetry.io/otel/oteltest` package to test TextMap type propagators and their use. (#1259)
 - `SpanContextFromContext` returns `SpanContext` from context. (#1255)
+- `MetricsLabelsEnricher` type is added to `go.opentelemetry.io/otel/sdk/metric` package. (#1271)
+- `WithMetricsLabelsEnricher` config option is added to `go.opentelemetry.io/otel/sdk/push` package to allow providing a function to enrich metrics labels based on context. (#1271)
 
 ### Changed
 

sdk/metric/config.go

@@ -0,0 +1,38 @@
+// 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 metric
+
+// Config contains configuration for an SDK.
+type Config struct {
+	// If provided, MetricsLabelsEnricher is executed each time a metric is recorded
+	// by the Accumulator's sync instrument implementation
+	MetricsLabelsEnricher MetricsLabelsEnricher
+}
+
+// Option is the interface that applies the value to a configuration option.
+type Option interface {
+	// Apply sets the Option value of a Config.
+	Apply(*Config)
+}
+
+func WithMetricsLabelsEnricher(e MetricsLabelsEnricher) Option {
+	return metricsLabelsEnricherOption(e)
+}
+
+type metricsLabelsEnricherOption MetricsLabelsEnricher
+
+func (e metricsLabelsEnricherOption) Apply(config *Config) {
+	config.MetricsLabelsEnricher = MetricsLabelsEnricher(e)
+}

sdk/metric/controller/pull/config.go

@@ -17,6 +17,7 @@ package pull // import "go.opentelemetry.io/otel/sdk/metric/controller/pull"
 import (
 	"time"
 
+	"go.opentelemetry.io/otel/sdk/metric"
 	"go.opentelemetry.io/otel/sdk/resource"
 )
 
@@ -33,6 +34,10 @@ type Config struct {
 	// If the period is zero, caching of the result is disabled.
 	// The default value is 10 seconds.
 	CachePeriod time.Duration
+
+	// MetricsLabelsEnricher is a function that enriches metrics labels based
+	// on kvs stored in context when metrics are recorded.
+	MetricsLabelsEnricher metric.MetricsLabelsEnricher
 }
 
 // Option is the interface that applies the value to a configuration option.
@@ -62,3 +67,13 @@ type cachePeriodOption time.Duration
 func (o cachePeriodOption) Apply(config *Config) {
 	config.CachePeriod = time.Duration(o)
 }
+
+func WithMetricsLabelsEnricher(e metric.MetricsLabelsEnricher) Option {
+	return metricsLabelsEnricherOption(e)
+}
+
+type metricsLabelsEnricherOption metric.MetricsLabelsEnricher
+
+func (e metricsLabelsEnricherOption) Apply(config *Config) {
+	config.MetricsLabelsEnricher = metric.MetricsLabelsEnricher(e)
+}

sdk/metric/controller/pull/pull.go

@@ -61,6 +61,7 @@ func New(checkpointer export.Checkpointer, options ...Option) *Controller {
 	accum := sdk.NewAccumulator(
 		checkpointer,
 		config.Resource,
+		sdk.WithMetricsLabelsEnricher(config.MetricsLabelsEnricher),
 	)
 	return &Controller{
 		accumulator:  accum,

sdk/metric/controller/pull/pull_test.go

@@ -117,3 +117,35 @@ func TestPullWithCache(t *testing.T) {
 	}, records.Map())
 
 }
+
+func TestPullWithMetricsLabelEnricher(t *testing.T) {
+	metricsLabelsEnricher := func(ctx context.Context, kvs []label.KeyValue) ([]label.KeyValue, error) {
+		baggage := otel.Baggage(ctx)
+		kvs = append(baggage.ToSlice(), kvs...)
+		return kvs, nil
+	}
+
+	puller := pull.New(
+		basic.New(
+			selector.NewWithExactDistribution(),
+			export.CumulativeExporter,
+			basic.WithMemory(true),
+		),
+		pull.WithCachePeriod(0),
+		pull.WithMetricsLabelsEnricher(metricsLabelsEnricher),
+	)
+
+	ctx := otel.ContextWithBaggageValues(context.Background(), label.String("A", "B"))
+	meter := puller.MeterProvider().Meter("withLabelEnricher")
+	counter := otel.Must(meter).NewInt64Counter("counter.sum")
+
+	counter.Add(ctx, 10)
+
+	require.NoError(t, puller.Collect(context.Background()))
+	records := processortest.NewOutput(label.DefaultEncoder())
+	require.NoError(t, puller.ForEach(export.CumulativeExporter, records.AddRecord))
+
+	require.EqualValues(t, map[string]float64{
+		"counter.sum/A=B/": 10,
+	}, records.Map())
+}

sdk/metric/controller/push/config.go

@@ -17,6 +17,7 @@ package push // import "go.opentelemetry.io/otel/sdk/metric/controller/push"
 import (
 	"time"
 
+	"go.opentelemetry.io/otel/sdk/metric"
 	"go.opentelemetry.io/otel/sdk/resource"
 )
 
@@ -33,6 +34,10 @@ type Config struct {
 	// integrate, and export) can last before it is canceled. Defaults to
 	// the controller push period.
 	Timeout time.Duration
+
+	// MetricsLabelsEnricher is a function that enriches metrics labels based
+	// on kvs stored in context when metrics are recorded.
+	MetricsLabelsEnricher metric.MetricsLabelsEnricher
 }
 
 // Option is the interface that applies the value to a configuration option.
@@ -73,3 +78,13 @@ type timeoutOption time.Duration
 func (o timeoutOption) Apply(config *Config) {
 	config.Timeout = time.Duration(o)
 }
+
+func WithMetricsLabelsEnricher(e metric.MetricsLabelsEnricher) Option {
+	return metricsLabelsEnricherOption(e)
+}
+
+type metricsLabelsEnricherOption metric.MetricsLabelsEnricher
+
+func (e metricsLabelsEnricherOption) Apply(config *Config) {
+	config.MetricsLabelsEnricher = metric.MetricsLabelsEnricher(e)
+}

sdk/metric/controller/push/push.go

@@ -62,6 +62,7 @@ func New(checkpointer export.Checkpointer, exporter export.Exporter, opts ...Opt
 	impl := sdk.NewAccumulator(
 		checkpointer,
 		c.Resource,
+		sdk.WithMetricsLabelsEnricher(c.MetricsLabelsEnricher),
 	)
 	return &Controller{
 		provider:     registry.NewMeterProvider(impl),

sdk/metric/controller/push/push_test.go

@@ -223,3 +223,39 @@ func TestPushExportError(t *testing.T) {
 		})
 	}
 }
+
+func TestWithMetricsLabelsEnricher(t *testing.T) {
+	exporter := newExporter()
+	checkpointer := newCheckpointer()
+	metricsLabelsEnricher := func(ctx context.Context, kvs []label.KeyValue) ([]label.KeyValue, error) {
+		baggage := otel.Baggage(ctx)
+		kvs = append(baggage.ToSlice(), kvs...)
+		return kvs, nil
+	}
+	p := push.New(
+		checkpointer,
+		exporter,
+		push.WithPeriod(time.Second),
+		push.WithMetricsLabelsEnricher(metricsLabelsEnricher),
+	)
+	meter := p.MeterProvider().Meter("name")
+
+	mock := controllertest.NewMockClock()
+	p.SetClock(mock)
+
+	counter := otel.Must(meter).NewInt64Counter("counter.sum")
+
+	p.Start()
+
+	ctx := otel.ContextWithBaggageValues(context.Background(), label.String("A", "B"))
+	counter.Add(ctx, 1)
+
+	require.EqualValues(t, map[string]float64{}, exporter.Values())
+
+	mock.Add(time.Second)
+	runtime.Gosched()
+
+	require.EqualValues(t, map[string]float64{
+		"counter.sum/A=B/": 1,
+	}, exporter.Values())
+}

sdk/metric/sdk.go

@@ -67,6 +67,9 @@ type (
 
 		// resource is applied to all records in this Accumulator.
 		resource *resource.Resource
+
+		// metricsLabelsEnricher is applied to all records in this Accumulator
+		metricsLabelsEnricher MetricsLabelsEnricher
 	}
 
 	syncInstrument struct {
@@ -139,6 +142,10 @@ type (
 		labels        *label.Set
 		observed      export.Aggregator
 	}
+
+	// MetricsLabelsEnricher can be provided as a config option to enrich metrics labels based on
+	// the context when the metrics are recorded
+	MetricsLabelsEnricher func(context.Context, []label.KeyValue) ([]label.KeyValue, error)
 )
 
 var (
@@ -291,6 +298,9 @@ func (s *syncInstrument) Bind(kvs []label.KeyValue) api.BoundSyncImpl {
 }
 
 func (s *syncInstrument) RecordOne(ctx context.Context, number api.Number, kvs []label.KeyValue) {
+	if s.meter.metricsLabelsEnricher != nil {
+		kvs, _ = s.meter.metricsLabelsEnricher(ctx, kvs)
+	}
 	h := s.acquireHandle(kvs, nil)
 	defer h.Unbind()
 	h.RecordOne(ctx, number)
@@ -305,11 +315,17 @@ func (s *syncInstrument) RecordOne(ctx context.Context, number api.Number, kvs [
 // processor will call Collect() when it receives a request to scrape
 // current metric values.  A push-based processor should configure its
 // own periodic collection.
-func NewAccumulator(processor export.Processor, resource *resource.Resource) *Accumulator {
+func NewAccumulator(processor export.Processor, resource *resource.Resource, opts ...Option) *Accumulator {
+	c := &Config{}
+	for _, opt := range opts {
+		opt.Apply(c)
+	}
+
 	return &Accumulator{
-		processor:        processor,
-		asyncInstruments: internal.NewAsyncInstrumentState(),
-		resource:         resource,
+		processor:             processor,
+		asyncInstruments:      internal.NewAsyncInstrumentState(),
+		resource:              resource,
+		metricsLabelsEnricher: c.MetricsLabelsEnricher,
 	}
 }