Batch Mutable State Metrics (#6655)

## What changed?
Extend metrics.Handler interface to include a batching feature which
enables implementations to send "wide events". Batch the mutable state metrics.

## Why?
Send a single metrics "event" where possible rather than several.

## How did you test it?
Unit tests.

## Potential risks
None expected.

## Documentation
Nothing in `docs/` to update. Release notes to come.

## Is hotfix candidate?
No

Author: njo

common/metrics/metrics.go

@@ -27,6 +27,7 @@
 package metrics
 
 import (
+	"io"
 	"time"
 
 	"go.temporal.io/server/common/log"
@@ -58,6 +59,15 @@ type (
 		Histogram(string, MetricUnit) HistogramIface
 
 		Stop(log.Logger)
+
+		// StartBatch returns a BatchHandler that can emit a series of metrics as a single "wide event".
+		// If wide events aren't supported in the underlying implementation, metrics can still be sent individually.
+		StartBatch(string) BatchHandler
+	}
+
+	BatchHandler interface {
+		Handler
+		io.Closer
 	}
 
 	// CounterIface is an ever-increasing counter.
@@ -66,7 +76,7 @@ type (
 		// Tags provided are merged with the source MetricsHandler
 		Record(int64, ...Tag)
 	}
-	// GaugeIface can be set to any float and repesents a latest value instrument.
+	// GaugeIface can be set to any float and represents a latest value instrument.
 	GaugeIface interface {
 		// Record updates the gauge value.
 		// Tags provided are merged with the source MetricsHandler

common/metrics/metrics_mock.go

@@ -142,5 +142,13 @@ func (c *CaptureHandler) Histogram(name string, unit metrics.MetricUnit) metrics
 	return metrics.HistogramFunc(func(v int64, tags ...metrics.Tag) { c.record(name, v, unit, tags...) })
 }
 
+func (c *CaptureHandler) Close() error {
+	return nil
+}
+
+func (c *CaptureHandler) StartBatch(_ string) metrics.BatchHandler {
+	return c
+}
+
 // Stop implements [metrics.Handler.Stop].
 func (*CaptureHandler) Stop(log.Logger) {}

common/metrics/metricstest/capture_handler.go

@@ -68,6 +68,14 @@ func (*noopMetricsHandler) Histogram(string, MetricUnit) HistogramIface {
 
 func (*noopMetricsHandler) Stop(log.Logger) {}
 
+func (*noopMetricsHandler) Close() error {
+	return nil
+}
+
+func (n *noopMetricsHandler) StartBatch(_ string) BatchHandler {
+	return n
+}
+
 var NoopCounterMetricFunc = CounterFunc(func(i int64, t ...Tag) {})
 var NoopGaugeMetricFunc = GaugeFunc(func(f float64, t ...Tag) {})
 var NoopTimerMetricFunc = TimerFunc(func(d time.Duration, t ...Tag) {})

common/metrics/noop_impl.go

@@ -204,6 +204,14 @@ func (omp *otelMetricsHandler) Stop(l log.Logger) {
 	omp.provider.Stop(l)
 }
 
+func (omp *otelMetricsHandler) Close() error {
+	return nil
+}
+
+func (omp *otelMetricsHandler) StartBatch(_ string) BatchHandler {
+	return omp
+}
+
 // makeSet returns an otel attribute.Set with the given tags merged with the
 // otelMetricsHandler's tags.
 func (omp *otelMetricsHandler) makeSet(tags []Tag) attribute.Set {

common/metrics/otel_metrics_handler.go

@@ -66,60 +66,68 @@ func NewTallyMetricsHandler(cfg ClientConfig, scope tally.Scope) *tallyMetricsHa
 
 // WithTags creates a new MetricProvder with provided []Tag
 // Tags are merged with registered Tags from the source MetricsHandler
-func (tmp *tallyMetricsHandler) WithTags(tags ...Tag) Handler {
+func (tmh *tallyMetricsHandler) WithTags(tags ...Tag) Handler {
 	return &tallyMetricsHandler{
-		scope:          tmp.scope.Tagged(tagsToMap(tags, tmp.excludeTags)),
-		perUnitBuckets: tmp.perUnitBuckets,
-		excludeTags:    tmp.excludeTags,
+		scope:          tmh.scope.Tagged(tagsToMap(tags, tmh.excludeTags)),
+		perUnitBuckets: tmh.perUnitBuckets,
+		excludeTags:    tmh.excludeTags,
 	}
 }
 
 // Counter obtains a counter for the given name.
-func (tmp *tallyMetricsHandler) Counter(counter string) CounterIface {
+func (tmh *tallyMetricsHandler) Counter(counter string) CounterIface {
 	return CounterFunc(func(i int64, t ...Tag) {
-		scope := tmp.scope
+		scope := tmh.scope
 		if len(t) > 0 {
-			scope = tmp.scope.Tagged(tagsToMap(t, tmp.excludeTags))
+			scope = tmh.scope.Tagged(tagsToMap(t, tmh.excludeTags))
 		}
 		scope.Counter(counter).Inc(i)
 	})
 }
 
 // Gauge obtains a gauge for the given name.
-func (tmp *tallyMetricsHandler) Gauge(gauge string) GaugeIface {
+func (tmh *tallyMetricsHandler) Gauge(gauge string) GaugeIface {
 	return GaugeFunc(func(f float64, t ...Tag) {
-		scope := tmp.scope
+		scope := tmh.scope
 		if len(t) > 0 {
-			scope = tmp.scope.Tagged(tagsToMap(t, tmp.excludeTags))
+			scope = tmh.scope.Tagged(tagsToMap(t, tmh.excludeTags))
 		}
 		scope.Gauge(gauge).Update(f)
 	})
 }
 
 // Timer obtains a timer for the given name.
-func (tmp *tallyMetricsHandler) Timer(timer string) TimerIface {
+func (tmh *tallyMetricsHandler) Timer(timer string) TimerIface {
 	return TimerFunc(func(d time.Duration, t ...Tag) {
-		scope := tmp.scope
+		scope := tmh.scope
 		if len(t) > 0 {
-			scope = tmp.scope.Tagged(tagsToMap(t, tmp.excludeTags))
+			scope = tmh.scope.Tagged(tagsToMap(t, tmh.excludeTags))
 		}
 		scope.Timer(timer).Record(d)
 	})
 }
 
 // Histogram obtains a histogram for the given name.
-func (tmp *tallyMetricsHandler) Histogram(histogram string, unit MetricUnit) HistogramIface {
+func (tmh *tallyMetricsHandler) Histogram(histogram string, unit MetricUnit) HistogramIface {
 	return HistogramFunc(func(i int64, t ...Tag) {
-		scope := tmp.scope
+		scope := tmh.scope
 		if len(t) > 0 {
-			scope = tmp.scope.Tagged(tagsToMap(t, tmp.excludeTags))
+			scope = tmh.scope.Tagged(tagsToMap(t, tmh.excludeTags))
 		}
-		scope.Histogram(histogram, tmp.perUnitBuckets[unit]).RecordValue(float64(i))
+		scope.Histogram(histogram, tmh.perUnitBuckets[unit]).RecordValue(float64(i))
 	})
 }
 
 func (*tallyMetricsHandler) Stop(log.Logger) {}
 
+func (*tallyMetricsHandler) Close() error {
+	return nil
+}
+
+func (tmh *tallyMetricsHandler) StartBatch(_ string) BatchHandler {
+	return tmh
+}
+
 func tagsToMap(t1 []Tag, e excludeTags) map[string]string {
 	if len(t1) == 0 {
 		return nil

common/metrics/tally_metrics_handler.go

@@ -66,6 +66,14 @@ func (h StubHandler) Histogram(_ string, _ metrics.MetricUnit) metrics.Histogram
 
 func (h StubHandler) Stop(_ log.Logger) {}
 
+func (h StubHandler) Close() error {
+	return nil
+}
+
+func (h StubHandler) StartBatch(_ string) metrics.BatchHandler {
+	return h
+}
+
 func TestHistoryBuilder_IsDirty(t *testing.T) {
 	hb := HistoryBuilder{EventStore: EventStore{}}
 	if hb.IsDirty() {