Hot shard metrics

common/aggregate/bench_moving_window_avg_test.go

@@ -0,0 +1,49 @@
+// 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.
+
+package aggregate
+
+import (
+	"math/rand"
+	"testing"
+	"time"
+)
+
+// BenchmarkArrayMovingWindowAvg
+// BenchmarkArrayMovingWindowAvg-10    	17021074	        66.27 ns/op
+
+const (
+	testWindowSize = 10 * time.Millisecond
+	testBufferSize = 200
+)
+
+func BenchmarkArrayMovingWindowAvg(b *testing.B) {
+	avg := NewMovingWindowAvgImpl(testWindowSize, testBufferSize)
+	for i := 0; i < b.N; i++ {
+		avg.Record(rand.Int63())
+		if i%10 == 0 {
+			avg.Average()
+		}
+	}
+}

common/aggregate/moving_window_average.go

@@ -0,0 +1,103 @@
+// 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.
+
+package aggregate
+
+import (
+	"sync"
+	"time"
+)
+
+type (
+	MovingWindowAverage interface {
+		Record(val int64)
+		Average() float64
+	}
+
+	timestampedData struct {
+		value     int64
+		timestamp time.Time
+	}
+
+	MovingWindowAvgImpl struct {
+		sync.Mutex
+		windowSize    time.Duration
+		maxBufferSize int
+		buffer        []timestampedData
+		headIdx       int
+		tailIdx       int
+		sum           int64
+		count         int64
+	}
+)
+
+func NewMovingWindowAvgImpl(
+	windowSize time.Duration,
+	maxBufferSize int,
+) *MovingWindowAvgImpl {
+	return &MovingWindowAvgImpl{
+		windowSize:    windowSize,
+		maxBufferSize: maxBufferSize,
+		buffer:        make([]timestampedData, maxBufferSize),
+	}
+}
+
+func (a *MovingWindowAvgImpl) Record(val int64) {
+	a.Lock()
+	defer a.Unlock()
+
+	a.buffer[a.tailIdx] = timestampedData{timestamp: time.Now(), value: val}
+	a.tailIdx = (a.tailIdx + 1) % a.maxBufferSize
+
+	a.sum += val
+	a.count++
+
+	if a.tailIdx == a.headIdx {
+		// buffer full, expire oldest element
+		a.sum -= a.buffer[a.headIdx].value
+		a.count--
+		a.headIdx = (a.headIdx + 1) % a.maxBufferSize
+	}
+}
+
+func (a *MovingWindowAvgImpl) Average() float64 {
+	a.Lock()
+	defer a.Unlock()
+
+	a.expireOldValuesLocked()
+	if a.count == 0 {
+		return 0
+	}
+	return float64(a.sum) / float64(a.count)
+}
+
+func (a *MovingWindowAvgImpl) expireOldValuesLocked() {
+	for ; a.headIdx != a.tailIdx; a.headIdx = (a.headIdx + 1) % a.maxBufferSize {
+		if time.Since(a.buffer[a.headIdx].timestamp) < a.windowSize {
+			break
+		}
+		a.sum -= a.buffer[a.headIdx].value
+		a.count--
+	}
+}

common/dynamicconfig/constants.go

@@ -106,6 +106,12 @@ const (
 	EnableEagerWorkflowStart = "system.enableEagerWorkflowStart"
 	// NamespaceCacheRefreshInterval is the key for namespace cache refresh interval dynamic config
 	NamespaceCacheRefreshInterval = "system.namespaceCacheRefreshInterval"
+	// PersistenceHealthSignalWindowSize is the time window size in seconds for aggregating persistence signals
+	PersistenceHealthSignalWindowSize = "system.persistenceHealthSignalWindowSize"
+	// PersistenceHealthSignalBufferSize is the maximum number of persistence signals to buffer in memory per signal key
+	PersistenceHealthSignalBufferSize = "system.persistenceHealthSignalBufferSize"
+	// ShardRPSWarnLimit is the per-shard RPS limit for warning
+	ShardRPSWarnLimit = "system.shardRPSWarnLimit"
 
 	// Whether the deadlock detector should dump goroutines
 	DeadlockDumpGoroutines = "system.deadlock.DumpGoroutines"

common/metrics/metric_defs.go

@@ -1648,6 +1648,7 @@ var (
 	PersistenceFailures                                 = NewCounterDef("persistence_errors")
 	PersistenceErrorWithType                            = NewCounterDef("persistence_error_with_type")
 	PersistenceLatency                                  = NewTimerDef("persistence_latency")
+	PersistenceShardRPS                                 = NewDimensionlessHistogramDef("persistence_shard_rps")
 	PersistenceErrShardExistsCounter                    = NewCounterDef("persistence_errors_shard_exists")
 	PersistenceErrShardOwnershipLostCounter             = NewCounterDef("persistence_errors_shard_ownership_lost")
 	PersistenceErrConditionFailedCounter                = NewCounterDef("persistence_errors_condition_failed")

common/persistence/client/factory.go

@@ -26,7 +26,6 @@ package client
 
 import (
 	"go.temporal.io/api/serviceerror"
-
 	"go.temporal.io/server/common"
 	"go.temporal.io/server/common/config"
 	"go.temporal.io/server/common/log"
@@ -69,6 +68,7 @@ type (
 		logger           log.Logger
 		clusterName      string
 		ratelimiter      quotas.RequestRateLimiter
+		healthSignals    p.HealthSignalAggregator
 	}
 )
 
@@ -87,6 +87,7 @@ func NewFactory(
 	clusterName string,
 	metricsHandler metrics.Handler,
 	logger log.Logger,
+	healthSignals p.HealthSignalAggregator,
 ) Factory {
 	return &factoryImpl{
 		dataStoreFactory: dataStoreFactory,
@@ -96,6 +97,7 @@ func NewFactory(
 		logger:           logger,
 		clusterName:      clusterName,
 		ratelimiter:      ratelimiter,
+		healthSignals:    healthSignals,
 	}
 }
 
@@ -110,8 +112,9 @@ func (f *factoryImpl) NewTaskManager() (p.TaskManager, error) {
 	if f.ratelimiter != nil {
 		result = p.NewTaskPersistenceRateLimitedClient(result, f.ratelimiter, f.logger)
 	}
-	if f.metricsHandler != nil {
-		result = p.NewTaskPersistenceMetricsClient(result, f.metricsHandler, f.logger)
+	if f.metricsHandler != nil || f.healthSignals != nil {
+		f.updateNilMetricsAndHealthSignals()
+		result = p.NewTaskPersistenceMetricsClient(result, f.metricsHandler, f.healthSignals, f.logger)
 	}
 	return result, nil
 }
@@ -127,8 +130,9 @@ func (f *factoryImpl) NewShardManager() (p.ShardManager, error) {
 	if f.ratelimiter != nil {
 		result = p.NewShardPersistenceRateLimitedClient(result, f.ratelimiter, f.logger)
 	}
-	if f.metricsHandler != nil {
-		result = p.NewShardPersistenceMetricsClient(result, f.metricsHandler, f.logger)
+	if f.metricsHandler != nil || f.healthSignals != nil {
+		f.updateNilMetricsAndHealthSignals()
+		result = p.NewShardPersistenceMetricsClient(result, f.metricsHandler, f.healthSignals, f.logger)
 	}
 	result = p.NewShardPersistenceRetryableClient(result, retryPolicy, IsPersistenceTransientError)
 	return result, nil
@@ -145,8 +149,9 @@ func (f *factoryImpl) NewMetadataManager() (p.MetadataManager, error) {
 	if f.ratelimiter != nil {
 		result = p.NewMetadataPersistenceRateLimitedClient(result, f.ratelimiter, f.logger)
 	}
-	if f.metricsHandler != nil {
-		result = p.NewMetadataPersistenceMetricsClient(result, f.metricsHandler, f.logger)
+	if f.metricsHandler != nil || f.healthSignals != nil {
+		f.updateNilMetricsAndHealthSignals()
+		result = p.NewMetadataPersistenceMetricsClient(result, f.metricsHandler, f.healthSignals, f.logger)
 	}
 	result = p.NewMetadataPersistenceRetryableClient(result, retryPolicy, IsPersistenceTransientError)
 	return result, nil
@@ -163,8 +168,9 @@ func (f *factoryImpl) NewClusterMetadataManager() (p.ClusterMetadataManager, err
 	if f.ratelimiter != nil {
 		result = p.NewClusterMetadataPersistenceRateLimitedClient(result, f.ratelimiter, f.logger)
 	}
-	if f.metricsHandler != nil {
-		result = p.NewClusterMetadataPersistenceMetricsClient(result, f.metricsHandler, f.logger)
+	if f.metricsHandler != nil || f.healthSignals != nil {
+		f.updateNilMetricsAndHealthSignals()
+		result = p.NewClusterMetadataPersistenceMetricsClient(result, f.metricsHandler, f.healthSignals, f.logger)
 	}
 	result = p.NewClusterMetadataPersistenceRetryableClient(result, retryPolicy, IsPersistenceTransientError)
 	return result, nil
@@ -181,8 +187,9 @@ func (f *factoryImpl) NewExecutionManager() (p.ExecutionManager, error) {
 	if f.ratelimiter != nil {
 		result = p.NewExecutionPersistenceRateLimitedClient(result, f.ratelimiter, f.logger)
 	}
-	if f.metricsHandler != nil {
-		result = p.NewExecutionPersistenceMetricsClient(result, f.metricsHandler, f.logger)
+	if f.metricsHandler != nil || f.healthSignals != nil {
+		f.updateNilMetricsAndHealthSignals()
+		result = p.NewExecutionPersistenceMetricsClient(result, f.metricsHandler, f.healthSignals, f.logger)
 	}
 	result = p.NewExecutionPersistenceRetryableClient(result, retryPolicy, IsPersistenceTransientError)
 	return result, nil
@@ -197,8 +204,9 @@ func (f *factoryImpl) NewNamespaceReplicationQueue() (p.NamespaceReplicationQueu
 	if f.ratelimiter != nil {
 		result = p.NewQueuePersistenceRateLimitedClient(result, f.ratelimiter, f.logger)
 	}
-	if f.metricsHandler != nil {
-		result = p.NewQueuePersistenceMetricsClient(result, f.metricsHandler, f.logger)
+	if f.metricsHandler != nil || f.healthSignals != nil {
+		f.updateNilMetricsAndHealthSignals()
+		result = p.NewQueuePersistenceMetricsClient(result, f.metricsHandler, f.healthSignals, f.logger)
 	}
 	result = p.NewQueuePersistenceRetryableClient(result, retryPolicy, IsPersistenceTransientError)
 	return p.NewNamespaceReplicationQueue(result, f.serializer, f.clusterName, f.metricsHandler, f.logger)
@@ -207,6 +215,9 @@ func (f *factoryImpl) NewNamespaceReplicationQueue() (p.NamespaceReplicationQueu
 // Close closes this factory
 func (f *factoryImpl) Close() {
 	f.dataStoreFactory.Close()
+	if f.healthSignals != nil {
+		f.healthSignals.Start()
+	}
 }
 
 func IsPersistenceTransientError(err error) bool {
@@ -217,3 +228,13 @@ func IsPersistenceTransientError(err error) bool {
 
 	return false
 }
+
+func (f *factoryImpl) updateNilMetricsAndHealthSignals() {
+	if f.metricsHandler == nil {
+		f.metricsHandler = metrics.NoopMetricsHandler
+	}
+	if f.healthSignals == nil {
+		f.healthSignals = p.NoopHealthSignalAggregator
+	}
+	f.healthSignals.Start()
+}

common/persistence/client/fx.go

@@ -32,6 +32,7 @@ import (
 	"go.temporal.io/server/common/dynamicconfig"
 	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/common/metrics"
+	"go.temporal.io/server/common/persistence"
 	"go.temporal.io/server/common/persistence/serialization"
 	"go.temporal.io/server/common/primitives"
 	"go.temporal.io/server/common/quotas"
@@ -57,6 +58,7 @@ type (
 		ServiceName                        primitives.ServiceName
 		MetricsHandler                     metrics.Handler
 		Logger                             log.Logger
+		HealthSignals                      persistence.HealthSignalAggregator
 	}
 
 	FactoryProviderFn func(NewFactoryParams) Factory
@@ -97,5 +99,6 @@ func FactoryProvider(
 		string(params.ClusterName),
 		params.MetricsHandler,
 		params.Logger,
+		params.HealthSignals,
 	)
 }