go.temporal.io/server/common/namespace/namespace.go
377 LOC · 158 covered · 219 uncovered · 53 ranges · 16051 concepts · 44 introducers · 8111 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
package namespace
2
3
import (
4
"fmt"
5
"maps"
6
"slices"
7
"time"
8
9
"github.com/google/uuid"
10
enumspb "go.temporal.io/api/enums/v1"
11
namespacepb "go.temporal.io/api/namespace/v1"
12
rulespb "go.temporal.io/api/rules/v1"
13
"go.temporal.io/api/serviceerror"
14
persistencespb "go.temporal.io/server/api/persistence/v1"
15
"go.temporal.io/server/common"
16
"go.temporal.io/server/common/util"
17
expmaps "golang.org/x/exp/maps"
18
)
19
20
type (
21
// Mutation changes a Namespace "in-flight" during a Clone operation.
22
Mutation interface {
23
apply(*Namespace)
24
}
25
26
// BadBinaryError is an error type carrying additional information about
27
// when/why/who configured a given checksum as being bad.
28
BadBinaryError struct {
29
cksum string
30
info *namespacepb.BadBinaryInfo
31
}
32
33
// ID is the unique identifier type for a Namespace.
34
ID string
35
36
// Name is a user-supplied nickname for a Namespace.
37
Name string
38
39
// Namespaces is a *Namespace slice
40
Namespaces []*Namespace
41
42
// Namespace contains the info and config for a namespace
43
Namespace struct {
44
info *persistencespb.NamespaceInfo
45
config *persistencespb.NamespaceConfig
46
configVersion int64
47
notificationVersion int64
48
49
customSearchAttributesMapper CustomSearchAttributesMapper
50
replicationResolver ReplicationResolver
51
}
52
53
CustomSearchAttributesMapper struct {
54
fieldToAlias map[string]string
55
aliasToField map[string]string
56
}
57
58
// ReplicationPolicy is the namespace's replication policy,
59
// derived from namespace's replication config
60
ReplicationPolicy int
61
)
62
63
const (
64
EmptyName Name = ""
65
EmptyID ID = ""
66
EmptyBusinessID = ""
67
68
// ReplicationPolicyOneCluster indicate that workflows does not need to be replicated
69
// applicable to local namespace & global namespace with one cluster
70
ReplicationPolicyOneCluster ReplicationPolicy = 0
71
// ReplicationPolicyMultiCluster indicate that workflows need to be replicated
72
ReplicationPolicyMultiCluster ReplicationPolicy = 1
73
)
74
76
>
return ID(uuid.NewString())
77
>
}
78
79
func FromPersistentState(
80
detail *persistencespb.NamespaceDetail,
81
resolver ReplicationResolver,
82
mutations ...Mutation,
84
>
if resolver == nil {
85
>
return nil, serviceerror.NewInvalidArgument("replicationResolver must be provided")
namespace.go ×1
86
>
}
88
>
info: detail.Info,
89
>
config: detail.Config,
90
>
configVersion: detail.ConfigVersion,
91
>
customSearchAttributesMapper: CustomSearchAttributesMapper{
92
>
fieldToAlias: detail.Config.CustomSearchAttributeAliases,
93
>
aliasToField: util.InverseMap(detail.Config.CustomSearchAttributeAliases),
94
>
},
95
>
replicationResolver: resolver,
96
>
}
97
>
98
>
for _, m := range mutations {
100
>
}
101
103
}
104
106
>
// Clone the resolver to get a deep copy of replication state
107
>
clonedResolver := ns.replicationResolver.Clone()
108
>
109
>
cloned := &Namespace{
110
>
info: common.CloneProto(ns.info),
111
>
config: common.CloneProto(ns.config),
112
>
configVersion: ns.configVersion,
113
>
customSearchAttributesMapper: CustomSearchAttributesMapper{
114
>
fieldToAlias: ns.customSearchAttributesMapper.fieldToAlias,
115
>
aliasToField: ns.customSearchAttributesMapper.aliasToField,
116
>
},
117
>
notificationVersion: ns.notificationVersion,
118
>
replicationResolver: clonedResolver,
119
>
}
120
>
121
>
for _, m := range mutations {
122
>
m.apply(cloned)
123
>
}
124
126
}
127
128
// VisibilityArchivalState observes the visibility archive configuration (state
129
// and URI) for this namespace.
131
>
return ArchivalConfigState{
132
>
State: ns.config.VisibilityArchivalState,
133
>
URI: ns.config.VisibilityArchivalUri,
134
>
}
135
>
}
136
137
// HistoryArchivalState observes the history archive configuration (state and
138
// URI) for this namespace.
140
>
return ArchivalConfigState{
141
>
State: ns.config.HistoryArchivalState,
142
>
URI: ns.config.HistoryArchivalUri,
143
>
}
144
>
}
145
146
// VerifyBinaryChecksum returns an error if the provided checksum is one of this
147
// namespace's configured bad binary checksums. The returned error (if any) will
148
// be unwrappable as BadBinaryError.
150
>
badBinMap := ns.config.GetBadBinaries().GetBinaries()
151
>
if badBinMap == nil {
153
>
}
156
>
}
158
}
159
160
// ID observes this namespace's permanent unique identifier in string form.
162
>
if ns.info == nil {
163
return ID("")
164
}
166
}
167
168
// Name observes this namespace's configured name.
170
>
if ns.info == nil {
172
>
}
174
}
175
177
>
if ns.info == nil {
178
return enumspb.NAMESPACE_STATE_UNSPECIFIED
179
}
181
}
182
183
func (ns *Namespace) ReplicationResolver() ReplicationResolver {
184
return ns.replicationResolver
185
}
186
187
>
func (ns *Namespace) ReplicationState(businessID string) enumspb.ReplicationState {
namespace.go ×1
188
>
return ns.replicationResolver.ReplicationState(businessID)
189
>
}
190
191
// ActiveClusterName observes the name of the cluster that is currently active
192
// for this namespace.
194
>
return ns.replicationResolver.ActiveClusterName(routingKey)
195
>
}
196
197
// ClusterNames observes the names of the clusters to which this namespace is
198
// replicated.
200
>
return ns.replicationResolver.ClusterNames(businessID)
201
>
}
202
203
// IsOnCluster returns true is namespace is registered on cluster otherwise false.
205
>
return slices.Contains(ns.ClusterNames(EmptyBusinessID), clusterName)
206
>
}
207
208
// ConfigVersion return the namespace config version
209
func (ns *Namespace) ConfigVersion() int64 {
210
return ns.configVersion
211
}
212
213
// FailoverVersion return the namespace failover version
215
>
return ns.replicationResolver.FailoverVersion(businessID)
216
>
}
217
218
// IsGlobalNamespace returns whether the namespace is a global namespace.
219
// Being a global namespace doesn't necessarily mean that there are multiple registered clusters for it, only that it
220
// has a failover version. To determine whether operations should be replicated for a namespace, see ReplicationPolicy.
222
>
return ns.replicationResolver.IsGlobalNamespace()
223
>
}
224
225
// FailoverNotificationVersion return the global notification version of when failover happened
226
func (ns *Namespace) FailoverNotificationVersion() int64 {
227
return ns.replicationResolver.FailoverNotificationVersion()
228
}
229
230
// Info returns the underlying NamespaceInfo proto. The returned value is the registry's own
231
// copy and must not be mutated.
233
>
return ns.info
234
>
}
235
236
// Config returns the underlying NamespaceConfig proto. The returned value is the registry's own
237
// copy and must not be mutated.
239
>
return ns.config
240
>
}
241
242
// ReplicationConfig returns the underlying NamespaceReplicationConfig proto. The returned value
243
// is the registry's own copy and must not be mutated.
244
>
func (ns *Namespace) ReplicationConfig() *persistencespb.NamespaceReplicationConfig {
namespace.go ×3
245
>
return ns.replicationResolver.ReplicationConfig()
246
>
}
247
248
// NotificationVersion return the global notification version of when namespace changed
250
>
return ns.notificationVersion
251
>
}
252
253
// ActiveInCluster returns whether the namespace is active in the given cluster.
254
// A namespace is considered active if it is either a local namespace or a global
255
// namespace whose active cluster matches the provided cluster.
256
// Note: Do not use this to determine if a workflow is active in the cluster.
257
// Use ActiveClusterName(businessID) instead.
259
>
return ns.replicationResolver.ActiveInCluster(clusterName)
260
>
}
261
262
// ReplicationPolicy return the derived workflow replication policy
264
>
// frontend guarantee that the clusters always contains the active
265
>
// namespace, so if the # of clusters is 1 then we do not need to send out
266
>
// any events for replication
267
>
if ns.replicationResolver.IsGlobalNamespace() && len(ns.ClusterNames(EmptyBusinessID)) > 1 {
269
>
}
271
}
272
273
// GetReplicationResolver return the replication resolover
274
func (ns *Namespace) GetReplicationResolver() ReplicationResolver {
275
return ns.replicationResolver
276
}
277
279
>
if ns.info.Data == nil {
280
return ""
281
}
283
}
284
285
// Retention returns retention duration for this namespace.
287
>
if ns.config.Retention == nil {
289
>
}
290
292
}
293
294
// CustomSearchAttributesMapper is a part of temporary solution. Do not use this method.
295
>
func (ns *Namespace) CustomSearchAttributesMapper() CustomSearchAttributesMapper {
handler.go ×25
296
>
return ns.customSearchAttributesMapper
297
>
}
298
300
>
if ns.config.WorkflowRules == nil {
301
>
return nil
302
>
}
303
return expmaps.Values(ns.config.WorkflowRules)
304
}
305
306
func (ns *Namespace) GetWorkflowRule(ruleID string) (*rulespb.WorkflowRule, bool) {
307
if ns.config.WorkflowRules == nil {
308
return nil, false
309
}
310
result, ok := ns.config.WorkflowRules[ruleID]
311
return result, ok
312
}
313
314
// Error returns the reason associated with this bad binary.
315
func (e BadBinaryError) Error() string {
316
return e.info.Reason
317
}
318
319
// Reason returns the reason associated with this bad binary.
320
func (e BadBinaryError) Reason() string {
321
return e.info.Reason
322
}
323
324
// Operator returns the operator associated with this bad binary.
325
func (e BadBinaryError) Operator() string {
326
return e.info.Operator
327
}
328
329
// Created returns the time at which this bad binary was declared to be bad.
330
func (e BadBinaryError) Created() time.Time {
331
return e.info.CreateTime.AsTime()
332
}
333
334
// Checksum observes the binary checksum that caused this error.
335
func (e BadBinaryError) Checksum() string {
336
return e.cksum
337
}
338
340
>
return string(id)
341
>
}
342
344
>
return id == EmptyID
345
>
}
346
348
>
return string(n)
349
>
}
350
352
>
return n == EmptyName
353
>
}
354
355
>
func (m *CustomSearchAttributesMapper) GetAlias(fieldName string, namespace string) (string, error) {
namespace.go ×1
356
>
alias, ok := m.fieldToAlias[fieldName]
357
>
if !ok {
359
>
fmt.Sprintf("Namespace %s has no mapping defined for field name %s", namespace, fieldName),
360
>
)
361
>
}
363
}
364
365
>
func (m *CustomSearchAttributesMapper) GetFieldName(alias string, namespace string) (string, error) {
namespace.go ×1
366
>
fieldName, ok := m.aliasToField[alias]
367
>
if !ok {
369
>
fmt.Sprintf("Namespace %s has no mapping defined for search attribute %s", namespace, alias),
370
>
)
371
>
}
373
}
374
375
func (m *CustomSearchAttributesMapper) FieldToAliasMap() map[string]string {
376
return maps.Clone(m.fieldToAlias)
377
}