Batch Metrics Handler for Mutable State Metrics

common/metrics/batchmetrics.go

@@ -0,0 +1,92 @@
+// The MIT License
+//
+// Copyright (c) 2020 Temporal Technologies Inc.  All rights reserved.
+//
+// Copyright (c) 2020 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+//go:generate mockgen -copyright_file ../../LICENSE -package $GOPACKAGE -source $GOFILE -destination batchmetrics_mock.go
+
+package metrics
+
+import "time"
+
+// The BatchMetrics interfaces provide a way to emit "wide events" within the codebase. If a wide event
+// implementation is not provided we default to the implementation provided below which will maintain
+// backwards compatibility by emitting each field as an individual metric. In custom implementations, fields
+// can be labeled using the metricDefinition.Name() accessor within the respective With() methods.
+
+type (
+	BatchMetricsHandler interface {
+		CreateBatch(string, ...Tag) BatchMetric
+	}
+
+	BatchMetric interface {
+		WithHistogram(histogramDefinition, int64) BatchMetric
+		WithTimer(timerDefinition, time.Duration) BatchMetric
+		WithCounter(counterDefinition, int64) BatchMetric
+		WithGauge(gaugeDefinition, float64) BatchMetric
+		Emit()
+	}
+
+	BatchHandlerImpl struct {
+		metricsHandler Handler
+	}
+	BatchMetricImpl struct {
+		emitter Handler
+	}
+)
+
+func NewBatchMetricsHandler(metricHandler Handler) BatchHandlerImpl {
+	return BatchHandlerImpl{metricHandler}
+}
+
+func (bh BatchHandlerImpl) CreateBatch(_ string, tags ...Tag) BatchMetric {
+	emitter := bh.metricsHandler
+	if len(tags) > 0 {
+		emitter = emitter.WithTags(tags...)
+	}
+	return &BatchMetricImpl{
+		emitter: emitter,
+	}
+}
+
+func (bm *BatchMetricImpl) WithHistogram(def histogramDefinition, value int64) BatchMetric {
+	def.With(bm.emitter).Record(value)
+	return bm
+}
+
+func (bm *BatchMetricImpl) WithTimer(def timerDefinition, value time.Duration) BatchMetric {
+	def.With(bm.emitter).Record(value)
+	return bm
+}
+
+func (bm *BatchMetricImpl) WithCounter(def counterDefinition, value int64) BatchMetric {
+	def.With(bm.emitter).Record(value)
+	return bm
+}
+
+func (bm *BatchMetricImpl) WithGauge(def gaugeDefinition, value float64) BatchMetric {
+	def.With(bm.emitter).Record(value)
+	return bm
+}
+
+// Emit is a no-op in this implementation since we don't actually send wide events.
+func (bm *BatchMetricImpl) Emit() {}

service/history/history_engine.go

@@ -128,6 +128,7 @@ type (
 		eventNotifier              events.Notifier
 		tokenSerializer            common.TaskTokenSerializer
 		metricsHandler             metrics.Handler
+		batchMetricsHandler        metrics.BatchMetricsHandler
 		logger                     log.Logger
 		throttledLogger            log.Logger
 		config                     *configs.Config
@@ -200,6 +201,7 @@ func NewEngineWithShardContext(
 		logger:                     log.With(logger, tag.ComponentHistoryEngine),
 		throttledLogger:            log.With(shard.GetThrottledLogger(), tag.ComponentHistoryEngine),
 		metricsHandler:             shard.GetMetricsHandler(),
+		batchMetricsHandler:        shard.GetBatchMetricsHandler(),
 		eventNotifier:              eventNotifier,
 		config:                     config,
 		sdkClientFactory:           sdkClientFactory,

service/history/shard/context.go

@@ -67,6 +67,7 @@ type (
 		GetLogger() log.Logger
 		GetThrottledLogger() log.Logger
 		GetMetricsHandler() metrics.Handler
+		GetBatchMetricsHandler() metrics.BatchMetricsHandler
 		GetTimeSource() clock.TimeSource
 
 		GetRemoteAdminClient(string) (adminservice.AdminServiceClient, error)

service/history/shard/context_factory.go

@@ -68,6 +68,7 @@ type (
 		HostInfoProvider            membership.HostInfoProvider
 		Logger                      log.Logger
 		MetricsHandler              metrics.Handler
+		BatchMetricsHandler         metrics.BatchMetricsHandler
 		NamespaceRegistry           namespace.Registry
 		PayloadSerializer           serialization.Serializer
 		PersistenceExecutionManager persistence.ExecutionManager
@@ -110,6 +111,7 @@ func (c *contextFactoryImpl) CreateContext(
 		c.ClientBean,
 		c.HistoryClient,
 		c.MetricsHandler,
+		c.BatchMetricsHandler,
 		c.PayloadSerializer,
 		c.TimeSource,
 		c.NamespaceRegistry,

service/history/shard/context_impl.go

@@ -110,6 +110,7 @@ type (
 		stringRepr          string
 		executionManager    persistence.ExecutionManager
 		metricsHandler      metrics.Handler
+		batchMetricsHandler metrics.BatchMetricsHandler
 		eventsCache         events.Cache
 		closeCallback       CloseCallback
 		config              *configs.Config
@@ -2063,6 +2064,7 @@ func newContext(
 	clientBean client.Bean,
 	historyClient historyservice.HistoryServiceClient,
 	metricsHandler metrics.Handler,
+	batchMetricsHandler metrics.BatchMetricsHandler,
 	payloadSerializer serialization.Serializer,
 	timeSource cclock.TimeSource,
 	namespaceRegistry namespace.Registry,
@@ -2097,6 +2099,7 @@ func newContext(
 		stringRepr:              fmt.Sprintf("Shard(%d)", shardID),
 		executionManager:        persistenceExecutionManager,
 		metricsHandler:          metricsHandler,
+		batchMetricsHandler:     batchMetricsHandler,
 		closeCallback:           closeCallback,
 		config:                  historyConfig,
 		finalizer:               finalizer.New(taggedLogger, metricsHandler),
@@ -2196,6 +2199,10 @@ func (s *ContextImpl) GetMetricsHandler() metrics.Handler {
 	return s.metricsHandler
 }
 
+func (s *ContextImpl) GetBatchMetricsHandler() metrics.BatchMetricsHandler {
+	return s.batchMetricsHandler
+}
+
 func (s *ContextImpl) GetTimeSource() cclock.TimeSource {
 	return s.timeSource
 }

service/history/shard/context_testutil.go

@@ -153,6 +153,7 @@ func newTestContext(t *resourcetest.Test, eventsCache events.Cache, config Conte
 		stringRepr:          fmt.Sprintf("Shard(%d)", config.ShardInfo.GetShardId()),
 		executionManager:    executionManager,
 		metricsHandler:      t.MetricsHandler,
+		batchMetricsHandler: metrics.NewBatchMetricsHandler(t.MetricsHandler),
 		eventsCache:         eventsCache,
 		config:              config.Config,
 		contextTaggedLogger: t.GetLogger(),