[receiver/statsd] Add an exponential histogram option (#12951)

Adds a "histogram" option to enable the OTLP v0.11 auto-scaling exponential histogram aggregator for Timing and Histogram measurements.

Co-authored-by: Pablo Baeyens <[email protected]>

.chloggen/expo_statsd_receiver.yaml

@@ -0,0 +1,4 @@
+change_type: enhancement
+component: receiver/statsdreceiver
+note: "Add OTLP exponential histogram aggregator support for high-resolution histogram and timing metrics"
+issues: [5742]

cmd/configschema/go.mod

@@ -271,6 +271,7 @@ require (
 	github.com/kylelemons/godebug v1.1.0 // indirect
 	github.com/leoluk/perflib_exporter v0.2.0 // indirect
 	github.com/lib/pq v1.10.7 // indirect
+	github.com/lightstep/go-expohisto v1.0.0 // indirect
 	github.com/linkedin/goavro/v2 v2.9.8 // indirect
 	github.com/linode/linodego v1.8.0 // indirect
 	github.com/lufia/plan9stats v0.0.0-20220517141722-cf486979b281 // indirect

go.mod

@@ -422,6 +422,7 @@ require (
 	github.com/kylelemons/godebug v1.1.0 // indirect
 	github.com/leoluk/perflib_exporter v0.2.0 // indirect
 	github.com/lib/pq v1.10.7 // indirect
+	github.com/lightstep/go-expohisto v1.0.0 // indirect
 	github.com/linkedin/goavro/v2 v2.9.8 // indirect
 	github.com/linode/linodego v1.8.0 // indirect
 	github.com/lufia/plan9stats v0.0.0-20220517141722-cf486979b281 // indirect

internal/coreinternal/goldendataset/metrics_gen.go

@@ -192,6 +192,7 @@ func setDoubleHistogramBounds(hdp pmetric.HistogramDataPoint, bounds ...float64)
 func addDoubleHistogramVal(hdp pmetric.HistogramDataPoint, val float64) {
 	hdp.SetCount(hdp.Count() + 1)
 	hdp.SetSum(hdp.Sum() + val)
+	// TODO: HasSum, Min, HasMin, Max, HasMax are not covered in tests.
 	buckets := hdp.BucketCounts()
 	bounds := hdp.ExplicitBounds()
 	for i := 0; i < bounds.Len(); i++ {
@@ -225,10 +226,12 @@ func populateExpoHistogram(cfg MetricsCfg, dh pmetric.ExponentialHistogram) {
 		pt.SetTimestamp(ts)
 		populatePtAttributes(cfg, pt.Attributes())
 
-		pt.SetSum(100)
+		pt.SetSum(100 * float64(cfg.PtVal))
 		pt.SetCount(uint64(cfg.PtVal))
-		pt.SetScale(0)
-		pt.SetZeroCount(0)
+		pt.SetScale(int32(cfg.PtVal))
+		pt.SetZeroCount(uint64(cfg.PtVal))
+		pt.SetMin(float64(cfg.PtVal))
+		pt.SetMax(float64(cfg.PtVal))
 		pt.Positive().SetOffset(int32(cfg.PtVal))
 		pt.Positive().BucketCounts().FromRaw([]uint64{uint64(cfg.PtVal)})
 	}

internal/coreinternal/metricstestutil/metric_diff.go

@@ -206,6 +206,7 @@ func diffHistogramPt(
 	diffs = diffMetricAttrs(diffs, expected.Attributes(), actual.Attributes())
 	diffs = diff(diffs, expected.Count(), actual.Count(), "HistogramDataPoint Count")
 	diffs = diff(diffs, expected.Sum(), actual.Sum(), "HistogramDataPoint Sum")
+	// TODO: HasSum, Min, HasMin, Max, HasMax are not covered in tests.
 	diffs = diff(diffs, expected.BucketCounts(), actual.BucketCounts(), "HistogramDataPoint BucketCounts")
 	diffs = diff(diffs, expected.ExplicitBounds(), actual.ExplicitBounds(), "HistogramDataPoint ExplicitBounds")
 	return diffExemplars(diffs, expected.Exemplars(), actual.Exemplars())
@@ -234,7 +235,12 @@ func diffExponentialHistogramPt(
 ) []*MetricDiff {
 	diffs = diffMetricAttrs(diffs, expected.Attributes(), actual.Attributes())
 	diffs = diff(diffs, expected.Count(), actual.Count(), "ExponentialHistogramDataPoint Count")
+	diffs = diff(diffs, expected.HasSum(), actual.HasSum(), "ExponentialHistogramDataPoint HasSum")
+	diffs = diff(diffs, expected.HasMin(), actual.HasMin(), "ExponentialHistogramDataPoint HasMin")
+	diffs = diff(diffs, expected.HasMax(), actual.HasMax(), "ExponentialHistogramDataPoint HasMax")
 	diffs = diff(diffs, expected.Sum(), actual.Sum(), "ExponentialHistogramDataPoint Sum")
+	diffs = diff(diffs, expected.Min(), actual.Min(), "ExponentialHistogramDataPoint Min")
+	diffs = diff(diffs, expected.Max(), actual.Max(), "ExponentialHistogramDataPoint Max")
 	diffs = diff(diffs, expected.ZeroCount(), actual.ZeroCount(), "ExponentialHistogramDataPoint ZeroCount")
 	diffs = diff(diffs, expected.Scale(), actual.Scale(), "ExponentialHistogramDataPoint Scale")
 

internal/coreinternal/metricstestutil/metric_diff_test.go

@@ -84,13 +84,13 @@ func TestAttributes(t *testing.T) {
 
 func TestExponentialHistogram(t *testing.T) {
 	cfg1 := goldendataset.DefaultCfg()
-	cfg1.MetricDescriptorType = pmetric.MetricTypeHistogram
+	cfg1.MetricDescriptorType = pmetric.MetricTypeExponentialHistogram
 	cfg1.PtVal = 1
 	expected := goldendataset.MetricsFromCfg(cfg1)
 	cfg2 := goldendataset.DefaultCfg()
-	cfg2.MetricDescriptorType = pmetric.MetricTypeHistogram
+	cfg2.MetricDescriptorType = pmetric.MetricTypeExponentialHistogram
 	cfg2.PtVal = 3
 	actual := goldendataset.MetricsFromCfg(cfg2)
 	diffs := DiffMetrics(nil, expected, actual)
-	assert.Len(t, diffs, 3)
+	assert.Len(t, diffs, 8)
 }

receiver/statsdreceiver/README.md

@@ -30,9 +30,9 @@ The Following settings are optional:
 
 `"statsd_type"` specifies received Statsd data type. Possible values for this setting are `"timing"`, `"timer"` and `"histogram"`.
 
-`"observer_type"` specifies OTLP data type to convert to. We support `"gauge"` and `"summary"`. For `"gauge"`, it does not perform any aggregation.
-For `"summary`, the statsD receiver will aggregate to one OTLP summary metric for one metric description(the same metric name with the same tags). It will send percentile 0, 10, 50, 90, 95, 100 to the downstream. 
-TODO: Add a new option to use a smoothed summary like Promethetheus: https://github.com/open-telemetry/opentelemetry-collector-contrib/pull/3261 
+`"observer_type"` specifies OTLP data type to convert to. We support `"gauge"`, `"summary"`, and `"histogram"`. For `"gauge"`, it does not perform any aggregation.
+For `"summary`, the statsD receiver will aggregate to one OTLP summary metric for one metric description (the same metric name with the same tags). It will send percentile 0, 10, 50, 90, 95, 100 to the downstream.  The `"histogram"` setting selects an [auto-scaling exponential histogram configured with only a maximum size](https://github.com/lightstep/go-expohisto#readme), as shown in the example below.
+TODO: Add a new option to use a smoothed summary like Prometheus: https://github.com/open-telemetry/opentelemetry-collector-contrib/pull/3261 
 
 Example:
 
@@ -48,7 +48,9 @@ receivers:
       - statsd_type: "histogram"
         observer_type: "gauge"
       - statsd_type: "timing"
-        observer_type: "gauge"
+        observer_type: "histogram"
+        histogram: 
+          max_size: 100
 ```
 
 The full list of settings exposed for this receiver are documented [here](./config.go)
@@ -123,9 +125,11 @@ receivers:
     is_monotonic_counter: false # default
     timer_histogram_mapping:
       - statsd_type: "histogram"
-        observer_type: "gauge"
+        observer_type: "histogram"
+        histogram:
+          max_size: 50
       - statsd_type: "timing"
-        observer_type: "gauge"
+        observer_type: "summary"
 
 exporters:
   file:

receiver/statsdreceiver/config.go

@@ -18,6 +18,7 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/lightstep/go-expohisto/structure"
 	"go.opentelemetry.io/collector/config"
 	"go.opentelemetry.io/collector/config/confignet"
 	"go.uber.org/multierr"
@@ -36,7 +37,6 @@ type Config struct {
 }
 
 func (c *Config) validate() error {
-
 	var errs error
 
 	if c.AggregationInterval <= 0 {
@@ -63,10 +63,22 @@ func (c *Config) validate() error {
 		}
 
 		switch eachMap.ObserverType {
-		case protocol.GaugeObserver, protocol.SummaryObserver:
+		case protocol.GaugeObserver, protocol.SummaryObserver, protocol.HistogramObserver:
 		default:
 			errs = multierr.Append(errs, fmt.Errorf("observer_type is not supported: %s", eachMap.ObserverType))
 		}
+
+		if eachMap.ObserverType == protocol.HistogramObserver {
+			if eachMap.Histogram.MaxSize != 0 && (eachMap.Histogram.MaxSize < structure.MinSize || eachMap.Histogram.MaxSize > structure.MaximumMaxSize) {
+				errs = multierr.Append(errs, fmt.Errorf("histogram max_size out of range: %v", eachMap.Histogram.MaxSize))
+			}
+		} else {
+			// Non-histogram observer w/ histogram config
+			var empty protocol.HistogramConfig
+			if eachMap.Histogram != empty {
+				errs = multierr.Append(errs, fmt.Errorf("histogram configuration requires observer_type: histogram"))
+			}
+		}
 	}
 
 	if TimerHistogramMappingMissingObjectName {

receiver/statsdreceiver/config_test.go

@@ -59,7 +59,10 @@ func TestLoadConfig(t *testing.T) {
 					},
 					{
 						StatsdType:   "timing",
-						ObserverType: "gauge",
+						ObserverType: "histogram",
+						Histogram: protocol.HistogramConfig{
+							MaxSize: 170,
+						},
 					},
 				},
 			},
@@ -153,5 +156,4 @@ func TestValidate(t *testing.T) {
 			require.EqualError(t, test.cfg.validate(), test.expectedErr)
 		})
 	}
-
 }

receiver/statsdreceiver/factory_test.go

@@ -17,15 +17,30 @@ package statsdreceiver
 import (
 	"context"
 	"testing"
+	"time"
 
+	"github.com/lightstep/go-expohisto/structure"
 	"github.com/stretchr/testify/assert"
+	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/config/configtest"
 	"go.opentelemetry.io/collector/consumer/consumertest"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/statsdreceiver/protocol"
 )
 
+type testHost struct {
+	component.Host
+	t *testing.T
+}
+
+// ReportFatalError causes the test to be run to fail.
+func (h *testHost) ReportFatalError(err error) {
+	h.t.Fatalf("receiver reported a fatal error: %v", err)
+}
+
+var _ component.Host = (*testHost)(nil)
+
 func TestCreateDefaultConfig(t *testing.T) {
 	factory := NewFactory()
 	cfg := factory.CreateDefaultConfig()
@@ -61,6 +76,83 @@ func TestCreateReceiverWithConfigErr(t *testing.T) {
 
 }
 
+func TestCreateReceiverWithHistogramConfigError(t *testing.T) {
+	for _, maxSize := range []int32{structure.MaximumMaxSize + 1, -1, -structure.MaximumMaxSize} {
+		cfg := &Config{
+			AggregationInterval: 20 * time.Second,
+			TimerHistogramMapping: []protocol.TimerHistogramMapping{
+				{
+					StatsdType:   "timing",
+					ObserverType: "histogram",
+					Histogram: protocol.HistogramConfig{
+						MaxSize: maxSize,
+					},
+				},
+			},
+		}
+		receiver, err := createMetricsReceiver(
+			context.Background(),
+			componenttest.NewNopReceiverCreateSettings(),
+			cfg,
+			consumertest.NewNop(),
+		)
+		assert.Error(t, err)
+		assert.Contains(t, err.Error(), "histogram max_size out of range")
+		assert.Nil(t, receiver)
+	}
+}
+
+func TestCreateReceiverWithHistogramGoodConfig(t *testing.T) {
+	for _, maxSize := range []int32{structure.MaximumMaxSize, 0, 2} {
+		cfg := &Config{
+			AggregationInterval: 20 * time.Second,
+			TimerHistogramMapping: []protocol.TimerHistogramMapping{
+				{
+					StatsdType:   "timing",
+					ObserverType: "histogram",
+					Histogram: protocol.HistogramConfig{
+						MaxSize: maxSize,
+					},
+				},
+			},
+		}
+		receiver, err := createMetricsReceiver(
+			context.Background(),
+			componenttest.NewNopReceiverCreateSettings(),
+			cfg,
+			consumertest.NewNop(),
+		)
+		assert.NoError(t, err)
+		assert.NotNil(t, receiver)
+		assert.NoError(t, receiver.Start(context.Background(), &testHost{t: t}))
+		assert.NoError(t, receiver.Shutdown(context.Background()))
+	}
+}
+
+func TestCreateReceiverWithInvalidHistogramConfig(t *testing.T) {
+	cfg := &Config{
+		AggregationInterval: 20 * time.Second,
+		TimerHistogramMapping: []protocol.TimerHistogramMapping{
+			{
+				StatsdType:   "timing",
+				ObserverType: "gauge",
+				Histogram: protocol.HistogramConfig{
+					MaxSize: 100,
+				},
+			},
+		},
+	}
+	receiver, err := createMetricsReceiver(
+		context.Background(),
+		componenttest.NewNopReceiverCreateSettings(),
+		cfg,
+		consumertest.NewNop(),
+	)
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "histogram configuration requires observer_type: histogram")
+	assert.Nil(t, receiver)
+}
+
 func TestCreateMetricsReceiverWithNilConsumer(t *testing.T) {
 	receiver, err := createMetricsReceiver(
 		context.Background(),

receiver/statsdreceiver/go.mod

@@ -3,7 +3,9 @@ module github.com/open-telemetry/opentelemetry-collector-contrib/receiver/statsd
 go 1.18
 
 require (
+	github.com/lightstep/go-expohisto v1.0.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/common v0.62.0
+	github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal v0.62.0
 	github.com/stretchr/testify v1.8.0
 	go.opencensus.io v0.23.0
 	go.opentelemetry.io/collector v0.62.0
@@ -16,7 +18,6 @@ require (
 
 require (
 	github.com/davecgh/go-spew v1.1.1 // indirect
-	github.com/fsnotify/fsnotify v1.5.4 // indirect
 	github.com/go-logr/logr v1.2.3 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
@@ -30,20 +31,21 @@ require (
 	github.com/mitchellh/reflectwalk v1.0.2 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
-	github.com/pelletier/go-toml v1.9.4 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	go.opentelemetry.io/otel/metric v0.32.1 // indirect
 	go.opentelemetry.io/otel/sdk v1.10.0 // indirect
 	go.opentelemetry.io/otel/trace v1.10.0 // indirect
 	go.uber.org/atomic v1.10.0 // indirect
 	golang.org/x/exp v0.0.0-20200224162631-6cc2880d07d6 // indirect
-	golang.org/x/net v0.0.0-20220225172249-27dd8689420f // indirect
+	golang.org/x/net v0.0.0-20220624214902-1bab6f366d9e // indirect
 	golang.org/x/sys v0.0.0-20220808155132-1c4a2a72c664 // indirect
 	golang.org/x/text v0.3.7 // indirect
-	google.golang.org/genproto v0.0.0-20211208223120-3a66f561d7aa // indirect
+	google.golang.org/genproto v0.0.0-20220822174746-9e6da59bd2fc // indirect
 	google.golang.org/grpc v1.50.0 // indirect
 	google.golang.org/protobuf v1.28.1 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
 
 replace github.com/open-telemetry/opentelemetry-collector-contrib/internal/common => ../../internal/common
+
+replace github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal => ../../internal/coreinternal