Delay task scheduler rate limiting on startup (#4824)

<!-- Describe what has changed in this PR -->
**What changed?**
Added a new dynamic config option,
`history.taskSchedulerRateLimiterStartupDelay`, to control a delay for
task scheduler rate limiting on service startup. Defaults to 10s

<!-- Tell your future self why have you made these changes -->
**Why?**
On service start, task processing has to catch up and replays a large
number of previously completed tasks and we do not want to rate limit
during this period.

<!-- How have you verified this change? Tested locally? Added a unit
test? Checked in staging env? -->
**How did you test it?**
Existing tests

<!-- Assuming the worst case, what can be broken when deploying this
change to production? -->
**Potential risks**
None

<!-- Is this PR a hotfix candidate or require that a notification be
sent to the broader community? (Yes/No) -->
**Is hotfix candidate?**
No

common/dynamicconfig/constants.go

@@ -567,6 +567,8 @@ const (
 	// TaskSchedulerEnableRateLimiterShadowMode indicates if task scheduler rate limiter should run in shadow mode
 	// i.e. through rate limiter and emit metrics but do not actually block/throttle task scheduling
 	TaskSchedulerEnableRateLimiterShadowMode = "history.taskSchedulerEnableRateLimiterShadowMode"
+	// TaskSchedulerRateLimiterStartupDelay is the duration to wait after startup before enforcing task scheduler rate limiting
+	TaskSchedulerRateLimiterStartupDelay = "history.taskSchedulerRateLimiterStartupDelay"
 	// TaskSchedulerThrottleDuration is the throttle duration when task scheduled exceeds max qps
 	TaskSchedulerThrottleDuration = "history.taskSchedulerThrottleDuration"
 	// TaskSchedulerMaxQPS is the max qps task schedulers on a host can schedule tasks

service/history/configs/config.go

@@ -101,6 +101,7 @@ type Config struct {
 
 	TaskSchedulerEnableRateLimiter           dynamicconfig.BoolPropertyFn
 	TaskSchedulerEnableRateLimiterShadowMode dynamicconfig.BoolPropertyFn
+	TaskSchedulerRateLimiterStartupDelay     dynamicconfig.DurationPropertyFn
 	TaskSchedulerThrottleDuration            dynamicconfig.DurationPropertyFn
 	TaskSchedulerMaxQPS                      dynamicconfig.IntPropertyFn
 	TaskSchedulerNamespaceMaxQPS             dynamicconfig.IntPropertyFnWithNamespaceFilter
@@ -380,6 +381,7 @@ func NewConfig(
 
 		TaskSchedulerEnableRateLimiter:           dc.GetBoolProperty(dynamicconfig.TaskSchedulerEnableRateLimiter, false),
 		TaskSchedulerEnableRateLimiterShadowMode: dc.GetBoolProperty(dynamicconfig.TaskSchedulerEnableRateLimiterShadowMode, true),
+		TaskSchedulerRateLimiterStartupDelay:     dc.GetDurationProperty(dynamicconfig.TaskSchedulerRateLimiterStartupDelay, 30*time.Second),
 		TaskSchedulerThrottleDuration:            dc.GetDurationProperty(dynamicconfig.TaskSchedulerThrottleDuration, time.Second),
 		TaskSchedulerMaxQPS:                      dc.GetIntProperty(dynamicconfig.TaskSchedulerMaxQPS, 0),
 		TaskSchedulerNamespaceMaxQPS:             dc.GetIntPropertyFilteredByNamespace(dynamicconfig.TaskSchedulerNamespaceMaxQPS, 0),

service/history/queue_factory_base.go

@@ -156,12 +156,15 @@ func getOptionalQueueFactories(
 
 func QueueSchedulerRateLimiterProvider(
 	config *configs.Config,
-) queues.SchedulerRateLimiter {
+	timeSource clock.TimeSource,
+) (queues.SchedulerRateLimiter, error) {
 	return queues.NewSchedulerRateLimiter(
 		config.TaskSchedulerNamespaceMaxQPS,
 		config.TaskSchedulerMaxQPS,
 		config.PersistenceNamespaceMaxQPS,
 		config.PersistenceMaxQPS,
+		config.TaskSchedulerRateLimiterStartupDelay,
+		timeSource,
 	)
 }
 

service/history/queue_factory_base_test.go

@@ -159,6 +159,7 @@ func getModuleDependencies(controller *gomock.Controller, c *moduleTestCase) fx.
 		fx.Annotate(archivalMetadata, fx.As(new(carchiver.ArchivalMetadata))),
 		fx.Annotate(metrics.NoopMetricsHandler, fx.As(new(metrics.Handler))),
 		fx.Annotate(clusterMetadata, fx.As(new(cluster.Metadata))),
+		fx.Annotate(clock.NewEventTimeSource(), fx.As(new(clock.TimeSource))),
 	)
 }
 
@@ -168,7 +169,6 @@ type compileTimeDependencies struct {
 	fx.Out
 
 	namespace.Registry
-	clock.TimeSource
 	log.SnTaggedLogger
 	client.Bean
 	sdk.ClientFactory

service/history/queues/queue_scheduled_test.go

@@ -83,18 +83,22 @@ func (s *scheduledQueueSuite) SetupTest() {
 	s.mockExecutionManager = s.mockShard.Resource.ExecutionMgr
 	s.mockShard.Resource.ClusterMetadata.EXPECT().GetCurrentClusterName().Return(cluster.TestCurrentClusterName).AnyTimes()
 
+	rateLimiter, _ := NewSchedulerRateLimiter(
+		s.mockShard.GetConfig().TaskSchedulerNamespaceMaxQPS,
+		s.mockShard.GetConfig().TaskSchedulerMaxQPS,
+		s.mockShard.GetConfig().PersistenceNamespaceMaxQPS,
+		s.mockShard.GetConfig().PersistenceMaxQPS,
+		s.mockShard.GetConfig().TaskSchedulerRateLimiterStartupDelay,
+		s.mockShard.GetTimeSource(),
+	)
+
 	scheduler := NewPriorityScheduler(
 		PrioritySchedulerOptions{
 			WorkerCount:                 dynamicconfig.GetIntPropertyFn(10),
 			EnableRateLimiter:           dynamicconfig.GetBoolPropertyFn(true),
 			EnableRateLimiterShadowMode: dynamicconfig.GetBoolPropertyFn(true),
 		},
-		NewSchedulerRateLimiter(
-			s.mockShard.GetConfig().TaskSchedulerNamespaceMaxQPS,
-			s.mockShard.GetConfig().TaskSchedulerMaxQPS,
-			s.mockShard.GetConfig().PersistenceNamespaceMaxQPS,
-			s.mockShard.GetConfig().PersistenceMaxQPS,
-		),
+		rateLimiter,
 		s.mockShard.GetTimeSource(),
 		log.NewTestLogger(),
 		metrics.NoopMetricsHandler,

service/history/queues/scheduler_quotas.go

@@ -25,6 +25,7 @@
 package queues
 
 import (
+	"go.temporal.io/server/common/clock"
 	"go.temporal.io/server/common/dynamicconfig"
 	"go.temporal.io/server/common/quotas"
 	"go.temporal.io/server/common/tasks"
@@ -37,7 +38,9 @@ func NewSchedulerRateLimiter(
 	hostMaxQPS dynamicconfig.IntPropertyFn,
 	persistenceNamespaceMaxQPS dynamicconfig.IntPropertyFnWithNamespaceFilter,
 	persistenceHostMaxQPS dynamicconfig.IntPropertyFn,
-) SchedulerRateLimiter {
+	startupDelay dynamicconfig.DurationPropertyFn,
+	timeSource clock.TimeSource,
+) (SchedulerRateLimiter, error) {
 	hostRateFn := func() float64 {
 		hostMaxQPS := float64(hostMaxQPS())
 		if hostMaxQPS > 0 {
@@ -75,10 +78,12 @@ func NewSchedulerRateLimiter(
 		priorityToRateLimiters[int(priority)] = requestRateLimiter
 	}
 
-	return quotas.NewPriorityRateLimiter(
-		requestPriorityFn,
-		priorityToRateLimiters,
-	)
+	priorityLimiter := quotas.NewPriorityRateLimiter(requestPriorityFn, priorityToRateLimiters)
+
+	if startupDelay != nil {
+		return quotas.NewDelayedRequestRateLimiter(priorityLimiter, startupDelay(), timeSource)
+	}
+	return priorityLimiter, nil
 }
 
 func newHighPriorityTaskRequestRateLimiter(