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)
+- `WithMetricsProcessor` config option is added to the `go.opentelemetry.io/otel/sdk/push` package to allow providing processors that implement `MetricsProcessor` interface, which is
+  added to `go.opentelemetry.io/otel/sdk/metric` package. (#1271)
 
 ### Changed
 

sdk/metric/config.go

@@ -23,6 +23,10 @@ type Config struct {
 	// Resource describes all the metric records processed by the
 	// Accumulator.
 	Resource *resource.Resource
+
+	// MetricProcessors are executed each time a metric is recorded
+	// by the Accumulator's sync instrument implementation
+	MetricsProcessors []MetricsProcessor
 }
 
 // Option is the interface that applies the value to a configuration option.
@@ -43,3 +47,13 @@ type resourceOption struct {
 func (o resourceOption) Apply(config *Config) {
 	config.Resource = o.Resource
 }
+
+func WithMetricsProcessors(processors []MetricsProcessor) Option {
+	return metricsProcessorsOption(processors)
+}
+
+type metricsProcessorsOption []MetricsProcessor
+
+func (p metricsProcessorsOption) Apply(config *Config) {
+	config.MetricsProcessors = p
+}

sdk/metric/controller/push/config.go

@@ -17,6 +17,7 @@ package push
 import (
 	"time"
 
+	"go.opentelemetry.io/otel/sdk/metric"
 	"go.opentelemetry.io/otel/sdk/resource"
 )
 
@@ -33,6 +34,8 @@ type Config struct {
 	// integrate, and export) can last before it is canceled. Defaults to
 	// the controller push period.
 	Timeout time.Duration
+
+	MetricsProcessors []metric.MetricsProcessor
 }
 
 // Option is the interface that applies the value to a configuration option.
@@ -73,3 +76,13 @@ type timeoutOption time.Duration
 func (o timeoutOption) Apply(config *Config) {
 	config.Timeout = time.Duration(o)
 }
+
+func WithMetricsProcessor(p metric.MetricsProcessor) Option {
+	return metricsProcessorOption{p}
+}
+
+type metricsProcessorOption struct{ metric.MetricsProcessor }
+
+func (m metricsProcessorOption) Apply(config *Config) {
+	config.MetricsProcessors = append(config.MetricsProcessors, m.MetricsProcessor)
+}

sdk/metric/controller/push/push.go

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

sdk/metric/sdk.go

@@ -67,6 +67,9 @@ type (
 
 		// resource is applied to all records in this Accumulator.
 		resource *resource.Resource
+
+		// metricsProcessors are applied to all records in this Accumulator
+		metricsProcessors []MetricsProcessor
 	}
 
 	syncInstrument struct {
@@ -123,8 +126,9 @@ type (
 	}
 
 	instrument struct {
-		meter      *Accumulator
-		descriptor otel.Descriptor
+		meter             *Accumulator
+		descriptor        otel.Descriptor
+		metricsProcessors []MetricsProcessor
 	}
 
 	asyncInstrument struct {
@@ -139,6 +143,15 @@ type (
 		labels        *label.Set
 		observed      export.Aggregator
 	}
+
+	// MetricsProcessor can be provided as an config option when creating
+	// a new push controller
+	MetricsProcessor interface {
+		// OnMetricRecorded is execute everytime a metric is recorded by
+		// the sync instrument implementation of an Accumulator, it generally
+		// provides ability to correlate the context with the metrics
+		OnMetricRecorded(context.Context, *[]label.KeyValue)
+	}
 )
 
 var (
@@ -154,6 +167,10 @@ func (inst *instrument) Descriptor() api.Descriptor {
 	return inst.descriptor
 }
 
+func (inst *instrument) getMetricsProcessors() []MetricsProcessor {
+	return inst.metricsProcessors
+}
+
 func (a *asyncInstrument) Implementation() interface{} {
 	return a
 }
@@ -291,6 +308,9 @@ func (s *syncInstrument) Bind(kvs []label.KeyValue) api.BoundSyncImpl {
 }
 
 func (s *syncInstrument) RecordOne(ctx context.Context, number api.Number, kvs []label.KeyValue) {
+	for _, processor := range s.getMetricsProcessors() {
+		processor.OnMetricRecorded(ctx, &kvs)
+	}
 	h := s.acquireHandle(kvs, nil)
 	defer h.Unbind()
 	h.RecordOne(ctx, number)
@@ -312,18 +332,20 @@ func NewAccumulator(processor export.Processor, opts ...Option) *Accumulator {
 	}
 
 	return &Accumulator{
-		processor:        processor,
-		asyncInstruments: internal.NewAsyncInstrumentState(),
-		resource:         c.Resource,
+		asyncInstruments:  internal.NewAsyncInstrumentState(),
+		processor:         processor,
+		resource:          c.Resource,
+		metricsProcessors: c.MetricsProcessors,
 	}
 }
 
 // NewSyncInstrument implements api.MetricImpl.
 func (m *Accumulator) NewSyncInstrument(descriptor api.Descriptor) (api.SyncImpl, error) {
 	return &syncInstrument{
 		instrument: instrument{
-			descriptor: descriptor,
-			meter:      m,
+			descriptor:        descriptor,
+			meter:             m,
+			metricsProcessors: m.metricsProcessors,
 		},
 	}, nil
 }