[poller autoscaler] fix logic to identify empty tasks (#1192)

Author: shijiesheng

internal/client.go

@@ -524,6 +524,7 @@ func getFeatureFlags(options *ClientOptions) FeatureFlags {
 	if options != nil {
 		return FeatureFlags{
 			WorkflowExecutionAlreadyCompletedErrorEnabled: options.FeatureFlags.WorkflowExecutionAlreadyCompletedErrorEnabled,
+			PollerAutoScalerEnabled:                       options.FeatureFlags.PollerAutoScalerEnabled,
 		}
 	}
 	return FeatureFlags{}

internal/internal_poller_autoscaler.go

@@ -171,17 +171,31 @@ func (m *pollerUsageEstimator) Reset() {
 
 // CollectUsage counts past poll results to estimate autoscaler.Usages
 func (m *pollerUsageEstimator) CollectUsage(data interface{}) error {
-	switch v := data.(type) {
-	case *polledTask:
-		if v == nil { // no-task poll
-			m.atomicBits.Add(1 << 32)
-		} else {
-			m.atomicBits.Add(1)
-		}
+	isEmpty, err := isTaskEmpty(data)
+	if err != nil {
+		return err
+	}
+	if isEmpty { // no-task poll
+		m.atomicBits.Add(1 << 32)
+	} else {
+		m.atomicBits.Add(1)
 	}
 	return nil
 }
 
+func isTaskEmpty(task interface{}) (bool, error) {
+	switch t := task.(type) {
+	case *workflowTask:
+		return t == nil || t.task == nil, nil
+	case *activityTask:
+		return t == nil || t.task == nil, nil
+	case *localActivityTask:
+		return t == nil || t.workflowTask == nil, nil
+	default:
+		return false, errors.New("unknown task type")
+	}
+}
+
 // Estimate is based on past poll counts
 func (m *pollerUsageEstimator) Estimate() (autoscaler.Usages, error) {
 	bits := m.atomicBits.Load()

internal/internal_poller_autoscaler_test.go

@@ -23,6 +23,7 @@ package internal
 import (
 	"github.com/stretchr/testify/assert"
 	"go.uber.org/atomic"
+	s "go.uber.org/cadence/.gen/go/shared"
 	"go.uber.org/cadence/internal/common/autoscaler"
 	"go.uber.org/zap/zaptest"
 	"math/rand"
@@ -270,10 +271,10 @@ type unrelatedPolledTask struct{}
 func generateRandomPollResults(noTaskPoll, taskPoll, unrelated int) <-chan interface{} {
 	var result []interface{}
 	for i := 0; i < noTaskPoll; i++ {
-		result = append(result, (*polledTask)(nil))
+		result = append(result, &activityTask{})
 	}
 	for i := 0; i < taskPoll; i++ {
-		result = append(result, &polledTask{})
+		result = append(result, &activityTask{task: &s.PollForActivityTaskResponse{}})
 	}
 	for i := 0; i < unrelated; i++ {
 		result = append(result, &unrelatedPolledTask{})

internal/internal_worker_base.go

@@ -306,7 +306,9 @@ func (bw *baseWorker) pollTask() {
 		} else {
 			if bw.pollerAutoScaler != nil {
 				if pErr := bw.pollerAutoScaler.CollectUsage(task); pErr != nil {
-					bw.logger.Warn("poller auto scaler collect usage error", zap.Error(pErr))
+					bw.logger.Sugar().Warnw("poller auto scaler collect usage error",
+						"error", pErr,
+						"task", task)
 				}
 			}
 			bw.retrier.Succeeded()