Add wait period between scan iterations (#1065)

Author: yycptt

internal/workflow_shadower.go

@@ -44,6 +44,10 @@ const (
 	statusStopped     int32 = 2
 )
 
+const (
+	defaultWaitDurationPerIteration = 5 * time.Minute
+)
+
 type (
 	// ShadowOptions is used to configure workflow shadowing.
 	ShadowOptions struct {
@@ -134,6 +138,8 @@ const (
 	ShadowModeNormal ShadowMode = iota
 	// ShadowModeContinuous mode will start a new round of shadowing
 	// after all workflows matches WorkflowQuery have been replayed.
+	// There will be a 5 min wait period between each round,
+	// currently this wait period is not configurable.
 	// Shadowing will complete only when ExitCondition is met.
 	// ExitCondition must be specified when using this mode
 	ShadowModeContinuous
@@ -268,8 +274,12 @@ func (s *WorkflowShadower) shadowWorker() error {
 			}
 		}
 
-		if len(scanResult.NextPageToken) == 0 && s.options.Mode == ShadowModeNormal {
-			return nil
+		if len(scanResult.NextPageToken) == 0 {
+			if s.options.Mode == ShadowModeNormal || s.clock.Now().Add(defaultWaitDurationPerIteration).After(expirationTime) {
+				return nil
+			}
+
+			s.clock.Sleep(defaultWaitDurationPerIteration)
 		}
 
 		scanRequest.NextPageToken = scanResult.NextPageToken

internal/workflow_shadower_test.go

@@ -21,6 +21,7 @@
 package internal
 
 import (
+	"sync"
 	"testing"
 	"time"
 
@@ -282,7 +283,7 @@ func (s *workflowShadowerSuite) TestShadowWorkerExitCondition_MaxShadowingCount(
 	s.NoError(s.testShadower.shadowWorker())
 }
 
-func (s *workflowShadowerSuite) TestShadowWorkerExitCondition_FullScan() {
+func (s *workflowShadowerSuite) TestShadowWorker_NormalMode() {
 	workflowExecutions := newTestWorkflowExecutions(10)
 	numScan := 3
 	totalWorkflows := len(workflowExecutions) * numScan
@@ -305,7 +306,49 @@ func (s *workflowShadowerSuite) TestShadowWorkerExitCondition_FullScan() {
 	s.NoError(s.testShadower.shadowWorker())
 }
 
-func (s *workflowShadowerSuite) TestShadowWorkerExitCondition_ReplayFailed() {
+func (s *workflowShadowerSuite) TestShadowWorker_ContinuousMode() {
+	workflowExecutions := newTestWorkflowExecutions(10)
+	numScan := 3
+	totalWorkflows := len(workflowExecutions) * numScan
+
+	s.testShadower.options.Mode = ShadowModeContinuous
+	s.testShadower.options.ExitCondition = ShadowExitCondition{
+		ShadowCount: totalWorkflows,
+	}
+
+	for i := 0; i != numScan; i++ {
+		scanResp := &shared.ListWorkflowExecutionsResponse{
+			Executions: workflowExecutions,
+		}
+		s.mockService.EXPECT().ScanWorkflowExecutions(gomock.Any(), gomock.Any(), callOptions...).Return(scanResp, nil).Times(1)
+	}
+
+	s.mockService.EXPECT().GetWorkflowExecutionHistory(gomock.Any(), gomock.Any(), callOptions...).Return(&shared.GetWorkflowExecutionHistoryResponse{
+		History: s.testWorkflowHistory,
+	}, nil).Times(totalWorkflows)
+
+	doneCh := make(chan struct{})
+	var advanceTimeWG sync.WaitGroup
+	advanceTimeWG.Add(1)
+	go func() {
+		defer advanceTimeWG.Done()
+		for {
+			time.Sleep(100 * time.Millisecond)
+			select {
+			case <-doneCh:
+				return
+			default:
+				s.testShadower.clock.(*clock.Mock).Add(defaultWaitDurationPerIteration)
+			}
+		}
+	}()
+
+	s.NoError(s.testShadower.shadowWorker())
+	close(doneCh)
+	advanceTimeWG.Wait()
+}
+
+func (s *workflowShadowerSuite) TestShadowWorker_ReplayFailed() {
 	successfullyReplayed := 5
 	s.mockService.EXPECT().ScanWorkflowExecutions(gomock.Any(), gomock.Any(), callOptions...).Return(&shared.ListWorkflowExecutionsResponse{
 		Executions:    newTestWorkflowExecutions(successfullyReplayed * 2),
@@ -321,7 +364,7 @@ func (s *workflowShadowerSuite) TestShadowWorkerExitCondition_ReplayFailed() {
 	s.Error(s.testShadower.shadowWorker())
 }
 
-func (s *workflowShadowerSuite) TestShadowWorkerExitCondition_ExpectedReplayError() {
+func (s *workflowShadowerSuite) TestShadowWorker_ExpectedReplayError() {
 	testCases := []struct {
 		msg                string
 		getHistoryErr      error

internal/workflow_shadower_worker.go

@@ -62,6 +62,12 @@ func newShadowWorker(
 	replayer := NewWorkflowReplayer()
 	replayer.registry = registry
 
+	if len(params.TaskList) != 0 {
+		// include domain name in tasklist to avoid confliction
+		// since all shadow workflow will be run in a single system domain
+		params.TaskList = generateShadowTaskList(domain, params.TaskList)
+	}
+
 	params.UserContext = context.WithValue(params.UserContext, serviceClientContextKey, service)
 	params.UserContext = context.WithValue(params.UserContext, workflowReplayerContextKey, replayer)
 
@@ -159,3 +165,7 @@ func (sw *shadowWorker) startShadowWorkflow() error {
 
 	return backoff.Retry(ctx, startWorkflowOp, createDynamicServiceRetryPolicy(ctx), isServiceTransientError)
 }
+
+func generateShadowTaskList(domain, taskList string) string {
+	return domain + "-" + taskList
+}

internal/workflow_shadower_worker_test.go

@@ -85,6 +85,9 @@ func (s *shadowWorkerSuite) TestNewShadowWorker() {
 	s.True(ok)
 	_, ok = userContext.Value(workflowReplayerContextKey).(*WorkflowReplayer)
 	s.True(ok)
+
+	taskList := shadowWorker.activityWorker.executionParameters.TaskList
+	s.Contains(taskList, testDomain)
 }
 
 func (s *shadowWorkerSuite) TestStartShadowWorker_Failed_ShadowOptionNotSpecified() {
@@ -226,7 +229,7 @@ func (s *shadowWorkerSuite) TestStartShadowWorker_Succeed() {
 	var workflowParams shadower.WorkflowParams
 	getDefaultDataConverter().FromData(startRequest.Input, &workflowParams)
 	s.Equal(testDomain, workflowParams.GetDomain())
-	s.Equal(testTaskList, workflowParams.GetTaskList())
+	s.Equal(generateShadowTaskList(testDomain, testTaskList), workflowParams.GetTaskList())
 	s.Equal(workflowQuery, workflowParams.GetWorkflowQuery())
 	s.Equal(samplingRate, workflowParams.GetSamplingRate())
 	s.Equal(shadowMode.toThriftPtr(), workflowParams.ShadowMode)

worker/worker.go

@@ -212,6 +212,8 @@ const (
 	ShadowModeNormal = internal.ShadowModeNormal
 	// ShadowModeContinuous mode will start a new round of shadowing
 	// after all workflows matches WorkflowQuery have been replayed.
+	// There will be a 5 min wait period between each round,
+	// currently this wait period is not configurable.
 	// Shadowing will complete only when ExitCondition is met.
 	// ExitCondition must be specified when using this mode
 	ShadowModeContinuous = internal.ShadowModeContinuous