[Wf-Diagnostics] retrieve workflow execution history within issue identification activity (#6607)

Author: sankari165

service/worker/diagnostics/activities.go

@@ -39,30 +39,26 @@ const (
 	WfDiagnosticsAppName  = "workflow-diagnostics"
 )
 
-type retrieveExecutionHistoryInputParams struct {
-	Domain    string
+type identifyIssuesParams struct {
 	Execution *types.WorkflowExecution
+	Domain    string
 }
 
-func (w *dw) retrieveExecutionHistory(ctx context.Context, info retrieveExecutionHistoryInputParams) (*types.GetWorkflowExecutionHistoryResponse, error) {
+func (w *dw) identifyIssues(ctx context.Context, info identifyIssuesParams) ([]invariant.InvariantCheckResult, error) {
+	result := make([]invariant.InvariantCheckResult, 0)
+
 	frontendClient := w.clientBean.GetFrontendClient()
-	return frontendClient.GetWorkflowExecutionHistory(ctx, &types.GetWorkflowExecutionHistoryRequest{
+	history, err := frontendClient.GetWorkflowExecutionHistory(ctx, &types.GetWorkflowExecutionHistoryRequest{
 		Domain:    info.Domain,
 		Execution: info.Execution,
 	})
-}
-
-type identifyIssuesParams struct {
-	History *types.GetWorkflowExecutionHistoryResponse
-	Domain  string
-}
-
-func (w *dw) identifyIssues(ctx context.Context, info identifyIssuesParams) ([]invariant.InvariantCheckResult, error) {
-	result := make([]invariant.InvariantCheckResult, 0)
+	if err != nil {
+		return nil, err
+	}
 
 	for _, inv := range w.invariants {
 		issues, err := inv.Check(ctx, invariant.InvariantCheckInput{
-			WorkflowExecutionHistory: info.History,
+			WorkflowExecutionHistory: history,
 			Domain:                   info.Domain,
 		})
 		if err != nil {

service/worker/diagnostics/activities_test.go

@@ -48,19 +48,6 @@ const (
 	timeUnit              = time.Second
 )
 
-func Test__retrieveExecutionHistory(t *testing.T) {
-	dwtest := testDiagnosticWorkflow(t)
-	result, err := dwtest.retrieveExecutionHistory(context.Background(), retrieveExecutionHistoryInputParams{
-		Domain: "test",
-		Execution: &types.WorkflowExecution{
-			WorkflowID: "123",
-			RunID:      "abc",
-		},
-	})
-	require.NoError(t, err)
-	require.Equal(t, testWorkflowExecutionHistoryResponse(), result)
-}
-
 func Test__identifyIssues(t *testing.T) {
 	dwtest := testDiagnosticWorkflow(t)
 	actMetadata := failure.FailureMetadata{
@@ -100,7 +87,10 @@ func Test__identifyIssues(t *testing.T) {
 			Metadata:      retryMetadataInBytes,
 		},
 	}
-	result, err := dwtest.identifyIssues(context.Background(), identifyIssuesParams{History: testWorkflowExecutionHistoryResponse()})
+	result, err := dwtest.identifyIssues(context.Background(), identifyIssuesParams{Execution: &types.WorkflowExecution{
+		WorkflowID: "123",
+		RunID:      "abc",
+	}})
 	require.NoError(t, err)
 	require.Equal(t, expectedResult, result)
 }

service/worker/diagnostics/module.go

@@ -91,7 +91,6 @@ func (w *dw) Start() error {
 	newWorker := worker.New(w.svcClient, common.SystemLocalDomainName, tasklist, workerOpts)
 	newWorker.RegisterWorkflowWithOptions(w.DiagnosticsWorkflow, workflow.RegisterOptions{Name: diagnosticsWorkflow})
 	newWorker.RegisterWorkflowWithOptions(w.DiagnosticsStarterWorkflow, workflow.RegisterOptions{Name: diagnosticsStarterWorkflow})
-	newWorker.RegisterActivityWithOptions(w.retrieveExecutionHistory, activity.RegisterOptions{Name: retrieveWfExecutionHistoryActivity})
 	newWorker.RegisterActivityWithOptions(w.identifyIssues, activity.RegisterOptions{Name: identifyIssuesActivity})
 	newWorker.RegisterActivityWithOptions(w.rootCauseIssues, activity.RegisterOptions{Name: rootCauseIssuesActivity})
 	newWorker.RegisterActivityWithOptions(w.emitUsageLogs, activity.RegisterOptions{Name: emitUsageLogsActivity})

service/worker/diagnostics/workflow.go

@@ -125,23 +125,14 @@ func (w *dw) DiagnosticsWorkflow(ctx workflow.Context, params DiagnosticsWorkflo
 		StartToCloseTimeout:    time.Second * 5,
 	}
 	activityCtx := workflow.WithActivityOptions(ctx, activityOptions)
+	timeoutsResult.Runbooks = []string{linkToTimeoutsRunbook}
 
-	var wfExecutionHistory *types.GetWorkflowExecutionHistoryResponse
-	err := workflow.ExecuteActivity(activityCtx, w.retrieveExecutionHistory, retrieveExecutionHistoryInputParams{
-		Domain: params.Domain,
+	err := workflow.ExecuteActivity(activityCtx, w.identifyIssues, identifyIssuesParams{
 		Execution: &types.WorkflowExecution{
 			WorkflowID: params.WorkflowID,
 			RunID:      params.RunID,
-		}}).Get(ctx, &wfExecutionHistory)
-	if err != nil {
-		return nil, fmt.Errorf("RetrieveExecutionHistory: %w", err)
-	}
-
-	timeoutsResult.Runbooks = []string{linkToTimeoutsRunbook}
-
-	err = workflow.ExecuteActivity(activityCtx, w.identifyIssues, identifyIssuesParams{
-		History: wfExecutionHistory,
-		Domain:  params.Domain,
+		},
+		Domain: params.Domain,
 	}).Get(ctx, &checkResult)
 	if err != nil {
 		return nil, fmt.Errorf("IdentifyIssues: %w", err)

service/worker/diagnostics/workflow_test.go

@@ -75,7 +75,6 @@ func (s *diagnosticsWorkflowTestSuite) SetupTest() {
 
 	s.workflowEnv.RegisterWorkflowWithOptions(s.dw.DiagnosticsStarterWorkflow, workflow.RegisterOptions{Name: diagnosticsStarterWorkflow})
 	s.workflowEnv.RegisterWorkflowWithOptions(s.dw.DiagnosticsWorkflow, workflow.RegisterOptions{Name: diagnosticsWorkflow})
-	s.workflowEnv.RegisterActivityWithOptions(s.dw.retrieveExecutionHistory, activity.RegisterOptions{Name: retrieveWfExecutionHistoryActivity})
 	s.workflowEnv.RegisterActivityWithOptions(s.dw.identifyIssues, activity.RegisterOptions{Name: identifyIssuesActivity})
 	s.workflowEnv.RegisterActivityWithOptions(s.dw.rootCauseIssues, activity.RegisterOptions{Name: rootCauseIssuesActivity})
 	s.workflowEnv.RegisterActivityWithOptions(s.dw.emitUsageLogs, activity.RegisterOptions{Name: emitUsageLogsActivity})
@@ -133,7 +132,6 @@ func (s *diagnosticsWorkflowTestSuite) TestWorkflow() {
 			PollersMetadata: &timeout.PollersMetadata{TaskListBacklog: taskListBacklog},
 		},
 	}
-	s.workflowEnv.OnActivity(retrieveWfExecutionHistoryActivity, mock.Anything, mock.Anything).Return(nil, nil)
 	s.workflowEnv.OnActivity(identifyIssuesActivity, mock.Anything, mock.Anything).Return(issues, nil)
 	s.workflowEnv.OnActivity(rootCauseIssuesActivity, mock.Anything, mock.Anything).Return(rootCause, nil)
 	s.workflowEnv.OnActivity(emitUsageLogsActivity, mock.Anything, mock.Anything).Return(nil)
@@ -157,7 +155,6 @@ func (s *diagnosticsWorkflowTestSuite) TestWorkflow_Error() {
 	}
 	mockErr := errors.New("mockErr")
 	errExpected := fmt.Errorf("IdentifyIssues: %w", mockErr)
-	s.workflowEnv.OnActivity(retrieveWfExecutionHistoryActivity, mock.Anything, mock.Anything).Return(nil, nil)
 	s.workflowEnv.OnActivity(identifyIssuesActivity, mock.Anything, mock.Anything).Return(nil, mockErr)
 	s.workflowEnv.ExecuteWorkflow(diagnosticsWorkflow, params)
 	s.True(s.workflowEnv.IsWorkflowCompleted())