-
Notifications
You must be signed in to change notification settings - Fork 787
/
metadata.go
216 lines (195 loc) · 8.48 KB
/
metadata.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
// Copyright (c) 2018 Uber Technologies, Inc.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
// THE SOFTWARE.
package cluster
import (
"fmt"
"github.com/uber/cadence/common"
"github.com/uber/cadence/common/config"
"github.com/uber/cadence/common/dynamicconfig"
"github.com/uber/cadence/common/log"
"github.com/uber/cadence/common/log/tag"
"github.com/uber/cadence/common/metrics"
)
type (
// Metadata provides information about clusters
Metadata struct {
log log.Logger
metrics metrics.Scope
// failoverVersionIncrement is the increment of each cluster's version when failover happen
failoverVersionIncrement int64
// primaryClusterName is the name of the primary cluster, only the primary cluster can register / update domain
// all clusters can do domain failover
primaryClusterName string
// currentClusterName is the name of the current cluster
currentClusterName string
// allClusters contains all cluster info
allClusters map[string]config.ClusterInformation
// enabledClusters contains enabled info
enabledClusters map[string]config.ClusterInformation
// remoteClusters contains enabled and remote info
remoteClusters map[string]config.ClusterInformation
// versionToClusterName contains all initial version -> corresponding cluster name
versionToClusterName map[int64]string
// allows for a new failover version migration
useNewFailoverVersionOverride dynamicconfig.BoolPropertyFnWithDomainFilter
}
)
// NewMetadata create a new instance of Metadata
func NewMetadata(
failoverVersionIncrement int64,
primaryClusterName string,
currentClusterName string,
clusterGroup map[string]config.ClusterInformation,
useMinFailoverVersionOverrideConfig dynamicconfig.BoolPropertyFnWithDomainFilter,
metricsClient metrics.Client,
logger log.Logger,
) Metadata {
versionToClusterName := make(map[int64]string)
for clusterName, info := range clusterGroup {
versionToClusterName[info.InitialFailoverVersion] = clusterName
}
// We never use disable clusters, filter them out on start
enabledClusters := map[string]config.ClusterInformation{}
for cluster, info := range clusterGroup {
if info.Enabled {
enabledClusters[cluster] = info
}
}
// Precompute remote clusters, they are used in multiple places
remoteClusters := map[string]config.ClusterInformation{}
for cluster, info := range enabledClusters {
if cluster != currentClusterName {
remoteClusters[cluster] = info
}
}
return Metadata{
log: logger,
metrics: metricsClient.Scope(metrics.ClusterMetadataScope),
failoverVersionIncrement: failoverVersionIncrement,
primaryClusterName: primaryClusterName,
currentClusterName: currentClusterName,
allClusters: clusterGroup,
enabledClusters: enabledClusters,
remoteClusters: remoteClusters,
versionToClusterName: versionToClusterName,
useNewFailoverVersionOverride: useMinFailoverVersionOverrideConfig,
}
}
// GetNextFailoverVersion return the next failover version based on input
func (m Metadata) GetNextFailoverVersion(cluster string, currentFailoverVersion int64, domainName string) int64 {
initialFailoverVersion := m.getInitialFailoverVersion(cluster, domainName)
failoverVersion := currentFailoverVersion/m.failoverVersionIncrement*m.failoverVersionIncrement + initialFailoverVersion
if failoverVersion < currentFailoverVersion {
return failoverVersion + m.failoverVersionIncrement
}
return failoverVersion
}
// IsVersionFromSameCluster return true if the new version is used for the same cluster
func (m Metadata) IsVersionFromSameCluster(version1 int64, version2 int64) bool {
v1Server, err := m.resolveServerName(version1)
if err != nil {
// preserving old behaviour however, this should never occur
m.metrics.IncCounter(metrics.ClusterMetadataFailureToResolveCounter)
m.log.Error("could not resolve an incoming version", tag.Dynamic("failover-version", version1))
return false
}
v2Server, err := m.resolveServerName(version2)
if err != nil {
m.log.Error("could not resolve an incoming version", tag.Dynamic("failover-version", version2))
return false
}
return v1Server == v2Server
}
func (m Metadata) IsPrimaryCluster() bool {
return m.primaryClusterName == m.currentClusterName
}
// GetCurrentClusterName return the current cluster name
func (m Metadata) GetCurrentClusterName() string {
return m.currentClusterName
}
// GetAllClusterInfo return all cluster info
func (m Metadata) GetAllClusterInfo() map[string]config.ClusterInformation {
return m.allClusters
}
// GetEnabledClusterInfo return enabled cluster info
func (m Metadata) GetEnabledClusterInfo() map[string]config.ClusterInformation {
return m.enabledClusters
}
// GetRemoteClusterInfo return enabled AND remote cluster info
func (m Metadata) GetRemoteClusterInfo() map[string]config.ClusterInformation {
return m.remoteClusters
}
// ClusterNameForFailoverVersion return the corresponding cluster name for a given failover version
func (m Metadata) ClusterNameForFailoverVersion(failoverVersion int64) (string, error) {
if failoverVersion == common.EmptyVersion {
return m.currentClusterName, nil
}
server, err := m.resolveServerName(failoverVersion)
if err != nil {
m.metrics.IncCounter(metrics.ClusterMetadataResolvingFailoverVersionCounter)
return "", fmt.Errorf("failed to resolve failover version: %v", err)
}
return server, nil
}
// gets the initial failover version for a cluster / domain
// along with some helpers for a migration - should it be necessary
func (m Metadata) getInitialFailoverVersion(cluster string, domainName string) int64 {
info, ok := m.allClusters[cluster]
if !ok {
panic(fmt.Sprintf(
"Unknown cluster name: %v with given cluster initial failover version map: %v.",
cluster,
m.allClusters,
))
}
// if using the minFailover Version during a cluster config, then return this from config
// (assuming it's safe to do so). This is not the normal state of things and intended only
// for when migrating versions.
usingNewFailoverVersion := m.useNewFailoverVersionOverride(domainName)
if usingNewFailoverVersion && info.NewInitialFailoverVersion != nil {
m.log.Debug("using new failover version for cluster", tag.ClusterName(cluster), tag.WorkflowDomainName(domainName))
m.metrics.IncCounter(metrics.ClusterMetadataGettingMinFailoverVersionCounter)
return *info.NewInitialFailoverVersion
}
// default behaviour - return the initial failover version - a marker to
// identify the cluster for all counters
m.log.Debug("getting failover version for cluster", tag.ClusterName(cluster), tag.WorkflowDomainName(domainName))
m.metrics.IncCounter(metrics.ClusterMetadataGettingFailoverVersionCounter)
return info.InitialFailoverVersion
}
// resolves the server name from a version number. Better to use this
// than to check versionToClusterName directly, as this also falls back to catch
// when there's a migration NewInitialFailoverVersion
func (m Metadata) resolveServerName(version int64) (string, error) {
moddedFoVersion := version % m.failoverVersionIncrement
// attempt a lookup first
server, ok := m.versionToClusterName[moddedFoVersion]
if ok {
return server, nil
}
// else fall back on checking for new failover versions
for name, cluster := range m.allClusters {
if cluster.NewInitialFailoverVersion != nil && *cluster.NewInitialFailoverVersion == moddedFoVersion {
return name, nil
}
}
m.metrics.IncCounter(metrics.ClusterMetadataFailureToResolveCounter)
return "", fmt.Errorf("could not resolve failover version: %d", version)
}