stats/otel: Add up down counter for A94

Author: mbissa

experimental/stats/metricregistry.go

@@ -75,6 +75,7 @@ const (
 	MetricTypeIntHisto
 	MetricTypeFloatHisto
 	MetricTypeIntGauge
+	MetricTypeIntUpDownCount
 )
 
 // Int64CountHandle is a typed handle for a int count metric. This handle
@@ -93,6 +94,23 @@ func (h *Int64CountHandle) Record(recorder MetricsRecorder, incr int64, labels .
 	recorder.RecordInt64Count(h, incr, labels...)
 }
 
+// Int64UpDownCountHandle is a typed handle for an int up-down counter metric.
+// This handle is passed at the recording point in order to know which metric
+// to record on.
+type Int64UpDownCountHandle MetricDescriptor
+
+// Descriptor returns the int64 up-down counter handle typecast to a pointer to a
+// MetricDescriptor.
+func (h *Int64UpDownCountHandle) Descriptor() *MetricDescriptor {
+	return (*MetricDescriptor)(h)
+}
+
+// Record records the int64 up-down counter value on the metrics recorder provided.
+// The value 'v' can be positive to increment or negative to decrement.
+func (h *Int64UpDownCountHandle) Record(recorder MetricsRecorder, v int64, labels ...string) {
+	recorder.RecordInt64UpDownCount(h, v, labels...)
+}
+
 // Float64CountHandle is a typed handle for a float count metric. This handle is
 // passed at the recording point in order to know which metric to record on.
 type Float64CountHandle MetricDescriptor
@@ -249,6 +267,21 @@ func RegisterInt64Gauge(descriptor MetricDescriptor) *Int64GaugeHandle {
 	return (*Int64GaugeHandle)(descPtr)
 }
 
+// RegisterInt64UpDownCounter registers the metric description onto the global registry.
+// It returns a typed handle to use for recording data.
+//
+// NOTE: this function must only be called during initialization time (i.e. in
+// an init() function), and is not thread-safe. If multiple metrics are
+// registered with the same name, this function will panic.
+func RegisterInt64UpDownCount(descriptor MetricDescriptor) *Int64UpDownCountHandle {
+	registerMetric(descriptor.Name, descriptor.Default)
+	// Set the specific metric type for the up-down counter
+	descriptor.Type = MetricTypeIntUpDownCount
+	descPtr := &descriptor
+	metricsRegistry[descriptor.Name] = descPtr
+	return (*Int64UpDownCountHandle)(descPtr)
+}
+
 // snapshotMetricsRegistryForTesting snapshots the global data of the metrics
 // registry. Returns a cleanup function that sets the metrics registry to its
 // original state.

experimental/stats/metricregistry_test.go

@@ -109,6 +109,14 @@ func (s) TestMetricRegistry(t *testing.T) {
 		OptionalLabels: []string{"int gauge optional label"},
 		Default:        false,
 	})
+	intUpDownCountHandle1 := RegisterInt64UpDownCount(MetricDescriptor{
+		Name:           "simple up down counter",
+		Description:    "current number of emissions from tests",
+		Unit:           "int",
+		Labels:         []string{"int up down counter label"},
+		OptionalLabels: []string{"int up down counter optional label"},
+		Default:        false,
+	})
 
 	fmr := newFakeMetricsRecorder(t)
 
@@ -120,6 +128,14 @@ func (s) TestMetricRegistry(t *testing.T) {
 		t.Fatalf("fmr.intValues[intCountHandle1.MetricDescriptor] got %v, want: %v", got, 1)
 	}
 
+	intUpDownCountHandle1.Record(fmr, 2, []string{"some label value", "some optional label value"}...)
+	// The Metric Descriptor in the handle should be able to identify the metric
+	// information. This is the key passed to metrics recorder to identify
+	// metric.
+	if got := fmr.intValues[intUpDownCountHandle1.Descriptor()]; got != 2 {
+		t.Fatalf("fmr.intValues[intUpDownCountHandle1.MetricDescriptor] got %v, want: %v", got, 2)
+	}
+
 	floatCountHandle1.Record(fmr, 1.2, []string{"some label value", "some optional label value"}...)
 	if got := fmr.floatValues[floatCountHandle1.Descriptor()]; got != 1.2 {
 		t.Fatalf("fmr.floatValues[floatCountHandle1.MetricDescriptor] got %v, want: %v", got, 1.2)
@@ -141,6 +157,28 @@ func (s) TestMetricRegistry(t *testing.T) {
 	}
 }
 
+func TestUpDownCounts(t *testing.T) {
+	cleanup := snapshotMetricsRegistryForTesting()
+	defer cleanup()
+
+	intUpDownCountHandle1 := RegisterInt64UpDownCount(MetricDescriptor{
+		Name:           "simple up down counter",
+		Description:    "current number of emissions from tests",
+		Unit:           "int",
+		Labels:         []string{"int up down counter label"},
+		OptionalLabels: []string{"int up down counter optional label"},
+		Default:        false,
+	})
+
+	fmr := newFakeMetricsRecorder(t)
+	intUpDownCountHandle1.Record(fmr, 2, []string{"up down value", "some optional label value"}...)
+	intUpDownCountHandle1.Record(fmr, -1, []string{"up down value", "some optional label value"}...)
+
+	if got := fmr.intValues[intUpDownCountHandle1.Descriptor()]; got != 1 {
+		t.Fatalf("fmr.intValues[intUpDownCountHandle1.MetricDescriptor] got %v, want: %v", got, 1)
+	}
+}
+
 // TestNumerousIntCounts tests numerous int count metrics registered onto the
 // metric registry. A component (simulated by test) should be able to record on
 // the different registered int count metrics.
@@ -265,3 +303,8 @@ func (r *fakeMetricsRecorder) RecordInt64Gauge(handle *Int64GaugeHandle, incr in
 	verifyLabels(r.t, handle.Descriptor().Labels, handle.Descriptor().OptionalLabels, labels)
 	r.intValues[handle.Descriptor()] += incr
 }
+
+func (r *fakeMetricsRecorder) RecordInt64UpDownCount(handle *Int64UpDownCountHandle, incr int64, labels ...string) {
+	verifyLabels(r.t, handle.Descriptor().Labels, handle.Descriptor().OptionalLabels, labels)
+	r.intValues[handle.Descriptor()] += incr
+}

experimental/stats/metrics.go

@@ -38,6 +38,9 @@ type MetricsRecorder interface {
 	// RecordInt64Gauge records the measurement alongside labels on the int
 	// gauge associated with the provided handle.
 	RecordInt64Gauge(handle *Int64GaugeHandle, incr int64, labels ...string)
+	// RecordInt64UpDownCounter records the measurement alongside labels on the int
+	// count associated with the provided handle.
+	RecordInt64UpDownCount(handle *Int64UpDownCountHandle, incr int64, labels ...string)
 }
 
 // Metrics is an experimental legacy alias of the now-stable stats.MetricSet.

internal/stats/metrics_recorder_list.go

@@ -64,6 +64,16 @@ func (l *MetricsRecorderList) RecordInt64Count(handle *estats.Int64CountHandle,
 	}
 }
 
+// RecordInt64UpDownCount records the measurement alongside labels on the int
+// count associated with the provided handle.
+func (l *MetricsRecorderList) RecordInt64UpDownCount(handle *estats.Int64UpDownCountHandle, incr int64, labels ...string) {
+	verifyLabels(handle.Descriptor(), labels...)
+
+	for _, metricRecorder := range l.metricsRecorders {
+		metricRecorder.RecordInt64UpDownCount(handle, incr, labels...)
+	}
+}
+
 // RecordFloat64Count records the measurement alongside labels on the float
 // count associated with the provided handle.
 func (l *MetricsRecorderList) RecordFloat64Count(handle *estats.Float64CountHandle, incr float64, labels ...string) {

internal/testutils/stats/test_metrics_recorder.go

@@ -35,11 +35,12 @@ import (
 // have taken place. It also persists metrics data keyed on the metrics
 // descriptor.
 type TestMetricsRecorder struct {
-	intCountCh   *testutils.Channel
-	floatCountCh *testutils.Channel
-	intHistoCh   *testutils.Channel
-	floatHistoCh *testutils.Channel
-	intGaugeCh   *testutils.Channel
+	intCountCh       *testutils.Channel
+	floatCountCh     *testutils.Channel
+	intHistoCh       *testutils.Channel
+	floatHistoCh     *testutils.Channel
+	intGaugeCh       *testutils.Channel
+	intUpDownCountCh *testutils.Channel
 
 	// mu protects data.
 	mu sync.Mutex
@@ -50,11 +51,12 @@ type TestMetricsRecorder struct {
 // NewTestMetricsRecorder returns a new TestMetricsRecorder.
 func NewTestMetricsRecorder() *TestMetricsRecorder {
 	return &TestMetricsRecorder{
-		intCountCh:   testutils.NewChannelWithSize(10),
-		floatCountCh: testutils.NewChannelWithSize(10),
-		intHistoCh:   testutils.NewChannelWithSize(10),
-		floatHistoCh: testutils.NewChannelWithSize(10),
-		intGaugeCh:   testutils.NewChannelWithSize(10),
+		intCountCh:       testutils.NewChannelWithSize(10),
+		floatCountCh:     testutils.NewChannelWithSize(10),
+		intHistoCh:       testutils.NewChannelWithSize(10),
+		floatHistoCh:     testutils.NewChannelWithSize(10),
+		intGaugeCh:       testutils.NewChannelWithSize(10),
+		intUpDownCountCh: testutils.NewChannelWithSize(10),
 
 		data: make(map[string]float64),
 	}
@@ -135,6 +137,22 @@ func (r *TestMetricsRecorder) RecordInt64Count(handle *estats.Int64CountHandle,
 	r.data[handle.Name] = float64(incr)
 }
 
+// RecordInt64UpDownCount sends the metrics data to the intUpDownCountCh channel and updates
+// the internal data map with the recorded value.
+func (r *TestMetricsRecorder) RecordInt64UpDownCount(handle *estats.Int64UpDownCountHandle, incr int64, labels ...string) {
+	r.intUpDownCountCh.ReceiveOrFail()
+	r.intUpDownCountCh.Send(MetricsData{
+		Handle:    handle.Descriptor(),
+		IntIncr:   incr,
+		LabelKeys: append(handle.Labels, handle.OptionalLabels...),
+		LabelVals: labels,
+	})
+
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.data[handle.Name] = float64(incr)
+}
+
 // WaitForFloat64Count waits for a float count metric to be recorded and
 // verifies that the recorded metrics data matches the expected metricsDataWant.
 // Returns an error if failed to wait or received wrong data.
@@ -294,3 +312,7 @@ func (r *NoopMetricsRecorder) RecordFloat64Histo(*estats.Float64HistoHandle, flo
 
 // RecordInt64Gauge is a noop implementation of RecordInt64Gauge.
 func (r *NoopMetricsRecorder) RecordInt64Gauge(*estats.Int64GaugeHandle, int64, ...string) {}
+
+// RecordInt64UpDownCount is a noop implementation of RecordInt64UpDownCount.
+func (r *NoopMetricsRecorder) RecordInt64UpDownCount(*estats.Int64UpDownCountHandle, int64, ...string) {
+}

stats/opentelemetry/opentelemetry.go

@@ -280,6 +280,18 @@ func createInt64Counter(setOfMetrics map[string]bool, metricName string, meter o
 	return ret
 }
 
+func createInt64UpDownCounter(setOfMetrics map[string]bool, metricName string, meter otelmetric.Meter, options ...otelmetric.Int64UpDownCounterOption) otelmetric.Int64UpDownCounter {
+	if _, ok := setOfMetrics[metricName]; !ok {
+		return noop.Int64UpDownCounter{}
+	}
+	ret, err := meter.Int64UpDownCounter(string(metricName), options...)
+	if err != nil {
+		logger.Errorf("failed to register metric \"%v\", will not record: %v", metricName, err)
+		return noop.Int64UpDownCounter{}
+	}
+	return ret
+}
+
 func createFloat64Counter(setOfMetrics map[string]bool, metricName string, meter otelmetric.Meter, options ...otelmetric.Float64CounterOption) otelmetric.Float64Counter {
 	if _, ok := setOfMetrics[metricName]; !ok {
 		return noop.Float64Counter{}
@@ -350,11 +362,12 @@ func optionFromLabels(labelKeys []string, optionalLabelKeys []string, optionalLa
 // registryMetrics implements MetricsRecorder for the client and server stats
 // handlers.
 type registryMetrics struct {
-	intCounts   map[*estats.MetricDescriptor]otelmetric.Int64Counter
-	floatCounts map[*estats.MetricDescriptor]otelmetric.Float64Counter
-	intHistos   map[*estats.MetricDescriptor]otelmetric.Int64Histogram
-	floatHistos map[*estats.MetricDescriptor]otelmetric.Float64Histogram
-	intGauges   map[*estats.MetricDescriptor]otelmetric.Int64Gauge
+	intCounts       map[*estats.MetricDescriptor]otelmetric.Int64Counter
+	floatCounts     map[*estats.MetricDescriptor]otelmetric.Float64Counter
+	intHistos       map[*estats.MetricDescriptor]otelmetric.Int64Histogram
+	floatHistos     map[*estats.MetricDescriptor]otelmetric.Float64Histogram
+	intGauges       map[*estats.MetricDescriptor]otelmetric.Int64Gauge
+	intUpDownCounts map[*estats.MetricDescriptor]otelmetric.Int64UpDownCounter
 
 	optionalLabels []string
 }
@@ -365,6 +378,7 @@ func (rm *registryMetrics) registerMetrics(metrics *stats.MetricSet, meter otelm
 	rm.intHistos = make(map[*estats.MetricDescriptor]otelmetric.Int64Histogram)
 	rm.floatHistos = make(map[*estats.MetricDescriptor]otelmetric.Float64Histogram)
 	rm.intGauges = make(map[*estats.MetricDescriptor]otelmetric.Int64Gauge)
+	rm.intUpDownCounts = make(map[*estats.MetricDescriptor]otelmetric.Int64UpDownCounter)
 
 	for metric := range metrics.Metrics() {
 		desc := estats.DescriptorForMetric(metric)
@@ -385,6 +399,8 @@ func (rm *registryMetrics) registerMetrics(metrics *stats.MetricSet, meter otelm
 			rm.floatHistos[desc] = createFloat64Histogram(metrics.Metrics(), desc.Name, meter, otelmetric.WithUnit(desc.Unit), otelmetric.WithDescription(desc.Description), otelmetric.WithExplicitBucketBoundaries(desc.Bounds...))
 		case estats.MetricTypeIntGauge:
 			rm.intGauges[desc] = createInt64Gauge(metrics.Metrics(), desc.Name, meter, otelmetric.WithUnit(desc.Unit), otelmetric.WithDescription(desc.Description))
+		case estats.MetricTypeIntUpDownCount:
+			rm.intUpDownCounts[desc] = createInt64UpDownCounter(metrics.Metrics(), desc.Name, meter, otelmetric.WithUnit(desc.Unit), otelmetric.WithDescription(desc.Description))
 		}
 	}
 }
@@ -397,6 +413,14 @@ func (rm *registryMetrics) RecordInt64Count(handle *estats.Int64CountHandle, inc
 	}
 }
 
+func (rm *registryMetrics) RecordInt64UpDownCount(handle *estats.Int64UpDownCountHandle, incr int64, labels ...string) {
+	desc := handle.Descriptor()
+	if ic, ok := rm.intUpDownCounts[desc]; ok {
+		ao := optionFromLabels(desc.Labels, desc.OptionalLabels, rm.optionalLabels, labels...)
+		ic.Add(context.TODO(), incr, ao)
+	}
+}
+
 func (rm *registryMetrics) RecordFloat64Count(handle *estats.Float64CountHandle, incr float64, labels ...string) {
 	desc := handle.Descriptor()
 	if fc, ok := rm.floatCounts[desc]; ok {