fix correct initial poller count for autoscaler (#1440)

What changed?

rename pollerCount to less confusing name pollerCountWithoutAutoscaling
ensure starting enough goroutines for pollers when autoscaler is enabled
Why?

Consider a case when autoscaler is enabled with 40 max, 10 initial, 2 min. We need to start 40 poller goroutines first and only allow 10 to pass initially.

This is achieved by starting 40 go bw.runPoller() goroutines and set PollerPermit to 10.

The bug we have now is we only start 10 goroutines at the beginning so it just won't scale up.

How did you test it?

bench test with stable tasklist traffic.

Pollers are 2 intially with 20 instances with autoscaler enabled.
After scaling down 20 -> 10 and 10 -> 2. We saw the poller quota increased to cover the loss of hosts. Schedule to start latency is maintained.

Potential risks

Additional goroutines will be created but it's ok because we are already creating a max of 10k goroutines by default for sticky executions.

Author: shijiesheng

internal/internal_worker.go

@@ -276,21 +276,17 @@ func newWorkflowTaskWorkerInternal(
 		domain,
 		params,
 	)
-	pollerCount := params.MaxConcurrentDecisionTaskPollers
-	if params.AutoScalerOptions.Enabled {
-		pollerCount = params.AutoScalerOptions.PollerInitCount
-	}
 	worker := newBaseWorker(baseWorkerOptions{
-		pollerAutoScaler:  params.AutoScalerOptions,
-		pollerCount:       pollerCount,
-		pollerRate:        defaultPollerRate,
-		maxConcurrentTask: params.MaxConcurrentDecisionTaskExecutionSize,
-		maxTaskPerSecond:  params.WorkerDecisionTasksPerSecond,
-		taskWorker:        poller,
-		identity:          params.Identity,
-		workerType:        "DecisionWorker",
-		shutdownTimeout:   params.WorkerStopTimeout,
-		pollerTracker:     params.WorkerStats.PollerTracker,
+		pollerAutoScaler:              params.AutoScalerOptions,
+		pollerCountWithoutAutoScaling: params.MaxConcurrentDecisionTaskPollers,
+		pollerRate:                    defaultPollerRate,
+		maxConcurrentTask:             params.MaxConcurrentDecisionTaskExecutionSize,
+		maxTaskPerSecond:              params.WorkerDecisionTasksPerSecond,
+		taskWorker:                    poller,
+		identity:                      params.Identity,
+		workerType:                    "DecisionWorker",
+		shutdownTimeout:               params.WorkerStopTimeout,
+		pollerTracker:                 params.WorkerStats.PollerTracker,
 	},
 		params.Logger,
 		params.MetricsScope,
@@ -308,14 +304,14 @@ func newWorkflowTaskWorkerInternal(
 	// 2) local activity task poller will poll from laTunnel, and result will be pushed to laTunnel
 	localActivityTaskPoller := newLocalActivityPoller(params, laTunnel)
 	localActivityWorker := newBaseWorker(baseWorkerOptions{
-		pollerCount:       1, // 1 poller (from local channel) is enough for local activity
-		maxConcurrentTask: params.MaxConcurrentLocalActivityExecutionSize,
-		maxTaskPerSecond:  params.WorkerLocalActivitiesPerSecond,
-		taskWorker:        localActivityTaskPoller,
-		identity:          params.Identity,
-		workerType:        "LocalActivityWorker",
-		shutdownTimeout:   params.WorkerStopTimeout,
-		pollerTracker:     params.WorkerStats.PollerTracker,
+		pollerCountWithoutAutoScaling: 1, // 1 poller (from local channel) is enough for local activity
+		maxConcurrentTask:             params.MaxConcurrentLocalActivityExecutionSize,
+		maxTaskPerSecond:              params.WorkerLocalActivitiesPerSecond,
+		taskWorker:                    localActivityTaskPoller,
+		identity:                      params.Identity,
+		workerType:                    "LocalActivityWorker",
+		shutdownTimeout:               params.WorkerStopTimeout,
+		pollerTracker:                 params.WorkerStats.PollerTracker,
 	},
 		params.Logger,
 		params.MetricsScope,
@@ -472,23 +468,19 @@ func newActivityTaskWorker(
 	workerType string,
 ) (worker *activityWorker) {
 	ensureRequiredParams(&workerParams)
-	pollerCount := workerParams.MaxConcurrentActivityTaskPollers
-	if workerParams.AutoScalerOptions.Enabled {
-		pollerCount = workerParams.AutoScalerOptions.PollerInitCount
-	}
 	base := newBaseWorker(
 		baseWorkerOptions{
-			pollerAutoScaler:  workerParams.AutoScalerOptions,
-			pollerCount:       pollerCount,
-			pollerRate:        defaultPollerRate,
-			maxConcurrentTask: workerParams.MaxConcurrentActivityExecutionSize,
-			maxTaskPerSecond:  workerParams.WorkerActivitiesPerSecond,
-			taskWorker:        poller,
-			identity:          workerParams.Identity,
-			workerType:        workerType,
-			shutdownTimeout:   workerParams.WorkerStopTimeout,
-			userContextCancel: workerParams.UserContextCancel,
-			pollerTracker:     workerParams.WorkerStats.PollerTracker,
+			pollerAutoScaler:              workerParams.AutoScalerOptions,
+			pollerCountWithoutAutoScaling: workerParams.MaxConcurrentActivityTaskPollers,
+			pollerRate:                    defaultPollerRate,
+			maxConcurrentTask:             workerParams.MaxConcurrentActivityExecutionSize,
+			maxTaskPerSecond:              workerParams.WorkerActivitiesPerSecond,
+			taskWorker:                    poller,
+			identity:                      workerParams.Identity,
+			workerType:                    workerType,
+			shutdownTimeout:               workerParams.WorkerStopTimeout,
+			userContextCancel:             workerParams.UserContextCancel,
+			pollerTracker:                 workerParams.WorkerStats.PollerTracker,
 		},
 
 		workerParams.Logger,

internal/internal_worker_base.go

@@ -115,18 +115,18 @@ type (
 
 	// baseWorkerOptions options to configure base worker.
 	baseWorkerOptions struct {
-		pollerAutoScaler  AutoScalerOptions
-		pollerCount       int
-		pollerRate        int
-		maxConcurrentTask int
-		maxTaskPerSecond  float64
-		taskWorker        taskPoller
-		identity          string
-		workerType        string
-		shutdownTimeout   time.Duration
-		userContextCancel context.CancelFunc
-		host              string
-		pollerTracker     debug.PollerTracker
+		pollerAutoScaler              AutoScalerOptions
+		pollerCountWithoutAutoScaling int
+		pollerRate                    int
+		maxConcurrentTask             int
+		maxTaskPerSecond              float64
+		taskWorker                    taskPoller
+		identity                      string
+		workerType                    string
+		shutdownTimeout               time.Duration
+		userContextCancel             context.CancelFunc
+		host                          string
+		pollerTracker                 debug.PollerTracker
 	}
 
 	// baseWorker that wraps worker activities.
@@ -172,12 +172,16 @@ func newBaseWorker(options baseWorkerOptions, logger *zap.Logger, metricsScope t
 	metricsScope = tagScope(metricsScope, tagWorkerType, options.workerType)
 
 	concurrency := &worker.ConcurrencyLimit{
-		PollerPermit: worker.NewResizablePermit(options.pollerCount),
+		PollerPermit: worker.NewResizablePermit(options.pollerCountWithoutAutoScaling),
 		TaskPermit:   worker.NewResizablePermit(options.maxConcurrentTask),
 	}
 
 	var concurrencyAS *worker.ConcurrencyAutoScaler
 	if pollerOptions := options.pollerAutoScaler; pollerOptions.Enabled {
+		concurrency = &worker.ConcurrencyLimit{
+			PollerPermit: worker.NewResizablePermit(pollerOptions.PollerInitCount),
+			TaskPermit:   worker.NewResizablePermit(options.maxConcurrentTask),
+		}
 		concurrencyAS = worker.NewConcurrencyAutoScaler(worker.ConcurrencyAutoScalerInput{
 			Concurrency:              concurrency,
 			Cooldown:                 pollerOptions.Cooldown,
@@ -221,7 +225,11 @@ func (bw *baseWorker) Start() {
 
 	bw.concurrencyAutoScaler.Start()
 
-	for i := 0; i < bw.options.pollerCount; i++ {
+	maxPollerCount := bw.options.pollerCountWithoutAutoScaling
+	if bw.options.pollerAutoScaler.Enabled {
+		maxPollerCount = bw.options.pollerAutoScaler.PollerMaxCount
+	}
+	for i := 0; i < maxPollerCount; i++ {
 		bw.shutdownWG.Add(1)
 		go bw.runPoller()
 	}
@@ -232,9 +240,13 @@ func (bw *baseWorker) Start() {
 	bw.isWorkerStarted = true
 	traceLog(func() {
 		bw.logger.Info("Started Worker",
-			zap.Int("PollerCount", bw.options.pollerCount),
+			zap.Int("PollerCountWithoutAutoScaling", bw.options.pollerCountWithoutAutoScaling),
 			zap.Int("MaxConcurrentTask", bw.options.maxConcurrentTask),
 			zap.Float64("MaxTaskPerSecond", bw.options.maxTaskPerSecond),
+			zap.Bool("AutoscalerEnabled", bw.options.pollerAutoScaler.Enabled),
+			zap.Int("AutoscalerPollerMaxCount", bw.options.pollerAutoScaler.PollerMaxCount),
+			zap.Int("AutoscalerPollerMinCount", bw.options.pollerAutoScaler.PollerMinCount),
+			zap.Int("AutoscalerPollerInitialCount", bw.options.pollerAutoScaler.PollerInitCount),
 		)
 	})
 }