go.temporal.io/server/tests/admin_test.go
148 LOC · 0 covered · 148 uncovered · 0 ranges · 0 concepts · 0 introducers · 0 tests
1
package tests
2
3
import (
4
"context"
5
"testing"
6
"time"
7
8
"github.com/google/uuid"
9
commonpb "go.temporal.io/api/common/v1"
10
sdkclient "go.temporal.io/sdk/client"
11
"go.temporal.io/sdk/workflow"
12
"go.temporal.io/server/api/adminservice/v1"
13
persistencespb "go.temporal.io/server/api/persistence/v1"
14
"go.temporal.io/server/chasm"
15
"go.temporal.io/server/common/dynamicconfig"
16
"go.temporal.io/server/common/primitives/timestamp"
17
"go.temporal.io/server/common/testing/parallelsuite"
18
"go.temporal.io/server/common/testing/testvars"
19
"go.temporal.io/server/tests/testcore"
20
)
21
22
type AdminTestSuite struct {
23
parallelsuite.Suite[*AdminTestSuite]
24
}
25
26
func TestAdminRebuildMutableState_ChasmDisabled(t *testing.T) {
27
parallelsuite.Run(t, &AdminTestSuite{}, false)
28
}
29
30
func TestAdminRebuildMutableState_ChasmEnabled(t *testing.T) {
31
parallelsuite.Run(t, &AdminTestSuite{}, true)
32
}
33
34
func (s *AdminTestSuite) TestAdminRebuildMutableState(testWithChasm bool) {
35
var opts []testcore.TestOption
36
if testWithChasm {
37
opts = append(opts, testcore.WithDynamicConfig(dynamicconfig.EnableChasm, true))
38
}
39
env := testcore.NewEnv(s.T(), opts...)
40
41
if testWithChasm {
42
configValues := env.GetTestCluster().Host().DcClient().GetValue(dynamicconfig.EnableChasm.Key())
43
s.NotEmpty(configValues, "EnableChasm config should be set")
44
configValue, _ := configValues[0].Value.(bool)
45
s.True(configValue, "EnableChasm config should be true")
46
}
47
48
tv := testvars.New(s.T())
49
workflowFn := func(ctx workflow.Context) error {
50
var randomUUID string
51
err := workflow.SideEffect(
52
ctx,
53
func(workflow.Context) any { return uuid.New().String() },
54
).Get(&randomUUID)
55
s.NoError(err)
56
57
_ = workflow.Sleep(ctx, 10*time.Minute)
58
return nil
59
}
60
61
env.SdkWorker().RegisterWorkflow(workflowFn)
62
63
workflowID := tv.Any().String()
64
workflowOptions := sdkclient.StartWorkflowOptions{
65
ID: workflowID,
66
TaskQueue: env.WorkerTaskQueue(),
67
WorkflowRunTimeout: 20 * time.Second,
68
}
69
ctx, cancel := context.WithTimeout(s.Context(), 30*time.Second)
70
defer cancel()
71
72
workflowRun, err := env.SdkClient().ExecuteWorkflow(s.Context(), workflowOptions, workflowFn)
73
s.NoError(err)
74
runID := workflowRun.GetRunID()
75
76
// there are total 6 events, 3 state transitions
77
// 1. WorkflowExecutionStarted
78
// 2. WorkflowTaskScheduled
79
//
80
// 3. WorkflowTaskStarted
81
//
82
// 4. WorkflowTaskCompleted
83
// 5. MarkerRecord
84
// 6. TimerStarted
85
86
var response1 *adminservice.DescribeMutableStateResponse
87
for {
88
response1, err = env.AdminClient().DescribeMutableState(ctx, &adminservice.DescribeMutableStateRequest{
89
Namespace: env.Namespace().String(),
90
Execution: &commonpb.WorkflowExecution{
91
WorkflowId: workflowID,
92
RunId: runID,
93
},
94
Archetype: chasm.WorkflowArchetype,
95
})
96
s.NoError(err)
97
if response1.DatabaseMutableState.ExecutionInfo.StateTransitionCount == 3 {
98
// Note: ChasmNodes may be empty even with CHASM enabled, so we only check if the rebuild can be performed,
99
// and not checking whether it is rebuildable because ChasmNodes are present.
100
if !testWithChasm {
101
s.Empty(response1.DatabaseMutableState.ChasmNodes, "CHASM-disabled workflows should not have ChasmNodes")
102
}
103
break
104
}
105
time.Sleep(20 * time.Millisecond) //nolint:forbidigo
106
}
107
108
_, err = env.AdminClient().RebuildMutableState(ctx, &adminservice.RebuildMutableStateRequest{
109
Namespace: env.Namespace().String(),
110
Execution: &commonpb.WorkflowExecution{
111
WorkflowId: workflowID,
112
RunId: runID,
113
},
114
})
115
s.NoError(err)
116
117
response2, err := env.AdminClient().DescribeMutableState(ctx, &adminservice.DescribeMutableStateRequest{
118
Namespace: env.Namespace().String(),
119
Execution: &commonpb.WorkflowExecution{
120
WorkflowId: workflowID,
121
RunId: runID,
122
},
123
Archetype: chasm.WorkflowArchetype,
124
})
125
s.NoError(err)
126
s.Equal(response1.DatabaseMutableState.ExecutionInfo.VersionHistories, response2.DatabaseMutableState.ExecutionInfo.VersionHistories)
127
s.Equal(response1.DatabaseMutableState.ExecutionInfo.StateTransitionCount, response2.DatabaseMutableState.ExecutionInfo.StateTransitionCount)
128
129
s.Equal(response1.DatabaseMutableState.ExecutionState.CreateRequestId, response2.DatabaseMutableState.ExecutionState.CreateRequestId)
130
s.Equal(response1.DatabaseMutableState.ExecutionState.RunId, response2.DatabaseMutableState.ExecutionState.RunId)
131
s.Equal(response1.DatabaseMutableState.ExecutionState.State, response2.DatabaseMutableState.ExecutionState.State)
132
s.Equal(response1.DatabaseMutableState.ExecutionState.Status, response2.DatabaseMutableState.ExecutionState.Status)
133
134
// From transition history perspective, Rebuild is considered as an update to the workflow and updates
135
// all sub state machines in the workflow, which includes the workflow ExecutionState.
136
s.Equal(&persistencespb.VersionedTransition{
137
NamespaceFailoverVersion: response1.DatabaseMutableState.ExecutionState.LastUpdateVersionedTransition.NamespaceFailoverVersion,
138
TransitionCount: response1.DatabaseMutableState.ExecutionInfo.StateTransitionCount + 1,
139
}, response2.DatabaseMutableState.ExecutionState.LastUpdateVersionedTransition)
140
141
// Rebuild explicitly sets start time, thus start time will change after rebuild.
142
s.NotNil(response1.DatabaseMutableState.ExecutionState.StartTime)
143
s.NotNil(response2.DatabaseMutableState.ExecutionState.StartTime)
144
145
timeBefore := timestamp.TimeValue(response1.DatabaseMutableState.ExecutionState.StartTime)
146
timeAfter := timestamp.TimeValue(response2.DatabaseMutableState.ExecutionState.StartTime)
147
s.False(timeAfter.Before(timeBefore))
148
}