go.temporal.io/server/common/cluster/metadata.go
612 LOC · 291 covered · 321 uncovered · 81 ranges · 1595 concepts · 32 introducers · 859 tests
File neighbourhood
The centred file is linked to every concept that introduces one of its ranges, every test that runs code from the file, and the gray connector concepts standing between those tests and the file's own introducer concepts. Undirected links join concepts to every file where they introduce source and concepts to the tests they introduce; arrows show specialization between the displayed concepts and bridge only concepts omitted from this view. Concept colors match the source ranges below; connector concepts have no source color and are shown in gray.
Focused file, its introducer and connector concepts, their introduced files, and tests that run code from the file
In the embedded map, ordinary wheel input scrolls the page; use the visible controls to zoom and drag to pan. Open the full-screen map for canvas navigation: wheel pans, Ctrl/Command plus wheel zooms, and arrow keys pan when this region is focused. On touch screens, open the full-screen map to pan or pinch. If JavaScript or WebGL is unavailable, use the related-file, concept, and source links on this page.
Graph controls are ready.
Interactive rendering requires JavaScript and WebGL. Use the related-file, concept, and source links on this page while the interactive map is unavailable.
1
//go:generate mockgen -package $GOPACKAGE -source $GOFILE -destination metadata_mock.go
2
3
package cluster
4
5
import (
6
"context"
7
"errors"
8
"fmt"
9
"maps"
10
"math"
11
"strconv"
12
"sync"
13
"sync/atomic"
14
"time"
15
16
"go.temporal.io/server/common"
17
"go.temporal.io/server/common/collection"
18
"go.temporal.io/server/common/dynamicconfig"
19
"go.temporal.io/server/common/goro"
20
"go.temporal.io/server/common/headers"
21
"go.temporal.io/server/common/log"
22
"go.temporal.io/server/common/log/tag"
23
"go.temporal.io/server/common/metrics"
24
"go.temporal.io/server/common/persistence"
25
"go.temporal.io/server/common/pingable"
26
)
27
28
const (
29
defaultClusterMetadataPageSize = 100
30
refreshInterval = time.Minute
31
32
unknownClusterNamePrefix = "unknown-cluster-"
33
)
34
35
type (
36
// Metadata provides information about the current cluster and other registered remote clusters.
37
Metadata interface {
38
pingable.Pingable
39
40
// IsGlobalNamespaceEnabled whether the global namespace is enabled,
41
// this attr should be discarded when cross DC is made public
42
IsGlobalNamespaceEnabled() bool
43
// IsMasterCluster whether current cluster is master cluster
44
IsMasterCluster() bool
45
// GetClusterID return the cluster ID, which is also the initial failover version
46
GetClusterID() int64
47
// GetNextFailoverVersion return the next failover version for namespace failover
48
GetNextFailoverVersion(string, int64) int64
49
// IsVersionFromSameCluster return true if 2 version are used for the same cluster
50
IsVersionFromSameCluster(version1 int64, version2 int64) bool
51
// GetMasterClusterName return the master cluster name
52
GetMasterClusterName() string
53
// GetCurrentClusterName return the current cluster name
54
GetCurrentClusterName() string
55
// GetAllClusterInfo return the all cluster name -> corresponding info
56
GetAllClusterInfo() map[string]ClusterInformation
57
// ClusterNameForFailoverVersion return the corresponding cluster name for a given failover version
58
ClusterNameForFailoverVersion(isGlobalNamespace bool, failoverVersion int64) string
59
// GetFailoverVersionIncrement return the Failover version increment value
60
GetFailoverVersionIncrement() int64
61
RegisterMetadataChangeCallback(callbackId any, cb CallbackFn)
62
UnRegisterMetadataChangeCallback(callbackId any)
63
Start()
64
Stop()
65
}
66
67
CallbackFn func(oldClusterMetadata map[string]*ClusterInformation, newClusterMetadata map[string]*ClusterInformation)
68
69
// Config contains the all cluster which participated in cross DC
70
Config struct {
71
EnableGlobalNamespace bool `yaml:"enableGlobalNamespace"`
72
// FailoverVersionIncrement is the increment of each cluster version when failover happens.
73
FailoverVersionIncrement int64 `yaml:"failoverVersionIncrement"`
74
// MasterClusterName is the master cluster name, only the master cluster can register / update namespace
75
// all clusters can do namespace failover.
76
MasterClusterName string `yaml:"masterClusterName"`
77
// CurrentClusterName is the name of the current cluster.
78
CurrentClusterName string `yaml:"currentClusterName"`
79
// ClusterInformation is a map from cluster name to corresponding information for each registered cluster.
80
ClusterInformation map[string]ClusterInformation `yaml:"clusterInformation"`
81
// Tags contains customized tags for the current cluster.
82
Tags map[string]string `yaml:"tags"`
83
}
84
85
// ClusterInformation contains information for a single cluster.
86
ClusterInformation struct {
87
Enabled bool `yaml:"enabled"`
88
InitialFailoverVersion int64 `yaml:"initialFailoverVersion"`
89
// RPCAddress indicate the remote service address(Host:Port). Host can be DNS name.
90
RPCAddress string `yaml:"rpcAddress"`
91
// HTTPAddress indicates the address of the [go.temporal.io/server/service/frontend.HTTPAPIServer].
92
// E.g. "localhost:7243".
93
HTTPAddress string `yaml:"httpAddress"`
94
// ClusterID allows to explicitly set the ID of the cluster. Optional.
95
ClusterID string `yaml:"-"`
96
ShardCount int32 `yaml:"-"` // Ignore this field when loading config.
97
Tags map[string]string `yaml:"-"` // Ignore this field. Use cluster.Config.Tags for customized tags.
98
// ReplicationEnabled controls whether replication streams are active.
99
ReplicationEnabled bool `yaml:"-"`
100
// private field to track cluster information updates
101
version int64
102
}
103
104
metadataImpl struct {
105
status int32
106
clusterMetadataStore persistence.ClusterMetadataManager
107
refresher *goro.Handle
108
refreshDuration dynamicconfig.DurationPropertyFn
109
logger log.Logger
110
111
// Immutable fields
112
113
// EnableGlobalNamespace whether the global namespace is enabled,
114
enableGlobalNamespace bool
115
// all clusters can do namespace failover
116
masterClusterName string
117
// currentClusterName is the name of the current cluster
118
currentClusterName string
119
// failoverVersionIncrement is the increment of each cluster's version when failover happen
120
failoverVersionIncrement int64
121
122
// Mutable fields
123
124
clusterLock sync.RWMutex
125
// clusterInfo contains all cluster name -> corresponding information
126
clusterInfo map[string]ClusterInformation
127
// versionToClusterName contains all initial version -> corresponding cluster name
128
versionToClusterName map[int64]string
129
130
clusterCallbackLock sync.RWMutex
131
clusterChangeCallback map[any]CallbackFn
132
}
133
)
134
135
func NewMetadata(
136
enableGlobalNamespace bool,
137
failoverVersionIncrement int64,
138
masterClusterName string,
139
currentClusterName string,
140
clusterInfo map[string]ClusterInformation,
141
clusterMetadataStore persistence.ClusterMetadataManager,
142
refreshDuration dynamicconfig.DurationPropertyFn,
143
logger log.Logger,
145
>
if len(clusterInfo) == 0 {
146
panic("Empty cluster information")
148
panic("Master cluster name is empty")
150
panic("Current cluster name is empty")
151
>
} else if failoverVersionIncrement == 0 || failoverVersionIncrement > math.MaxInt32 {
metadata.go ×13
152
panic("Version increment <= 0 or > 2147483647")
153
}
154
155
>
versionToClusterName, err := updateVersionToClusterName(clusterInfo, failoverVersionIncrement)
metadata.go ×13
156
>
if err != nil {
157
// nolint:forbidigo // matches the other startup-config panics in this constructor
158
panic(err.Error())
159
}
161
panic("Current cluster is not specified in cluster info")
162
}
164
panic("Master cluster is not specified in cluster info")
165
}
166
168
>
maps.Copy(copyClusterInfo, clusterInfo)
169
>
if refreshDuration == nil {
171
>
}
173
>
status: common.DaemonStatusInitialized,
174
>
enableGlobalNamespace: enableGlobalNamespace,
175
>
failoverVersionIncrement: failoverVersionIncrement,
176
>
masterClusterName: masterClusterName,
177
>
currentClusterName: currentClusterName,
178
>
clusterInfo: copyClusterInfo,
179
>
versionToClusterName: versionToClusterName,
180
>
clusterChangeCallback: make(map[any]CallbackFn),
181
>
clusterMetadataStore: clusterMetadataStore,
182
>
logger: logger,
183
>
refreshDuration: refreshDuration,
184
>
}
185
}
186
187
func NewMetadataFromConfig(
188
config *Config,
189
clusterMetadataStore persistence.ClusterMetadataManager,
190
dynamicCollection *dynamicconfig.Collection,
191
logger log.Logger,
193
>
return NewMetadata(
194
>
config.EnableGlobalNamespace,
195
>
config.FailoverVersionIncrement,
196
>
config.MasterClusterName,
197
>
config.CurrentClusterName,
198
>
config.ClusterInformation,
199
>
clusterMetadataStore,
200
>
dynamicconfig.ClusterMetadataRefreshInterval.Get(dynamicCollection),
201
>
logger,
202
>
)
203
>
}
204
206
>
if !atomic.CompareAndSwapInt32(&m.status, common.DaemonStatusInitialized, common.DaemonStatusStarted) {
207
return
208
}
209
210
// TODO: specify a timeout for the context
212
>
context.TODO(),
213
>
headers.SystemBackgroundHighCallerInfo,
214
>
)
215
>
err := m.refreshClusterMetadata(ctx)
216
>
if err != nil {
217
// Crash rather than start with partial cluster metadata (e.g. an invalid
218
// or missing row in cluster_metadata): replication and failover routing
219
// would be incorrect. The Fatal forces operators to fix or remove the bad
220
// row before the next start can succeed.
221
m.logger.Fatal("Unable to initialize cluster metadata cache", tag.Error(err))
222
}
224
}
225
227
>
if !atomic.CompareAndSwapInt32(&m.status, common.DaemonStatusStarted, common.DaemonStatusStopped) {
228
return
229
}
230
232
>
<-m.refresher.Done()
233
}
234
236
>
return []pingable.Check{
237
>
{
238
>
Name: "cluster metadata lock",
239
>
// we don't do any persistence ops under clusterLock, use a short timeout
240
>
Timeout: 10 * time.Second,
241
>
Ping: func() []pingable.Pingable {
242
>
m.clusterLock.Lock()
243
>
// nolint:staticcheck
244
>
m.clusterLock.Unlock()
245
>
return nil
246
>
},
247
MetricsName: metrics.DDClusterMetadataLockLatency.Name(),
248
},
249
{
250
Name: "cluster metadata callback lock",
251
// listeners get called under clusterCallbackLock, they may do some more work, but
252
// not persistence ops.
253
Timeout: 10 * time.Second,
255
>
m.clusterCallbackLock.Lock()
256
>
// nolint:staticcheck
257
>
m.clusterCallbackLock.Unlock()
258
>
return nil
259
>
},
260
MetricsName: metrics.DDClusterMetadataCallbackLockLatency.Name(),
261
},
262
}
263
}
264
266
>
return m.enableGlobalNamespace
267
>
}
268
270
>
return m.masterClusterName == m.currentClusterName
271
>
}
272
274
>
m.clusterLock.RLock()
275
>
defer m.clusterLock.RUnlock()
276
>
277
>
info, ok := m.clusterInfo[m.currentClusterName]
278
>
if !ok {
279
panic(fmt.Sprintf(
280
"Unknown cluster name: %v with given cluster initial failover version map: %v.",
281
m.currentClusterName,
282
m.clusterInfo,
283
))
284
}
286
}
287
288
>
func (m *metadataImpl) GetNextFailoverVersion(clusterName string, currentFailoverVersion int64) int64 {
metadata.go ×2
289
>
m.clusterLock.RLock()
290
>
defer m.clusterLock.RUnlock()
291
>
292
>
info, ok := m.clusterInfo[clusterName]
293
>
if !ok {
294
panic(fmt.Sprintf(
295
"Unknown cluster name: %v with given cluster initial failover version map: %v.",
296
clusterName,
297
m.clusterInfo,
298
))
299
}
300
>
failoverVersion := currentFailoverVersion/m.failoverVersionIncrement*m.failoverVersionIncrement + info.InitialFailoverVersion
metadata.go ×2
301
>
if failoverVersion < currentFailoverVersion {
302
>
return failoverVersion + m.failoverVersionIncrement
303
>
}
304
return failoverVersion
305
}
306
307
>
func (m *metadataImpl) IsVersionFromSameCluster(version1 int64, version2 int64) bool {
metadata.go ×1
308
>
return (version1-version2)%m.failoverVersionIncrement == 0
309
>
}
310
312
>
return m.masterClusterName
313
>
}
314
316
>
return m.currentClusterName
317
>
}
318
320
>
m.clusterLock.RLock()
321
>
defer m.clusterLock.RUnlock()
322
>
323
>
result := make(map[string]ClusterInformation, len(m.clusterInfo))
324
>
maps.Copy(result, m.clusterInfo)
325
>
return result
326
>
}
327
328
>
func (m *metadataImpl) ClusterNameForFailoverVersion(isGlobalNamespace bool, failoverVersion int64) string {
metadata.go ×1
329
>
if failoverVersion == common.EmptyVersion {
330
>
// Local namespace uses EmptyVersion. But local namespace could be promoted to global namespace. Once promoted,
handler.go ×25
331
>
// workflows with EmptyVersion could be replicated to other clusters. The receiving cluster needs to know that
332
>
// those workflows are not from their current cluster.
333
>
if isGlobalNamespace {
334
return unknownClusterNamePrefix + strconv.Itoa(int(failoverVersion))
335
}
337
}
338
340
panic(fmt.Sprintf(
341
"ClusterMetadata encountered local namesapce with failover version %v",
342
failoverVersion,
343
))
344
}
345
347
>
// Failover version starts with 1. Zero is an invalid value for failover version
348
>
if initialFailoverVersion == common.EmptyVersion {
349
initialFailoverVersion = m.failoverVersionIncrement
350
}
351
353
>
defer m.clusterLock.RUnlock()
354
>
clusterName, ok := m.versionToClusterName[initialFailoverVersion]
355
>
if !ok {
356
>
m.logger.Warn(fmt.Sprintf(
357
>
"Unknown initial failover version %v with given cluster initial failover version map: %v and failover version increment %v.",
358
>
initialFailoverVersion,
359
>
m.clusterInfo,
360
>
m.failoverVersionIncrement,
361
>
))
362
>
return unknownClusterNamePrefix + strconv.Itoa(int(initialFailoverVersion))
363
>
}
364
>
return clusterName
365
}
366
368
>
return m.failoverVersionIncrement
369
>
}
370
371
>
func (m *metadataImpl) RegisterMetadataChangeCallback(callbackId any, cb CallbackFn) {
metadata.go ×1
372
>
m.clusterCallbackLock.Lock()
373
>
m.clusterChangeCallback[callbackId] = cb
374
>
m.clusterCallbackLock.Unlock()
375
>
376
>
oldEntries := make(map[string]*ClusterInformation)
377
>
newEntries := make(map[string]*ClusterInformation)
378
>
m.clusterLock.RLock()
379
>
for clusterName, clusterInfo := range m.clusterInfo {
380
>
oldEntries[clusterName] = nil
381
>
newEntries[clusterName] = ShallowCopyClusterInformation(&clusterInfo)
382
>
}
383
>
m.clusterLock.RUnlock()
384
>
cb(oldEntries, newEntries)
385
}
386
388
>
m.clusterCallbackLock.Lock()
389
>
delete(m.clusterChangeCallback, callbackId)
390
>
m.clusterCallbackLock.Unlock()
391
>
}
392
394
>
timer := time.NewTicker(m.refreshDuration())
395
>
defer timer.Stop()
396
>
397
>
for {
398
>
select {
400
>
return nil
402
>
for err := m.refreshClusterMetadata(ctx); err != nil; err = m.refreshClusterMetadata(ctx) {
403
m.logger.Error("Error refreshing remote cluster metadata", tag.Error(err))
404
refreshTimer := time.NewTimer(m.refreshDuration() / 2)
405
406
select {
407
case <-refreshTimer.C:
408
case <-ctx.Done():
409
refreshTimer.Stop()
410
return nil
411
}
412
}
413
}
414
}
415
}
416
418
>
clusterMetadataMap, err := m.listAllClusterMetadataFromDB(ctx)
419
>
if err != nil {
420
return err
421
}
422
424
>
newEntries := make(map[string]*ClusterInformation)
425
>
426
>
clusterInfoMap := m.GetAllClusterInfo()
427
>
for clusterName, newClusterInfo := range clusterMetadataMap {
428
>
oldClusterInfo, ok := clusterInfoMap[clusterName]
429
>
if !ok {
431
>
oldEntries[clusterName] = nil
432
>
newEntries[clusterName] = ShallowCopyClusterInformation(newClusterInfo)
435
>
newClusterInfo.ReplicationEnabled == oldClusterInfo.ReplicationEnabled &&
436
>
newClusterInfo.RPCAddress == oldClusterInfo.RPCAddress &&
437
>
newClusterInfo.HTTPAddress == oldClusterInfo.HTTPAddress &&
438
>
newClusterInfo.InitialFailoverVersion == oldClusterInfo.InitialFailoverVersion &&
439
>
newClusterInfo.ClusterID == oldClusterInfo.ClusterID &&
440
>
maps.Equal(newClusterInfo.Tags, oldClusterInfo.Tags) {
441
// key cluster info does not change
442
continue
443
}
444
// handle updated cluster registry
446
>
newEntries[clusterName] = ShallowCopyClusterInformation(newClusterInfo)
447
}
448
}
450
>
if _, ok := clusterMetadataMap[clusterName]; !ok {
452
>
oldEntries[clusterName] = &oldClusterInfo
453
>
newEntries[clusterName] = nil
454
>
}
455
}
456
459
>
// A bad row in cluster_metadata must not be able to crash the refresher
460
>
// or corrupt the in-memory state.
461
>
candidate := maps.Clone(clusterInfoMap)
462
>
applyClusterInfoUpdates(candidate, oldEntries, newEntries)
463
>
newVersionMap, err := updateVersionToClusterName(candidate, m.failoverVersionIncrement)
464
>
if err != nil {
466
>
}
467
469
>
m.clusterInfo = candidate
470
>
m.versionToClusterName = newVersionMap
471
>
m.clusterLock.Unlock()
472
>
473
>
m.clusterCallbackLock.RLock()
474
>
defer m.clusterCallbackLock.RUnlock()
475
>
for _, cb := range m.clusterChangeCallback {
476
>
cb(oldEntries, newEntries)
477
>
}
478
}
480
}
481
482
func applyClusterInfoUpdates(
483
clusterInfo map[string]ClusterInformation,
484
oldClusterMetadata map[string]*ClusterInformation,
485
newClusterMetadata map[string]*ClusterInformation,
487
>
for clusterName := range oldClusterMetadata {
488
>
if oldClusterMetadata[clusterName] != nil && newClusterMetadata[clusterName] == nil {
491
>
clusterInfo[clusterName] = *newClusterMetadata[clusterName]
492
>
}
493
}
494
}
495
496
// ValidateClusterInformation checks the invariants that NewMetadata and the
497
// runtime refresh both depend on. It is also used at admin/operator RPC
498
// boundaries so that bad input is rejected before it can be persisted and
499
// later crash the metadata refresher.
500
func ValidateClusterInformation(
501
clusterName string,
502
info ClusterInformation,
503
failoverVersionIncrement int64,
505
>
if clusterName == "" {
507
>
}
510
>
clusterName, info.InitialFailoverVersion)
511
>
}
513
>
return fmt.Errorf("cluster %q: InitialFailoverVersion (%d) must be < FailoverVersionIncrement (%d)",
metadata.go ×1
514
>
clusterName, info.InitialFailoverVersion, failoverVersionIncrement)
515
>
}
517
>
return fmt.Errorf("cluster %q: RPCAddress must not be empty when Enabled=true", clusterName)
metadata.go ×1
518
>
}
520
}
521
522
>
func updateVersionToClusterName(clusterInfo map[string]ClusterInformation, failoverVersionIncrement int64) (map[int64]string, error) {
metadata.go ×13
523
>
versionToClusterName := make(map[int64]string)
524
>
for clusterName, info := range clusterInfo {
525
>
if err := ValidateClusterInformation(clusterName, info, failoverVersionIncrement); err != nil {
527
>
}
529
return nil, fmt.Errorf(
530
"duplicate InitialFailoverVersion %d for clusters %q and %q",
531
info.InitialFailoverVersion, existing, clusterName)
532
}
534
}
536
}
537
538
func (m *metadataImpl) listAllClusterMetadataFromDB(
539
ctx context.Context,
541
>
result := make(map[string]*ClusterInformation)
542
>
metadataStore := m.clusterMetadataStore
543
>
if metadataStore == nil {
544
return result, nil
545
}
546
548
>
for iterator.HasNext() {
549
>
item, err := iterator.Next()
550
>
if err != nil {
551
return nil, err
552
}
554
}
556
}
557
558
// GetAllClustersIter returns an iterator that can be used to iterate over all clusters in the metadata store.
559
func GetAllClustersIter(
560
ctx context.Context,
561
metadataStore persistence.ClusterMetadataManager,
563
>
paginationFn := func(paginationToken []byte) ([]*persistence.GetClusterMetadataResponse, []byte, error) {
564
>
resp, err := metadataStore.ListClusterMetadata(
565
>
ctx,
566
>
&persistence.ListClusterMetadataRequest{
567
>
PageSize: defaultClusterMetadataPageSize,
568
>
NextPageToken: paginationToken,
569
>
},
570
>
)
571
>
if err != nil {
572
return nil, nil, err
573
}
575
}
576
578
>
return iterator
579
}
580
581
>
func ClusterInformationFromDB(getClusterResp *persistence.GetClusterMetadataResponse) *ClusterInformation {
metadata.go ×4
582
>
return &ClusterInformation{
583
>
Enabled: getClusterResp.GetIsConnectionEnabled(),
584
>
InitialFailoverVersion: getClusterResp.GetInitialFailoverVersion(),
585
>
RPCAddress: getClusterResp.GetClusterAddress(),
586
>
HTTPAddress: getClusterResp.GetHttpAddress(),
587
>
ClusterID: getClusterResp.GetClusterId(),
588
>
ShardCount: getClusterResp.GetHistoryShardCount(),
589
>
Tags: getClusterResp.GetTags(),
590
>
ReplicationEnabled: getClusterResp.GetIsReplicationEnabled(),
591
>
version: getClusterResp.Version,
592
>
}
593
>
}
594
595
// ShallowCopyClusterInformation returns a shallow copy of the given ClusterInformation. The [ClusterInformation.Tags]
596
// field is not deep-copied, so you must be careful when modifying it.
597
>
func ShallowCopyClusterInformation(information *ClusterInformation) *ClusterInformation {
metadata.go ×1
598
>
tmp := *information
599
>
return &tmp
600
>
}
601
602
// IsReplicationEnabledForCluster checks if replication is enabled for a cluster, considering the feature flag.
603
// When enableSeparateReplicationFlag is false, it falls back to only checking the Enabled flag.
604
// This is a shared helper function used across history service components.
605
>
func IsReplicationEnabledForCluster(clusterInfo ClusterInformation, enableSeparateReplicationFlag bool) bool {
metadata.go ×2
606
>
if enableSeparateReplicationFlag {
607
// New behavior: check both Enabled (for connectivity) and ReplicationEnabled (for replication streams)
608
return clusterInfo.Enabled && clusterInfo.ReplicationEnabled
609
}
610
// Old behavior: only check Enabled flag
612
}