Exponential histogram aggregator & export support

CHANGELOG.md

@@ -11,6 +11,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 ### Added
 
 - Support `OTEL_EXPORTER_ZIPKIN_ENDPOINT` env to specify zipkin collector endpoint (#2490)
+- Add an exponential histogram aggregator, a corresponding aggregation.Kind, and OTLP exporter support. (TODO)
 
 ### Changed
 

exporters/otlp/otlpmetric/internal/metrictransform/metric.go

@@ -246,6 +246,13 @@ func Record(temporalitySelector aggregation.TemporalitySelector, r export.Record
 		}
 		return histogramPoint(r, temporalitySelector.TemporalityFor(r.Descriptor(), aggregation.HistogramKind), h)
 
+	case aggregation.ExponentialHistogramKind:
+		h, ok := agg.(aggregation.ExponentialHistogram)
+		if !ok {
+			return nil, fmt.Errorf("%w: %T", ErrIncompatibleAgg, agg)
+		}
+		return exponentialHistogramPoint(r, temporalitySelector.TemporalityFor(r.Descriptor(), aggregation.ExponentialHistogramKind), h)
+
 	case aggregation.SumKind:
 		s, ok := agg.(aggregation.Sum)
 		if !ok {
@@ -438,3 +445,75 @@ func histogramPoint(record export.Record, temporality aggregation.Temporality, a
 	}
 	return m, nil
 }
+
+// exponentialHistogramPoint transforms an ExponentialHistogram Aggregator into an OTLP Metric.
+func exponentialHistogramPoint(record export.Record, temporality aggregation.Temporality, a aggregation.ExponentialHistogram) (*metricpb.Metric, error) {
+	desc := record.Descriptor()
+	labels := record.Labels()
+
+	count, err := a.Count()
+	if err != nil {
+		return nil, err
+	}
+
+	sum, err := a.Sum()
+	if err != nil {
+		return nil, err
+	}
+
+	zeros, err := a.ZeroCount()
+	if err != nil {
+		return nil, err
+	}
+
+	scale, err := a.Scale()
+	if err != nil {
+		return nil, err
+	}
+
+	point := &metricpb.ExponentialHistogramDataPoint{
+		Attributes:        Iterator(labels.Iter()),
+		StartTimeUnixNano: toNanos(record.StartTime()),
+		TimeUnixNano:      toNanos(record.EndTime()),
+		Count:             count,
+		Sum:               sum.CoerceToFloat64(desc.NumberKind()),
+		ZeroCount:         zeros,
+		Scale:             scale, // Note: will be zero if all zeros.
+	}
+
+	if pos, err := a.Positive(); pos != nil && err == nil && pos.Len() != 0 {
+		point.Positive = exponentialHistogramBucket(pos)
+	} else if err != nil {
+		return nil, err
+	}
+
+	if neg, err := a.Negative(); neg != nil && err == nil && neg.Len() != 0 {
+		point.Negative = exponentialHistogramBucket(neg)
+	} else if err != nil {
+		return nil, err
+	}
+
+	m := &metricpb.Metric{
+		Name:        desc.Name(),
+		Description: desc.Description(),
+		Unit:        string(desc.Unit()),
+		Data: &metricpb.Metric_ExponentialHistogram{
+			ExponentialHistogram: &metricpb.ExponentialHistogram{
+				AggregationTemporality: sdkTemporalityToTemporality(temporality),
+				DataPoints:             []*metricpb.ExponentialHistogramDataPoint{point},
+			},
+		},
+	}
+	return m, nil
+}
+
+func exponentialHistogramBucket(b aggregation.ExponentialBuckets) *metricpb.ExponentialHistogramDataPoint_Buckets {
+	cnts := make([]uint64, b.Len())
+	for i := range cnts {
+		cnts[i] = b.At(uint32(i))
+	}
+	return &metricpb.ExponentialHistogramDataPoint_Buckets{
+		Offset:       b.Offset(),
+		BucketCounts: cnts,
+	}
+}

exporters/otlp/otlpmetric/internal/metrictransform/metric_test.go

@@ -29,6 +29,7 @@ import (
 	"go.opentelemetry.io/otel/metric/number"
 	"go.opentelemetry.io/otel/metric/sdkapi"
 	"go.opentelemetry.io/otel/sdk/metric/aggregator"
+	"go.opentelemetry.io/otel/sdk/metric/aggregator/exponential"
 	"go.opentelemetry.io/otel/sdk/metric/aggregator/lastvalue"
 	"go.opentelemetry.io/otel/sdk/metric/aggregator/sum"
 	"go.opentelemetry.io/otel/sdk/metric/export"
@@ -321,3 +322,167 @@ func TestRecordAggregatorUnexpectedErrors(t *testing.T) {
 	require.Nil(t, mpb)
 	require.True(t, errors.Is(err, errEx))
 }
+
+func TestExponentialHistogramDataPoints(t *testing.T) {
+	type testCase struct {
+		name        string
+		values      []float64
+		temporality aggregation.Temporality
+		numberKind  number.Kind
+		expect      *metricpb.ExponentialHistogram
+	}
+	useAttrs := []attribute.KeyValue{
+		attribute.String("one", "1"),
+	}
+	expectAttrs := []*commonpb.KeyValue{
+		{Key: "one", Value: &commonpb.AnyValue{Value: &commonpb.AnyValue_StringValue{StringValue: "1"}}},
+	}
+
+	for _, test := range []testCase{
+		{
+			"empty",
+			[]float64{},
+			aggregation.DeltaTemporality,
+			number.Float64Kind,
+			&metricpb.ExponentialHistogram{
+				AggregationTemporality: otelDelta,
+				DataPoints: []*metricpb.ExponentialHistogramDataPoint{{
+					Attributes:        expectAttrs,
+					StartTimeUnixNano: uint64(intervalStart.UnixNano()),
+					TimeUnixNano:      uint64(intervalEnd.UnixNano()),
+					Count:             0,
+					ZeroCount:         0,
+					Sum:               0,
+				}},
+			},
+		},
+		{
+			"positive",
+			[]float64{1, 2, 4, 8},
+			aggregation.DeltaTemporality,
+			number.Float64Kind,
+			&metricpb.ExponentialHistogram{
+				AggregationTemporality: otelDelta,
+				DataPoints: []*metricpb.ExponentialHistogramDataPoint{{
+					Attributes:        expectAttrs,
+					StartTimeUnixNano: uint64(intervalStart.UnixNano()),
+					TimeUnixNano:      uint64(intervalEnd.UnixNano()),
+
+					// 1..8 spans 3 orders of magnitide, max-size 2, thus scale=-1
+					Scale:     -1,
+					Count:     4,
+					ZeroCount: 0,
+					Sum:       15,
+					Positive: &metricpb.ExponentialHistogramDataPoint_Buckets{
+						Offset:       0,
+						BucketCounts: []uint64{2, 2},
+					},
+				}},
+			},
+		},
+		{
+			"positive_and_negative",
+			[]float64{2, 3, -100},
+			aggregation.DeltaTemporality,
+			number.Float64Kind,
+			&metricpb.ExponentialHistogram{
+				AggregationTemporality: otelDelta,
+				DataPoints: []*metricpb.ExponentialHistogramDataPoint{{
+					Attributes:        expectAttrs,
+					StartTimeUnixNano: uint64(intervalStart.UnixNano()),
+					TimeUnixNano:      uint64(intervalEnd.UnixNano()),
+
+					// Scale 1 has boundaries at 1, sqrt(2), 2, 2*sqrt(2), ...
+					Scale:     1,
+					Count:     3,
+					ZeroCount: 0,
+					Sum:       -95,
+					Positive: &metricpb.ExponentialHistogramDataPoint_Buckets{
+						// Index 2 => 2, Index 3 => 2*sqrt(2)
+						Offset:       2,
+						BucketCounts: []uint64{1, 1},
+					},
+					Negative: &metricpb.ExponentialHistogramDataPoint_Buckets{
+						// Index 13 = 2^floor(13/2) * sqrt(2) ~= 90
+						Offset:       13,
+						BucketCounts: []uint64{1},
+					},
+				}},
+			},
+		},
+		{
+			// Note: (2**(2**-10))**-100 = 0.9345
+			// Note: (2**(2**-10))**-101 = 0.9339
+			// Note: (2**(2**-10))**-102 = 0.9333
+			"negative and zero",
+			[]float64{-0.9343, -0.9342, -0.9341, -0.9338, -0.9337, -0.9336, 0, 0, 0, 0},
+			aggregation.DeltaTemporality,
+			number.Float64Kind,
+			&metricpb.ExponentialHistogram{
+				AggregationTemporality: otelDelta,
+				DataPoints: []*metricpb.ExponentialHistogramDataPoint{{
+					Attributes:        expectAttrs,
+					StartTimeUnixNano: uint64(intervalStart.UnixNano()),
+					TimeUnixNano:      uint64(intervalEnd.UnixNano()),
+
+					// Scale 1 has boundaries at 1, sqrt(2), 2, 2*sqrt(2), ...
+					Scale:     10,
+					Count:     10,
+					ZeroCount: 4,
+					Sum:       -0.9343 + -0.9342 + -0.9341 + -0.9338 + -0.9337 + -0.9336,
+					Negative: &metricpb.ExponentialHistogramDataPoint_Buckets{
+						Offset:       -102,
+						BucketCounts: []uint64{3, 3},
+					},
+				}},
+			},
+		},
+		{
+			"integers cumulative",
+			// Scale=-1 has base 4,
+			// index 0 holds values [1, 4), has count 2
+			// index 1 holds values [4, 15), has count 12
+			[]float64{2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15},
+			aggregation.CumulativeTemporality,
+			number.Float64Kind,
+			&metricpb.ExponentialHistogram{
+				AggregationTemporality: otelCumulative,
+				DataPoints: []*metricpb.ExponentialHistogramDataPoint{{
+					Attributes:        expectAttrs,
+					StartTimeUnixNano: uint64(intervalStart.UnixNano()),
+					TimeUnixNano:      uint64(intervalEnd.UnixNano()),
+					Count:             14,
+					ZeroCount:         0,
+					Sum:               119,
+					Scale:             -1,
+					Positive: &metricpb.ExponentialHistogramDataPoint_Buckets{
+						Offset:       0,
+						BucketCounts: []uint64{2, 12},
+					},
+				}},
+			},
+		},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			desc := metrictest.NewDescriptor("ignore", sdkapi.HistogramInstrumentKind, test.numberKind)
+			labels := attribute.NewSet(useAttrs...)
+			agg := &exponential.New(1, &desc, exponential.WithMaxSize(2))[0]
+
+			for _, value := range test.values {
+				var num number.Number
+				if test.numberKind == number.Float64Kind {
+					num = number.NewFloat64Number(value)
+				} else {
+					num = number.NewInt64Number(int64(value))
+				}
+				assert.NoError(t, agg.Update(context.Background(), num, &desc))
+			}
+
+			record := export.NewRecord(&desc, &labels, agg, intervalStart, intervalEnd)
+
+			if m, err := exponentialHistogramPoint(record, test.temporality, agg); assert.NoError(t, err) {
+				assert.Equal(t, test.expect, m.GetExponentialHistogram())
+			}
+		})
+	}
+}

sdk/export/metric/aggregation/aggregation_test.go

@@ -0,0 +1,27 @@
+// 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 aggregation
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestAggregationKind(t *testing.T) {
+	require.Equal(t, "Sum", SumKind.String())
+	require.Equal(t, "Lastvalue", LastValueKind.String())
+	require.Equal(t, "Histogram", HistogramKind.String())
+}

sdk/export/metric/go.mod

@@ -42,6 +42,7 @@ replace go.opentelemetry.io/otel/sdk/metric => ../../metric
 replace go.opentelemetry.io/otel/trace => ../../../trace
 
 require (
+	github.com/stretchr/testify v1.7.0
 	go.opentelemetry.io/otel v1.3.0
 	go.opentelemetry.io/otel/metric v0.26.0
 	go.opentelemetry.io/otel/sdk/metric v0.0.0-00010101000000-000000000000

sdk/export/metric/go.sum

@@ -15,6 +15,7 @@ github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/
 golang.org/x/sys v0.0.0-20210423185535-09eb48e85fd7 h1:iGu644GcxtEcrInvDsQRCwJjtCIOlT2V7IRt6ah2Whw=
 golang.org/x/sys v0.0.0-20210423185535-09eb48e85fd7/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=