go.temporal.io/server/tools/cassandra/handler.go
233 LOC · 102 covered · 131 uncovered · 39 ranges · 20 concepts · 14 introducers · 9 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 cassandra
2
3
import (
4
"fmt"
5
"strings"
6
7
"github.com/urfave/cli"
8
"go.temporal.io/server/common/auth"
9
c "go.temporal.io/server/common/config"
10
"go.temporal.io/server/common/log"
11
"go.temporal.io/server/common/log/tag"
12
"go.temporal.io/server/temporal/environment"
13
"go.temporal.io/server/tools/common/schema"
14
)
15
16
const defaultNumReplicas = 1
17
18
// SetupSchemaConfig contains the configuration params needed to setup schema tables
19
type SetupSchemaConfig struct {
20
CQLClientConfig
21
schema.SetupConfig
22
}
23
24
// setupSchema executes the setupSchemaTask
25
// using the given command line arguments
26
// as input
28
>
config, err := newCQLClientConfig(cli)
29
>
if err != nil {
30
logger.Error("Unable to read config.", tag.Error(schema.NewConfigError(err.Error())))
31
return err
32
}
34
>
if err != nil {
35
logger.Error("Unable to establish CQL session.", tag.Error(err))
36
return err
37
}
39
>
if err := schema.Setup(cli, client, logger); err != nil {
41
>
return err
42
>
}
44
}
45
46
// updateSchema executes the updateSchemaTask
47
// using the given command line args as input
49
>
config, err := newCQLClientConfig(cli)
50
>
if err != nil {
51
logger.Error("Unable to read config.", tag.Error(schema.NewConfigError(err.Error())))
52
return err
53
}
55
>
if err != nil {
56
logger.Error("Unable to establish CQL session.", tag.Error(err))
57
return err
58
}
60
>
if err := schema.Update(cli, client, logger); err != nil {
61
logger.Error("Unable to update CQL schema.", tag.Error(err))
62
return err
63
}
65
}
66
68
>
config, err := newCQLClientConfig(cli)
69
>
if err != nil {
70
logger.Error("Unable to read config.", tag.Error(schema.NewConfigError(err.Error())))
71
return err
72
}
74
>
if keyspace == "" {
76
>
logger.Error("Unable to read config.", tag.Error(schema.NewConfigError(err.Error())))
77
>
return err
78
>
}
80
>
if err != nil {
81
logger.Error("Unable to create keyspace.", tag.Error(err))
82
return err
83
}
85
}
86
88
>
config, err := newCQLClientConfig(cli)
89
>
if err != nil {
90
logger.Error("Unable to read config.", tag.Error(schema.NewConfigError(err.Error())))
91
return err
92
}
94
>
if keyspace == "" {
95
>
err := fmt.Errorf("missing %s argument", flag(schema.CLIOptKeyspace))
96
>
logger.Error("Unable to read config.", tag.Error(schema.NewConfigError(err.Error())))
97
>
return err
98
>
}
99
err = doDropKeyspace(config, keyspace, logger)
100
if err != nil {
101
logger.Error("Unable to drop keyspace.", tag.Error(err))
102
return err
103
}
104
return nil
105
}
106
107
func validateHealth(cli *cli.Context, logger log.Logger) error {
108
config, err := newCQLClientConfig(cli)
109
if err != nil {
110
logger.Error("Unable to read config.", tag.Error(schema.NewConfigError(err.Error())))
111
return err
112
}
113
114
config.Keyspace = systemKeyspace
115
116
client, err := newCQLClient(config, logger)
117
if err != nil {
118
logger.Error("Unable to establish CQL session.", tag.Error(err))
119
return err
120
}
121
122
defer client.Close()
123
return nil
124
}
125
126
>
func doCreateKeyspace(cfg *CQLClientConfig, name string, logger log.Logger) error {
handler.go ×4
127
>
cfg.Keyspace = systemKeyspace
128
>
client, err := newCQLClient(cfg, logger)
129
>
if err != nil {
130
return err
131
}
133
>
return client.createKeyspace(name)
134
}
135
136
func doDropKeyspace(cfg *CQLClientConfig, name string, logger log.Logger) error {
137
cfg.Keyspace = systemKeyspace
138
client, err := newCQLClient(cfg, logger)
139
if err != nil {
140
return err
141
}
142
defer client.Close()
143
return client.dropKeyspace(name)
144
}
145
147
>
config := &CQLClientConfig{
148
>
Hosts: cli.GlobalString(schema.CLIOptEndpoint),
149
>
Port: cli.GlobalInt(schema.CLIOptPort),
150
>
User: cli.GlobalString(schema.CLIOptUser),
151
>
Password: cli.GlobalString(schema.CLIOptPassword),
152
>
AllowedAuthenticators: cli.GlobalStringSlice(schema.CLIOptAllowedAuthenticators),
153
>
Timeout: cli.GlobalInt(schema.CLIOptTimeout),
154
>
Keyspace: cli.GlobalString(schema.CLIOptKeyspace),
155
>
numReplicas: cli.Int(schema.CLIOptReplicationFactor),
156
>
Datacenter: cli.String(schema.CLIOptDatacenter),
157
>
Consistency: cli.String(schema.CLIOptConsistency),
158
>
DisableInitialHostLookup: cli.GlobalBool(schema.CLIFlagDisableInitialHostLookup),
159
>
}
160
>
161
>
if cli.GlobalBool(schema.CLIFlagEnableTLS) {
162
config.TLS = &auth.TLS{
163
Enabled: true,
164
CertFile: cli.GlobalString(schema.CLIFlagTLSCertFile),
165
KeyFile: cli.GlobalString(schema.CLIFlagTLSKeyFile),
166
CaFile: cli.GlobalString(schema.CLIFlagTLSCaFile),
167
ServerName: cli.GlobalString(schema.CLIFlagTLSHostName),
168
EnableHostVerification: !cli.GlobalBool(schema.CLIFlagTLSDisableHostVerification),
169
}
170
}
171
173
>
Translator: cli.GlobalString(schema.CLIOptAddressTranslator),
174
>
Options: parseOptionsMap(cli.GlobalString(schema.CLIOptAddressTranslatorOptions)),
175
>
}
176
>
177
>
if err := validateCQLClientConfig(config); err != nil {
178
return nil, err
179
}
181
}
182
184
>
if len(value) == 0 {
186
>
}
187
189
>
190
>
for pair := range strings.SplitSeq(value, ",") {
191
>
trimmedPair := strings.ReplaceAll(pair, " ", "")
192
>
if len(trimmedPair) == 0 {
193
continue
194
}
196
>
if len(splitPair) != 2 {
197
continue
198
}
200
>
continue
201
}
203
}
204
206
}
207
209
>
if len(config.Hosts) == 0 {
210
>
return schema.NewConfigError("missing cassandra endpoint argument " + flag(schema.CLIOptEndpoint))
handler.go ×3
211
>
}
213
>
return schema.NewConfigError("missing " + flag(schema.CLIOptKeyspace) + " argument ")
handler.go ×3
214
>
}
217
>
}
220
>
}
221
222
>
if config.AddressTranslator != nil && len(config.AddressTranslator.Options) != 0 {
handler.go ×6
223
if len(config.AddressTranslator.Translator) == 0 {
224
return schema.NewConfigError("missing address translator argument " + flag(schema.CLIOptAddressTranslator))
225
}
226
}
227
229
}
230
232
>
return "(-" + opt + ")"
233
>
}