Add dynamic property Fixer Enabled

uber · Dec 9, 2020 · e046c9b · e046c9b
1 parent 1e057eb
commit e046c9b
Show file tree

Hide file tree

Showing 7 changed files with 26 additions and 36 deletions.
diff --git a/service/worker/scanner/executions/activities.go b/service/worker/scanner/executions/activities.go
diff --git a/service/worker/scanner/executions/concrete_execution.go b/service/worker/scanner/executions/concrete_execution.go
@@ -162,7 +162,8 @@ func ConcreteExecutionScannerConfig(dc *dynamicconfig.Collection) *shardscanner.
 		ScannerWFTypeName: ConcreteExecutionsScannerWFTypeName,
 		FixerWFTypeName:   ConcreteExecutionsFixerWFTypeName,
 		DynamicParams: shardscanner.DynamicParams{
-			Enabled:                 dc.GetBoolProperty(dynamicconfig.ConcreteExecutionsScannerEnabled, false),
+			ScannerEnabled:          dc.GetBoolProperty(dynamicconfig.ConcreteExecutionsScannerEnabled, false),
+			FixerEnabled:            dc.GetBoolProperty(dynamicconfig.ConcreteExecutionFixerEnabled, false),
 			Concurrency:             dc.GetIntProperty(dynamicconfig.ConcreteExecutionsScannerConcurrency, 25),
 			PageSize:                dc.GetIntProperty(dynamicconfig.ConcreteExecutionsScannerPersistencePageSize, 1000),
 			BlobstoreFlushThreshold: dc.GetIntProperty(dynamicconfig.ConcreteExecutionsScannerBlobstoreFlushThreshold, 100),

diff --git a/service/worker/scanner/executions/current_execution.go b/service/worker/scanner/executions/current_execution.go
@@ -128,7 +128,8 @@ func CurrentExecutionScannerConfig(dc *dynamicconfig.Collection) *shardscanner.S
 		FixerWFTypeName:   CurrentExecutionsFixerWFTypeName,
 		DynamicCollection: dc,
 		DynamicParams: shardscanner.DynamicParams{
-			Enabled:                 dc.GetBoolProperty(dynamicconfig.CurrentExecutionsScannerEnabled, false),
+			ScannerEnabled:          dc.GetBoolProperty(dynamicconfig.CurrentExecutionsScannerEnabled, false),
+			FixerEnabled:            dc.GetBoolProperty(dynamicconfig.CurrentExecutionFixerEnabled, false),
 			Concurrency:             dc.GetIntProperty(dynamicconfig.CurrentExecutionsScannerConcurrency, 25),
 			PageSize:                dc.GetIntProperty(dynamicconfig.CurrentExecutionsScannerPersistencePageSize, 1000),
 			BlobstoreFlushThreshold: dc.GetIntProperty(dynamicconfig.CurrentExecutionsScannerBlobstoreFlushThreshold, 100),

diff --git a/service/worker/scanner/scanner.go b/service/worker/scanner/scanner.go
@@ -28,7 +28,6 @@ import (
 	"go.uber.org/cadence/.gen/go/shared"
 	cclient "go.uber.org/cadence/client"
 	"go.uber.org/cadence/worker"
-	"go.uber.org/cadence/workflow"
 	"go.uber.org/zap"
 
 	"github.com/uber/cadence/common"
@@ -39,7 +38,6 @@ import (
 	"github.com/uber/cadence/common/resource"
 	"github.com/uber/cadence/common/service/config"
 	"github.com/uber/cadence/common/service/dynamicconfig"
-	"github.com/uber/cadence/service/worker/scanner/executions"
 	"github.com/uber/cadence/service/worker/scanner/shardscanner"
 )
 
@@ -165,9 +163,10 @@ func (s *Scanner) startShardScanner(
 ) (context.Context, []string) {
 	scannerContextKey := shardscanner.ScannerContextKey(config.ScannerWFTypeName)
 	fixerContextKey := shardscanner.ScannerContextKey(config.FixerWFTypeName)
-
+	workerTaskListNames := []string{}
 	backgroundActivityContext := context.WithValue(ctx, scannerContextKey, s.context)
-	if config.DynamicParams.Enabled() {
+
+	if config.DynamicParams.ScannerEnabled() {
 		backgroundActivityContext = context.WithValue(
 			backgroundActivityContext,
 			scannerContextKey,
@@ -180,19 +179,21 @@ func (s *Scanner) startShardScanner(
 			})
 	}
 
-	backgroundActivityContext = context.WithValue(
-		backgroundActivityContext,
-		fixerContextKey,
-		shardscanner.FixerContext{
-			Resource:   s.context.Resource,
-			Scope:      s.context.Resource.GetMetricsClient().Scope(metrics.ExecutionsFixerScope),
-			ContextKey: fixerContextKey,
-			Hooks:      config.FixerHooks(),
-		})
+	if config.DynamicParams.FixerEnabled() {
+		backgroundActivityContext = context.WithValue(
+			backgroundActivityContext,
+			fixerContextKey,
+			shardscanner.FixerContext{
+				Resource:   s.context.Resource,
+				Scope:      s.context.Resource.GetMetricsClient().Scope(metrics.ExecutionsFixerScope),
+				ContextKey: fixerContextKey,
+				Hooks:      config.FixerHooks(),
+			})
 
-	workerTaskListNames := []string{config.FixerTLName}
+		workerTaskListNames = append(workerTaskListNames, config.FixerTLName)
+	}
 
-	if config.DynamicParams.Enabled() {
+	if config.DynamicParams.ScannerEnabled() {
 		workerTaskListNames = append(workerTaskListNames, config.StartWorkflowOptions.TaskList)
 
 		go s.startWorkflowWithRetry(config.StartWorkflowOptions, config.ScannerWFTypeName, shardscanner.ScannerWorkflowParams{
@@ -205,20 +206,6 @@ func (s *Scanner) startShardScanner(
 		})
 	}
 
-	if fixerConfig.Enabled() {
-		backgroundActivityContext = context.WithValue(backgroundActivityContext, fixerContextKey, executions.FixerContext{
-			Resource:                   s.context.Resource,
-			Scope:                      s.context.Resource.GetMetricsClient().Scope(metrics.ExecutionsFixerScope),
-			FixerWorkflowDynamicConfig: fixerConfig,
-		})
-		workerTaskListNames = append(workerTaskListNames, fixerTaskListName)
-		go s.startWorkflowWithRetry(fixerStartWorkflowOptions, fixerWFTypeName, executions.FixerWorkflowParams{
-			ScanExecution: workflow.Execution{
-				ID: scannerStartWorkflowOptions.ID,
-			},
-			ScanType: scanType,
-		})
-	}
 	return backgroundActivityContext, workerTaskListNames
 }
 

diff --git a/service/worker/scanner/shardscanner/activities.go b/service/worker/scanner/shardscanner/activities.go
@@ -61,7 +61,7 @@ func ScannerConfigActivity(
 
 	result := ResolvedScannerWorkflowConfig{
 		GenericScannerConfig: GenericScannerConfig{
-			Enabled:                 dc.Enabled(),
+			Enabled:                 dc.ScannerEnabled(),
 			Concurrency:             dc.Concurrency(),
 			PageSize:                dc.PageSize(),
 			BlobstoreFlushThreshold: dc.BlobstoreFlushThreshold(),

diff --git a/service/worker/scanner/shardscanner/activities_test.go b/service/worker/scanner/shardscanner/activities_test.go
@@ -284,7 +284,7 @@ func (s *activitiesSuite) TestScannerConfigActivity() {
 	}{
 		{
 			dynamicParams: &DynamicParams{
-				Enabled:                 dynamicconfig.GetBoolPropertyFn(true),
+				ScannerEnabled:          dynamicconfig.GetBoolPropertyFn(true),
 				Concurrency:             dynamicconfig.GetIntPropertyFn(10),
 				PageSize:                dynamicconfig.GetIntPropertyFn(100),
 				ActivityBatchSize:       dynamicconfig.GetIntPropertyFn(10),
@@ -310,7 +310,7 @@ func (s *activitiesSuite) TestScannerConfigActivity() {
 		},
 		{
 			dynamicParams: &DynamicParams{
-				Enabled:                 dynamicconfig.GetBoolPropertyFn(true),
+				ScannerEnabled:          dynamicconfig.GetBoolPropertyFn(true),
 				Concurrency:             dynamicconfig.GetIntPropertyFn(10),
 				PageSize:                dynamicconfig.GetIntPropertyFn(100),
 				ActivityBatchSize:       dynamicconfig.GetIntPropertyFn(10),
@@ -332,7 +332,7 @@ func (s *activitiesSuite) TestScannerConfigActivity() {
 		},
 		{
 			dynamicParams: &DynamicParams{
-				Enabled:                 dynamicconfig.GetBoolPropertyFn(true),
+				ScannerEnabled:          dynamicconfig.GetBoolPropertyFn(true),
 				Concurrency:             dynamicconfig.GetIntPropertyFn(10),
 				ActivityBatchSize:       dynamicconfig.GetIntPropertyFn(100),
 				PageSize:                dynamicconfig.GetIntPropertyFn(100),

diff --git a/service/worker/scanner/shardscanner/types.go b/service/worker/scanner/shardscanner/types.go
@@ -240,7 +240,8 @@ type (
 
 	// DynamicParams is the dynamic config for scanner workflow.
 	DynamicParams struct {
-		Enabled                 dynamicconfig.BoolPropertyFn
+		ScannerEnabled          dynamicconfig.BoolPropertyFn
+		FixerEnabled            dynamicconfig.BoolPropertyFn
 		Concurrency             dynamicconfig.IntPropertyFn
 		PageSize                dynamicconfig.IntPropertyFn
 		BlobstoreFlushThreshold dynamicconfig.IntPropertyFn