Add aggregation package and reader/view options

sdk/metric/aggregation/aggregation.go

@@ -0,0 +1,119 @@
+// 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.
+
+//go:build go1.17
+// +build go1.17
+
+// Package aggregation contains configuration types that define the
+// aggregation operation used to summarizes recorded measurements.
+package aggregation // import "go.opentelemetry.io/otel/sdk/metric/aggregation"
+
+import (
+	"errors"
+	"fmt"
+)
+
+// Aggregation is the aggregation used to summarize recorded measurements.
+type Aggregation struct {
+	// Operation is the kind of operation performed by the aggregation and the
+	// configuration for that operation. This can be Drop, Sum, LastValue, or
+	// ExplicitBucketHistogram.
+	Operation operation
+}
+
+var errAgg = errors.New("aggregation")
+
+// Err returns an error if Aggregation a is invalid, nil otherwise.
+func (a Aggregation) Err() error {
+	if a.Operation == nil {
+		return fmt.Errorf("%w: unset operation", errAgg)
+	}
+	switch v := a.Operation.(type) {
+	case Drop, Sum, LastValue, ExplicitBucketHistogram:
+		if err := v.err(); err != nil {
+			return fmt.Errorf("%w: %v", errAgg, err)
+		}
+		return nil
+	}
+	return fmt.Errorf("%w: unknown %T", errAgg, a.Operation)
+}
+
+// operation is an aggregation operation. The OTel specification does not
+// allow user-defined aggregations, therefore, this is not exported.
+type operation interface {
+	// err returns an error for misconfigured operations.
+	err() error
+}
+
+// Drop drops all data recorded.
+type Drop struct{} // The Drop operation has no parameters.
+
+func (Drop) err() error { return nil }
+
+// Default selects an aggregation based on the instrument type. It will use
+// the following selection mapping: Counter ⇨ Sum, Asynchronous Counter ⇨ Sum,
+// UpDownCounter ⇨ Sum, Asynchronous UpDownCounter ⇨ Sum, Asynchronous Gauge ⇨
+// LastValue, Histogram ⇨ ExplicitBucketHistogram.
+type Default struct{} // The Default operation has no parameters.
+
+func (Default) err() error { return nil }
+
+// Sum summarizes a set of measurements as their arithmetic sum.
+type Sum struct{} // The Sum operation has no parameters.
+
+func (Sum) err() error { return nil }
+
+// LastValues summarizes a set of measurements as the last one made.
+type LastValue struct{} // The LastValue operation has no parameters.
+
+func (LastValue) err() error { return nil }
+
+// ExplicitBucketHistogram summarizes a set of measurements as an histogram
+// with explicitly defined buckets.
+type ExplicitBucketHistogram struct {
+	// Boundaries are the increasing bucket boundary values. Boundary values
+	// define bucket upper bounds. Buckets are exclusive of their lower
+	// boundary and inclusive of their upper bound (except at positive
+	// infinity). A measurement is defined to fall into the greatest-numbered
+	// bucket with a boundary that is greater than or equal to the
+	// measurement. As an example, boundaries defined as:
+	//
+	// []float64{0, 5, 10, 25, 50, 75, 100, 250, 500, 1000}
+	//
+	// Will define these buckets:
+	//
+	// (-∞, 0], (0, 5.0], (5.0, 10.0], (10.0, 25.0], (25.0, 50.0],
+	// (50.0, 75.0], (75.0, 100.0], (100.0, 250.0], (250.0, 500.0],
+	// (500.0, 1000.0], (1000.0, +∞)
+	Boundaries []float64
+	// RecordMinMax indicates whether to record the min and max of the
+	// distribution.
+	RecordMinMax bool
+}
+
+func (h ExplicitBucketHistogram) err() error {
+	// Check boundaries are monotonic.
+	if len(h.Boundaries) <= 1 {
+		return nil
+	}
+
+	i := h.Boundaries[0]
+	for _, j := range h.Boundaries[1:] {
+		if i >= j {
+			return fmt.Errorf("non-monotonic boundaries: %v", h.Boundaries)
+		}
+	}
+
+	return nil
+}

sdk/metric/aggregation/aggregation_test.go

@@ -0,0 +1,79 @@
+// 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.
+
+//go:build go1.17
+// +build go1.17
+
+package aggregation
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+type invalidOperation struct {
+	operation
+}
+
+func TestAggregationErr(t *testing.T) {
+	t.Run("DropOperation", func(t *testing.T) {
+		agg := Aggregation{Operation: Drop{}}
+		assert.NoError(t, agg.Err())
+	})
+
+	t.Run("SumOperation", func(t *testing.T) {
+		agg := Aggregation{Operation: Sum{}}
+		assert.NoError(t, agg.Err())
+	})
+
+	t.Run("LastValueOperation", func(t *testing.T) {
+		agg := Aggregation{Operation: LastValue{}}
+		assert.NoError(t, agg.Err())
+	})
+
+	t.Run("ExplicitBucketHistogramOperation", func(t *testing.T) {
+		agg := Aggregation{Operation: ExplicitBucketHistogram{}}
+		assert.NoError(t, agg.Err())
+
+		agg = Aggregation{Operation: ExplicitBucketHistogram{
+			Boundaries:   []float64{0},
+			RecordMinMax: true,
+		}}
+		assert.NoError(t, agg.Err())
+
+		agg = Aggregation{Operation: ExplicitBucketHistogram{
+			Boundaries:   []float64{0, 5, 10, 25, 50, 75, 100, 250, 500, 1000},
+			RecordMinMax: true,
+		}}
+		assert.NoError(t, agg.Err())
+	})
+
+	t.Run("UnsetOperation", func(t *testing.T) {
+		agg := Aggregation{}
+		assert.ErrorIs(t, agg.Err(), errAgg)
+	})
+
+	t.Run("UnknownOperation", func(t *testing.T) {
+		agg := Aggregation{Operation: invalidOperation{}}
+		assert.ErrorIs(t, agg.Err(), errAgg)
+	})
+
+	t.Run("NonmonotonicHistogramBoundaries", func(t *testing.T) {
+		agg := Aggregation{Operation: ExplicitBucketHistogram{
+			Boundaries: []float64{2, 1},
+		}}
+		assert.ErrorIs(t, agg.Err(), errAgg)
+	})
+}

sdk/metric/aggregation/example_test.go

@@ -0,0 +1,37 @@
+// 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.
+
+//go:build go1.17
+// +build go1.17
+
+package aggregation_test
+
+import "go.opentelemetry.io/otel/sdk/metric/aggregation"
+
+func ExampleAggregation() {
+	// An aggregation that drops measurements.
+	_ = aggregation.Aggregation{Operation: aggregation.Drop{}}
+
+	// An aggregation that sums measurements.
+	_ = aggregation.Aggregation{Operation: aggregation.Sum{}}
+
+	// An aggregation that uses the last value for measurements.
+	_ = aggregation.Aggregation{Operation: aggregation.LastValue{}}
+
+	// An aggregation that bins measurements in a histogram.
+	_ = aggregation.Aggregation{Operation: aggregation.ExplicitBucketHistogram{
+		Boundaries:   []float64{0, 5, 10, 25, 50, 75, 100, 250, 500, 1000},
+		RecordMinMax: true,
+	}}
+}

sdk/metric/manual_reader.go

@@ -24,6 +24,7 @@ import (
 	"sync/atomic"
 
 	"go.opentelemetry.io/otel/internal/global"
+	"go.opentelemetry.io/otel/sdk/metric/aggregation"
 	"go.opentelemetry.io/otel/sdk/metric/export"
 )
 
@@ -103,12 +104,14 @@ func (mr *manualReader) Collect(ctx context.Context) (export.Metrics, error) {
 // manualReaderConfig contains configuration options for a ManualReader.
 type manualReaderConfig struct {
 	temporalitySelector func(InstrumentKind) Temporality
+	aggregationSelector func(InstrumentKind) aggregation.Aggregation
 }
 
 // newManualReaderConfig returns a manualReaderConfig configured with options.
 func newManualReaderConfig(opts []ManualReaderOption) manualReaderConfig {
 	cfg := manualReaderConfig{
 		temporalitySelector: defaultTemporalitySelector,
+		aggregationSelector: defaultAggregationSelector,
 	}
 	for _, opt := range opts {
 		cfg = opt.applyManual(cfg)

sdk/metric/periodic_reader.go

@@ -26,6 +26,7 @@ import (
 
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/internal/global"
+	"go.opentelemetry.io/otel/sdk/metric/aggregation"
 	"go.opentelemetry.io/otel/sdk/metric/export"
 )
 
@@ -40,6 +41,7 @@ type periodicReaderConfig struct {
 	interval            time.Duration
 	timeout             time.Duration
 	temporalitySelector func(InstrumentKind) Temporality
+	aggregationSelector func(InstrumentKind) aggregation.Aggregation
 }
 
 // newPeriodicReaderConfig returns a periodicReaderConfig configured with
@@ -49,6 +51,7 @@ func newPeriodicReaderConfig(options []PeriodicReaderOption) periodicReaderConfi
 		interval:            defaultInterval,
 		timeout:             defaultTimeout,
 		temporalitySelector: defaultTemporalitySelector,
+		aggregationSelector: defaultAggregationSelector,
 	}
 	for _, o := range options {
 		c = o.applyPeriodic(c)

sdk/metric/reader.go

@@ -21,6 +21,7 @@ import (
 	"context"
 	"fmt"
 
+	"go.opentelemetry.io/otel/sdk/metric/aggregation"
 	"go.opentelemetry.io/otel/sdk/metric/export"
 )
 
@@ -139,3 +140,47 @@ func (t temporalitySelectorOption) applyPeriodic(prc periodicReaderConfig) perio
 func defaultTemporalitySelector(InstrumentKind) Temporality {
 	return CumulativeTemporality
 }
+
+// WithAggregation sets the default aggregation a reader will use for an
+// instrument based on the returned value from the selector. If this option is
+// not used, the reader will use the default selector which uses the following
+// selection mapping: Counter ⇨ Sum, Asynchronous Counter ⇨ Sum, UpDownCounter
+// ⇨ Sum, Asynchronous UpDownCounter ⇨ Sum, Asynchronous Gauge ⇨ LastValue,
+// Histogram ⇨ ExplicitBucketHistogram.
+func WithAggregation(selector func(InstrumentKind) aggregation.Aggregation) ReaderOption {
+	return aggregationSelectorOption{selector: selector}
+}
+
+type aggregationSelectorOption struct {
+	selector func(InstrumentKind) aggregation.Aggregation
+}
+
+// applyManual returns a manualReaderConfig with option applied.
+func (t aggregationSelectorOption) applyManual(c manualReaderConfig) manualReaderConfig {
+	c.aggregationSelector = t.selector
+	return c
+}
+
+// applyPeriodic returns a periodicReaderConfig with option applied.
+func (t aggregationSelectorOption) applyPeriodic(c periodicReaderConfig) periodicReaderConfig {
+	c.aggregationSelector = t.selector
+	return c
+}
+
+// defaultAggregationSelector returns the default aggregation measurements
+// from instrument i should be summarized with.
+func defaultAggregationSelector(i InstrumentKind) aggregation.Aggregation {
+	var a aggregation.Aggregation
+	switch i {
+	case SyncCounter, SyncUpDownCounter, AsyncCounter, AsyncUpDownCounter:
+		a.Operation = aggregation.Sum{}
+	case AsyncGauge:
+		a.Operation = aggregation.LastValue{}
+	case SyncHistogram:
+		a.Operation = aggregation.ExplicitBucketHistogram{
+			Boundaries:   []float64{0, 5, 10, 25, 50, 75, 100, 250, 500, 1000},
+			RecordMinMax: true,
+		}
+	}
+	return a
+}

sdk/metric/view/instrument.go

@@ -17,12 +17,16 @@
 
 package view // import "go.opentelemetry.io/otel/sdk/metric/view"
 
-import "go.opentelemetry.io/otel/sdk/instrumentation"
+import (
+	"go.opentelemetry.io/otel/sdk/instrumentation"
+	"go.opentelemetry.io/otel/sdk/metric/aggregation"
+)
 
 // Instrument uniquely identifies an instrument within a meter.
 type Instrument struct {
 	Scope instrumentation.Library
 
 	Name        string
 	Description string
+	Aggregation aggregation.Aggregation
 }

sdk/metric/view/view.go

@@ -24,6 +24,7 @@ import (
 
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/sdk/instrumentation"
+	"go.opentelemetry.io/otel/sdk/metric/aggregation"
 )
 
 // View provides users with the flexibility to customize the metrics that are
@@ -43,7 +44,7 @@ type View struct {
 	filter      attribute.Filter
 	name        string
 	description string
-	// TODO: Aggregation selection
+	agg         aggregation.Aggregation
 }
 
 // New returns a new configured View. If there are any duplicate Options passed,
@@ -80,6 +81,9 @@ func (v View) TransformInstrument(inst Instrument) (transformed Instrument, matc
 	if v.description != "" {
 		inst.Description = v.description
 	}
+	if v.agg.Operation != nil {
+		inst.Aggregation = v.agg
+	}
 	return inst, true
 }
 
@@ -199,4 +203,12 @@ func WithFilterAttributes(keys ...attribute.Key) Option {
 	})
 }
 
-// TODO (#2816): Implement when WithAggregation when Aggregations are defined
+// WithSetAggregation will use the aggregation a for matching instruments. If
+// this option is not provided, the reader defined aggregation for the
+// instrument will be used.
+func WithSetAggregation(a aggregation.Aggregation) Option {
+	return optionFunc(func(v View) View {
+		v.agg = a
+		return v
+	})
+}