Dynamic rate limiter

common/dynamicconfig/constants.go

@@ -413,6 +413,9 @@ const (
 	HistoryPersistencePerShardNamespaceMaxQPS = "history.persistencePerShardNamespaceMaxQPS"
 	// HistoryEnablePersistencePriorityRateLimiting indicates if priority rate limiting is enabled in history persistence client
 	HistoryEnablePersistencePriorityRateLimiting = "history.enablePersistencePriorityRateLimiting"
+	// HistoryPersistenceDynamicRateLimitingParams is a map that contains all adjustable dynamic rate limiting params
+	// Available options are: enabled, refreshInterval, latencyThreshold, errorThreshold, rateBackoffStepSize, and rateIncreaseStepSize
+	HistoryPersistenceDynamicRateLimitingParams = "history.persistenceDynamicRateLimitingParams"
 	// HistoryLongPollExpirationInterval is the long poll expiration interval in the history service
 	HistoryLongPollExpirationInterval = "history.longPollExpirationInterval"
 	// HistoryCacheInitialSize is initial size of history cache

common/dynamicconfig/shared_constants.go

@@ -45,4 +45,5 @@ var defaultNumTaskQueuePartitions = []ConstrainedValue{
 	},
 }
 
-var DefaultPerShardNamespaceRPSMax = func(namespace string) int { return 0 }
+var DefaultPerShardNamespaceRPSMax = GetIntPropertyFilteredByNamespace(0)
+var DisabledDynamicRateLimiting = GetMapPropertyFn(map[string]interface{}{"enabled": false})

common/persistence/client/fx.go

@@ -45,7 +45,10 @@ type (
 	PersistenceNamespaceMaxQps         dynamicconfig.IntPropertyFnWithNamespaceFilter
 	PersistencePerShardNamespaceMaxQPS dynamicconfig.IntPropertyFnWithNamespaceFilter
 	EnablePriorityRateLimiting         dynamicconfig.BoolPropertyFn
-	ClusterName                        string
+
+	DynamicRateLimitingParams dynamicconfig.MapPropertyFn
+
+	ClusterName string
 
 	NewFactoryParams struct {
 		fx.In
@@ -61,6 +64,7 @@ type (
 		MetricsHandler                     metrics.Handler
 		Logger                             log.Logger
 		HealthSignals                      persistence.HealthSignalAggregator
+		DynamicRateLimitingParams          DynamicRateLimitingParams
 	}
 
 	FactoryProviderFn func(NewFactoryParams) Factory
@@ -88,6 +92,9 @@ func FactoryProvider(
 				params.PersistenceMaxQPS,
 				params.PersistencePerShardNamespaceMaxQPS,
 				RequestPriorityFn,
+				params.HealthSignals,
+				params.DynamicRateLimitingParams,
+				params.Logger,
 			)
 		} else {
 			requestRatelimiter = NewNoopPriorityRateLimiter(params.PersistenceMaxQPS)
@@ -113,8 +120,8 @@ func HealthSignalAggregatorProvider(
 ) persistence.HealthSignalAggregator {
 	if dynamicCollection.GetBoolProperty(dynamicconfig.PersistenceHealthSignalCollectionEnabled, true)() {
 		return persistence.NewHealthSignalAggregatorImpl(
-			dynamicCollection.GetDurationProperty(dynamicconfig.PersistenceHealthSignalWindowSize, 3*time.Second)(),
-			dynamicCollection.GetIntProperty(dynamicconfig.PersistenceHealthSignalBufferSize, 500)(),
+			dynamicCollection.GetDurationProperty(dynamicconfig.PersistenceHealthSignalWindowSize, 10*time.Second)(),
+			dynamicCollection.GetIntProperty(dynamicconfig.PersistenceHealthSignalBufferSize, 5000)(),
 			metricsHandler,
 			dynamicCollection.GetIntProperty(dynamicconfig.ShardRPSWarnLimit, 50),
 			logger,

common/persistence/client/health_request_rate_limiter.go

@@ -0,0 +1,196 @@
+// 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 client
+
+import (
+	"context"
+	"encoding/json"
+	"math"
+	"sync/atomic"
+	"time"
+
+	"go.temporal.io/server/common/log"
+	"go.temporal.io/server/common/log/tag"
+	"go.temporal.io/server/common/persistence"
+	"go.temporal.io/server/common/primitives/timestamp"
+	"go.temporal.io/server/common/quotas"
+)
+
+const (
+	DefaultRefreshInterval   = 10 * time.Second
+	DefaultRateBurstRatio    = 1.0
+	DefaultMinRateMultiplier = 0.1
+	DefaultMaxRateMultiplier = 1.0
+)
+
+type (
+	HealthRequestRateLimiterImpl struct {
+		enabled    *atomic.Bool
+		params     DynamicRateLimitingParams  // dynamic config map
+		curOptions dynamicRateLimitingOptions // current dynamic config values (updated on refresh)
+
+		rateLimiter   *quotas.RateLimiterImpl
+		healthSignals persistence.HealthSignalAggregator
+
+		refreshTimer *time.Ticker
+
+		rateFn           quotas.RateFn
+		rateToBurstRatio float64
+
+		minRateMultiplier float64
+		maxRateMultiplier float64
+		curRateMultiplier float64
+
+		logger log.Logger
+	}
+
+	dynamicRateLimitingOptions struct {
+		Enabled bool
+
+		RefreshInterval string // string returned by json.Unmarshal will be parsed into a duration
+
+		// thresholds which should trigger backoff if exceeded
+		LatencyThreshold float64
+		ErrorThreshold   float64
+
+		// if either threshold is exceeded, the current rate multiplier will be reduced by this amount
+		RateBackoffStepSize float64
+		// when the system is healthy and current rate < max rate, the current rate multiplier will be
+		// increased by this amount
+		RateIncreaseStepSize float64
+	}
+)
+
+var _ quotas.RequestRateLimiter = (*HealthRequestRateLimiterImpl)(nil)
+
+func NewHealthRequestRateLimiterImpl(
+	healthSignals persistence.HealthSignalAggregator,
+	rateFn quotas.RateFn,
+	params DynamicRateLimitingParams,
+	logger log.Logger,
+) *HealthRequestRateLimiterImpl {
+	limiter := &HealthRequestRateLimiterImpl{
+		enabled:           &atomic.Bool{},
+		rateLimiter:       quotas.NewRateLimiter(rateFn(), int(DefaultRateBurstRatio*rateFn())),
+		healthSignals:     healthSignals,
+		rateFn:            rateFn,
+		params:            params,
+		refreshTimer:      time.NewTicker(DefaultRefreshInterval),
+		rateToBurstRatio:  DefaultRateBurstRatio,
+		minRateMultiplier: DefaultMinRateMultiplier,
+		maxRateMultiplier: DefaultMaxRateMultiplier,
+		curRateMultiplier: DefaultMaxRateMultiplier,
+		logger:            logger,
+	}
+	limiter.refreshDynamicParams()
+	return limiter
+}
+
+func (rl *HealthRequestRateLimiterImpl) Allow(now time.Time, request quotas.Request) bool {
+	rl.maybeRefresh()
+	if !rl.enabled.Load() {
+		return true
+	}
+	return rl.rateLimiter.AllowN(now, request.Token)
+}
+
+func (rl *HealthRequestRateLimiterImpl) Reserve(now time.Time, request quotas.Request) quotas.Reservation {
+	rl.maybeRefresh()
+	if !rl.enabled.Load() {
+		return quotas.NoopReservation
+	}
+	return rl.rateLimiter.ReserveN(now, request.Token)
+}
+
+func (rl *HealthRequestRateLimiterImpl) Wait(ctx context.Context, request quotas.Request) error {
+	rl.maybeRefresh()
+	if !rl.enabled.Load() {
+		return nil
+	}
+	return rl.rateLimiter.WaitN(ctx, request.Token)
+}
+
+func (rl *HealthRequestRateLimiterImpl) maybeRefresh() {
+	select {
+	case <-rl.refreshTimer.C:
+		rl.refreshDynamicParams()
+		if rl.enabled.Load() {
+			rl.refreshRate()
+		}
+
+	default:
+		// no-op
+	}
+}
+
+func (rl *HealthRequestRateLimiterImpl) refreshRate() {
+	if rl.latencyThresholdExceeded() || rl.errorThresholdExceeded() {
+		// limit exceeded, do backoff
+		rl.curRateMultiplier = math.Max(rl.minRateMultiplier, rl.curRateMultiplier-rl.curOptions.RateBackoffStepSize)
+		rl.rateLimiter.SetRate(rl.curRateMultiplier * rl.rateFn())
+		rl.rateLimiter.SetBurst(int(rl.rateToBurstRatio * rl.rateFn()))
+	} else if rl.curRateMultiplier < rl.maxRateMultiplier {
+		// already doing backoff and under thresholds, increase limit
+		rl.curRateMultiplier = math.Min(rl.maxRateMultiplier, rl.curRateMultiplier+rl.curOptions.RateIncreaseStepSize)
+		rl.rateLimiter.SetRate(rl.curRateMultiplier * rl.rateFn())
+		rl.rateLimiter.SetBurst(int(rl.rateToBurstRatio * rl.rateFn()))
+	}
+}
+
+func (rl *HealthRequestRateLimiterImpl) refreshDynamicParams() {
+	var options dynamicRateLimitingOptions
+	b, err := json.Marshal(rl.params())
+	if err != nil {
+		rl.logger.Warn("Error marshalling dynamic rate limiting params. Dynamic rate limiting is disabled.", tag.Error(err))
+		rl.enabled.Store(false)
+		return
+	}
+
+	err = json.Unmarshal(b, &options)
+	if err != nil {
+		rl.logger.Warn("Error unmarshalling dynamic rate limiting params. Dynamic rate limiting is disabled.", tag.Error(err))
+		rl.enabled.Store(false)
+		return
+	}
+
+	if len(options.RefreshInterval) > 0 {
+		if refreshDuration, parseErr := timestamp.ParseDuration(options.RefreshInterval); parseErr != nil {
+			rl.logger.Warn("Error parsing dynamic rate limit refreshInterval timestamp. Using previous value.", tag.Error(err))
+		} else {
+			rl.refreshTimer.Reset(refreshDuration)
+		}
+	}
+
+	rl.enabled.Store(options.Enabled)
+	rl.curOptions = options
+}
+
+func (rl *HealthRequestRateLimiterImpl) latencyThresholdExceeded() bool {
+	return rl.curOptions.LatencyThreshold > 0 && rl.healthSignals.AverageLatency() > rl.curOptions.LatencyThreshold
+}
+
+func (rl *HealthRequestRateLimiterImpl) errorThresholdExceeded() bool {
+	return rl.curOptions.ErrorThreshold > 0 && rl.healthSignals.ErrorRatio() > rl.curOptions.ErrorThreshold
+}

common/persistence/client/quotas.go

@@ -26,6 +26,7 @@ package client
 
 import (
 	"go.temporal.io/server/common/headers"
+	"go.temporal.io/server/common/log"
 	p "go.temporal.io/server/common/persistence"
 	"go.temporal.io/server/common/quotas"
 	"go.temporal.io/server/service/history/tasks"
@@ -85,16 +86,21 @@ func NewPriorityRateLimiter(
 	hostMaxQPS PersistenceMaxQps,
 	perShardNamespaceMaxQPS PersistencePerShardNamespaceMaxQPS,
 	requestPriorityFn quotas.RequestPriorityFn,
+	healthSignals p.HealthSignalAggregator,
+	dynamicParams DynamicRateLimitingParams,
+	logger log.Logger,
 ) quotas.RequestRateLimiter {
-	hostRequestRateLimiter := newPriorityRateLimiter(
-		func() float64 { return float64(hostMaxQPS()) },
-		requestPriorityFn,
-	)
+	hostRateFn := func() float64 { return float64(hostMaxQPS()) }
 
 	return quotas.NewMultiRequestRateLimiter(
+		// host-level dynamic rate limiter
+		newPriorityDynamicRateLimiter(hostRateFn, requestPriorityFn, healthSignals, dynamicParams, logger),
+		// per shardID+namespaceID rate limiters
 		newPerShardPerNamespacePriorityRateLimiter(perShardNamespaceMaxQPS, hostMaxQPS, requestPriorityFn),
+		// per namespaceID rate limiters
 		newPriorityNamespaceRateLimiter(namespaceMaxQPS, hostMaxQPS, requestPriorityFn),
-		hostRequestRateLimiter,
+		// basic host-level rate limiter
+		newPriorityRateLimiter(hostRateFn, requestPriorityFn),
 	)
 }
 
@@ -169,6 +175,24 @@ func newPriorityRateLimiter(
 	)
 }
 
+func newPriorityDynamicRateLimiter(
+	rateFn quotas.RateFn,
+	requestPriorityFn quotas.RequestPriorityFn,
+	healthSignals p.HealthSignalAggregator,
+	dynamicParams DynamicRateLimitingParams,
+	logger log.Logger,
+) quotas.RequestRateLimiter {
+	rateLimiters := make(map[int]quotas.RequestRateLimiter)
+	for priority := range RequestPrioritiesOrdered {
+		rateLimiters[priority] = NewHealthRequestRateLimiterImpl(healthSignals, rateFn, dynamicParams, logger)
+	}
+
+	return quotas.NewPriorityRateLimiter(
+		requestPriorityFn,
+		rateLimiters,
+	)
+}
+
 func NewNoopPriorityRateLimiter(
 	maxQPS PersistenceMaxQps,
 ) quotas.RequestRateLimiter {

common/persistence/client/store.go

@@ -90,7 +90,8 @@ func DataStoreFactoryProvider(
 
 	var faultInjection *FaultInjectionDataStoreFactory
 	if defaultCfg.FaultInjection != nil {
-		dataStoreFactory = NewFaultInjectionDatastoreFactory(defaultCfg.FaultInjection, dataStoreFactory)
+		faultInjection = NewFaultInjectionDatastoreFactory(defaultCfg.FaultInjection, dataStoreFactory)
+		dataStoreFactory = faultInjection
 	}
 
 	return dataStoreFactory, faultInjection

common/persistence/health_signal_aggregator.go

@@ -29,6 +29,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	"go.temporal.io/api/serviceerror"
 	"go.temporal.io/server/common"
 	"go.temporal.io/server/common/aggregate"
 	"go.temporal.io/server/common/dynamicconfig"
@@ -103,14 +104,13 @@ func (s *HealthSignalAggregatorImpl) Stop() {
 }
 
 func (s *HealthSignalAggregatorImpl) Record(callerSegment int32, latency time.Duration, err error) {
-	// TODO: uncomment when adding dynamic rate limiter
-	//s.latencyAverage.Record(latency.Milliseconds())
-	//
-	//if isUnhealthyError(err) {
-	//	s.errorRatio.Record(1)
-	//} else {
-	//	s.errorRatio.Record(0)
-	//}
+	s.latencyAverage.Record(latency.Milliseconds())
+
+	if isUnhealthyError(err) {
+		s.errorRatio.Record(1)
+	} else {
+		s.errorRatio.Record(0)
+	}
 
 	if callerSegment != CallerSegmentMissing {
 		s.incrementShardRequestCount(callerSegment)
@@ -153,18 +153,18 @@ func (s *HealthSignalAggregatorImpl) emitMetricsLoop() {
 	}
 }
 
-// TODO: uncomment when adding dynamic rate limiter
-//func isUnhealthyError(err error) bool {
-//	if err == nil {
-//		return false
-//	}
-//	switch err.(type) {
-//	case *ShardOwnershipLostError,
-//		*AppendHistoryTimeoutError,
-//		*TimeoutError:
-//		return true
-//
-//	default:
-//		return false
-//	}
-//}
+func isUnhealthyError(err error) bool {
+	if err == nil {
+		return false
+	}
+	switch err.(type) {
+	case *ShardOwnershipLostError,
+		*AppendHistoryTimeoutError,
+		*TimeoutError,
+		*serviceerror.Unavailable:
+		return true
+
+	default:
+		return false
+	}
+}