Vihren Atlas

file_based_client.go ×30

Code frontier

unlabeled · c_e77d8b42196b

49 tests · 1905 LOC · 70 files · introduces 0 tests · 225 LOC · 5 files

Introduces — evidence that enters the hierarchy at this concept

Code
68 ranges225 lines · 5 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
285 ranges1905 lines · 70 files · Browse complete extent
All tests (intent)
49 testsBrowse complete intent

Neighbourhood graph

The orange circle is the focus. Violet and green circles are every ancestor and descendant, broader and narrower, at any distance; blue squares and pink diamonds are the introduced files and exact introduced tests of every visible concept, not only the focus's. Arrows point from broader to narrower concepts and bridge only concepts omitted from this view. Undirected links show source or test introduction. Concept and file size follows LOC; exact test nodes use test-count units.

Introduced files, introduced tests, and structurally relevant concept specialization

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 native relationship evidence on this page.

Graph controls are ready.

Interactive rendering requires JavaScript and WebGL. Use the native relationship evidence on this page while the interactive map is unavailable.

Native relationship evidence

Every exact file and test below is linked only from the concept that introduces it.

Introduced tests

Every collected test enters the hierarchy at exactly one concept.

No tests are introduced at this concept. Its intent tests are introduced by other concepts.

Introduced code

Every collected source range enters the hierarchy at exactly one concept.

5 files ranked by introduced lines: 225 introduced LOC across 68 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/common/dynamicconfig/yaml_loader.go 91 introduced LOC · 28 ranges

Open complete file

35
36 // LoadYamlFile is a convenience function to create a YamlLoader and call LoadFile.
37 > func LoadYamlFile(contents []byte) *YamlLoader { yaml_loader.go
38 > var lr YamlLoader
39 > lr.LoadFile(contents)
40 > return &lr
41 > }
42
43 // Err returns a joined error of lr.Errors.
47
48 // LoadFile parses and processes the given file contents into lr.Map (initialized if nil).
49 > func (lr *YamlLoader) LoadFile(contents []byte) { yaml_loader.go
50 > var yamlValues map[string]yamlConstrainedValue
51 > if err := yaml.Unmarshal(contents, &yamlValues); err != nil {
52 lr.errorf("decode error: %w", err)
53 return
54 }
55
56 > if lr.Map == nil { yaml_loader.go
57 > lr.Map = make(ConfigValueMap, len(yamlValues))
58 > }
59
60 > for key, yamlCV := range yamlValues { yaml_loader.go
61 > lr.add(MakeKey(key), yamlCV)
62 > }
63 }
64
77 }
78
79 > func (lr *YamlLoader) add(key Key, yamlCV yamlConstrainedValue) { yaml_loader.go
80 > precedence := PrecedenceUnknown
81 > setting := queryRegistry(key)
82 > if setting == nil {
83 > lr.warnf("unregistered key %q", key)
84 > } else {
85 > precedence = setting.Precedence()
86 > }
87
88 > cvs := make([]ConstrainedValue, len(yamlCV)) yaml_loader.go
89 > for i, cv := range yamlCV {
90 > // yaml will unmarshal map into map[interface{}]interface{} instead of map[string]interface{}
91 > // manually convert key type to string for all values here
92 > val, err := convertKeyTypeToString(cv.Value)
93 > if err != nil {
94 lr.error(err)
95 continue
97
98 // try validating if known setting
99 > if setting != nil { yaml_loader.go
100 > if valErr := setting.Validate(val); valErr != nil {
101 > // TODO: raise this to error level
102 > lr.warnf("validation failed: key %q value %v: %w", key, cv.Value, valErr)
103 > }
104 }
105
106 > cvs[i].Value = val yaml_loader.go
107 > cvs[i].Constraints = convertYamlConstraints(key, cv.Constraints, precedence, lr)
108 }
109 > lr.Map[key] = cvs yaml_loader.go
110 }
111
112 > func (lr *YamlLoader) warn(err error) { yaml_loader.go
113 > lr.Warnings = append(lr.Warnings, err)
114 > }
115
116 > func (lr *YamlLoader) warnf(format string, args ...any) { yaml_loader.go
117 > lr.warn(fmt.Errorf(format, args...))
118 > }
119
120 func (lr *YamlLoader) error(err error) {
126 }
127
128 > func convertKeyTypeToString(v any) (any, error) { yaml_loader.go
129 > switch v := v.(type) {
130 case map[any]any:
131 return convertKeyTypeToStringMap(v)
132 case []any:
133 return convertKeyTypeToStringSlice(v)
134 > default: yaml_loader.go
135 > return v, nil
136 }
137 }
166
167 // nolint:revive // cognitive-complexity, it's just a big switch
168 > func convertYamlConstraints(key Key, m map[string]any, precedence Precedence, lr *YamlLoader) Constraints { yaml_loader.go
169 > var cs Constraints
170 > for k, v := range m {
171 > validConstraint := true
172 > switch strings.ToLower(k) {
173 > case "namespace":
174 > if v, ok := v.(string); ok {
175 > cs.Namespace = v
176 > } else {
177 lr.errorf("namespace constraint must be string")
178 }
179 > validConstraint = precedence == PrecedenceNamespace || precedence == PrecedenceTaskQueue || precedence == PrecedenceDestination yaml_loader.go
180 case "namespaceid":
181 if v, ok := v.(string); ok {
185 }
186 validConstraint = precedence == PrecedenceNamespaceID
187 > case "taskqueuename": yaml_loader.go
188 > if v, ok := v.(string); ok {
189 > cs.TaskQueueName = v
190 > } else {
191 lr.errorf("taskQueueName constraint must be string")
192 }
193 > validConstraint = precedence == PrecedenceTaskQueue yaml_loader.go
194 > case "tasktype":
195 > switch v := v.(type) {
196 > case string:
197 > i, err := enumspb.TaskQueueTypeFromString(v)
198 > if err != nil {
199 lr.errorf("invalid value for taskType: %w", err)
200 > } else if i <= enumspb.TASK_QUEUE_TYPE_UNSPECIFIED { yaml_loader.go
201 lr.errorf("taskType constraint must be Workflow/Activity")
202 }
203 > cs.TaskQueueType = i yaml_loader.go
204 case int:
205 if v > int(enumspb.TASK_QUEUE_TYPE_UNSPECIFIED) {
211 lr.errorf("taskType constraint must be Workflow/Activity")
212 }
213 > validConstraint = precedence == PrecedenceTaskQueue yaml_loader.go
214 > case "historytasktype":
215 > switch v := v.(type) {
216 > case string:
217 > tt, err := enumsspb.TaskTypeFromString(v)
218 > if err != nil {
219 lr.errorf("invalid value for historytasktype constraint: %w", err)
220 > } else if tt <= enumsspb.TASK_TYPE_UNSPECIFIED { yaml_loader.go
221 lr.errorf("historytasktype %s constraint is not supported", v)
222 }
223 > cs.TaskType = tt yaml_loader.go
224 > case int:
225 > cs.TaskType = enumsspb.TaskType(v)
226 default:
227 lr.errorf("historytasktype %T constraint is not supported", v)
228 }
229 > validConstraint = precedence == PrecedenceTaskType yaml_loader.go
230 case "shardid":
231 if v, ok := v.(int); ok {
235 }
236 validConstraint = precedence == PrecedenceShardID
237 > case "destination": yaml_loader.go
238 > if v, ok := v.(string); ok {
239 > cs.Destination = v
240 > } else {
241 lr.errorf("destination constraint must be string")
242 }
243 > validConstraint = precedence == PrecedenceDestination yaml_loader.go
244 > case "chasmtasktype":
245 > if v, ok := v.(string); ok {
246 > cs.ChasmTaskType = v
247 > } else {
248 lr.errorf("chasmtasktype constraint must be string")
249 }
250 > validConstraint = precedence == PrecedenceChasmTaskType yaml_loader.go
251 default:
252 lr.errorf("unknown constraint type %q", k)
256 // unregistered key above
257 // TODO: raise this to error level
258 > if !validConstraint && precedence != PrecedenceUnknown { yaml_loader.go
259 lr.warnf("constraint %q isn't valid for dynamic config key %q", k, key)
260 }
261 }
262 > return cs yaml_loader.go
263 }
go.temporal.io/server/common/dynamicconfig/file_based_client.go 77 introduced LOC · 30 ranges

Open complete file

59 }
60
61 > func NewFileBasedClientWithMetrics(config *FileBasedClientConfig, logger log.Logger, doneCh <-chan any, metricsHandler metrics.Handler) (*FileBasedClient, error) { file_based_client.go
62 > if config == nil {
63 return nil, errors.New("configuration for dynamic config client is nil")
64 }
65 > reader := &osReader{path: config.Filepath} file_based_client.go
66 > return NewFileBasedClientWithReader(reader, config, logger, doneCh, metricsHandler)
67 }
68
69 > func NewFileBasedClientWithReader(reader FileReader, config *FileBasedClientConfig, logger log.Logger, doneCh <-chan any, metricsHandler metrics.Handler) (*FileBasedClient, error) { file_based_client.go
70 > if config == nil {
71 return nil, errors.New("configuration for dynamic config client is nil")
72 }
73 > if reader == nil { file_based_client.go
74 return nil, errors.New("file reader for dynamic config client is nil")
75 }
76 > if logger == nil { file_based_client.go
77 return nil, errors.New("logger for dynamic config client is nil")
78 }
79 > if metricsHandler == nil { file_based_client.go
80 metricsHandler = metrics.NoopMetricsHandler
81 logger.Warn("metrics handler is nil, using noop metrics handler")
82 }
83
84 > client := &FileBasedClient{ file_based_client.go
85 > logger: logger,
86 > reader: reader,
87 > config: config,
88 > doneCh: doneCh,
89 > NotifyingClientImpl: NewNotifyingClientImpl(),
90 > }
91 > client.metricsHandler.Store(&metricsHandler)
92 > err := client.init()
93 > if err != nil {
94 return nil, err
95 }
96 > return client, nil file_based_client.go
97 }
98
107 }
108
109 > func (fc *FileBasedClient) getMetricsHandler() metrics.Handler { file_based_client.go
110 > h := fc.metricsHandler.Load() // nolint:revive // unchecked-type-assertion
111 > if h == nil {
112 // this should never happen, but we'll log a warning if it does
113 fc.logger.Warn("dynamic config is missing correct metrics handler, using noop metrics handler")
114 return metrics.NoopMetricsHandler
115 }
116 > return *h file_based_client.go
117 }
118
122 }
123
124 > func (fc *FileBasedClient) init() error { file_based_client.go
125 > if err := fc.validateStaticConfig(fc.config); err != nil {
126 return fmt.Errorf("unable to validate dynamic config: %w", err)
127 }
128
129 > if err := fc.Update(); err != nil { file_based_client.go
130 return fmt.Errorf("unable to read dynamic config: %w", err)
131 }
132
133 > go func() { file_based_client.go
134 > ticker := time.NewTicker(fc.config.PollInterval)
135 > for {
136 > select {
137 case <-ticker.C:
138 err := fc.Update()
140 fc.logger.Error("Unable to update dynamic config.", tag.Error(err))
141 }
142 > case <-fc.doneCh: file_based_client.go
143 > ticker.Stop()
144 > return
145 }
146 }
147 }()
148
149 > return nil file_based_client.go
150 }
151
152 // This is public mainly for testing. The update loop will call this periodically, you don't
153 // have to call it explicitly.
154 > func (fc *FileBasedClient) Update() (updateErr error) { file_based_client.go
155 > modtime, updateErr := fc.reader.GetModTime()
156 > retryOnErr := true
157 > defer func() {
158 > h := fc.getMetricsHandler()
159 > // gauge value 1 refers to a failed update state, and should trigger alerts
160 > if updateErr != nil {
161 metrics.DynamicConfigUpdateFailure.With(h).Record(1)
162 > } else { file_based_client.go
163 > metrics.DynamicConfigUpdateFailure.With(h).Record(0)
164 > }
165 > if updateErr == nil || !retryOnErr {
166 > fc.lastCheckedTime = modtime
167 > }
168 }()
169 > if updateErr != nil { file_based_client.go
170 return fmt.Errorf("dynamic config file: %s: %w", fc.config.Filepath, updateErr)
171 }
172 > if !modtime.After(fc.lastCheckedTime) { file_based_client.go
173 return nil
174 }
175
176 > contents, err := fc.reader.ReadFile() file_based_client.go
177 > if err != nil {
178 return fmt.Errorf("dynamic config file: %s: %w", fc.config.Filepath, err)
179 }
180
181 > lr := LoadYamlFile(contents) file_based_client.go
182 > for _, e := range lr.Errors {
183 fc.logger.Error("dynamic config error", tag.Error(e))
184 }
185 > for _, w := range lr.Warnings { file_based_client.go
186 > fc.logger.Warn("dynamic config warning", tag.Error(w))
187 > }
188 > if len(lr.Errors) > 0 {
189 // we don't retry on parsing errors which will fail deterministically until the file is fixed
190 retryOnErr = false
193 }
194
195 > prev := fc.values.Swap(lr.Map) file_based_client.go
196 > oldValues, _ := prev.(ConfigValueMap) // nolint:revive // unchecked-type-assertion
197 > changedMap := DiffAndLogConfigs(fc.logger, oldValues, lr.Map)
198 > fc.logger.Info("Updated dynamic config")
199 >
200 > fc.PublishUpdates(changedMap)
201 > return nil
202 }
203
204 > func (fc *FileBasedClient) validateStaticConfig(config *FileBasedClientConfig) error { file_based_client.go
205 > if config == nil {
206 return errors.New("configuration for dynamic config client is nil")
207 }
208 > if _, err := fc.reader.GetModTime(); err != nil { file_based_client.go
209 return fmt.Errorf("dynamic config: %s: %w", config.Filepath, err)
210 }
211 > if config.PollInterval < minPollInterval { file_based_client.go
212 return fmt.Errorf("poll interval should be at least %v", minPollInterval)
213 }
214 > return nil file_based_client.go
215 }
216
217 > func (r *osReader) ReadFile() ([]byte, error) { file_based_client.go
218 > return os.ReadFile(r.path)
219 > }
220
221 > func (r *osReader) GetModTime() (time.Time, error) { file_based_client.go
222 > fi, err := os.Stat(r.path)
223 > if err != nil {
224 return time.Time{}, err
225 }
226 > return fi.ModTime(), nil file_based_client.go
227 }
go.temporal.io/server/common/dynamicconfig/client_diff.go 47 introduced LOC · 7 ranges

Open complete file

14 // returned as a ConfigValueMap that can be merged with old to produce new, except with deleted
15 // keys mapped to nil. It also logs the differences to a logger.
16 > func DiffAndLogConfigs(logger log.Logger, oldValues ConfigValueMap, newValues ConfigValueMap) ConfigValueMap { client_diff.go
17 > changedMap := make(map[Key][]ConstrainedValue)
18 >
19 > for key, newValues := range newValues {
20 > oldValues, ok := oldValues[key]
21 > if !ok {
22 > for _, newValue := range newValues {
23 > // new key added
24 > diffAndLogValue(logger, key, nil, &newValue)
25 > }
26 > changedMap[Key(key)] = newValues
27 } else {
28 // compare existing keys
35
36 // check for removed values
37 > for key, oldValues := range oldValues { client_diff.go
38 if _, ok := newValues[key]; !ok {
39 for _, oldValue := range oldValues {
81 }
82
83 > func diffAndLogValue(logger log.Logger, key Key, oldValue *ConstrainedValue, newValue *ConstrainedValue) { client_diff.go
84 > logLine := &strings.Builder{}
85 > logLine.Grow(128)
86 > logLine.WriteString("dynamic config changed for the key: ")
87 > logLine.WriteString(key.String())
88 > logLine.WriteString(" oldValue: ")
89 > appendConstrainedValue(logLine, oldValue)
90 > logLine.WriteString(" newValue: ")
91 > appendConstrainedValue(logLine, newValue)
92 > logger.Info(logLine.String())
93 > }
94
95 > func appendConstrainedValue(logLine *strings.Builder, value *ConstrainedValue) { client_diff.go
96 > if value == nil {
97 > logLine.WriteString("nil")
98 > } else {
99 > logLine.WriteString("{ constraints: {")
100 > if value.Constraints.Namespace != "" {
101 > fmt.Fprintf(logLine, "{Namespace:%s}", value.Constraints.Namespace)
102 > }
103 > if value.Constraints.NamespaceID != "" {
104 fmt.Fprintf(logLine, "{NamespaceID:%s}", value.Constraints.NamespaceID)
105 }
106 > if value.Constraints.TaskQueueName != "" { client_diff.go
107 > fmt.Fprintf(logLine, "{TaskQueueName:%s}", value.Constraints.TaskQueueName)
108 > }
109 > if value.Constraints.TaskQueueType != enumspb.TASK_QUEUE_TYPE_UNSPECIFIED {
110 > fmt.Fprintf(logLine, "{TaskQueueType:%s}", value.Constraints.TaskQueueType)
111 > }
112 > if value.Constraints.ShardID != 0 {
113 fmt.Fprintf(logLine, "{ShardID:%d}", value.Constraints.ShardID)
114 }
115 > if value.Constraints.TaskType != enumsspb.TASK_TYPE_UNSPECIFIED { client_diff.go
116 > fmt.Fprintf(logLine, "{HistoryTaskType:%s}", value.Constraints.TaskType)
117 > }
118 > if value.Constraints.Destination != "" {
119 > fmt.Fprintf(logLine, "{Destination:%s}", value.Constraints.Destination)
120 > }
121 > fmt.Fprint(logLine, "} value: ", value.Value, " }")
122 }
123 }
go.temporal.io/server/api/enums/v1/task.go-helpers.pb.go 5 introduced LOC · 2 ranges

Open complete file

66 // TaskTypeFromString parses a TaskType value from either the protojson
67 // canonical SCREAMING_CASE enum or the traditional temporal PascalCase enum to TaskType
68 > func TaskTypeFromString(s string) (TaskType, error) { task.go-helpers.pb.go
69 > if v, ok := TaskType_value[s]; ok {
70 return TaskType(v), nil
71 > } else if v, ok := TaskType_shorthandValue[s]; ok { task.go-helpers.pb.go
72 > return TaskType(v), nil
73 > }
74 return TaskType(0), fmt.Errorf("%s is not a valid TaskType", s)
75 }
go.temporal.io/server/common/dynamicconfig/setting_gen.go 5 introduced LOC · 1 range

Open complete file

1022
1023 func (s NamespaceTypedSetting[T]) Key() Key { return s.key }
1024 > func (s NamespaceTypedSetting[T]) Precedence() Precedence { return PrecedenceNamespace } setting_gen.go
1025 > func (s NamespaceTypedSetting[T]) Validate(v any) error {
1026 > _, err := s.convert(v)
1027 > return err
1028 > }
1029
1030 func (s NamespaceTypedConstrainedDefaultSetting[T]) Key() Key { return s.key }

Direct hierarchy

Parents

Children