Merge branch 'master' into jjtimmons/fix/jsonEncoding-Unmarshal

Joshua Timmons · web-flow · commit e40084791177 · 2022-10-01T23:39:07.000+02:00
diff --git a/activity/activity.go b/activity/activity.go
@@ -23,7 +23,7 @@ package activity
 import (
 	"context"
 
-	"github.com/uber-go/tally/v4"
+	"github.com/uber-go/tally"
 	"go.uber.org/cadence/internal"
 	"go.uber.org/zap"
 )
diff --git a/go.mod b/go.mod
@@ -3,23 +3,21 @@ module go.uber.org/cadence
 go 1.13
 
 require (
-	github.com/apache/thrift v0.0.0-20161221203622-b2a4d4ae21c7
+	github.com/apache/thrift v0.16.0
 	github.com/cristalhq/jwt/v3 v3.1.0
 	github.com/facebookgo/clock v0.0.0-20150410010913-600d898af40a
 	github.com/gogo/protobuf v1.3.2
-	github.com/golang/mock v1.4.4
+	github.com/golang/mock v1.5.0
 	github.com/kisielk/errcheck v1.5.0
 	github.com/marusama/semaphore/v2 v2.5.0
 	github.com/opentracing/opentracing-go v1.1.0
 	github.com/pborman/uuid v0.0.0-20160209185913-a97ce2ca70fa
 	github.com/robfig/cron v1.2.0
-	github.com/streadway/quantile v0.0.0-20220407130108-4246515d968d // indirect
-	github.com/stretchr/testify v1.4.0
-	github.com/twmb/murmur3 v1.1.6 // indirect
-	github.com/uber-go/tally/v4 v4.1.1
+	github.com/stretchr/testify v1.5.1
+	github.com/uber-go/tally v3.3.15+incompatible
 	github.com/uber/cadence-idl v0.0.0-20220713235846-fda89e95df1e
 	github.com/uber/jaeger-client-go v2.22.1+incompatible
-	github.com/uber/tchannel-go v1.16.0
+	github.com/uber/tchannel-go v1.32.1
 	go.uber.org/atomic v1.9.0
 	go.uber.org/fx v1.13.1 // indirect
 	go.uber.org/goleak v1.0.0
@@ -28,7 +26,7 @@ require (
 	go.uber.org/yarpc v1.55.0
 	go.uber.org/zap v1.13.0
 	golang.org/x/lint v0.0.0-20200130185559-910be7a94367
-	golang.org/x/net v0.0.0-20201021035429-f5854403a974
+	golang.org/x/net v0.0.0-20220121210141-e204ce36a2ba
 	golang.org/x/sys v0.0.0-20220403205710-6acee93ad0eb // indirect
 	golang.org/x/time v0.0.0-20170927054726-6dc17368e09b
 	honnef.co/go/tools v0.0.1-2019.2.3
diff --git a/go.sum b/go.sum
diff --git a/internal/activity.go b/internal/activity.go
@@ -27,7 +27,7 @@ import (
 	"time"
 
 	"github.com/opentracing/opentracing-go"
-	"github.com/uber-go/tally/v4"
+	"github.com/uber-go/tally"
 	"go.uber.org/cadence/.gen/go/shared"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
diff --git a/internal/client.go b/internal/client.go
@@ -27,7 +27,7 @@ import (
 	"time"
 
 	"github.com/opentracing/opentracing-go"
-	"github.com/uber-go/tally/v4"
+	"github.com/uber-go/tally"
 	"go.uber.org/cadence/.gen/go/cadence/workflowserviceclient"
 	s "go.uber.org/cadence/.gen/go/shared"
 	"go.uber.org/cadence/internal/common/auth"
@@ -524,6 +524,7 @@ func getFeatureFlags(options *ClientOptions) FeatureFlags {
 	if options != nil {
 		return FeatureFlags{
 			WorkflowExecutionAlreadyCompletedErrorEnabled: options.FeatureFlags.WorkflowExecutionAlreadyCompletedErrorEnabled,
+			PollerAutoScalerEnabled:                       options.FeatureFlags.PollerAutoScalerEnabled,
 		}
 	}
 	return FeatureFlags{}
diff --git a/internal/common/metrics/capturingStatsReporter.go b/internal/common/metrics/capturingStatsReporter.go
@@ -24,7 +24,7 @@ import (
 	"io"
 	"time"
 
-	"github.com/uber-go/tally/v4"
+	"github.com/uber-go/tally"
 )
 
 // NewMetricsScope returns a new metric scope
diff --git a/internal/common/metrics/scope.go b/internal/common/metrics/scope.go
@@ -24,7 +24,7 @@ import (
 	"sync"
 	"time"
 
-	"github.com/uber-go/tally/v4"
+	"github.com/uber-go/tally"
 )
 
 type (
diff --git a/internal/common/metrics/scope_test.go b/internal/common/metrics/scope_test.go
@@ -27,7 +27,7 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/require"
-	"github.com/uber-go/tally/v4"
+	"github.com/uber-go/tally"
 )
 
 func Test_Counter(t *testing.T) {
diff --git a/internal/common/metrics/service_wrapper.go b/internal/common/metrics/service_wrapper.go
@@ -25,7 +25,7 @@ import (
 	"sync"
 	"time"
 
-	"github.com/uber-go/tally/v4"
+	"github.com/uber-go/tally"
 	"go.uber.org/cadence/.gen/go/cadence/workflowserviceclient"
 	"go.uber.org/cadence/.gen/go/shared"
 	"go.uber.org/yarpc"
diff --git a/internal/common/metrics/service_wrapper_test.go b/internal/common/metrics/service_wrapper_test.go
@@ -30,7 +30,7 @@ import (
 
 	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/require"
-	"github.com/uber-go/tally/v4"
+	"github.com/uber-go/tally"
 	"github.com/uber/tchannel-go/thrift"
 	"go.uber.org/cadence/.gen/go/cadence/workflowserviceclient"
 	"go.uber.org/cadence/.gen/go/cadence/workflowservicetest"
diff --git a/internal/common/thrift_util.go b/internal/common/thrift_util.go
@@ -21,14 +21,15 @@
 package common
 
 import (
+	"context"
 	"reflect"
 
 	"github.com/apache/thrift/lib/go/thrift"
 )
 
 // TSerialize is used to serialize thrift TStruct to []byte
-func TSerialize(t thrift.TStruct) (b []byte, err error) {
-	return thrift.NewTSerializer().Write(t)
+func TSerialize(ctx context.Context, t thrift.TStruct) (b []byte, err error) {
+	return thrift.NewTSerializer().Write(ctx, t)
 }
 
 // TListSerialize is used to serialize list of thrift TStruct to []byte
@@ -43,18 +44,20 @@ func TListSerialize(ts []thrift.TStruct) (b []byte, err error) {
 	// NOTE: we don't write any markers as thrift by design being a streaming protocol doesn't
 	// recommend writing length.
 
+	// TODO populate context from argument
+	ctx := context.Background()
 	for _, v := range ts {
-		if e := v.Write(t.Protocol); e != nil {
+		if e := v.Write(ctx, t.Protocol); e != nil {
 			err = thrift.PrependError("error writing TStruct: ", e)
 			return
 		}
 	}
 
-	if err = t.Protocol.Flush(); err != nil {
+	if err = t.Protocol.Flush(ctx); err != nil {
 		return
 	}
 
-	if err = t.Transport.Flush(); err != nil {
+	if err = t.Transport.Flush(ctx); err != nil {
 		return
 	}
 
@@ -63,8 +66,8 @@ func TListSerialize(ts []thrift.TStruct) (b []byte, err error) {
 }
 
 // TDeserialize is used to deserialize []byte to thrift TStruct
-func TDeserialize(t thrift.TStruct, b []byte) (err error) {
-	return thrift.NewTDeserializer().Read(t, b)
+func TDeserialize(ctx context.Context, t thrift.TStruct, b []byte) (err error) {
+	return thrift.NewTDeserializer().Read(ctx, t, b)
 }
 
 // TListDeserialize is used to deserialize []byte to list of thrift TStruct
@@ -75,8 +78,10 @@ func TListDeserialize(ts []thrift.TStruct, b []byte) (err error) {
 		return
 	}
 
+	// TODO populate context from argument
+	ctx := context.Background()
 	for i := 0; i < len(ts); i++ {
-		if e := ts[i].Read(t.Protocol); e != nil {
+		if e := ts[i].Read(ctx, t.Protocol); e != nil {
 			err = thrift.PrependError("error reading TStruct: ", e)
 			return
 		}
diff --git a/internal/interceptors.go b/internal/interceptors.go
@@ -24,7 +24,7 @@ package internal
 import (
 	"time"
 
-	"github.com/uber-go/tally/v4"
+	"github.com/uber-go/tally"
 	"go.uber.org/zap"
 )
 
diff --git a/internal/internal_activity.go b/internal/internal_activity.go
@@ -31,7 +31,7 @@ import (
 	"time"
 
 	"github.com/opentracing/opentracing-go"
-	"github.com/uber-go/tally/v4"
+	"github.com/uber-go/tally"
 	"go.uber.org/cadence/.gen/go/shared"
 	"go.uber.org/cadence/internal/common"
 	"go.uber.org/zap"
diff --git a/internal/internal_event_handlers.go b/internal/internal_event_handlers.go
@@ -32,7 +32,7 @@ import (
 	"time"
 
 	"github.com/opentracing/opentracing-go"
-	"github.com/uber-go/tally/v4"
+	"github.com/uber-go/tally"
 	"go.uber.org/cadence/.gen/go/shared"
 	m "go.uber.org/cadence/.gen/go/shared"
 	"go.uber.org/cadence/internal/common"
diff --git a/internal/internal_poller_autoscaler.go b/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()
diff --git a/internal/internal_poller_autoscaler_test.go b/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{})
diff --git a/internal/internal_task_handlers_test.go b/internal/internal_task_handlers_test.go
@@ -1279,23 +1279,26 @@ func (t *TaskHandlersTestSuite) TestHeartBeat_Interleaved() {
 
 	heartbeatErr := cadenceInvoker.BatchHeartbeat([]byte("1"))
 	t.NoError(heartbeatErr)
-	time.Sleep(1 * time.Second)
+	time.Sleep(1000 * time.Millisecond)
 
 	for i := 0; i < 4; i++ {
 		heartbeatErr = cadenceInvoker.BackgroundHeartbeat()
 		t.NoError(heartbeatErr)
-		time.Sleep(time.Second)
+		time.Sleep(800 * time.Millisecond)
 	}
 
+	time.Sleep(time.Second)
+
 	heartbeatErr = cadenceInvoker.BatchHeartbeat([]byte("2"))
 	t.NoError(heartbeatErr)
-	time.Sleep(1 * time.Second)
+	time.Sleep(1000 * time.Millisecond)
 
 	for i := 0; i < 4; i++ {
 		heartbeatErr = cadenceInvoker.BackgroundHeartbeat()
 		t.NoError(heartbeatErr)
-		time.Sleep(time.Second)
+		time.Sleep(800 * time.Millisecond)
 	}
+	time.Sleep(1 * time.Second)
 }
 
 func (t *TaskHandlersTestSuite) TestHeartBeat_NilResponseWithError() {
diff --git a/internal/internal_task_pollers.go b/internal/internal_task_pollers.go
@@ -31,7 +31,7 @@ import (
 
 	"github.com/opentracing/opentracing-go"
 	"github.com/pborman/uuid"
-	"github.com/uber-go/tally/v4"
+	"github.com/uber-go/tally"
 	"go.uber.org/cadence/.gen/go/cadence/workflowserviceclient"
 	s "go.uber.org/cadence/.gen/go/shared"
 	"go.uber.org/cadence/internal/common"
diff --git a/internal/internal_utils.go b/internal/internal_utils.go
@@ -33,7 +33,7 @@ import (
 	"time"
 
 	"github.com/pborman/uuid"
-	"github.com/uber-go/tally/v4"
+	"github.com/uber-go/tally"
 	s "go.uber.org/cadence/.gen/go/shared"
 	"go.uber.org/cadence/internal/common"
 	"go.uber.org/cadence/internal/common/metrics"
diff --git a/internal/internal_worker.go b/internal/internal_worker.go
@@ -40,7 +40,7 @@ import (
 
 	"github.com/opentracing/opentracing-go"
 	"github.com/pborman/uuid"
-	"github.com/uber-go/tally/v4"
+	"github.com/uber-go/tally"
 	"go.uber.org/cadence/.gen/go/cadence/workflowserviceclient"
 	"go.uber.org/cadence/.gen/go/shared"
 	"go.uber.org/cadence/internal/common/auth"
diff --git a/internal/internal_worker_base.go b/internal/internal_worker_base.go
@@ -32,7 +32,7 @@ import (
 	"syscall"
 	"time"
 
-	"github.com/uber-go/tally/v4"
+	"github.com/uber-go/tally"
 	"go.uber.org/cadence/.gen/go/shared"
 	"go.uber.org/cadence/internal/common/backoff"
 	"go.uber.org/cadence/internal/common/metrics"
@@ -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()
diff --git a/internal/internal_worker_test.go b/internal/internal_worker_test.go
@@ -35,7 +35,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
-	"github.com/uber-go/tally/v4"
+	"github.com/uber-go/tally"
 	"go.uber.org/cadence/.gen/go/cadence/workflowservicetest"
 	"go.uber.org/cadence/.gen/go/shared"
 	"go.uber.org/cadence/internal/common"
diff --git a/internal/internal_workflow_client.go b/internal/internal_workflow_client.go
@@ -33,7 +33,7 @@ import (
 
 	"github.com/opentracing/opentracing-go"
 	"github.com/pborman/uuid"
-	"github.com/uber-go/tally/v4"
+	"github.com/uber-go/tally"
 
 	"go.uber.org/cadence/.gen/go/cadence/workflowserviceclient"
 	s "go.uber.org/cadence/.gen/go/shared"
diff --git a/internal/internal_workflow_testsuite.go b/internal/internal_workflow_testsuite.go
@@ -35,7 +35,7 @@ import (
 	"github.com/opentracing/opentracing-go"
 	"github.com/robfig/cron"
 	"github.com/stretchr/testify/mock"
-	"github.com/uber-go/tally/v4"
+	"github.com/uber-go/tally"
 	"go.uber.org/cadence/.gen/go/cadence/workflowserviceclient"
 	"go.uber.org/cadence/.gen/go/cadence/workflowservicetest"
 	"go.uber.org/cadence/.gen/go/shared"
diff --git a/internal/worker.go b/internal/worker.go
@@ -26,7 +26,7 @@ import (
 	"time"
 
 	"github.com/opentracing/opentracing-go"
-	"github.com/uber-go/tally/v4"
+	"github.com/uber-go/tally"
 	"go.uber.org/cadence/.gen/go/cadence/workflowserviceclient"
 	"go.uber.org/cadence/.gen/go/shared"
 	"go.uber.org/cadence/internal/common/auth"
diff --git a/internal/workflow.go b/internal/workflow.go
@@ -28,7 +28,7 @@ import (
 	"strings"
 	"time"
 
-	"github.com/uber-go/tally/v4"
+	"github.com/uber-go/tally"
 	s "go.uber.org/cadence/.gen/go/shared"
 	"go.uber.org/cadence/internal/common"
 	"go.uber.org/cadence/internal/common/backoff"
diff --git a/internal/workflow_replayer.go b/internal/workflow_replayer.go
diff --git a/internal/workflow_testsuite.go b/internal/workflow_testsuite.go
diff --git a/workflow/workflow.go b/workflow/workflow.go

Original file line number	Diff line number	Diff line change
`@@ -23,7 +23,7 @@ package activity`
`23`	`23`	`import (`
`24`	`24`	`"context"`
`25`	`25`
`26`		`- "github.com/uber-go/tally/v4"`
	`26`	`+ "github.com/uber-go/tally"`
`27`	`27`	`"go.uber.org/cadence/internal"`
`28`	`28`	`"go.uber.org/zap"`
`29`	`29`	`)`
Original file line number	Diff line number	Diff line change
`@@ -24,7 +24,7 @@ import (`
`24`	`24`	`"io"`
`25`	`25`	`"time"`
`26`	`26`
`27`		`- "github.com/uber-go/tally/v4"`
	`27`	`+ "github.com/uber-go/tally"`
`28`	`28`	`)`
`29`	`29`
`30`	`30`	`// NewMetricsScope returns a new metric scope`
Original file line number	Diff line number	Diff line change
`@@ -27,7 +27,7 @@ import (`
`27`	`27`	`"time"`
`28`	`28`
`29`	`29`	`"github.com/stretchr/testify/require"`
`30`		`- "github.com/uber-go/tally/v4"`
	`30`	`+ "github.com/uber-go/tally"`
`31`	`31`	`)`
`32`	`32`
`33`	`33`	`func Test_Counter(t *testing.T) {`