go.temporal.io/server/common/config/persistence.go
323 LOC · 143 covered · 180 uncovered · 65 ranges · 3012 concepts · 40 introducers · 2174 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 config
2
3
import (
4
"bytes"
5
"context"
6
"errors"
7
"fmt"
8
"os/exec"
9
"reflect"
10
"strings"
11
"time"
12
13
"github.com/gocql/gocql"
14
"go.temporal.io/server/common/persistence/visibility/store/elasticsearch/client"
15
)
16
17
const (
18
// StoreTypeSQL refers to sql based storage as persistence store
19
StoreTypeSQL = "sql"
20
// StoreTypeNoSQL refers to nosql based storage as persistence store
21
StoreTypeNoSQL = "nosql"
22
)
23
24
var ErrPersistenceConfig = errors.New("persistence config error")
25
26
// DefaultStoreType returns the storeType for the default persistence store
28
>
if c.DataStores[c.DefaultStore].SQL != nil {
30
>
}
32
}
33
34
// Validate validates the persistence config
36
>
stores := []string{c.DefaultStore}
37
>
if c.VisibilityStore != "" {
38
>
stores = append(stores, c.VisibilityStore)
39
>
}
40
>
if c.SecondaryVisibilityStore != "" {
41
stores = append(stores, c.SecondaryVisibilityStore)
42
}
43
44
// There are 3 config keys:
45
// - visibilityStore: can set any data store
46
// - secondaryVisibilityStore: can set any data store
47
// If visibilityStore is set, then it's always the primary.
48
// If secondaryVisibilityStore is set, it's always the secondary.
49
//
50
// Valid dual visibility combinations (order: primary, secondary):
51
// - visibilityStore (advanced sql), secondaryVisibilityStore (advanced sql)
52
// - visibilityStore (es), visibilityStore (es) [via elasticsearch.indices config]
53
// - visibilityStore (es), secondaryVisibilityStore (es)
54
//
55
// Invalid dual visibility combinations:
56
// - visibilityStore (advanced sql), secondaryVisibilityStore (es)
57
// - visibilityStore (es), secondaryVisibilityStore (advanced sql)
58
60
return fmt.Errorf("%w: visibilityStore must be specified", ErrPersistenceConfig)
61
}
63
isAnyCustom := c.DataStores[c.VisibilityStore].CustomDataStoreConfig != nil ||
64
c.DataStores[c.SecondaryVisibilityStore].CustomDataStoreConfig != nil
65
isPrimaryEs := c.DataStores[c.VisibilityStore].Elasticsearch != nil
66
isSecondaryEs := c.DataStores[c.SecondaryVisibilityStore].Elasticsearch != nil
67
if !isAnyCustom && isPrimaryEs != isSecondaryEs {
68
return fmt.Errorf(
69
"%w: cannot set visibilityStore and secondaryVisibilityStore with different datastore types",
70
ErrPersistenceConfig)
71
}
72
if c.DataStores[c.VisibilityStore].Elasticsearch.GetSecondaryVisibilityIndex() != "" {
73
return fmt.Errorf(
74
"%w: cannot set secondaryVisibilityStore "+
75
"when visibilityStore is setting Elasticsearch secondary visibility index",
76
ErrPersistenceConfig)
77
}
78
if c.DataStores[c.SecondaryVisibilityStore].Elasticsearch.GetSecondaryVisibilityIndex() != "" {
79
return fmt.Errorf(
80
"%w: secondary visibility datastore %q cannot set secondary_visibility",
81
ErrPersistenceConfig,
82
c.SecondaryVisibilityStore)
83
}
84
}
85
87
>
ds, ok := c.DataStores[st]
88
>
if !ok {
89
return fmt.Errorf("%w: missing config for datastore %q", ErrPersistenceConfig, st)
90
}
92
return fmt.Errorf("%w: datastore %q: %s", ErrPersistenceConfig, st, err.Error())
93
}
94
}
96
}
97
98
// VisibilityConfigExist returns whether user specified visibilityStore in config
100
>
return c.VisibilityStore != ""
101
>
}
102
103
// SecondaryVisibilityConfigExist returns whether user specified secondaryVisibilityStore in config
105
>
return c.SecondaryVisibilityStore != ""
106
>
}
107
108
func (c *Persistence) IsSQLVisibilityStore() bool {
109
return (c.VisibilityConfigExist() && c.DataStores[c.VisibilityStore].SQL != nil) ||
110
(c.SecondaryVisibilityConfigExist() && c.DataStores[c.SecondaryVisibilityStore].SQL != nil)
111
}
112
113
func (c *Persistence) IsCustomVisibilityStore() bool {
114
return c.GetVisibilityStoreConfig().CustomDataStoreConfig != nil ||
115
c.GetSecondaryVisibilityStoreConfig().CustomDataStoreConfig != nil
116
}
117
119
>
return c.DataStores[c.VisibilityStore]
120
>
}
121
123
>
if c.SecondaryVisibilityStore != "" {
124
return c.DataStores[c.SecondaryVisibilityStore]
125
}
127
>
ds := c.DataStores[c.VisibilityStore]
128
>
if ds.Elasticsearch != nil && ds.Elasticsearch.GetSecondaryVisibilityIndex() != "" {
129
esConfig := *ds.Elasticsearch
130
esConfig.Indices = map[string]string{
131
client.VisibilityAppName: ds.Elasticsearch.GetSecondaryVisibilityIndex(),
132
}
133
ds.Elasticsearch = &esConfig
134
return ds
135
}
136
}
138
}
139
141
>
switch {
142
>
case ds.SQL != nil:
143
>
return ds.SQL.DatabaseName
144
case ds.Cassandra != nil:
145
return ds.Cassandra.Keyspace
146
case ds.Elasticsearch != nil:
147
return ds.Elasticsearch.GetVisibilityIndex()
148
case ds.CustomDataStoreConfig != nil:
149
return ds.CustomDataStoreConfig.IndexName
151
>
return ""
152
}
153
}
154
155
// Validate validates the data store config
157
>
storeConfigCount := 0
158
>
if ds.SQL != nil {
159
>
storeConfigCount++
160
>
}
161
>
if ds.Cassandra != nil {
162
storeConfigCount++
163
}
165
storeConfigCount++
166
}
168
storeConfigCount++
169
}
171
return errors.New(
172
"must provide config for one and only one datastore: " +
173
"elasticsearch, cassandra, sql or custom store",
174
)
175
}
176
178
>
if ds.SQL.TaskScanPartitions == 0 {
179
>
ds.SQL.TaskScanPartitions = 1
180
>
}
181
>
if err := ds.SQL.validate(); err != nil {
182
return err
183
}
184
}
186
if err := ds.Cassandra.validate(); err != nil {
187
return err
188
}
189
}
191
if err := ds.Elasticsearch.Validate(); err != nil {
192
return err
193
}
194
}
196
}
197
198
// GetConsistency returns the gosql.Consistency setting from the configuration for the given store type
200
>
return gocql.ParseConsistency(c.getConsistencySettings().Consistency)
201
>
}
202
203
// GetSerialConsistency returns the gosql.SerialConsistency setting from the configuration for the store
204
>
func (c *CassandraStoreConsistency) GetSerialConsistency() gocql.SerialConsistency {
persistence.go ×2
205
>
res, err := parseSerialConsistency(c.getConsistencySettings().SerialConsistency)
206
>
if err != nil {
207
panic(fmt.Sprintf("unable to decode cassandra serial consistency: %v", err))
208
}
210
}
211
212
>
func (c *CassandraStoreConsistency) getConsistencySettings() *CassandraConsistencySettings {
persistence.go ×6
213
>
return ensureStoreConsistencyNotNil(c).Default
214
>
}
215
216
>
func ensureStoreConsistencyNotNil(c *CassandraStoreConsistency) *CassandraStoreConsistency {
persistence.go ×6
217
>
if c == nil {
219
>
}
222
>
}
225
>
}
228
>
}
229
231
}
232
233
func (c *Cassandra) validate() error {
234
return c.Consistency.validate()
235
}
236
238
>
if c == nil {
240
>
}
241
243
>
244
>
for _, field := range v.Fields() {
245
>
s, ok := field.Interface().(*CassandraConsistencySettings)
246
>
if ok {
247
>
if err := s.validate(); err != nil {
249
>
}
250
}
251
}
252
254
}
255
257
>
if c == nil {
259
>
}
260
263
>
if err != nil {
265
>
}
266
}
267
270
>
if err != nil {
272
>
}
273
}
274
276
}
277
278
>
func parseSerialConsistency(serialConsistency string) (gocql.SerialConsistency, error) {
persistence.go ×1
279
>
var s gocql.SerialConsistency
280
>
err := s.UnmarshalText([]byte(strings.ToUpper(serialConsistency)))
281
>
return s, err
282
>
}
283
285
>
if c.PasswordCommand != nil && c.Password != "" {
287
>
}
290
>
}
292
}
293
294
const (
295
defaultPasswordCommandTimeout = 30 * time.Second
296
passwordCommandWaitDelay = 5 * time.Second
297
)
298
299
// ResolvePassword returns the database password, either from the static Password
300
// field or by executing PasswordCommand. If neither is set, it returns an empty string.
302
>
if c.PasswordCommand == nil {
304
>
}
306
>
if timeout == 0 {
308
>
}
310
>
defer cancel()
311
>
cmd := exec.CommandContext(ctx, c.PasswordCommand.Command, c.PasswordCommand.Args...) //nolint:gosec
312
>
// WaitDelay caps how long we block on the stdout pipe after the process is killed.
313
>
// Without it, a subprocess that inherits the pipe could keep it open indefinitely.
314
>
cmd.WaitDelay = passwordCommandWaitDelay
315
>
var stderr bytes.Buffer
316
>
cmd.Stderr = &stderr
317
>
out, err := cmd.Output()
318
>
if err != nil {
320
>
c.PasswordCommand.Command, c.PasswordCommand.Args, err, stderr.String())
321
>
}
323
}