diff --git a/exporters/otlp/otlpmetric/internal/oconf/options.go b/exporters/otlp/otlpmetric/internal/oconf/options.go index f5a82d6db17..1cf9131e203 100644 --- a/exporters/otlp/otlpmetric/internal/oconf/options.go +++ b/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, } @@ -312,3 +325,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 + }) +} diff --git a/exporters/otlp/otlpmetric/otlpmetricgrpc/client.go b/exporters/otlp/otlpmetric/otlpmetricgrpc/client.go index 4c5beb8f384..0f1a0040f61 100644 --- a/exporters/otlp/otlpmetric/otlpmetricgrpc/client.go +++ b/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() } diff --git a/exporters/otlp/otlpmetric/otlpmetricgrpc/config.go b/exporters/otlp/otlpmetric/otlpmetricgrpc/config.go index 5f7a66e559e..3b9539c7187 100644 --- a/exporters/otlp/otlpmetric/otlpmetricgrpc/config.go +++ b/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)} +} diff --git a/exporters/otlp/otlpmetric/otlpmetrichttp/client.go b/exporters/otlp/otlpmetric/otlpmetrichttp/client.go index 7a8d7e14707..04302356056 100644 --- a/exporters/otlp/otlpmetric/otlpmetrichttp/client.go +++ b/exporters/otlp/otlpmetric/otlpmetrichttp/client.go @@ -33,6 +33,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" ) @@ -54,6 +57,9 @@ type client struct { compression Compression requestFunc retry.RequestFunc httpClient *http.Client + + temporalitySelector metric.TemporalitySelector + aggregationSelector metric.AggregationSelector } // Keep it in sync with golang's DefaultTransport from net/http! We @@ -113,9 +119,22 @@ func newClient(opts ...Option) (otlpmetric.Client, error) { req: req, requestFunc: cfg.RetryConfig.RequestFunc(evaluate), httpClient: httpClient, + + temporalitySelector: cfg.Metrics.TemporalitySelector, + aggregationSelector: cfg.Metrics.AggregationSelector, }, 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() } diff --git a/exporters/otlp/otlpmetric/otlpmetrichttp/config.go b/exporters/otlp/otlpmetric/otlpmetrichttp/config.go index c320c25d406..e9f7c774642 100644 --- a/exporters/otlp/otlpmetric/otlpmetrichttp/config.go +++ b/exporters/otlp/otlpmetric/otlpmetrichttp/config.go @@ -20,6 +20,7 @@ import ( "go.opentelemetry.io/otel/exporters/otlp/internal/retry" "go.opentelemetry.io/otel/exporters/otlp/otlpmetric/internal/oconf" + "go.opentelemetry.io/otel/sdk/metric" ) // Compression describes the compression used for payloads sent to the @@ -179,3 +180,20 @@ func WithTimeout(duration time.Duration) Option { func WithRetry(rc RetryConfig) Option { return wrappedOption{oconf.WithRetry(retry.Config(rc))} } + +// 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)} +}