Add flag to enable execution scanner event Id validator (#4114)

* Add flag to enable execution scanner event id validator
temporalio · Mar 28, 2023 · e4a49b2 · e4a49b2
1 parent 47424c0
commit e4a49b2
Show file tree

Hide file tree

Showing 6 changed files with 52 additions and 34 deletions.
diff --git a/common/dynamicconfig/constants.go b/common/dynamicconfig/constants.go
@@ -707,6 +707,8 @@ const (
 	ExecutionDataDurationBuffer = "worker.executionDataDurationBuffer"
 	// ExecutionScannerWorkerCount is the execution scavenger worker count
 	ExecutionScannerWorkerCount = "worker.executionScannerWorkerCount"
+	// ExecutionScannerHistoryEventIdValidator is the flag to enable history event id validator
+	ExecutionScannerHistoryEventIdValidator = "worker.executionEnableHistoryEventIdValidator"
 	// TaskQueueScannerEnabled indicates if task queue scanner should be started as part of worker.Scanner
 	TaskQueueScannerEnabled = "worker.taskQueueScannerEnabled"
 	// HistoryScannerEnabled indicates if history scanner should be started as part of worker.Scanner

diff --git a/service/worker/scanner/executions/scavenger.go b/service/worker/scanner/executions/scavenger.go
@@ -56,16 +56,17 @@ type (
 		numHistoryShards int32
 		activityContext  context.Context
 
-		executionManager            persistence.ExecutionManager
-		registry                    namespace.Registry
-		historyClient               historyservice.HistoryServiceClient
-		adminClient                 adminservice.AdminServiceClient
-		executor                    executor.Executor
-		rateLimiter                 quotas.RateLimiter
-		perShardQPS                 dynamicconfig.IntPropertyFn
-		executionDataDurationBuffer dynamicconfig.DurationPropertyFn
-		metricsHandler              metrics.Handler
-		logger                      log.Logger
+		executionManager              persistence.ExecutionManager
+		registry                      namespace.Registry
+		historyClient                 historyservice.HistoryServiceClient
+		adminClient                   adminservice.AdminServiceClient
+		executor                      executor.Executor
+		rateLimiter                   quotas.RateLimiter
+		perShardQPS                   dynamicconfig.IntPropertyFn
+		executionDataDurationBuffer   dynamicconfig.DurationPropertyFn
+		enableHistoryEventIDValidator dynamicconfig.BoolPropertyFn
+		metricsHandler                metrics.Handler
+		logger                        log.Logger
 
 		stopC  chan struct{}
 		stopWG sync.WaitGroup
@@ -89,6 +90,7 @@ func NewScavenger(
 	perShardQPS dynamicconfig.IntPropertyFn,
 	executionDataDurationBuffer dynamicconfig.DurationPropertyFn,
 	executionTaskWorker dynamicconfig.IntPropertyFn,
+	enableHistoryEventIDValidator dynamicconfig.BoolPropertyFn,
 	executionManager persistence.ExecutionManager,
 	registry namespace.Registry,
 	historyClient historyservice.HistoryServiceClient,
@@ -112,10 +114,11 @@ func NewScavenger(
 		rateLimiter: quotas.NewDefaultOutgoingRateLimiter(
 			func() float64 { return float64(perHostQPS()) },
 		),
-		perShardQPS:                 perShardQPS,
-		executionDataDurationBuffer: executionDataDurationBuffer,
-		metricsHandler:              metricsHandler.WithTags(metrics.OperationTag(metrics.ExecutionsScavengerScope)),
-		logger:                      logger,
+		perShardQPS:                   perShardQPS,
+		executionDataDurationBuffer:   executionDataDurationBuffer,
+		enableHistoryEventIDValidator: enableHistoryEventIDValidator,
+		metricsHandler:                metricsHandler.WithTags(metrics.OperationTag(metrics.ExecutionsScavengerScope)),
+		logger:                        logger,
 
 		stopC: make(chan struct{}),
 	}
@@ -185,6 +188,7 @@ func (s *Scavenger) run() {
 				s.rateLimiter,
 			}),
 			s.executionDataDurationBuffer,
+			s.enableHistoryEventIDValidator,
 		))
 		if !submitted {
 			s.logger.Error("unable to submit task to executor", tag.ShardID(shardID))

diff --git a/service/worker/scanner/executions/task.go b/service/worker/scanner/executions/task.go
@@ -66,10 +66,11 @@ type (
 		logger           log.Logger
 		scavenger        *Scavenger
 
-		ctx                         context.Context
-		rateLimiter                 quotas.RateLimiter
-		executionDataDurationBuffer dynamicconfig.DurationPropertyFn
-		paginationToken             []byte
+		ctx                           context.Context
+		rateLimiter                   quotas.RateLimiter
+		executionDataDurationBuffer   dynamicconfig.DurationPropertyFn
+		enableHistoryEventIDValidator dynamicconfig.BoolPropertyFn
+		paginationToken               []byte
 	}
 )
 
@@ -86,6 +87,7 @@ func newTask(
 	scavenger *Scavenger,
 	rateLimiter quotas.RateLimiter,
 	executionDataDurationBuffer dynamicconfig.DurationPropertyFn,
+	enableHistoryEventIDValidator dynamicconfig.BoolPropertyFn,
 ) executor.Task {
 	return &task{
 		shardID:          shardID,
@@ -98,9 +100,10 @@ func newTask(
 		logger:         logger,
 		scavenger:      scavenger,
 
-		ctx:                         ctx,
-		rateLimiter:                 rateLimiter,
-		executionDataDurationBuffer: executionDataDurationBuffer,
+		ctx:                           ctx,
+		rateLimiter:                   rateLimiter,
+		executionDataDurationBuffer:   executionDataDurationBuffer,
+		enableHistoryEventIDValidator: enableHistoryEventIDValidator,
 	}
 }
 
@@ -186,19 +189,21 @@ func (t *task) validate(
 		return results
 	}
 
-	if validationResults, err := NewHistoryEventIDValidator(
-		t.shardID,
-		t.executionManager,
-	).Validate(t.ctx, mutableState); err != nil {
-		t.logger.Error("unable to validate history event ID being contiguous",
-			tag.ShardID(t.shardID),
-			tag.WorkflowNamespaceID(mutableState.GetExecutionInfo().GetNamespaceId()),
-			tag.WorkflowID(mutableState.GetExecutionInfo().GetWorkflowId()),
-			tag.WorkflowRunID(mutableState.GetExecutionState().GetRunId()),
-			tag.Error(err),
-		)
-	} else {
-		results = append(results, validationResults...)
+	if t.enableHistoryEventIDValidator() {
+		if validationResults, err := NewHistoryEventIDValidator(
+			t.shardID,
+			t.executionManager,
+		).Validate(t.ctx, mutableState); err != nil {
+			t.logger.Error("unable to validate history event ID being contiguous",
+				tag.ShardID(t.shardID),
+				tag.WorkflowNamespaceID(mutableState.GetExecutionInfo().GetNamespaceId()),
+				tag.WorkflowID(mutableState.GetExecutionInfo().GetWorkflowId()),
+				tag.WorkflowRunID(mutableState.GetExecutionState().GetRunId()),
+				tag.Error(err),
+			)
+		} else {
+			results = append(results, validationResults...)
+		}
 	}
 
 	return results

diff --git a/service/worker/scanner/scanner.go b/service/worker/scanner/scanner.go
@@ -82,6 +82,8 @@ type (
 		ExecutionDataDurationBuffer dynamicconfig.DurationPropertyFn
 		// ExecutionScannerWorkerCount is the execution scavenger task worker number
 		ExecutionScannerWorkerCount dynamicconfig.IntPropertyFn
+		// ExecutionScannerHistoryEventIdValidator indicates if the execution scavenger to validate history event id.
+		ExecutionScannerHistoryEventIdValidator dynamicconfig.BoolPropertyFn
 	}
 
 	// scannerContext is the context object that get's

diff --git a/service/worker/scanner/workflow.go b/service/worker/scanner/workflow.go
@@ -196,6 +196,7 @@ func ExecutionsScavengerActivity(
 		ctx.cfg.ExecutionScannerPerShardQPS,
 		ctx.cfg.ExecutionDataDurationBuffer,
 		ctx.cfg.ExecutionScannerWorkerCount,
+		ctx.cfg.ExecutionScannerHistoryEventIdValidator,
 		ctx.executionManager,
 		ctx.namespaceRegistry,
 		ctx.historyClient,

diff --git a/service/worker/service.go b/service/worker/service.go
@@ -302,6 +302,10 @@ func NewConfig(dc *dynamicconfig.Collection, persistenceConfig *config.Persisten
 				dynamicconfig.ExecutionScannerWorkerCount,
 				8,
 			),
+			ExecutionScannerHistoryEventIdValidator: dc.GetBoolProperty(
+				dynamicconfig.ExecutionScannerHistoryEventIdValidator,
+				true,
+			),
 		},
 		EnableBatcher:      dc.GetBoolProperty(dynamicconfig.EnableBatcher, true),
 		BatcherRPS:         dc.GetIntPropertyFilteredByNamespace(dynamicconfig.BatcherRPS, batcher.DefaultRPS),