Skip to content

Commit

Permalink
Add domain to pendingChildExecutionInfo (cadence-workflow#4611)
Browse files Browse the repository at this point in the history
- Include pending child domain name in describeWorkflowExecution response
  • Loading branch information
yycptt authored Nov 25, 2021
1 parent 80700d8 commit d3d0682
Show file tree
Hide file tree
Showing 17 changed files with 1,169 additions and 1,038 deletions.
72 changes: 68 additions & 4 deletions .gen/go/shared/shared.go

Large diffs are not rendered by default.

287 changes: 144 additions & 143 deletions .gen/proto/admin/v1/service.pb.yarpc.go

Large diffs are not rendered by default.

287 changes: 144 additions & 143 deletions .gen/proto/history/v1/service.pb.yarpc.go

Large diffs are not rendered by default.

287 changes: 144 additions & 143 deletions .gen/proto/matching/v1/service.pb.yarpc.go

Large diffs are not rendered by default.

287 changes: 144 additions & 143 deletions .gen/proto/shared/v1/error.pb.yarpc.go

Large diffs are not rendered by default.

287 changes: 144 additions & 143 deletions .gen/proto/shared/v1/history.pb.yarpc.go

Large diffs are not rendered by default.

287 changes: 144 additions & 143 deletions .gen/proto/shared/v1/queue.pb.yarpc.go

Large diffs are not rendered by default.

287 changes: 144 additions & 143 deletions .gen/proto/shared/v1/replication.pb.yarpc.go

Large diffs are not rendered by default.

6 changes: 4 additions & 2 deletions common/types/mapper/proto/api.go
Original file line number Diff line number Diff line change
Expand Up @@ -1996,8 +1996,9 @@ func FromPendingChildExecutionInfo(t *types.PendingChildExecutionInfo) *apiv1.Pe
return nil
}
return &apiv1.PendingChildExecutionInfo{
Domain: t.Domain,
WorkflowExecution: FromWorkflowRunPair(t.WorkflowID, t.RunID),
WorkflowTypeName: t.WorkflowTypName,
WorkflowTypeName: t.WorkflowTypeName,
InitiatedId: t.InitiatedID,
ParentClosePolicy: FromParentClosePolicy(t.ParentClosePolicy),
}
Expand All @@ -2008,9 +2009,10 @@ func ToPendingChildExecutionInfo(t *apiv1.PendingChildExecutionInfo) *types.Pend
return nil
}
return &types.PendingChildExecutionInfo{
Domain: t.Domain,
WorkflowID: ToWorkflowID(t.WorkflowExecution),
RunID: ToRunID(t.WorkflowExecution),
WorkflowTypName: t.WorkflowTypeName,
WorkflowTypeName: t.WorkflowTypeName,
InitiatedID: t.InitiatedId,
ParentClosePolicy: ToParentClosePolicy(t.ParentClosePolicy),
}
Expand Down
6 changes: 4 additions & 2 deletions common/types/mapper/thrift/shared.go
Original file line number Diff line number Diff line change
Expand Up @@ -3370,9 +3370,10 @@ func FromPendingChildExecutionInfo(t *types.PendingChildExecutionInfo) *shared.P
return nil
}
return &shared.PendingChildExecutionInfo{
Domain: &t.Domain,
WorkflowID: &t.WorkflowID,
RunID: &t.RunID,
WorkflowTypName: &t.WorkflowTypName,
WorkflowTypName: &t.WorkflowTypeName,
InitiatedID: &t.InitiatedID,
ParentClosePolicy: FromParentClosePolicy(t.ParentClosePolicy),
}
Expand All @@ -3384,9 +3385,10 @@ func ToPendingChildExecutionInfo(t *shared.PendingChildExecutionInfo) *types.Pen
return nil
}
return &types.PendingChildExecutionInfo{
Domain: t.GetDomain(),
WorkflowID: t.GetWorkflowID(),
RunID: t.GetRunID(),
WorkflowTypName: t.GetWorkflowTypName(),
WorkflowTypeName: t.GetWorkflowTypName(),
InitiatedID: t.GetInitiatedID(),
ParentClosePolicy: ToParentClosePolicy(t.ParentClosePolicy),
}
Expand Down
17 changes: 13 additions & 4 deletions common/types/shared.go
Original file line number Diff line number Diff line change
Expand Up @@ -5010,13 +5010,22 @@ const (

// PendingChildExecutionInfo is an internal type (TBD...)
type PendingChildExecutionInfo struct {
Domain string `json:"domain,omitempty"`
WorkflowID string `json:"workflowID,omitempty"`
RunID string `json:"runID,omitempty"`
WorkflowTypName string `json:"workflowTypName,omitempty"`
WorkflowTypeName string `json:"workflowTypeName,omitempty"`
InitiatedID int64 `json:"initiatedID,omitempty"`
ParentClosePolicy *ParentClosePolicy `json:"parentClosePolicy,omitempty"`
}

// GetDomain is an internal getter (TBD...)
func (v *PendingChildExecutionInfo) GetDomain() (o string) {
if v != nil {
return v.Domain
}
return
}

// GetWorkflowID is an internal getter (TBD...)
func (v *PendingChildExecutionInfo) GetWorkflowID() (o string) {
if v != nil {
Expand All @@ -5033,10 +5042,10 @@ func (v *PendingChildExecutionInfo) GetRunID() (o string) {
return
}

// GetWorkflowTypName is an internal getter (TBD...)
func (v *PendingChildExecutionInfo) GetWorkflowTypName() (o string) {
// GetWorkflowTypeName is an internal getter (TBD...)
func (v *PendingChildExecutionInfo) GetWorkflowTypeName() (o string) {
if v != nil {
return v.WorkflowTypName
return v.WorkflowTypeName
}
return
}
Expand Down
3 changes: 2 additions & 1 deletion common/types/testdata/common.go
Original file line number Diff line number Diff line change
Expand Up @@ -379,9 +379,10 @@ var (
&PendingActivityInfo,
}
PendingChildExecutionInfo = types.PendingChildExecutionInfo{
Domain: DomainName,
WorkflowID: WorkflowID,
RunID: RunID,
WorkflowTypName: WorkflowTypeName,
WorkflowTypeName: WorkflowTypeName,
InitiatedID: EventID1,
ParentClosePolicy: &ParentClosePolicy,
}
Expand Down
2 changes: 1 addition & 1 deletion go.mod
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,7 @@ require (
github.com/sirupsen/logrus v1.4.2
github.com/stretchr/testify v1.6.1
github.com/uber-go/tally v3.3.15+incompatible
github.com/uber/cadence-idl v0.0.0-20211111093939-c54ab85c0e0a
github.com/uber/cadence-idl v0.0.0-20211125000611-a6d3e3393ba6
github.com/uber/ringpop-go v0.8.5
github.com/uber/tchannel-go v1.22.0
github.com/uber/tcheck v1.1.0
Expand Down
4 changes: 2 additions & 2 deletions go.sum
Original file line number Diff line number Diff line change
Expand Up @@ -425,8 +425,8 @@ github.com/uber-go/mapdecode v1.0.0/go.mod h1:b5nP15FwXTgpjTjeA9A2uTHXV5UJCl4arw
github.com/uber-go/tally v3.3.12+incompatible/go.mod h1:YDTIBxdXyOU/sCWilKB4bgyufu1cEi0jdVnRdxvjnmU=
github.com/uber-go/tally v3.3.15+incompatible h1:9hLSgNBP28CjIaDmAuRTq9qV+UZY+9PcvAkXO4nNMwg=
github.com/uber-go/tally v3.3.15+incompatible/go.mod h1:YDTIBxdXyOU/sCWilKB4bgyufu1cEi0jdVnRdxvjnmU=
github.com/uber/cadence-idl v0.0.0-20211111093939-c54ab85c0e0a h1:wTcqQwjWrtol5Dy0yVTzGkjoxIKrWvv/8B7mJSFteZM=
github.com/uber/cadence-idl v0.0.0-20211111093939-c54ab85c0e0a/go.mod h1:oyUK7GCNCRHCCyWyzifSzXpVrRYVBbAMHAzF5dXiKws=
github.com/uber/cadence-idl v0.0.0-20211125000611-a6d3e3393ba6 h1:eH1W60bvlJCN0KFJMJuxNLxtR+hM73/FlwkaMlyV/D4=
github.com/uber/cadence-idl v0.0.0-20211125000611-a6d3e3393ba6/go.mod h1:oyUK7GCNCRHCCyWyzifSzXpVrRYVBbAMHAzF5dXiKws=
github.com/uber/jaeger-client-go v2.22.1+incompatible h1:NHcubEkVbahf9t3p75TOCR83gdUHXjRJvjoBh1yACsM=
github.com/uber/jaeger-client-go v2.22.1+incompatible/go.mod h1:WVhlPFC8FDjOFMMWRy2pZqQJSXxYSwNYOkTr/Z6d3Kk=
github.com/uber/jaeger-lib v2.2.0+incompatible h1:MxZXOiR2JuoANZ3J6DE/U0kSFv/eJ/GfSYVCjK7dyaw=
Expand Down
66 changes: 49 additions & 17 deletions host/integration_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -1789,6 +1789,10 @@ func (s *IntegrationSuite) TestDescribeWorkflowExecution() {
workflowType := &types.WorkflowType{Name: wt}
taskList := &types.TaskList{Name: tl}

childID := id + "-child"
childType := wt + "-child"
childTaskList := tl + "-child"

// Start workflow execution
request := &types.StartWorkflowExecutionRequest{
RequestID: uuid.New(),
Expand Down Expand Up @@ -1831,19 +1835,33 @@ func (s *IntegrationSuite) TestDescribeWorkflowExecution() {
signalSent = true

s.NoError(err)
return nil, []*types.Decision{{
DecisionType: types.DecisionTypeScheduleActivityTask.Ptr(),
ScheduleActivityTaskDecisionAttributes: &types.ScheduleActivityTaskDecisionAttributes{
ActivityID: "1",
ActivityType: &types.ActivityType{Name: "test-activity-type"},
TaskList: &types.TaskList{Name: tl},
Input: []byte("test-input"),
ScheduleToCloseTimeoutSeconds: common.Int32Ptr(100),
ScheduleToStartTimeoutSeconds: common.Int32Ptr(2),
StartToCloseTimeoutSeconds: common.Int32Ptr(50),
HeartbeatTimeoutSeconds: common.Int32Ptr(5),
return nil, []*types.Decision{
{
DecisionType: types.DecisionTypeScheduleActivityTask.Ptr(),
ScheduleActivityTaskDecisionAttributes: &types.ScheduleActivityTaskDecisionAttributes{
ActivityID: "1",
ActivityType: &types.ActivityType{Name: "test-activity-type"},
TaskList: &types.TaskList{Name: tl},
Input: []byte("test-input"),
ScheduleToCloseTimeoutSeconds: common.Int32Ptr(100),
ScheduleToStartTimeoutSeconds: common.Int32Ptr(2),
StartToCloseTimeoutSeconds: common.Int32Ptr(50),
HeartbeatTimeoutSeconds: common.Int32Ptr(5),
},
},
}}, nil
{
DecisionType: types.DecisionTypeStartChildWorkflowExecution.Ptr(),
StartChildWorkflowExecutionDecisionAttributes: &types.StartChildWorkflowExecutionDecisionAttributes{
WorkflowID: childID,
WorkflowType: &types.WorkflowType{Name: childType},
TaskList: &types.TaskList{Name: childTaskList},
Input: []byte("child-workflow-input"),
ExecutionStartToCloseTimeoutSeconds: common.Int32Ptr(200),
TaskStartToCloseTimeoutSeconds: common.Int32Ptr(2),
Control: nil,
},
},
}, nil
}

workflowComplete = true
Expand Down Expand Up @@ -1876,21 +1894,35 @@ func (s *IntegrationSuite) TestDescribeWorkflowExecution() {
s.Logger.Info("PollAndProcessDecisionTask", tag.Error(err))
s.Nil(err)

dweResponse, err = describeWorkflowExecution()
s.Nil(err)
// wait for child workflow to start
for i := 0; i != 10; i++ {
dweResponse, err = describeWorkflowExecution()
s.Nil(err)
if len(dweResponse.PendingChildren) == 1 &&
dweResponse.PendingChildren[0].GetRunID() != "" {
break
}
time.Sleep(100 * time.Millisecond)
}
s.NotEmpty(dweResponse.PendingChildren[0].GetRunID(), "unable to start child workflow")
s.True(nil == dweResponse.WorkflowExecutionInfo.CloseStatus)
s.Equal(int64(5), dweResponse.WorkflowExecutionInfo.HistoryLength) // DecisionStarted, DecisionCompleted, ActivityScheduled
// DecisionStarted, DecisionCompleted, ActivityScheduled, ChildWorkflowInit, ChildWorkflowStarted, DecisionTaskScheduled
s.Equal(int64(8), dweResponse.WorkflowExecutionInfo.HistoryLength)
s.Equal(1, len(dweResponse.PendingActivities))
s.Equal("test-activity-type", dweResponse.PendingActivities[0].ActivityType.GetName())
s.Equal(int64(0), dweResponse.PendingActivities[0].GetLastHeartbeatTimestamp())
s.Equal(1, len(dweResponse.PendingChildren))
s.Equal(s.domainName, dweResponse.PendingChildren[0].GetDomain())
s.Equal(childID, dweResponse.PendingChildren[0].GetWorkflowID())
s.Equal(childType, dweResponse.PendingChildren[0].GetWorkflowTypeName())

// process activity task
err = poller.PollAndProcessActivityTask(false)

dweResponse, err = describeWorkflowExecution()
s.Nil(err)
s.True(nil == dweResponse.WorkflowExecutionInfo.CloseStatus)
s.Equal(int64(8), dweResponse.WorkflowExecutionInfo.HistoryLength) // ActivityTaskStarted, ActivityTaskCompleted, DecisionTaskScheduled
s.Equal(int64(10), dweResponse.WorkflowExecutionInfo.HistoryLength) // ActivityTaskStarted, ActivityTaskCompleted
s.Equal(0, len(dweResponse.PendingActivities))

// Process signal in decider
Expand All @@ -1901,7 +1933,7 @@ func (s *IntegrationSuite) TestDescribeWorkflowExecution() {
dweResponse, err = describeWorkflowExecution()
s.Nil(err)
s.Equal(types.WorkflowExecutionCloseStatusCompleted, *dweResponse.WorkflowExecutionInfo.CloseStatus)
s.Equal(int64(11), dweResponse.WorkflowExecutionInfo.HistoryLength) // DecisionStarted, DecisionCompleted, WorkflowCompleted
s.Equal(int64(13), dweResponse.WorkflowExecutionInfo.HistoryLength) // DecisionStarted, DecisionCompleted, WorkflowCompleted
}

func (s *IntegrationSuite) TestVisibility() {
Expand Down
2 changes: 1 addition & 1 deletion idls
Submodule idls updated from d6b70b to a6d3e3
20 changes: 17 additions & 3 deletions service/history/historyEngine.go
Original file line number Diff line number Diff line change
Expand Up @@ -1491,9 +1491,9 @@ func (e *historyEngineImpl) DescribeWorkflowExecution(
}

domainID := request.DomainUUID
execution := *request.Request.Execution
wfExecution := *request.Request.Execution

wfContext, release, err0 := e.executionCache.GetOrCreateWorkflowExecution(ctx, domainID, execution)
wfContext, release, err0 := e.executionCache.GetOrCreateWorkflowExecution(ctx, domainID, wfExecution)
if err0 != nil {
return nil, err0
}
Expand Down Expand Up @@ -1607,10 +1607,24 @@ func (e *historyEngineImpl) DescribeWorkflowExecution(

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,
WorkflowTypName: ch.WorkflowTypeName,
WorkflowTypeName: ch.WorkflowTypeName,
InitiatedID: ch.InitiatedID,
ParentClosePolicy: &ch.ParentClosePolicy,
}
Expand Down

0 comments on commit d3d0682

Please sign in to comment.