Return ActiveClusterSelectionPolicy on DescribeWorkflow requests (#7181)

Author: c-warren

service/history/engine/engineimpl/describe_workflow_execution.go

@@ -26,6 +26,7 @@ import (
 	"time"
 
 	"github.com/uber/cadence/common"
+	"github.com/uber/cadence/common/cache"
 	"github.com/uber/cadence/common/constants"
 	"github.com/uber/cadence/common/persistence"
 	"github.com/uber/cadence/common/types"
@@ -37,8 +38,7 @@ func (e *historyEngineImpl) DescribeWorkflowExecution(
 	ctx context.Context,
 	request *types.HistoryDescribeWorkflowExecutionRequest,
 ) (retResp *types.DescribeWorkflowExecutionResponse, retError error) {
-
-	if err := common.ValidateDomainUUID(request.DomainUUID); err != nil {
+	if err := validateDescribeWorkflowExecutionRequest(request); err != nil {
 		return nil, err
 	}
 
@@ -55,41 +55,37 @@ func (e *historyEngineImpl) DescribeWorkflowExecution(
 	if err1 != nil {
 		return nil, err1
 	}
+
 	// If history is corrupted, return an error to the end user
 	if corrupted, err := e.checkForHistoryCorruptions(ctx, mutableState); err != nil {
 		return nil, err
 	} else if corrupted {
 		return nil, &types.EntityNotExistsError{Message: "Workflow execution corrupted."}
 	}
 
+	domainCache := e.shard.GetDomainCache()
+	result, err := createDescribeWorkflowExecutionResponse(ctx, mutableState, domainCache)
+	if err != nil {
+		return nil, err
+	}
+
+	return result, nil
+}
+
+// validateDescribeWorkflowExecutionRequest validates the input request
+func validateDescribeWorkflowExecutionRequest(request *types.HistoryDescribeWorkflowExecutionRequest) error {
+	return common.ValidateDomainUUID(request.DomainUUID)
+}
+
+func createDescribeWorkflowExecutionResponse(ctx context.Context, mutableState execution.MutableState, domainCache cache.DomainCache) (*types.DescribeWorkflowExecutionResponse, error) {
 	executionInfo := mutableState.GetExecutionInfo()
+	executionConfiguration, err := mapWorkflowExecutionConfiguration(executionInfo)
+	if err != nil {
+		return nil, err
+	}
 
 	result := &types.DescribeWorkflowExecutionResponse{
-		ExecutionConfiguration: &types.WorkflowExecutionConfiguration{
-			TaskList:                            &types.TaskList{Name: executionInfo.TaskList},
-			ExecutionStartToCloseTimeoutSeconds: common.Int32Ptr(executionInfo.WorkflowTimeout),
-			TaskStartToCloseTimeoutSeconds:      common.Int32Ptr(executionInfo.DecisionStartToCloseTimeout),
-		},
-		WorkflowExecutionInfo: &types.WorkflowExecutionInfo{
-			Execution: &types.WorkflowExecution{
-				WorkflowID: executionInfo.WorkflowID,
-				RunID:      executionInfo.RunID,
-			},
-			TaskList: &types.TaskList{
-				Name: executionInfo.TaskList,
-				Kind: executionInfo.TaskListKind.Ptr(),
-			},
-			Type:              &types.WorkflowType{Name: executionInfo.WorkflowTypeName},
-			StartTime:         common.Int64Ptr(executionInfo.StartTimestamp.UnixNano()),
-			HistoryLength:     mutableState.GetNextEventID() - constants.FirstEventID,
-			AutoResetPoints:   executionInfo.AutoResetPoints,
-			Memo:              &types.Memo{Fields: executionInfo.CopyMemo()},
-			IsCron:            len(executionInfo.CronSchedule) > 0,
-			UpdateTime:        common.Int64Ptr(executionInfo.LastUpdatedTimestamp.UnixNano()),
-			SearchAttributes:  &types.SearchAttributes{IndexedFields: executionInfo.CopySearchAttributes()},
-			PartitionConfig:   executionInfo.CopyPartitionConfig(),
-			CronOverlapPolicy: &executionInfo.CronOverlapPolicy,
-		},
+		ExecutionConfiguration: executionConfiguration,
 	}
 
 	// TODO: we need to consider adding execution time to mutable state
@@ -99,125 +95,197 @@ func (e *historyEngineImpl) DescribeWorkflowExecution(
 	if err != nil {
 		return nil, err
 	}
+
+	var completionEvent *types.HistoryEvent
+	if executionInfo.State == persistence.WorkflowStateCompleted {
+		completionEvent, err = mutableState.GetCompletionEvent(ctx)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	historyLength := mutableState.GetNextEventID() - constants.FirstEventID
+	workflowExecutionInfo, err := mapWorkflowExecutionInfo(executionInfo, startEvent, domainCache, historyLength, completionEvent)
+	if err != nil {
+		return nil, err
+	}
+	result.WorkflowExecutionInfo = workflowExecutionInfo
+
+	pendingActivityInfos := mutableState.GetPendingActivityInfos()
+	for _, ai := range pendingActivityInfos {
+		scheduledEvent, err := mutableState.GetActivityScheduledEvent(ctx, ai.ScheduleID)
+		if err != nil {
+			return nil, err
+		}
+
+		p := mapPendingActivityInfo(ai, scheduledEvent)
+		result.PendingActivities = append(result.PendingActivities, p)
+	}
+
+	childExecutions := mutableState.GetPendingChildExecutionInfos()
+	domainEntry := mutableState.GetDomainEntry()
+	for _, childExecution := range childExecutions {
+		pendingChild, err := mapPendingChildExecutionInfo(childExecution, domainEntry, domainCache)
+		if err != nil {
+			return nil, err
+		}
+		result.PendingChildren = append(result.PendingChildren, pendingChild)
+	}
+
+	if di, ok := mutableState.GetPendingDecision(); ok {
+		result.PendingDecision = mapPendingDecisionInfo(di)
+	}
+
+	return result, nil
+}
+
+func mapWorkflowExecutionConfiguration(executionInfo *persistence.WorkflowExecutionInfo) (*types.WorkflowExecutionConfiguration, error) {
+	return &types.WorkflowExecutionConfiguration{
+		TaskList:                            mapDecisionInfoToTaskList(executionInfo),
+		ExecutionStartToCloseTimeoutSeconds: common.Int32Ptr(executionInfo.WorkflowTimeout),
+		TaskStartToCloseTimeoutSeconds:      common.Int32Ptr(executionInfo.DecisionStartToCloseTimeout),
+	}, nil
+}
+
+func mapWorkflowExecutionInfo(executionInfo *persistence.WorkflowExecutionInfo, startEvent *types.HistoryEvent, domainCache cache.DomainCache, historyLength int64, completionEvent *types.HistoryEvent) (*types.WorkflowExecutionInfo, error) {
+	result := &types.WorkflowExecutionInfo{
+		Execution: &types.WorkflowExecution{
+			WorkflowID: executionInfo.WorkflowID,
+			RunID:      executionInfo.RunID,
+		},
+		TaskList:                     mapDecisionInfoToTaskList(executionInfo),
+		Type:                         &types.WorkflowType{Name: executionInfo.WorkflowTypeName},
+		StartTime:                    common.Int64Ptr(executionInfo.StartTimestamp.UnixNano()),
+		HistoryLength:                historyLength,
+		AutoResetPoints:              executionInfo.AutoResetPoints,
+		Memo:                         &types.Memo{Fields: executionInfo.CopyMemo()},
+		IsCron:                       len(executionInfo.CronSchedule) > 0,
+		UpdateTime:                   common.Int64Ptr(executionInfo.LastUpdatedTimestamp.UnixNano()),
+		SearchAttributes:             &types.SearchAttributes{IndexedFields: executionInfo.CopySearchAttributes()},
+		PartitionConfig:              executionInfo.CopyPartitionConfig(),
+		CronOverlapPolicy:            &executionInfo.CronOverlapPolicy,
+		ActiveClusterSelectionPolicy: executionInfo.ActiveClusterSelectionPolicy,
+	}
+
 	backoffDuration := time.Duration(startEvent.GetWorkflowExecutionStartedEventAttributes().GetFirstDecisionTaskBackoffSeconds()) * time.Second
-	result.WorkflowExecutionInfo.ExecutionTime = common.Int64Ptr(result.WorkflowExecutionInfo.GetStartTime() + backoffDuration.Nanoseconds())
+	result.ExecutionTime = common.Int64Ptr(result.GetStartTime() + backoffDuration.Nanoseconds())
 
 	if executionInfo.ParentRunID != "" {
-		result.WorkflowExecutionInfo.ParentExecution = &types.WorkflowExecution{
+		result.ParentExecution = &types.WorkflowExecution{
 			WorkflowID: executionInfo.ParentWorkflowID,
 			RunID:      executionInfo.ParentRunID,
 		}
-		result.WorkflowExecutionInfo.ParentDomainID = common.StringPtr(executionInfo.ParentDomainID)
-		result.WorkflowExecutionInfo.ParentInitiatedID = common.Int64Ptr(executionInfo.InitiatedID)
-		parentDomain, err := e.shard.GetDomainCache().GetDomainName(executionInfo.ParentDomainID)
+		result.ParentDomainID = common.StringPtr(executionInfo.ParentDomainID)
+		result.ParentInitiatedID = common.Int64Ptr(executionInfo.InitiatedID)
+		parentDomain, err := domainCache.GetDomainName(executionInfo.ParentDomainID)
 		if err != nil {
 			return nil, err
 		}
-		result.WorkflowExecutionInfo.ParentDomain = common.StringPtr(parentDomain)
+		result.ParentDomain = common.StringPtr(parentDomain)
 	}
+
 	if executionInfo.State == persistence.WorkflowStateCompleted {
-		// for closed workflow
-		result.WorkflowExecutionInfo.CloseStatus = persistence.ToInternalWorkflowExecutionCloseStatus(executionInfo.CloseStatus)
-		completionEvent, err := mutableState.GetCompletionEvent(ctx)
-		if err != nil {
-			return nil, err
-		}
-		result.WorkflowExecutionInfo.CloseTime = common.Int64Ptr(completionEvent.GetTimestamp())
-	}
-
-	if len(mutableState.GetPendingActivityInfos()) > 0 {
-		for _, ai := range mutableState.GetPendingActivityInfos() {
-			p := &types.PendingActivityInfo{
-				ActivityID: ai.ActivityID,
-				ScheduleID: ai.ScheduleID,
-			}
-			state := types.PendingActivityStateScheduled
-			if ai.CancelRequested {
-				state = types.PendingActivityStateCancelRequested
-			} else if ai.StartedID != constants.EmptyEventID {
-				state = types.PendingActivityStateStarted
-			}
-			p.State = &state
-			lastHeartbeatUnixNano := ai.LastHeartBeatUpdatedTime.UnixNano()
-			if lastHeartbeatUnixNano > 0 {
-				p.LastHeartbeatTimestamp = common.Int64Ptr(lastHeartbeatUnixNano)
-				p.HeartbeatDetails = ai.Details
-			}
-			// TODO: move to mutable state instead of loading it from event
-			scheduledEvent, err := mutableState.GetActivityScheduledEvent(ctx, ai.ScheduleID)
-			if err != nil {
-				return nil, err
-			}
-			p.ActivityType = scheduledEvent.ActivityTaskScheduledEventAttributes.ActivityType
-			if state == types.PendingActivityStateScheduled {
-				p.ScheduledTimestamp = common.Int64Ptr(ai.ScheduledTime.UnixNano())
-			} else {
-				p.LastStartedTimestamp = common.Int64Ptr(ai.StartedTime.UnixNano())
-			}
-			if ai.HasRetryPolicy {
-				p.Attempt = ai.Attempt
-				if !ai.ExpirationTime.IsZero() {
-					p.ExpirationTimestamp = common.Int64Ptr(ai.ExpirationTime.UnixNano())
-				}
-				if ai.MaximumAttempts != 0 {
-					p.MaximumAttempts = ai.MaximumAttempts
-				}
-				if ai.LastFailureReason != "" {
-					p.LastFailureReason = common.StringPtr(ai.LastFailureReason)
-					p.LastFailureDetails = ai.LastFailureDetails
-				}
-				if ai.LastWorkerIdentity != "" {
-					p.LastWorkerIdentity = ai.LastWorkerIdentity
-				}
-				if ai.StartedIdentity != "" {
-					p.StartedWorkerIdentity = ai.StartedIdentity
-				}
-			}
-			result.PendingActivities = append(result.PendingActivities, p)
+		result.CloseStatus = persistence.ToInternalWorkflowExecutionCloseStatus(executionInfo.CloseStatus)
+		if completionEvent != nil {
+			result.CloseTime = common.Int64Ptr(completionEvent.GetTimestamp())
 		}
 	}
 
-	if len(mutableState.GetPendingChildExecutionInfos()) > 0 {
-		for _, ch := range mutableState.GetPendingChildExecutionInfos() {
-			childDomainName, err := execution.GetChildExecutionDomainName(
-				ch,
-				e.shard.GetDomainCache(),
-				mutableState.GetDomainEntry(),
-			)
-			if err != nil {
-				if !common.IsEntityNotExistsError(err) {
-					return nil, err
-				}
-				// child domain already deleted, instead of failing the request,
-				// return domainID instead since this field is only for information purpose
-				childDomainName = ch.DomainID
-			}
-			p := &types.PendingChildExecutionInfo{
-				Domain:            childDomainName,
-				WorkflowID:        ch.StartedWorkflowID,
-				RunID:             ch.StartedRunID,
-				WorkflowTypeName:  ch.WorkflowTypeName,
-				InitiatedID:       ch.InitiatedID,
-				ParentClosePolicy: &ch.ParentClosePolicy,
-			}
-			result.PendingChildren = append(result.PendingChildren, p)
-		}
+	return result, nil
+}
+
+func mapPendingActivityInfo(ai *persistence.ActivityInfo, activityScheduledEvent *types.HistoryEvent) *types.PendingActivityInfo {
+	p := &types.PendingActivityInfo{
+		ActivityID: ai.ActivityID,
+		ScheduleID: ai.ScheduleID,
 	}
 
-	if di, ok := mutableState.GetPendingDecision(); ok {
-		pendingDecision := &types.PendingDecisionInfo{
-			State:                      types.PendingDecisionStateScheduled.Ptr(),
-			ScheduledTimestamp:         common.Int64Ptr(di.ScheduledTimestamp),
-			Attempt:                    di.Attempt,
-			OriginalScheduledTimestamp: common.Int64Ptr(di.OriginalScheduledTimestamp),
-			ScheduleID:                 di.ScheduleID,
+	state := types.PendingActivityStateScheduled
+	if ai.CancelRequested {
+		state = types.PendingActivityStateCancelRequested
+	} else if ai.StartedID != constants.EmptyEventID {
+		state = types.PendingActivityStateStarted
+	}
+	p.State = &state
+
+	lastHeartbeatUnixNano := ai.LastHeartBeatUpdatedTime.UnixNano()
+	if lastHeartbeatUnixNano > 0 {
+		p.LastHeartbeatTimestamp = common.Int64Ptr(lastHeartbeatUnixNano)
+		p.HeartbeatDetails = ai.Details
+	}
+
+	p.ActivityType = activityScheduledEvent.ActivityTaskScheduledEventAttributes.ActivityType
+	if state == types.PendingActivityStateScheduled {
+		p.ScheduledTimestamp = common.Int64Ptr(ai.ScheduledTime.UnixNano())
+	} else {
+		p.LastStartedTimestamp = common.Int64Ptr(ai.StartedTime.UnixNano())
+	}
+
+	if ai.HasRetryPolicy {
+		p.Attempt = ai.Attempt
+		if !ai.ExpirationTime.IsZero() {
+			p.ExpirationTimestamp = common.Int64Ptr(ai.ExpirationTime.UnixNano())
+		}
+		if ai.MaximumAttempts != 0 {
+			p.MaximumAttempts = ai.MaximumAttempts
+		}
+		if ai.LastFailureReason != "" {
+			p.LastFailureReason = common.StringPtr(ai.LastFailureReason)
+			p.LastFailureDetails = ai.LastFailureDetails
 		}
-		if di.StartedID != constants.EmptyEventID {
-			pendingDecision.State = types.PendingDecisionStateStarted.Ptr()
-			pendingDecision.StartedTimestamp = common.Int64Ptr(di.StartedTimestamp)
+		if ai.LastWorkerIdentity != "" {
+			p.LastWorkerIdentity = ai.LastWorkerIdentity
+		}
+		if ai.StartedIdentity != "" {
+			p.StartedWorkerIdentity = ai.StartedIdentity
 		}
-		result.PendingDecision = pendingDecision
 	}
 
-	return result, nil
+	return p
+}
+
+func mapPendingChildExecutionInfo(childExecution *persistence.ChildExecutionInfo, domainEntry *cache.DomainCacheEntry, domainCache cache.DomainCache) (*types.PendingChildExecutionInfo, error) {
+	childDomainName, err := execution.GetChildExecutionDomainName(
+		childExecution,
+		domainCache,
+		domainEntry,
+	)
+	if err != nil {
+		if !common.IsEntityNotExistsError(err) {
+			return nil, err
+		}
+		// child domain already deleted, instead of failing the request,
+		// return domainID instead since this field is only for information purpose
+		childDomainName = childExecution.DomainID
+	}
+	return &types.PendingChildExecutionInfo{
+		Domain:            childDomainName,
+		WorkflowID:        childExecution.StartedWorkflowID,
+		RunID:             childExecution.StartedRunID,
+		WorkflowTypeName:  childExecution.WorkflowTypeName,
+		InitiatedID:       childExecution.InitiatedID,
+		ParentClosePolicy: &childExecution.ParentClosePolicy,
+	}, nil
+}
+
+func mapPendingDecisionInfo(di *execution.DecisionInfo) *types.PendingDecisionInfo {
+	pendingDecision := &types.PendingDecisionInfo{
+		State:                      types.PendingDecisionStateScheduled.Ptr(),
+		ScheduledTimestamp:         common.Int64Ptr(di.ScheduledTimestamp),
+		Attempt:                    di.Attempt,
+		OriginalScheduledTimestamp: common.Int64Ptr(di.OriginalScheduledTimestamp),
+		ScheduleID:                 di.ScheduleID,
+	}
+	if di.StartedID != constants.EmptyEventID {
+		pendingDecision.State = types.PendingDecisionStateStarted.Ptr()
+		pendingDecision.StartedTimestamp = common.Int64Ptr(di.StartedTimestamp)
+	}
+	return pendingDecision
+}
+
+func mapDecisionInfoToTaskList(executionInfo *persistence.WorkflowExecutionInfo) *types.TaskList {
+	return &types.TaskList{
+		Name: executionInfo.TaskList,
+		Kind: executionInfo.TaskListKind.Ptr(),
+	}
 }