Domain Deprecation: Implement workflow listing and termination activities (#6819)

* GitButler Workspace Commit

This is a merge commit the virtual branches in your workspace.

Due to GitButler managing multiple virtual branches, you cannot switch back and
forth between git branches and virtual branches easily. 

If you switch to another branch, GitButler will need to be reinitialized.
If you commit on this branch, GitButler will throw it away.

Here are the branches that are currently applied:
 - domain-deprecation/list_terminate_activities (refs/gitbutler/domain-deprecation/list_terminate_activities)
   branch head: 4bfa8c5
   - service/worker/domaindeprecation/workflow.go
   - config/dynamicconfig/development_es.yaml
   - service/worker/domaindeprecation/module.go
   - service/worker/domaindeprecation/activities.go
   - docker/docker-compose-es.yml
   - docker/docker-compose-es-v7.yml
For more information about what we're doing here, check out our docs:
https://docs.gitbutler.com/features/virtual-branches/integration-branch

* [domain-deletion]Add activities to list and terminate workflows

* [domain-deletion]Add activities to list and terminate workflows

* Replace the entityNotExistsError with cadence customer error

* Add rate limiter

* Fix workflow tests

* Add listAndTerminate activity

* Update comments

* Run batch terminate workflow as child workflow

* Update child workflow ID

---------

Co-authored-by: GitButler <[email protected]>

Author: gazi-yestemirova

service/worker/batcher/batcher.go

@@ -89,7 +89,7 @@ func New(params *BootstrapParams) *Batcher {
 // Start starts the scanner
 func (s *Batcher) Start() error {
 	// start worker for batch operation workflows
-	ctx := context.WithValue(context.Background(), batcherContextKey, s)
+	ctx := context.WithValue(context.Background(), BatcherContextKey, s)
 	workerOpts := worker.Options{
 		MetricsScope:              s.tallyScope,
 		BackgroundActivityContext: ctx,

service/worker/batcher/workflow.go

@@ -45,7 +45,7 @@ type (
 )
 
 const (
-	batcherContextKey contextKey = "batcherContext"
+	BatcherContextKey contextKey = "batcherContext"
 	// BatcherTaskListName is the tasklist name
 	BatcherTaskListName = "cadence-sys-batcher-tasklist"
 	// BatchWFTypeName is the workflow type
@@ -183,7 +183,7 @@ func setDefaultParams(params BatchParams) BatchParams {
 
 // BatchActivity is activity for processing batch operation
 func BatchActivity(ctx context.Context, batchParams BatchParams) (HeartBeatDetails, error) {
-	batcher := ctx.Value(batcherContextKey).(*Batcher)
+	batcher := ctx.Value(BatcherContextKey).(*Batcher)
 	client := batcher.clientBean.GetFrontendClient()
 	var adminClient admin.Client
 	if batchParams.BatchType == BatchTypeReplicate {
@@ -284,7 +284,7 @@ func BatchActivity(ctx context.Context, batchParams BatchParams) (HeartBeatDetai
 func getHeartBeatDetails(ctx context.Context) (hbd HeartBeatDetails, ok bool) {
 	if activity.HasHeartbeatDetails(ctx) {
 		if err := activity.GetHeartbeatDetails(ctx, &hbd); err != nil {
-			batcher := ctx.Value(batcherContextKey).(*Batcher)
+			batcher := ctx.Value(BatcherContextKey).(*Batcher)
 			batcher.metricsClient.IncCounter(metrics.BatcherScope, metrics.BatcherProcessorFailures)
 			getActivityLogger(ctx).Error("Failed to recover from last heartbeat, start over from beginning", tag.Error(err))
 			return HeartBeatDetails{}, false
@@ -305,7 +305,7 @@ func startTaskProcessor(
 	client frontend.Client,
 	adminClient admin.Client,
 ) {
-	batcher := ctx.Value(batcherContextKey).(*Batcher)
+	batcher := ctx.Value(BatcherContextKey).(*Batcher)
 	for {
 		select {
 		case <-ctx.Done():
@@ -461,7 +461,7 @@ func isDone(ctx context.Context) bool {
 }
 
 func getActivityLogger(ctx context.Context) log.Logger {
-	batcher := ctx.Value(batcherContextKey).(*Batcher)
+	batcher := ctx.Value(BatcherContextKey).(*Batcher)
 	wfInfo := activity.GetInfo(ctx)
 	return batcher.logger.WithTags(
 		tag.WorkflowID(wfInfo.WorkflowExecution.ID),

service/worker/batcher/workflow_retry_test.go

@@ -72,7 +72,7 @@ func (s *workflowRetrySuite) SetupTest() {
 	mockResource.FrontendClient.EXPECT().RequestCancelWorkflowExecution(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 	mockResource.FrontendClient.EXPECT().DescribeWorkflowExecution(gomock.Any(), gomock.Any()).Return(&types.DescribeWorkflowExecutionResponse{}, nil).AnyTimes()
 
-	ctx := context.WithValue(context.Background(), batcherContextKey, batcher)
+	ctx := context.WithValue(context.Background(), BatcherContextKey, batcher)
 	workerOpts := worker.Options{
 		MetricsScope:              tally.TestScope(nil),
 		BackgroundActivityContext: ctx,

service/worker/batcher/workflow_test.go

@@ -77,7 +77,7 @@ func (s *workflowSuite) SetupTest() {
 
 	mockResource.RemoteAdminClient.EXPECT().ResendReplicationTasks(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 
-	ctx := context.WithValue(context.Background(), batcherContextKey, batcher)
+	ctx := context.WithValue(context.Background(), BatcherContextKey, batcher)
 	workerOpts := worker.Options{
 		MetricsScope:              tally.TestScope(nil),
 		BackgroundActivityContext: ctx,

service/worker/domaindeprecation/activities.go

@@ -25,38 +25,38 @@ import (
 	"errors"
 	"fmt"
 
+	"go.uber.org/cadence"
+
 	"github.com/uber/cadence/common/log/tag"
 	"github.com/uber/cadence/common/types"
 )
 
 // DisableArchivalActivity disables archival for the domain
-func (w *domainDeprecator) DisableArchivalActivity(ctx context.Context, domainName string) error {
+func (w *domainDeprecator) DisableArchivalActivity(ctx context.Context, params DomainActivityParams) error {
 	client := w.clientBean.GetFrontendClient()
 	disabled := types.ArchivalStatusDisabled
 
 	describeRequest := &types.DescribeDomainRequest{
-		Name: &domainName,
+		Name: &params.DomainName,
 	}
 	domainResp, err := client.DescribeDomain(ctx, describeRequest)
 	if err != nil {
 		var entityNotExistsError *types.EntityNotExistsError
 		if errors.As(err, &entityNotExistsError) {
-			return types.EntityNotExistsError{Message: errDomainDoesNotExistNonRetryable}
+			return cadence.NewCustomError(ErrDomainDoesNotExistNonRetryable)
 		}
-
 		return fmt.Errorf("failed to describe domain: %v", err)
-
 	}
 
 	// Check if archival is already disabled
 	if *domainResp.Configuration.VisibilityArchivalStatus == disabled &&
 		*domainResp.Configuration.HistoryArchivalStatus == disabled {
-		w.logger.Info("Archival is already disabled for domain", tag.WorkflowDomainName(domainName))
+		w.logger.Info("Archival is already disabled for domain", tag.WorkflowDomainName(params.DomainName))
 		return nil
 	}
 
 	updateRequest := &types.UpdateDomainRequest{
-		Name:                     domainName,
+		Name:                     params.DomainName,
 		HistoryArchivalStatus:    &disabled,
 		VisibilityArchivalStatus: &disabled,
 		SecurityToken:            w.cfg.AdminOperationToken(),
@@ -68,19 +68,19 @@ func (w *domainDeprecator) DisableArchivalActivity(ctx context.Context, domainNa
 
 	if *updateResp.Configuration.VisibilityArchivalStatus != disabled ||
 		*updateResp.Configuration.HistoryArchivalStatus != disabled {
-		return fmt.Errorf("failed to disable archival for domain %s", domainName)
+		return fmt.Errorf("failed to disable archival for domain %s", params.DomainName)
 	}
 
-	w.logger.Info("Disabled archival for domain", tag.WorkflowDomainName(domainName))
+	w.logger.Info("Disabled archival for domain", tag.WorkflowDomainName(params.DomainName))
 	return nil
 }
 
 // DeprecateDomainActivity deprecates the domain
-func (w *domainDeprecator) DeprecateDomainActivity(ctx context.Context, domainName string) error {
+func (w *domainDeprecator) DeprecateDomainActivity(ctx context.Context, params DomainActivityParams) error {
 	client := w.clientBean.GetFrontendClient()
 
 	err := client.DeprecateDomain(ctx, &types.DeprecateDomainRequest{
-		Name:          domainName,
+		Name:          params.DomainName,
 		SecurityToken: w.cfg.AdminOperationToken(),
 	})
 	if err != nil {

service/worker/domaindeprecation/activities_test.go

@@ -129,7 +129,9 @@ func TestDisableArchivalActivity(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			tt.setupMocks()
-			err := deprecator.DisableArchivalActivity(context.Background(), testDomain)
+			err := deprecator.DisableArchivalActivity(context.Background(), DomainActivityParams{
+				DomainName: testDomain,
+			})
 			if tt.expectedError != nil {
 				assert.Error(t, err)
 			} else {
@@ -163,14 +165,14 @@ func TestDeprecateDomainActivity(t *testing.T) {
 		expectedError error
 	}{
 		{
-			name: "Success - Deprecate domain",
+			name: "Success",
 			setupMocks: func() {
 				mockClient.EXPECT().DeprecateDomain(gomock.Any(), gomock.Any()).Return(nil)
 			},
 			expectedError: nil,
 		},
 		{
-			name: "Error - Deprecate domain",
+			name: "Error",
 			setupMocks: func() {
 				mockClient.EXPECT().DeprecateDomain(gomock.Any(), gomock.Any()).Return(assert.AnError)
 			},
@@ -181,7 +183,9 @@ func TestDeprecateDomainActivity(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			tt.setupMocks()
-			err := deprecator.DeprecateDomainActivity(context.Background(), testDomain)
+			err := deprecator.DeprecateDomainActivity(context.Background(), DomainActivityParams{
+				DomainName: testDomain,
+			})
 			if tt.expectedError != nil {
 				assert.Error(t, err)
 			} else {

service/worker/domaindeprecation/entities.go

@@ -0,0 +1,28 @@
+// The MIT License (MIT)
+
+// Copyright (c) 2017-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 domaindeprecation
+
+// DomainActivityParams contains the domain name parameter used by domain-related activities.
+type DomainActivityParams struct {
+	DomainName string `json:"domain_name"`
+}

service/worker/domaindeprecation/helpers.go

@@ -22,7 +22,26 @@
 
 package domaindeprecation
 
-// Error reason strings used for Cadence non-retryable errors
+import "time"
+
 const (
-	errDomainDoesNotExistNonRetryable = "domain does not exist"
+	// ErrDomainDoesNotExistNonRetryable is error reason used for Cadence non-retryable errors
+	ErrDomainDoesNotExistNonRetryable = "domain does not exist"
+	// ErrAccessDeniedNonRetryable is permission error that require manual intervention
+	ErrAccessDeniedNonRetryable = "AccessDeniedError"
+	// ErrWorkflowAlreadyCompletedNonRetryable is error that indicates workflow execution already completed
+	ErrWorkflowAlreadyCompletedNonRetryable = "WorkflowExecutionAlreadyCompletedError"
+
+	// DefaultRPS is the default RPS
+	DefaultRPS = 50
+	// DefaultConcurrency is the default concurrency
+	DefaultConcurrency = 5
+	// DefaultPageSize is the default page size
+	DefaultPageSize = 1000
+	// DefaultAttemptsOnRetryableError is the default value for AttemptsOnRetryableError
+	DefaultAttemptsOnRetryableError = 50
+	// DefaultActivityHeartBeatTimeout is the default value for ActivityHeartBeatTimeout
+	DefaultActivityHeartBeatTimeout = time.Second * 10
+	// DefaultMaxActivityRetries is the default value for MaxActivityRetries
+	DefaultMaxActivityRetries = 4
 )

service/worker/domaindeprecation/module.go

@@ -36,6 +36,8 @@ import (
 	"github.com/uber/cadence/common/constants"
 	"github.com/uber/cadence/common/dynamicconfig/dynamicproperties"
 	"github.com/uber/cadence/common/log"
+	"github.com/uber/cadence/common/metrics"
+	"github.com/uber/cadence/service/worker/batcher"
 )
 
 type (
@@ -51,18 +53,20 @@ type (
 	}
 
 	domainDeprecator struct {
-		cfg        Config
-		svcClient  workflowserviceclient.Interface
-		clientBean client.Bean
-		worker     worker.Worker
-		tally      tally.Scope
-		logger     log.Logger
+		cfg           Config
+		svcClient     workflowserviceclient.Interface
+		clientBean    client.Bean
+		metricsClient metrics.Client
+		worker        worker.Worker
+		tally         tally.Scope
+		logger        log.Logger
 	}
 
 	Params struct {
 		Config        Config
 		ServiceClient workflowserviceclient.Interface
 		ClientBean    client.Bean
+		MetricsClient metrics.Client
 		Tally         tally.Scope
 		Logger        log.Logger
 	}
@@ -71,27 +75,42 @@ type (
 // New creates a new domain deprecation workflow.
 func New(params Params) DomainDeprecationWorker {
 	return &domainDeprecator{
-		cfg:        params.Config,
-		svcClient:  params.ServiceClient,
-		clientBean: params.ClientBean,
-		tally:      params.Tally,
-		logger:     params.Logger,
+		cfg:           params.Config,
+		svcClient:     params.ServiceClient,
+		clientBean:    params.ClientBean,
+		metricsClient: params.MetricsClient,
+		tally:         params.Tally,
+		logger:        params.Logger,
 	}
 }
 
 // Start starts the worker
 func (w *domainDeprecator) Start() error {
+	batcherParams := &batcher.BootstrapParams{
+		Config: batcher.Config{
+			AdminOperationToken: w.cfg.AdminOperationToken,
+		},
+		ServiceClient: w.svcClient,
+		ClientBean:    w.clientBean,
+		MetricsClient: w.metricsClient,
+		Logger:        w.logger,
+		TallyScope:    w.tally,
+	}
+	batcherInstance := batcher.New(batcherParams)
+
+	ctx := context.WithValue(context.Background(), batcher.BatcherContextKey, batcherInstance)
+
 	workerOpts := worker.Options{
 		MetricsScope:                     w.tally,
-		BackgroundActivityContext:        context.Background(),
+		BackgroundActivityContext:        ctx,
 		Tracer:                           opentracing.GlobalTracer(),
 		MaxConcurrentActivityTaskPollers: 10,
 		MaxConcurrentDecisionTaskPollers: 10,
 	}
 	newWorker := worker.New(w.svcClient, constants.SystemLocalDomainName, domainDeprecationTaskListName, workerOpts)
 	newWorker.RegisterWorkflowWithOptions(w.DomainDeprecationWorkflow, workflow.RegisterOptions{Name: domainDeprecationWorkflowTypeName})
-	newWorker.RegisterActivityWithOptions(w.DisableArchivalActivity, activity.RegisterOptions{Name: disableArchivalActivity})
-	newWorker.RegisterActivityWithOptions(w.DeprecateDomainActivity, activity.RegisterOptions{Name: deprecateDomainActivity})
+	newWorker.RegisterActivityWithOptions(w.DisableArchivalActivity, activity.RegisterOptions{Name: disableArchivalActivity, EnableAutoHeartbeat: true})
+	newWorker.RegisterActivityWithOptions(w.DeprecateDomainActivity, activity.RegisterOptions{Name: deprecateDomainActivity, EnableAutoHeartbeat: true})
 	w.worker = newWorker
 	return newWorker.Start()
 }