Move Aggregation/Temporality selection to the Exporter interface

CHANGELOG.md

@@ -18,6 +18,10 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 - The `"go.opentelemetry.io/otel/sdk/metric".WithReader` option no longer accepts views to associate with the `Reader`.
    Instead, views are now registered directly with the `MeterProvider` via the new `WithView` option.
    The views registered with the `MeterProvider` apply to all `Reader`s. (#3387)
+- The `Temporality(view.InstrumentKind) metricdata.Temporality` and `Aggregation(view.InstrumentKind) aggregation.Aggregation` methods are added to the `"go.opentelemetry.io/otel/sdk/metric".Exporter` interface. (#3260)
+- The `Temporality(view.InstrumentKind) metricdata.Temporality` and `Aggregation(view.InstrumentKind) aggregation.Aggregation` methods are added to the `"go.opentelemetry.io/otel/exporters/otlp/otlpmetric".Client` interface. (#3260)
+- The `WithTemporalitySelector` and `WithAggregationSelector` `ReaderOption`s have been changed to `ManualReaderOption`s in the `go.opentelemetry.io/otel/sdk/metric` package. (#3260)
+- The periodic reader in the `go.opentelemetry.io/otel/sdk/metric` package now uses the temporality and aggregation selectors from its configured exporter instead of accepting them as options. (#3260)
 
 ### Fixed
 

exporters/otlp/otlpmetric/client.go

@@ -17,11 +17,20 @@ package otlpmetric // import "go.opentelemetry.io/otel/exporters/otlp/otlpmetric
 import (
 	"context"
 
+	"go.opentelemetry.io/otel/sdk/metric/aggregation"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata"
+	"go.opentelemetry.io/otel/sdk/metric/view"
 	mpb "go.opentelemetry.io/proto/otlp/metrics/v1"
 )
 
 // Client handles the transmission of OTLP data to an OTLP receiving endpoint.
 type Client interface {
+	// Temporality returns the Temporality to use for an instrument kind.
+	Temporality(view.InstrumentKind) metricdata.Temporality
+
+	// Aggregation returns the Aggregation to use for an instrument kind.
+	Aggregation(view.InstrumentKind) aggregation.Aggregation
+
 	// UploadMetrics transmits metric data to an OTLP receiver.
 	//
 	// All retry logic must be handled by UploadMetrics alone, the Exporter

exporters/otlp/otlpmetric/exporter.go

@@ -21,7 +21,9 @@ import (
 
 	"go.opentelemetry.io/otel/exporters/otlp/otlpmetric/internal/transform"
 	"go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/metric/aggregation"
 	"go.opentelemetry.io/otel/sdk/metric/metricdata"
+	"go.opentelemetry.io/otel/sdk/metric/view"
 	mpb "go.opentelemetry.io/proto/otlp/metrics/v1"
 )
 
@@ -34,6 +36,20 @@ type exporter struct {
 	shutdownOnce sync.Once
 }
 
+// Temporality returns the Temporality to use for an instrument kind.
+func (e *exporter) Temporality(k view.InstrumentKind) metricdata.Temporality {
+	e.clientMu.Lock()
+	defer e.clientMu.Unlock()
+	return e.client.Temporality(k)
+}
+
+// Aggregation returns the Aggregation to use for an instrument kind.
+func (e *exporter) Aggregation(k view.InstrumentKind) aggregation.Aggregation {
+	e.clientMu.Lock()
+	defer e.clientMu.Unlock()
+	return e.client.Aggregation(k)
+}
+
 // Export transforms and transmits metric data to an OTLP receiver.
 func (e *exporter) Export(ctx context.Context, rm metricdata.ResourceMetrics) error {
 	otlpRm, err := transform.ResourceMetrics(rm)
@@ -68,7 +84,10 @@ func (e *exporter) Shutdown(ctx context.Context) error {
 	e.shutdownOnce.Do(func() {
 		e.clientMu.Lock()
 		client := e.client
-		e.client = shutdownClient{}
+		e.client = shutdownClient{
+			temporalitySelector: client.Temporality,
+			aggregationSelector: client.Aggregation,
+		}
 		e.clientMu.Unlock()
 		err = client.Shutdown(ctx)
 	})
@@ -82,7 +101,10 @@ func New(client Client) metric.Exporter {
 	return &exporter{client: client}
 }
 
-type shutdownClient struct{}
+type shutdownClient struct {
+	temporalitySelector metric.TemporalitySelector
+	aggregationSelector metric.AggregationSelector
+}
 
 func (c shutdownClient) err(ctx context.Context) error {
 	if err := ctx.Err(); err != nil {
@@ -91,6 +113,14 @@ func (c shutdownClient) err(ctx context.Context) error {
 	return errShutdown
 }
 
+func (c shutdownClient) Temporality(k view.InstrumentKind) metricdata.Temporality {
+	return c.temporalitySelector(k)
+}
+
+func (c shutdownClient) Aggregation(k view.InstrumentKind) aggregation.Aggregation {
+	return c.aggregationSelector(k)
+}
+
 func (c shutdownClient) UploadMetrics(ctx context.Context, _ *mpb.ResourceMetrics) error {
 	return c.err(ctx)
 }

exporters/otlp/otlpmetric/exporter_test.go

@@ -21,7 +21,10 @@ import (
 
 	"github.com/stretchr/testify/assert"
 
+	"go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/metric/aggregation"
 	"go.opentelemetry.io/otel/sdk/metric/metricdata"
+	"go.opentelemetry.io/otel/sdk/metric/view"
 	mpb "go.opentelemetry.io/proto/otlp/metrics/v1"
 )
 
@@ -31,6 +34,14 @@ type client struct {
 	n int
 }
 
+func (c *client) Temporality(k view.InstrumentKind) metricdata.Temporality {
+	return metric.DefaultTemporalitySelector(k)
+}
+
+func (c *client) Aggregation(k view.InstrumentKind) aggregation.Aggregation {
+	return metric.DefaultAggregationSelector(k)
+}
+
 func (c *client) UploadMetrics(context.Context, *mpb.ResourceMetrics) error {
 	c.n++
 	return nil

exporters/otlp/otlpmetric/internal/oconf/options.go

@@ -27,6 +27,10 @@ import (
 
 	"go.opentelemetry.io/otel/exporters/otlp/internal"
 	"go.opentelemetry.io/otel/exporters/otlp/internal/retry"
+	"go.opentelemetry.io/otel/internal/global"
+	"go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/metric/aggregation"
+	"go.opentelemetry.io/otel/sdk/metric/view"
 )
 
 const (
@@ -57,6 +61,9 @@ type (
 
 		// gRPC configurations
 		GRPCCredentials credentials.TransportCredentials
+
+		TemporalitySelector metric.TemporalitySelector
+		AggregationSelector metric.AggregationSelector
 	}
 
 	Config struct {
@@ -82,6 +89,9 @@ func NewHTTPConfig(opts ...HTTPOption) Config {
 			URLPath:     DefaultMetricsPath,
 			Compression: NoCompression,
 			Timeout:     DefaultTimeout,
+
+			TemporalitySelector: metric.DefaultTemporalitySelector,
+			AggregationSelector: metric.DefaultAggregationSelector,
 		},
 		RetryConfig: retry.DefaultConfig,
 	}
@@ -102,6 +112,9 @@ func NewGRPCConfig(opts ...GRPCOption) Config {
 			URLPath:     DefaultMetricsPath,
 			Compression: NoCompression,
 			Timeout:     DefaultTimeout,
+
+			TemporalitySelector: metric.DefaultTemporalitySelector,
+			AggregationSelector: metric.DefaultAggregationSelector,
 		},
 		RetryConfig: retry.DefaultConfig,
 		DialOptions: []grpc.DialOption{grpc.WithUserAgent(internal.GetUserAgentHeader())},
@@ -313,3 +326,32 @@ func WithTimeout(duration time.Duration) GenericOption {
 		return cfg
 	})
 }
+
+func WithTemporalitySelector(selector metric.TemporalitySelector) GenericOption {
+	return newGenericOption(func(cfg Config) Config {
+		cfg.Metrics.TemporalitySelector = selector
+		return cfg
+	})
+}
+
+func WithAggregationSelector(selector metric.AggregationSelector) GenericOption {
+	// Deep copy and validate before using.
+	wrapped := func(ik view.InstrumentKind) aggregation.Aggregation {
+		a := selector(ik)
+		cpA := a.Copy()
+		if err := cpA.Err(); err != nil {
+			cpA = metric.DefaultAggregationSelector(ik)
+			global.Error(
+				err, "using default aggregation instead",
+				"aggregation", a,
+				"replacement", cpA,
+			)
+		}
+		return cpA
+	}
+
+	return newGenericOption(func(cfg Config) Config {
+		cfg.Metrics.AggregationSelector = wrapped
+		return cfg
+	})
+}

exporters/otlp/otlpmetric/internal/oconf/options_test.go

@@ -23,6 +23,9 @@ import (
 
 	"go.opentelemetry.io/otel/exporters/otlp/internal/envconfig"
 	"go.opentelemetry.io/otel/exporters/otlp/otlpmetric/internal/oconf"
+	"go.opentelemetry.io/otel/sdk/metric/aggregation"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata"
+	"go.opentelemetry.io/otel/sdk/metric/view"
 )
 
 const (
@@ -383,6 +386,38 @@ func TestConfigs(t *testing.T) {
 				assert.Equal(t, c.Metrics.Timeout, 5*time.Second)
 			},
 		},
+
+		// Temporality Selector Tests
+		{
+			name: "WithTemporalitySelector",
+			opts: []oconf.GenericOption{
+				oconf.WithTemporalitySelector(deltaSelector),
+			},
+			asserts: func(t *testing.T, c *oconf.Config, grpcOption bool) {
+				// Function value comparisons are disallowed, test non-default
+				// behavior of a TemporalitySelector here to ensure our "catch
+				// all" was set.
+				var undefinedKind view.InstrumentKind
+				got := c.Metrics.TemporalitySelector
+				assert.Equal(t, metricdata.DeltaTemporality, got(undefinedKind))
+			},
+		},
+
+		// Aggregation Selector Tests
+		{
+			name: "WithAggregationSelector",
+			opts: []oconf.GenericOption{
+				oconf.WithAggregationSelector(dropSelector),
+			},
+			asserts: func(t *testing.T, c *oconf.Config, grpcOption bool) {
+				// Function value comparisons are disallowed, test non-default
+				// behavior of a AggregationSelector here to ensure our "catch
+				// all" was set.
+				var undefinedKind view.InstrumentKind
+				got := c.Metrics.AggregationSelector
+				assert.Equal(t, aggregation.Drop{}, got(undefinedKind))
+			},
+		},
 	}
 
 	for _, tt := range tests {
@@ -406,6 +441,14 @@ func TestConfigs(t *testing.T) {
 	}
 }
 
+func dropSelector(view.InstrumentKind) aggregation.Aggregation {
+	return aggregation.Drop{}
+}
+
+func deltaSelector(view.InstrumentKind) metricdata.Temporality {
+	return metricdata.DeltaTemporality
+}
+
 func asHTTPOptions(opts []oconf.GenericOption) []oconf.HTTPOption {
 	converted := make([]oconf.HTTPOption, len(opts))
 	for i, o := range opts {

exporters/otlp/otlpmetric/internal/otest/client_test.go

@@ -19,6 +19,10 @@ import (
 	"testing"
 
 	"go.opentelemetry.io/otel/exporters/otlp/otlpmetric"
+	"go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/metric/aggregation"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata"
+	"go.opentelemetry.io/otel/sdk/metric/view"
 	cpb "go.opentelemetry.io/proto/otlp/collector/metrics/v1"
 	mpb "go.opentelemetry.io/proto/otlp/metrics/v1"
 )
@@ -27,6 +31,14 @@ type client struct {
 	storage *Storage
 }
 
+func (c *client) Temporality(k view.InstrumentKind) metricdata.Temporality {
+	return metric.DefaultTemporalitySelector(k)
+}
+
+func (c *client) Aggregation(k view.InstrumentKind) aggregation.Aggregation {
+	return metric.DefaultAggregationSelector(k)
+}
+
 func (c *client) Collect() *Storage {
 	return c.storage
 }

exporters/otlp/otlpmetric/otlpmetricgrpc/client.go

@@ -28,6 +28,9 @@ import (
 	"go.opentelemetry.io/otel/exporters/otlp/otlpmetric"
 	"go.opentelemetry.io/otel/exporters/otlp/otlpmetric/internal/oconf"
 	"go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/metric/aggregation"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata"
+	"go.opentelemetry.io/otel/sdk/metric/view"
 	colmetricpb "go.opentelemetry.io/proto/otlp/collector/metrics/v1"
 	metricpb "go.opentelemetry.io/proto/otlp/metrics/v1"
 )
@@ -53,6 +56,9 @@ type client struct {
 	exportTimeout time.Duration
 	requestFunc   retry.RequestFunc
 
+	temporalitySelector metric.TemporalitySelector
+	aggregationSelector metric.AggregationSelector
+
 	// ourConn keeps track of where conn was created: true if created here in
 	// NewClient, or false if passed with an option. This is important on
 	// Shutdown as the conn should only be closed if we created it. Otherwise,
@@ -70,6 +76,9 @@ func newClient(ctx context.Context, options ...Option) (otlpmetric.Client, error
 		exportTimeout: cfg.Metrics.Timeout,
 		requestFunc:   cfg.RetryConfig.RequestFunc(retryable),
 		conn:          cfg.GRPCConn,
+
+		temporalitySelector: cfg.Metrics.TemporalitySelector,
+		aggregationSelector: cfg.Metrics.AggregationSelector,
 	}
 
 	if len(cfg.Metrics.Headers) > 0 {
@@ -94,6 +103,16 @@ func newClient(ctx context.Context, options ...Option) (otlpmetric.Client, error
 	return c, nil
 }
 
+// Temporality returns the Temporality to use for an instrument kind.
+func (c *client) Temporality(k view.InstrumentKind) metricdata.Temporality {
+	return c.temporalitySelector(k)
+}
+
+// Aggregation returns the Aggregation to use for an instrument kind.
+func (c *client) Aggregation(k view.InstrumentKind) aggregation.Aggregation {
+	return c.aggregationSelector(k)
+}
+
 // ForceFlush does nothing, the client holds no state.
 func (c *client) ForceFlush(ctx context.Context) error { return ctx.Err() }
 

exporters/otlp/otlpmetric/otlpmetricgrpc/config.go

@@ -24,6 +24,7 @@ import (
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/exporters/otlp/internal/retry"
 	"go.opentelemetry.io/otel/exporters/otlp/otlpmetric/internal/oconf"
+	"go.opentelemetry.io/otel/sdk/metric"
 )
 
 // Option applies a configuration option to the Exporter.
@@ -236,3 +237,20 @@ func WithTimeout(duration time.Duration) Option {
 func WithRetry(settings RetryConfig) Option {
 	return wrappedOption{oconf.WithRetry(retry.Config(settings))}
 }
+
+// WithTemporalitySelector sets the TemporalitySelector the client will use to
+// determine the Temporality of an instrument based on its kind. If this option
+// is not used, the client will use the DefaultTemporalitySelector from the
+// go.opentelemetry.io/otel/sdk/metric package.
+func WithTemporalitySelector(selector metric.TemporalitySelector) Option {
+	return wrappedOption{oconf.WithTemporalitySelector(selector)}
+}
+
+// WithAggregationSelector sets the AggregationSelector the client will use to
+// determine the aggregation to use for an instrument based on its kind. If
+// this option is not used, the reader will use the DefaultAggregationSelector
+// from the go.opentelemetry.io/otel/sdk/metric package, or the aggregation
+// explicitly passed for a view matching an instrument.
+func WithAggregationSelector(selector metric.AggregationSelector) Option {
+	return wrappedOption{oconf.WithAggregationSelector(selector)}
+}