go.temporal.io/server/tests/xdc/base.go
477 LOC · 0 covered · 477 uncovered · 0 ranges · 0 concepts · 0 introducers · 0 tests
1
package xdc
2
3
import (
4
"cmp"
5
"context"
6
"errors"
7
"sync"
8
"time"
9
10
"github.com/google/uuid"
11
"github.com/stretchr/testify/assert"
12
"github.com/stretchr/testify/require"
13
"github.com/stretchr/testify/suite"
14
"go.temporal.io/api/operatorservice/v1"
15
replicationpb "go.temporal.io/api/replication/v1"
16
"go.temporal.io/api/serviceerror"
17
"go.temporal.io/api/workflowservice/v1"
18
sdkclient "go.temporal.io/sdk/client"
19
sdkworker "go.temporal.io/sdk/worker"
20
"go.temporal.io/server/api/adminservice/v1"
21
"go.temporal.io/server/api/historyservice/v1"
22
"go.temporal.io/server/common"
23
"go.temporal.io/server/common/cluster"
24
"go.temporal.io/server/common/dynamicconfig"
25
"go.temporal.io/server/common/log"
26
"go.temporal.io/server/common/log/tag"
27
"go.temporal.io/server/common/searchattribute"
28
"go.temporal.io/server/common/testing/historyrequire"
29
"go.temporal.io/server/common/testing/protorequire"
30
"go.temporal.io/server/tests/testcore"
31
"google.golang.org/protobuf/types/known/durationpb"
32
)
33
34
const (
35
namespaceCacheWaitTime = 2 * testcore.NamespaceCacheRefreshInterval
36
namespaceCacheCheckInterval = testcore.NamespaceCacheRefreshInterval / 2
37
replicationWaitTime = 15 * time.Second
38
replicationCheckInterval = 500 * time.Millisecond
39
40
testTimeout = 30 * time.Second
41
)
42
43
type (
44
xdcBaseSuite struct {
45
// TODO (alex): use FunctionalTestBase instead.
46
suite.Suite
47
// override suite.Suite.Assertions with require.Assertions; this means that s.NotNil(nil) will stop the test,
48
// not merely log an error
49
*require.Assertions
50
protorequire.ProtoAssertions
51
historyrequire.HistoryRequire
52
53
clusters []*testcore.TestCluster
54
logger log.Logger
55
dynamicConfigOverrides map[dynamicconfig.Key]any
56
57
startTime time.Time
58
onceClusterConnect sync.Once
59
60
enableTransitionHistory bool
61
62
// TODO: add sdkClient and worker here and remove its creation in many tests.
63
}
64
)
65
66
// TODO (alex): this should be gone.
67
func (s *xdcBaseSuite) clusterReplicationConfig() []*replicationpb.ClusterReplicationConfig {
68
config := make([]*replicationpb.ClusterReplicationConfig, 2)
69
for ci, c := range s.clusters {
70
config[ci] = &replicationpb.ClusterReplicationConfig{
71
ClusterName: c.ClusterName(),
72
}
73
}
74
return config
75
}
76
77
func (s *xdcBaseSuite) setupSuite(opts ...testcore.TestClusterOption) {
78
79
params := testcore.ApplyTestClusterOptions(opts)
80
81
if s.logger == nil {
82
s.logger = log.NewTestLogger()
83
}
84
if s.dynamicConfigOverrides == nil {
85
s.dynamicConfigOverrides = make(map[dynamicconfig.Key]any)
86
}
87
s.dynamicConfigOverrides[dynamicconfig.ClusterMetadataRefreshInterval.Key()] = time.Second * 5
88
s.dynamicConfigOverrides[dynamicconfig.NamespaceCacheRefreshInterval.Key()] = testcore.NamespaceCacheRefreshInterval
89
s.dynamicConfigOverrides[dynamicconfig.EnableTransitionHistory.Key()] = s.enableTransitionHistory
90
// TODO (prathyush): remove this after setting it to true by default.
91
s.dynamicConfigOverrides[dynamicconfig.SendRawHistoryBetweenInternalServices.Key()] = true
92
// Override checkpoint intervals to 3 seconds for faster testing
93
s.dynamicConfigOverrides[dynamicconfig.TransferProcessorUpdateAckInterval.Key()] = time.Second * 3
94
s.dynamicConfigOverrides[dynamicconfig.TimerProcessorUpdateAckInterval.Key()] = time.Second * 3
95
s.dynamicConfigOverrides[dynamicconfig.VisibilityProcessorUpdateAckInterval.Key()] = time.Second * 3
96
s.dynamicConfigOverrides[dynamicconfig.OutboundProcessorUpdateAckInterval.Key()] = time.Second * 3
97
s.dynamicConfigOverrides[dynamicconfig.ArchivalProcessorUpdateAckInterval.Key()] = time.Second * 3
98
// Override max poll intervals to 3 seconds for faster task discovery in tests
99
s.dynamicConfigOverrides[dynamicconfig.TransferProcessorMaxPollInterval.Key()] = time.Second * 3
100
s.dynamicConfigOverrides[dynamicconfig.TimerProcessorMaxPollInterval.Key()] = time.Second * 3
101
s.dynamicConfigOverrides[dynamicconfig.VisibilityProcessorMaxPollInterval.Key()] = time.Second * 3
102
s.dynamicConfigOverrides[dynamicconfig.OutboundProcessorMaxPollInterval.Key()] = time.Second * 3
103
104
persistenceDefaults := testcore.GetPersistenceTestDefaults()
105
clusterConfigs := []*testcore.TestClusterConfig{
106
{
107
ClusterMetadata: cluster.Config{
108
EnableGlobalNamespace: true,
109
FailoverVersionIncrement: 10,
110
},
111
HistoryConfig: testcore.HistoryConfig{
112
NumHistoryShards: cmp.Or(params.NumHistoryShards, 1),
113
},
114
Persistence: persistenceDefaults,
115
},
116
{
117
ClusterMetadata: cluster.Config{
118
EnableGlobalNamespace: true,
119
FailoverVersionIncrement: 10,
120
},
121
HistoryConfig: testcore.HistoryConfig{
122
NumHistoryShards: cmp.Or(params.NumHistoryShards, 1),
123
},
124
Persistence: persistenceDefaults,
125
},
126
}
127
128
s.clusters = make([]*testcore.TestCluster, len(clusterConfigs))
129
suffix := common.GenerateRandomString(5)
130
131
testClusterFactory := testcore.NewTestClusterFactory()
132
for clusterIndex, clusterName := range []string{"active_" + suffix, "standby_" + suffix} {
133
clusterConfigs[clusterIndex].DynamicConfigOverrides = s.dynamicConfigOverrides
134
clusterConfigs[clusterIndex].DCRedirectionPolicy = params.DCRedirectionPolicy
135
clusterConfigs[clusterIndex].ClusterMetadata.MasterClusterName = clusterName
136
clusterConfigs[clusterIndex].ClusterMetadata.CurrentClusterName = clusterName
137
clusterConfigs[clusterIndex].ClusterMetadata.EnableGlobalNamespace = true
138
clusterConfigs[clusterIndex].Persistence.DBName += "_" + clusterName
139
clusterConfigs[clusterIndex].ClusterMetadata.ClusterInformation = map[string]cluster.ClusterInformation{
140
clusterName: {
141
Enabled: true,
142
InitialFailoverVersion: int64(clusterIndex + 1),
143
// RPCAddress and HTTPAddress will be filled in
144
},
145
}
146
clusterConfigs[clusterIndex].EnableMetricsCapture = true
147
clusterConfigs[clusterIndex].EnableHistoryTaskRecorder = params.EnableHistoryTaskRecorder
148
149
var err error
150
s.clusters[clusterIndex], err = testClusterFactory.NewCluster(s.T(), clusterConfigs[clusterIndex], log.With(s.logger, tag.ClusterName(clusterName)))
151
s.Require().NoError(err)
152
}
153
154
s.startTime = time.Now()
155
156
for ci, c := range s.clusters {
157
for remoteCi, remoteC := range s.clusters {
158
if ci != remoteCi {
159
_, err := c.AdminClient().AddOrUpdateRemoteCluster(
160
testcore.NewContext(),
161
&adminservice.AddOrUpdateRemoteClusterRequest{
162
FrontendAddress: remoteC.Host().RemoteFrontendGRPCAddress(),
163
FrontendHttpAddress: remoteC.Host().FrontendHTTPAddress(),
164
EnableRemoteClusterConnection: true,
165
EnableReplication: true,
166
})
167
s.Require().NoError(err)
168
}
169
}
170
}
171
// TODO (alex): This looks suspicious. Why 200ms?
172
// Wait for cluster metadata to refresh new added clusters
173
time.Sleep(time.Millisecond * 200)
174
}
175
176
func (s *xdcBaseSuite) waitForClusterConnected(
177
sourceCluster *testcore.TestCluster,
178
targetClusterName string,
179
) {
180
s.logger.Info("wait for clusters to be synced", tag.SourceCluster(sourceCluster.ClusterName()), tag.TargetCluster(targetClusterName))
181
s.EventuallyWithT(func(c *assert.CollectT) {
182
s.logger.Info("check if clusters are synced", tag.SourceCluster(sourceCluster.ClusterName()), tag.TargetCluster(targetClusterName))
183
resp, err := sourceCluster.HistoryClient().GetReplicationStatus(context.Background(), &historyservice.GetReplicationStatusRequest{})
184
require.NoError(c, err)
185
require.Lenf(c, resp.Shards, 1, "test cluster has only one history shard")
186
187
shard := resp.Shards[0]
188
require.NotNil(c, shard)
189
require.Positive(c, shard.MaxReplicationTaskId)
190
require.NotNil(c, shard.ShardLocalTime)
191
require.WithinRange(c, shard.ShardLocalTime.AsTime(), s.startTime, time.Now())
192
require.NotNil(c, shard.RemoteClusters)
193
194
standbyAckInfo, ok := shard.RemoteClusters[targetClusterName]
195
require.True(c, ok)
196
require.NotNil(c, standbyAckInfo)
197
require.LessOrEqual(c, shard.MaxReplicationTaskId, standbyAckInfo.AckedTaskId)
198
require.NotNil(c, standbyAckInfo.AckedTaskVisibilityTime)
199
require.WithinRange(c, standbyAckInfo.AckedTaskVisibilityTime.AsTime(), s.startTime, time.Now())
200
}, 90*time.Second, 1*time.Second)
201
s.logger.Info("clusters synced", tag.SourceCluster(sourceCluster.ClusterName()), tag.TargetCluster(targetClusterName))
202
}
203
204
func (s *xdcBaseSuite) tearDownSuite() {
205
for _, c := range s.clusters {
206
s.NoError(c.TearDownCluster())
207
}
208
}
209
210
func (s *xdcBaseSuite) waitForClusterSynced() {
211
for sourceClusterI, sourceCluster := range s.clusters {
212
for targetClusterI, targetCluster := range s.clusters {
213
if sourceClusterI != targetClusterI {
214
s.waitForClusterConnected(sourceCluster, targetCluster.ClusterName())
215
}
216
}
217
}
218
}
219
220
func (s *xdcBaseSuite) setupTest() {
221
// Have to define our overridden assertions in the test setup. If we did it earlier, s.T() will return nil
222
s.Assertions = require.New(s.T())
223
s.ProtoAssertions = protorequire.New(s.T())
224
s.HistoryRequire = historyrequire.New(s.T())
225
226
s.onceClusterConnect.Do(func() {
227
s.waitForClusterSynced()
228
})
229
}
230
231
func (s *xdcBaseSuite) createGlobalNamespace() string {
232
return s.createNamespace(true, s.clusters)
233
}
234
235
func (s *xdcBaseSuite) registerTestSearchAttributes(ns string) {
236
expectedSearchAttributes := searchattribute.TestSearchAttributesToRegister()
237
// For SQL: call AddSearchAttributes on the active cluster only. It calls UpdateNamespace
238
// internally, so the alias mapping replicates to all clusters. Calling it on each cluster
239
// independently is unsafe — alias assignment uses non-deterministic Go map iteration and
240
// can produce different field→alias mappings per cluster, corrupting standby visibility.
241
// For ES: each cluster has its own index and cluster metadata, so register on each.
242
clusters := s.clusters
243
if testcore.UseSQLVisibility() {
244
clusters = s.clusters[:1]
245
}
246
for _, cl := range clusters {
247
_, err := cl.OperatorClient().AddSearchAttributes(testcore.NewContext(), &operatorservice.AddSearchAttributesRequest{
248
Namespace: ns,
249
SearchAttributes: expectedSearchAttributes,
250
})
251
var alreadyExistsErr *serviceerror.AlreadyExists
252
if err != nil && !errors.As(err, &alreadyExistsErr) {
253
s.Require().NoError(err)
254
}
255
}
256
for _, cl := range s.clusters {
257
s.EventuallyWithT(func(t *assert.CollectT) {
258
resp, err := cl.OperatorClient().ListSearchAttributes(testcore.NewContext(), &operatorservice.ListSearchAttributesRequest{
259
Namespace: ns,
260
})
261
require.NoError(t, err)
262
for attrName, attrType := range expectedSearchAttributes {
263
gotType, ok := resp.GetCustomAttributes()[attrName]
264
require.True(t, ok, "expected search attribute %q to be registered", attrName)
265
require.Equal(t, attrType, gotType)
266
}
267
}, replicationWaitTime, replicationCheckInterval)
268
}
269
}
270
271
// TODO (alex): rename this to createLocalNamespace, and everywhere where it is called with isGlobal == true, add call to promoteNamespace.
272
func (s *xdcBaseSuite) createNamespaceInCluster0(isGlobal bool) string {
273
return s.createNamespace(isGlobal, s.clusters[:1])
274
}
275
276
func (s *xdcBaseSuite) createNamespace(
277
isGlobal bool,
278
clusters []*testcore.TestCluster,
279
) string {
280
ctx := testcore.NewContext()
281
ns := "test-namespace-" + uuid.NewString()
282
var replicationConfigs []*replicationpb.ClusterReplicationConfig
283
var clusterNames []string
284
if isGlobal {
285
replicationConfigs = make([]*replicationpb.ClusterReplicationConfig, len(clusters))
286
clusterNames = make([]string, len(clusters))
287
for ci, c := range clusters {
288
replicationConfigs[ci] = &replicationpb.ClusterReplicationConfig{ClusterName: c.ClusterName()}
289
clusterNames[ci] = c.ClusterName()
290
}
291
}
292
293
regReq := &workflowservice.RegisterNamespaceRequest{
294
Namespace: ns,
295
IsGlobalNamespace: isGlobal,
296
Clusters: replicationConfigs,
297
ActiveClusterName: clusters[0].ClusterName(), // cluster 0 is always active.
298
WorkflowExecutionRetentionPeriod: durationpb.New(7 * time.Hour * 24),
299
}
300
// namespace is always created in cluster 0.
301
_, err := clusters[0].FrontendClient().RegisterNamespace(ctx, regReq)
302
s.NoError(err)
303
304
s.EventuallyWithT(func(t *assert.CollectT) {
305
s.describeNamespace(t, clusters[0], ns, isGlobal)
306
}, namespaceCacheWaitTime, namespaceCacheCheckInterval)
307
308
if len(clusters) > 1 && isGlobal {
309
// If namespace is global and config has more than 1 cluster, it should be replicated to these other clusters.
310
// Check other clusters too.
311
s.EventuallyWithT(func(t *assert.CollectT) {
312
for _, c := range clusters[1:] {
313
resp := s.describeNamespace(t, c, ns, isGlobal)
314
require.ElementsMatch(t, clusterNames, s.namespaceClusterNames(resp))
315
}
316
}, replicationWaitTime, replicationCheckInterval)
317
}
318
319
s.waitForNamespaceCacheRefresh()
320
return ns
321
}
322
323
func (s *xdcBaseSuite) updateNamespaceClusters(
324
ns string,
325
inClusterIndex int,
326
clusters []*testcore.TestCluster,
327
) {
328
replicationConfigs := make([]*replicationpb.ClusterReplicationConfig, len(clusters))
329
clusterNames := make([]string, len(clusters))
330
for ci, c := range clusters {
331
replicationConfigs[ci] = &replicationpb.ClusterReplicationConfig{ClusterName: c.ClusterName()}
332
clusterNames[ci] = c.ClusterName()
333
}
334
335
_, err := clusters[inClusterIndex].FrontendClient().UpdateNamespace(testcore.NewContext(), &workflowservice.UpdateNamespaceRequest{
336
Namespace: ns,
337
ReplicationConfig: &replicationpb.NamespaceReplicationConfig{
338
Clusters: replicationConfigs,
339
}})
340
s.NoError(err)
341
342
var isGlobalNamespace bool
343
s.EventuallyWithT(func(t *assert.CollectT) {
344
resp := s.describeNamespace(t, clusters[inClusterIndex], ns, true)
345
require.ElementsMatch(t, clusterNames, s.namespaceClusterNames(resp))
346
isGlobalNamespace = resp.GetIsGlobalNamespace()
347
}, namespaceCacheWaitTime, namespaceCacheCheckInterval)
348
349
if len(clusters) > 1 && isGlobalNamespace {
350
// If namespace is global and config has more than 1 cluster, it should be replicated to these other clusters.
351
// Check other clusters too.
352
s.EventuallyWithT(func(t *assert.CollectT) {
353
for ci, c := range clusters {
354
if ci == inClusterIndex {
355
continue
356
}
357
resp := s.describeNamespace(t, c, ns, true)
358
require.ElementsMatch(t, clusterNames, s.namespaceClusterNames(resp))
359
}
360
}, replicationWaitTime, replicationCheckInterval)
361
}
362
s.waitForNamespaceCacheRefresh()
363
}
364
365
func (s *xdcBaseSuite) promoteNamespace(
366
ns string,
367
inClusterIndex int,
368
) {
369
_, err := s.clusters[inClusterIndex].FrontendClient().UpdateNamespace(testcore.NewContext(), &workflowservice.UpdateNamespaceRequest{
370
Namespace: ns,
371
PromoteNamespace: true,
372
})
373
s.NoError(err)
374
375
s.EventuallyWithT(func(t *assert.CollectT) {
376
s.describeNamespace(t, s.clusters[inClusterIndex], ns, true)
377
}, namespaceCacheWaitTime, namespaceCacheCheckInterval)
378
s.waitForNamespaceCacheRefresh()
379
}
380
381
func (s *xdcBaseSuite) failover(
382
ns string,
383
inClusterIndex int,
384
targetCluster string,
385
targetFailoverVersion int64,
386
) {
387
s.waitForClusterSynced()
388
389
// update namespace to fail over
390
updateReq := &workflowservice.UpdateNamespaceRequest{
391
Namespace: ns,
392
ReplicationConfig: &replicationpb.NamespaceReplicationConfig{
393
ActiveClusterName: targetCluster,
394
},
395
}
396
updateResp, err := s.clusters[inClusterIndex].FrontendClient().UpdateNamespace(testcore.NewContext(), updateReq)
397
s.NoError(err)
398
// TODO (alex): not clear why it matters.
399
s.Equal(targetFailoverVersion, updateResp.GetFailoverVersion())
400
401
// check local and remote clusters
402
s.EventuallyWithT(func(t *assert.CollectT) {
403
for _, c := range s.clusters {
404
resp := s.describeNamespace(t, c, ns, true)
405
require.Equal(t, targetCluster, resp.GetReplicationConfig().GetActiveClusterName())
406
}
407
}, replicationWaitTime, replicationCheckInterval)
408
409
s.waitForClusterSynced()
410
s.waitForNamespaceCacheRefresh()
411
}
412
413
func (s *xdcBaseSuite) waitForNamespaceCacheRefresh() {
414
time.Sleep(namespaceCacheWaitTime) //nolint:forbidigo
415
}
416
417
func (s *xdcBaseSuite) describeNamespace(
418
t require.TestingT,
419
testCluster *testcore.TestCluster,
420
ns string,
421
isGlobal bool,
422
) *workflowservice.DescribeNamespaceResponse {
423
resp, err := testCluster.FrontendClient().DescribeNamespace(testcore.NewContext(), &workflowservice.DescribeNamespaceRequest{
424
Namespace: ns,
425
})
426
require.NoError(t, err)
427
require.NotNil(t, resp)
428
require.Equal(t, isGlobal, resp.GetIsGlobalNamespace())
429
if isGlobal {
430
require.NotNil(t, resp.GetReplicationConfig())
431
}
432
return resp
433
}
434
435
func (s *xdcBaseSuite) namespaceClusterNames(resp *workflowservice.DescribeNamespaceResponse) []string {
436
replicationConfig := resp.GetReplicationConfig()
437
if replicationConfig == nil {
438
return nil
439
}
440
clusters := replicationConfig.GetClusters()
441
clusterNames := make([]string, len(clusters))
442
for i, replicationCluster := range clusters {
443
clusterNames[i] = replicationCluster.GetClusterName()
444
}
445
return clusterNames
446
}
447
448
func (s *xdcBaseSuite) newClientAndWorker(hostport, ns, taskqueue, identity string) (sdkclient.Client, sdkworker.Worker) {
449
sdkClient, err := sdkclient.Dial(sdkclient.Options{
450
HostPort: hostport,
451
Namespace: ns,
452
})
453
s.NoError(err)
454
455
worker := sdkworker.New(sdkClient, taskqueue, sdkworker.Options{
456
Identity: identity,
457
})
458
459
return sdkClient, worker
460
}
461
462
// waitForVisibilityCount waits for the visibility store to index the expected number of workflow
463
// executions in the given namespace before proceeding. This is important before starting
464
// force-replication, which uses ListWorkflowExecutions with an empty query to discover all
465
// workflows in a namespace.
466
func (s *xdcBaseSuite) waitForVisibilityCount(ctx context.Context, ns string, expectedCount int64) {
467
frontendClient := s.clusters[0].FrontendClient()
468
s.Eventually(func() bool {
469
countResp, err := frontendClient.CountWorkflowExecutions(ctx, &workflowservice.CountWorkflowExecutionsRequest{
470
Namespace: ns,
471
})
472
if err != nil {
473
return false
474
}
475
return countResp.GetCount() == expectedCount
476
}, 15*time.Second, 200*time.Millisecond, "visibility should index %d workflow runs before force-replication", expectedCount)
477
}