go.temporal.io/server/common/dynamicconfig/collection.go
779 LOC · 404 covered · 375 uncovered · 123 ranges · 22919 concepts · 81 introducers · 11231 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 dynamicconfig
2
3
import (
4
"context"
5
"errors"
6
"fmt"
7
"math"
8
"reflect"
9
"runtime"
10
"strconv"
11
"strings"
12
"sync"
13
"sync/atomic"
14
"time"
15
"weak"
16
17
"github.com/mitchellh/mapstructure"
18
"go.temporal.io/server/common/goro"
19
"go.temporal.io/server/common/log"
20
"go.temporal.io/server/common/log/tag"
21
"go.temporal.io/server/common/pingable"
22
"go.temporal.io/server/common/primitives/timestamp"
23
"go.temporal.io/server/common/util"
24
"google.golang.org/protobuf/reflect/protoreflect"
25
)
26
27
type (
28
// Collection implements lookup and constraint logic on top of a Client.
29
// The rest of the server code should use Collection as the interface to dynamic config,
30
// instead of the low-level Client.
31
Collection struct {
32
client Client
33
logger log.Logger
34
errCount int64
35
36
cancelClientSubscription func()
37
38
subscriptionLock sync.Mutex // protects subscriptions and subscriptionIdx
39
subscriptions map[Key]map[int]any // final "any" is *subscription[T]
40
subscriptionIdx int
41
42
poller goro.Group
43
44
// cache converted values. use weak pointers to avoid holding on to values in the cache
45
// that are no longer in use. this must be a pointer since the cleanup closures need to
46
// reference this without referencing Collection.
47
convertCache *sync.Map // map[weak.Pointer[ConstrainedValue]]any
48
49
// index by constraints
50
indexCache *sync.Map // map[weak.Pointer[ConstrainedValue]]map[Constraints]int32
51
}
52
53
subscription[T any] struct {
54
// constant:
55
prec []Constraints
56
f func(T)
57
def T
58
cdef []TypedConstrainedValue[T] // nil for regular settings, populated for constrained default settings
59
// protected by subscriptionLock in Collection:
60
raw any // raw value that last sent value was converted from
61
}
62
63
subscriptionCallbackSettings struct {
64
MinWorkers int
65
MaxWorkers int
66
TargetDelay time.Duration
67
ShrinkFactor float64
68
}
69
70
// sentinel type that doesn't compare equal to anything else
71
defaultValue struct{}
72
73
// These function types follow a similar pattern:
74
// {X}PropertyFn - returns a value of type X that is global (no filters)
75
// {X}PropertyFnWith{Y}Filter - returns a value of type X with the given filters
76
// Available value types:
77
// Bool: bool
78
// Duration: time.Duration
79
// Float: float64
80
// Int: int
81
// Map: map[string]any
82
// String: string
83
// Available filters:
84
// Namespace func(namespace string)
85
// NamespaceID func(namespaceID string)
86
// TaskQueue func(namespace string, taskQueue string, taskType enumspb.TaskQueueType) (matching task queue)
87
// TaskType func(taskType enumspsb.TaskType) (history task type)
88
// ShardID func(shardID int32)
89
)
90
91
const (
92
errCountLogThreshold = 1000
93
// After this many constraints, switch to a cached lookup. This value was determined
94
// empirically on my machine using BenchmarkCollectionIndexed.
95
constraintsCacheThreshold = 32
96
)
97
98
var (
99
errKeyNotPresent = errors.New("key not present")
100
errNoMatchingConstraint = errors.New("no matching constraint in key")
101
102
protoEnumType = reflect.TypeFor[protoreflect.Enum]()
103
errorType = reflect.TypeFor[error]()
104
durationType = reflect.TypeFor[time.Duration]()
105
timeType = reflect.TypeFor[time.Time]()
106
stringType = reflect.TypeFor[string]()
107
108
usingDefaultValue any = defaultValue{}
109
)
110
111
// NewCollection creates a new collection. For subscriptions to work, you must call Start/Stop.
112
// Get will work without Start/Stop.
114
>
// Do this at the first convenient place we have a logger:
115
>
logSharedStructureWarnings(logger)
116
>
117
>
return &Collection{
118
>
client: client,
119
>
logger: logger,
120
>
errCount: -1,
121
>
subscriptions: make(map[Key]map[int]any),
122
>
convertCache: new(sync.Map),
123
>
indexCache: new(sync.Map),
124
>
}
125
>
}
126
128
>
c.subscriptionLock.Lock()
129
>
defer c.subscriptionLock.Unlock()
130
>
if notifyingClient, ok := c.client.(NotifyingClient); ok {
134
>
}
135
}
136
138
>
c.poller.Cancel()
139
>
c.poller.Wait()
140
>
if c.cancelClientSubscription != nil {
142
>
}
143
}
144
145
// Implement pingable.Pingable
147
>
return []pingable.Check{
148
>
{
149
>
Name: "dynamic config callbacks",
150
>
Timeout: 5 * time.Second,
151
>
Ping: func() []pingable.Pingable {
152
>
c.subscriptionLock.Lock()
153
>
//nolint:staticcheck // SA2001 just checking if we can acquire the lock
154
>
c.subscriptionLock.Unlock()
155
>
return nil
156
>
},
157
},
158
}
159
}
160
162
>
interval := DynamicConfigSubscriptionPollInterval.Get(c)
163
>
for ctx.Err() == nil {
164
>
util.InterruptibleSleep(ctx, interval())
165
>
c.pollOnce()
166
>
}
168
}
169
171
>
c.subscriptionLock.Lock()
172
>
defer c.subscriptionLock.Unlock()
173
>
174
>
for key, subs := range c.subscriptions {
175
>
setting := queryRegistry(key)
176
>
if setting == nil {
177
continue
178
}
180
cvs := c.client.GetValue(key)
181
setting.dispatchUpdate(c, sub, cvs)
182
}
183
}
184
}
185
187
>
c.subscriptionLock.Lock()
188
>
defer c.subscriptionLock.Unlock()
189
>
190
>
for key, cvs := range changed {
191
>
setting := queryRegistry(key)
192
>
if setting == nil {
193
continue
194
}
195
// use setting.Key instead of key to avoid changing case again
198
>
}
199
}
200
}
201
203
>
// TODO: This is a lot of unnecessary contention with little benefit. Consider using
204
>
// https://github.com/cespare/percpu here.
205
>
errCount := atomic.AddInt64(&c.errCount, 1)
206
>
// log only the first x errors and then one every x after that to reduce log noise
207
>
return errCount < errCountLogThreshold || errCount%errCountLogThreshold == 0
208
>
}
209
210
func findMatch(
211
cache *sync.Map,
212
cvs []ConstrainedValue,
213
precedence []Constraints,
215
>
if len(cvs) == 0 {
217
>
} else if len(cvs) > constraintsCacheThreshold && len(cvs) <= math.MaxInt32 {
collection.go ×2
219
>
}
220
222
>
for idx, cv := range cvs {
223
>
if m == cv.Constraints {
224
>
// Note: cvs here is the slice returned by Client.GetValue. We want to return a
225
>
// pointer into that slice so that the converted value is cached as long as the
226
>
// Client keeps the []ConstrainedValue alive. See the comment on
227
>
// Client.GetValue.
228
>
return &cvs[idx], nil
229
>
}
230
}
231
}
232
// key is present but no constraint section matches
234
}
235
236
func findMatchWithCache(
237
cache *sync.Map,
238
cvs []ConstrainedValue,
239
precedence []Constraints,
241
>
var cached map[Constraints]int32
242
>
weakcvp := weak.Make(&cvs[0])
243
>
if v, ok := cache.Load(weakcvp); ok {
244
>
cached = v.(map[Constraints]int32) // nolint:revive // unchecked-type-assertion
245
>
} else {
246
>
cached = make(map[Constraints]int32, len(cvs))
247
>
for i := range cvs {
248
>
// pick first one to match behavior if multiple match
249
>
if _, ok := cached[cvs[i].Constraints]; !ok {
250
>
cached[cvs[i].Constraints] = int32(i)
251
>
}
252
}
254
>
runtime.AddCleanup(&cvs[0], func(w weak.Pointer[ConstrainedValue]) {
255
cache.Delete(w)
256
}, weakcvp)
257
}
258
}
259
261
>
if i, ok := cached[m]; ok {
262
>
// Note: cvs here is the slice returned by Client.GetValue. We want to return a
263
>
// pointer into that slice so that the converted value is cached as long as the
264
>
// Client keeps the []ConstrainedValue alive. See the comment on
265
>
// Client.GetValue.
266
>
return &cvs[i], nil
267
>
}
268
}
269
// key is present but no constraint section matches
271
}
272
273
// matchAndConvert can't be a method of Collection because methods can't be generic, but we can
274
// take a *Collection as an argument.
275
func matchAndConvert[T any](
276
c *Collection,
277
key Key,
278
def T,
279
convert func(value any) (T, error),
280
precedence []Constraints,
282
>
cvs := c.client.GetValue(key)
283
>
v, _ := matchAndConvertCvs(c, key, def, convert, precedence, cvs)
284
>
return v
285
>
}
286
287
func matchAndConvertCvs[T any](
288
c *Collection,
289
key Key,
290
def T,
291
convert func(value any) (T, error),
292
precedence []Constraints,
293
cvs []ConstrainedValue,
295
>
cvp, err := findMatch(c.indexCache, cvs, precedence)
296
>
if err != nil {
298
>
return def, usingDefaultValue
299
>
}
300
302
>
if err != nil {
304
>
if c.throttleLog() {
305
>
c.logger.Warn("Failed to convert value, using default", tag.Key(key.String()), tag.IgnoredValue(cvp), tag.Error(err))
306
>
}
307
>
return def, usingDefaultValue
308
}
310
}
311
312
// Returns matched value out of cvs, matched default out of defaultCVs, and also the priorities
313
// of each of the matches (lower matched first). For no match, order will be 0.
314
func findMatchWithConstrainedDefaults[T any](cvs []ConstrainedValue, defaultCVs []TypedConstrainedValue[T], precedence []Constraints) (
315
matchedValue *ConstrainedValue,
316
matchedDefault T,
317
valueOrder int,
318
defaultOrder int,
320
>
order := 0
321
>
for _, m := range precedence {
322
>
for idx, cv := range cvs {
324
>
if m == cv.Constraints {
325
>
if valueOrder == 0 {
326
>
valueOrder = order
327
>
// Note: cvs here is the slice returned by Client.GetValue. We want to
328
>
// return a pointer into that slice instead of copying the ConstrainedValue.
329
>
// See findMatch.
330
>
matchedValue = &cvs[idx]
331
>
}
332
}
333
}
335
>
order++
336
>
if m == cv.Constraints {
337
>
if defaultOrder == 0 {
338
>
defaultOrder = order
339
>
matchedDefault = cv.Value
340
>
}
341
}
342
}
343
}
345
}
346
347
func findAndResolveWithConstrainedDefaults[T any](
348
c *Collection,
349
key Key,
350
convert func(value any) (T, error),
351
cvs []ConstrainedValue,
352
defaultCVs []TypedConstrainedValue[T],
353
precedence []Constraints,
355
>
cvp, defVal, valOrder, defOrder := findMatchWithConstrainedDefaults(cvs, defaultCVs, precedence)
356
>
357
>
if defOrder == 0 {
358
// This is a server bug: all precedence lists must end with no-constraints, and all
359
// constrained defaults must have a no-constraints value, so we should have gotten a match.
360
c.logger.Warn("Constrained defaults had no match (this is a bug; fix server code)", tag.Key(key.String()))
361
// leave value as the zero value, that's the best we can do
362
return value, usingDefaultValue
367
>
return defVal, usingDefaultValue // use sentinel since we're using default
368
>
}
370
>
if err != nil {
371
// We failed to convert the value to the desired type. Use the default.
372
if c.throttleLog() {
373
c.logger.Warn("Failed to convert value, using default", tag.Key(key.String()), tag.IgnoredValue(cvp), tag.Error(err))
374
}
375
return defVal, usingDefaultValue
376
}
378
}
379
380
func matchAndConvertWithConstrainedDefault[T any](
381
c *Collection,
382
key Key,
383
cdef []TypedConstrainedValue[T],
384
convert func(value any) (T, error),
385
precedence []Constraints,
387
>
cvs := c.client.GetValue(key)
388
>
value, _ := findAndResolveWithConstrainedDefaults(c, key, convert, cvs, cdef, precedence)
389
>
return value
390
>
}
391
392
func subscribe[T any](
393
c *Collection,
394
key Key,
395
def T,
396
convert func(value any) (T, error),
397
prec []Constraints,
398
callback func(T),
400
>
c.subscriptionLock.Lock()
401
>
defer c.subscriptionLock.Unlock()
402
>
403
>
// get one value immediately (note that subscriptionLock is held here so we can't race with
404
>
// an update)
405
>
cvs := c.client.GetValue(key)
406
>
init, raw := matchAndConvertCvs(c, key, def, convert, prec, cvs)
407
>
408
>
// As a convenience (and for efficiency), you can pass in a nil callback; we just return the
409
>
// current value and skip the subscription. The cancellation func returned is also nil.
410
>
if callback == nil {
412
>
}
413
415
>
id := c.subscriptionIdx
416
>
417
>
if c.subscriptions[key] == nil {
418
>
c.subscriptions[key] = make(map[int]any)
419
>
}
420
422
>
prec: prec,
423
>
f: callback,
424
>
def: def,
425
>
raw: raw,
426
>
}
427
>
428
>
return init, func() {
430
>
defer c.subscriptionLock.Unlock()
431
>
delete(c.subscriptions[key], id)
432
>
}
433
}
434
435
func subscribeWithConstrainedDefault[T any](
436
c *Collection,
437
key Key,
438
cdef []TypedConstrainedValue[T],
439
convert func(value any) (T, error),
440
prec []Constraints,
441
callback func(T),
443
>
c.subscriptionLock.Lock()
444
>
defer c.subscriptionLock.Unlock()
445
>
446
>
// get one value immediately (note that subscriptionLock is held here so we can't race with
447
>
// an update)
448
>
cvs := c.client.GetValue(key)
449
>
init, raw := findAndResolveWithConstrainedDefaults(c, key, convert, cvs, cdef, prec)
450
>
451
>
// As a convenience (and for efficiency), you can pass in a nil callback; we just return the
452
>
// current value and skip the subscription. The cancellation func returned is also nil.
453
>
if callback == nil {
454
return init, nil
455
}
456
458
>
id := c.subscriptionIdx
459
>
460
>
if c.subscriptions[key] == nil {
461
>
c.subscriptions[key] = make(map[int]any)
462
>
}
463
465
>
prec: prec,
466
>
f: callback,
467
>
cdef: cdef,
468
>
raw: raw,
469
>
}
470
>
471
>
return init, func() {
473
>
defer c.subscriptionLock.Unlock()
474
>
delete(c.subscriptions[key], id)
475
>
}
476
}
477
478
// called with subscriptionLock
479
func dispatchUpdate[T any](
480
c *Collection,
481
key Key,
482
convert func(value any) (T, error),
483
sub *subscription[T],
484
cvs []ConstrainedValue,
486
>
var raw any
487
>
cvp, err := findMatch(c.indexCache, cvs, sub.prec)
488
>
if err != nil {
492
>
}
493
494
// compare raw (pre-conversion) values, if unchanged, skip this update. note that
495
// `usingDefaultValue` is equal to itself but nothing else.
497
>
// make raw field point to new one, not old one, so that old loaded files can get
collection.go ×1
498
>
// garbage collected.
499
>
sub.raw = raw
500
>
return
501
>
}
502
503
// raw value changed, need to dispatch default or converted value
505
>
if cvp == nil {
509
>
if err != nil {
510
// We failed to convert the value to the desired type. Use the default.
511
if c.throttleLog() {
512
c.logger.Warn("Failed to convert value, using default", tag.Key(key.String()), tag.IgnoredValue(cvp), tag.Error(err))
513
}
514
newVal, raw = sub.def, usingDefaultValue
515
}
516
}
517
519
>
go sub.f(newVal)
520
}
521
522
// called with subscriptionLock
523
func dispatchUpdateWithConstrainedDefault[T any](
524
c *Collection,
525
key Key,
526
convert func(value any) (T, error),
527
sub *subscription[T],
528
cvs []ConstrainedValue,
530
>
// Note: This performs the conversion even if the raw value is unchanged. This isn't ideal,
531
>
// but so far constrained default settings are only used for primitive values so it's okay.
532
>
// If we have a constrained default value with a complex conversion function, this could be
533
>
// optimized to delay conversion until after we check DeepEqual.
534
>
newVal, raw := findAndResolveWithConstrainedDefaults(c, key, convert, cvs, sub.cdef, sub.prec)
535
>
536
>
// compare raw (pre-conversion) values, if unchanged, skip this update. note that
537
>
// `usingDefaultValue` is equal to itself but nothing else.
538
>
if reflect.DeepEqual(sub.raw, raw) {
539
>
// make raw field point to new one, not old one, so that old loaded files can get
540
>
// garbage collected.
541
>
sub.raw = raw
542
>
return
543
>
}
544
546
>
go sub.f(newVal)
547
}
548
549
>
func convertWithCache[T any](c *Collection, key Key, convert func(any) (T, error), cvp *ConstrainedValue) (T, error) {
collection.go ×2
550
>
weakcvp := weak.Make(cvp)
551
>
552
>
if converted, ok := c.convertCache.Load(weakcvp); ok {
554
>
return t, nil
555
>
}
556
// Each key can only be used with a single type, so this shouldn't happen
557
c.logger.Warn("Cached converted value has wrong type", tag.Key(key.String()))
558
// Fall through to regular conversion
559
}
560
562
>
if err != nil {
564
>
return zero, err
565
>
}
566
568
>
cc := c.convertCache // capture only this pointer, not the whole Collection
569
>
runtime.AddCleanup(cvp, func(w weak.Pointer[ConstrainedValue]) {
571
>
}, weakcvp)
572
}
573
575
}
576
578
>
switch val := val.(type) {
580
>
return int(val), nil
581
case int8:
582
return int(val), nil
584
>
return int(val), nil
585
case int32:
586
return int(val), nil
588
>
return int(val), nil
589
case uint:
590
return int(val), nil
591
case uint8:
592
return int(val), nil
593
case uint16:
594
return int(val), nil
596
>
return int(val), nil
597
case uint64:
598
return int(val), nil
599
case uintptr:
600
return int(val), nil
602
>
return 0, errors.New("value type is not int")
603
}
604
}
605
607
>
switch val := val.(type) {
609
>
return float64(val), nil
611
>
return float64(val), nil
612
}
615
>
}
617
}
618
620
>
switch v := val.(type) {
622
>
return v, nil
624
>
d, err := timestamp.ParseDurationDefaultSeconds(v)
625
>
if err != nil {
627
>
}
629
}
630
// treat numeric values as seconds
635
>
}
637
}
638
640
>
if stringVal, ok := val.(string); ok {
641
>
return stringVal, nil
642
>
}
643
return "", errors.New("value type is not string")
644
}
645
647
>
switch v := val.(type) {
649
>
return v, nil
651
>
return strconv.ParseBool(v)
653
>
return false, errors.New("value type is not bool")
654
}
655
}
656
658
>
if mapVal, ok := val.(map[string]any); ok {
660
>
}
662
}
663
664
// ConvertStructure can be used as a conversion function for New*TypedSettingWithConverter.
665
// The value from dynamic config will be converted to T, on top of the given default.
666
//
667
// Note that any failure in conversion of _any_ field will result in the overall default being used,
668
// ignoring the fields that successfully converted.
669
//
670
// Note that the default value will be deep-copied and then passed to mapstructure with the
671
// ZeroFields setting false, so the config value will be _merged_ on top of it. Be very careful
672
// when using non-empty maps or slices, the result may not be what you want.
673
//
674
// To avoid confusion, the default passed to ConvertStructure should be either the same as the
675
// overall default for the setting (if you want any value set to be merged over the default, i.e.
676
// treat the fields independently), or the zero value of its type (if you want to treat the fields
677
// as a group and default unset fields to zero).
679
>
return func(v any) (T, error) {
680
>
// if we already have the right type, no conversion is necessary
681
>
if typedV, ok := v.(T); ok {
683
>
}
684
685
// Deep-copy the default and decode over it. This allows using e.g. a struct with some
686
// default fields filled in and a config that only set some fields.
688
>
689
>
dec, err := mapstructure.NewDecoder(&mapstructure.DecoderConfig{
690
>
Result: &out,
691
>
DecodeHook: mapstructure.ComposeDecodeHookFunc(
692
>
mapstructureHookDuration,
693
>
mapstructureHookTimestamp,
694
>
mapstructureHookProtoEnum,
695
>
mapstructureHookGeneric,
696
>
),
697
>
})
698
>
if err != nil {
699
return out, err
700
}
702
>
return out, err
703
}
704
}
705
706
// Parses string into time.Duration. mapstructure has an implementation of this already but it
707
// calls time.ParseDuration and we want to use our own method.
709
>
if t != durationType {
710
>
return data, nil
711
>
}
713
}
714
715
// Parses string or int into time.Time.
717
>
if t != timeType {
718
>
return data, nil
719
>
}
721
case time.Time:
722
return v, nil
724
>
ts, err := time.Parse(time.RFC3339, v)
725
>
if err != nil {
726
return time.Time{}, fmt.Errorf("failed to parse time: %v", err)
727
}
729
}
730
// treat numeric values as seconds
731
if ival, err := convertInt(data); err == nil {
732
return time.Unix(int64(ival), 0), nil
733
} else if fval, err := convertFloat(data); err == nil {
734
ipart, fpart := math.Modf(fval)
735
return time.Unix(int64(ipart), int64(fpart*float64(time.Second))), nil
736
}
737
return time.Time{}, errors.New("value not convertible to Time")
738
}
739
740
// Parses proto enum values from strings.
742
>
if f != stringType || !t.Implements(protoEnumType) {
744
>
}
746
>
str := strings.ToLower(data.(string)) // we checked f above so this can't fail
747
>
for i := 0; i < vals.Len(); i++ {
748
>
val := vals.Get(i)
749
>
if str == strings.ToLower(string(val.Name())) {
751
>
}
752
}
754
}
755
756
// Parses generic values. See GenericParseHook.
758
>
if mth, ok := t.MethodByName("DynamicConfigParseHook"); ok &&
759
>
mth.Func.IsValid() &&
760
>
mth.Type != nil &&
761
>
mth.Type.NumIn() == 2 &&
762
>
mth.Type.In(1) == f &&
763
>
mth.Type.NumOut() == 2 &&
764
>
mth.Type.Out(0) == t &&
765
>
mth.Type.Out(1) == errorType {
767
>
out := mth.Func.Call([]reflect.Value{reflect.Zero(t), reflect.ValueOf(data)})
768
>
if !out[1].IsNil() {
770
>
return nil, err
771
>
}
772
return nil, errors.New("failed to convert DynamicConfigParseHook error")
773
}
775
}
776
777
// pass through
779
}