go.temporal.io/server/tools/cassandra/cqlclient.go
305 LOC · 143 covered · 162 uncovered · 41 ranges · 17 concepts · 6 introducers · 7 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
"context"
5
"fmt"
6
"time"
7
8
"github.com/gocql/gocql"
9
"go.temporal.io/server/common/auth"
10
"go.temporal.io/server/common/config"
11
"go.temporal.io/server/common/log"
12
"go.temporal.io/server/common/log/tag"
13
"go.temporal.io/server/common/metrics"
14
commongocql "go.temporal.io/server/common/persistence/nosql/nosqlplugin/cassandra/gocql"
15
"go.temporal.io/server/common/resolver"
16
"go.temporal.io/server/tools/common/schema"
17
)
18
19
type (
20
cqlClient struct {
21
nReplicas int
22
datacenter string
23
keyspace string
24
timeout time.Duration
25
session commongocql.Session
26
logger log.Logger
27
}
28
// CQLClientConfig contains the configuration for cql client
29
CQLClientConfig struct {
30
Hosts string
31
Port int
32
User string
33
Password string
34
AllowedAuthenticators []string
35
Keyspace string
36
Timeout int
37
numReplicas int
38
Datacenter string
39
Consistency string
40
TLS *auth.TLS
41
DisableInitialHostLookup bool
42
AddressTranslator *config.CassandraAddressTranslator
43
}
44
)
45
46
const (
47
defaultTimeout = 30 // Timeout in seconds
48
systemKeyspace = "system"
49
dbType = "cassandra"
50
)
51
52
const (
53
readSchemaVersionCQL = `SELECT curr_version from schema_version where keyspace_name=?`
54
listTablesCQL = `SELECT table_name from system_schema.tables where keyspace_name=?`
55
listTypesCQL = `SELECT type_name from system_schema.types where keyspace_name=?`
56
writeSchemaVersionCQL = `INSERT into schema_version(keyspace_name, creation_time, curr_version, min_compatible_version) VALUES (?,?,?,?)`
57
writeSchemaUpdateHistoryCQL = `INSERT into schema_update_history(year, month, update_time, old_version, new_version, manifest_md5, description) VALUES(?,?,?,?,?,?,?)`
58
59
createSchemaVersionTableCQL = `CREATE TABLE IF NOT EXISTS schema_version(keyspace_name text PRIMARY KEY, ` +
60
`creation_time timestamp, ` +
61
`curr_version text, ` +
62
`min_compatible_version text);`
63
64
createSchemaUpdateHistoryTableCQL = `CREATE TABLE IF NOT EXISTS schema_update_history(` +
65
`year int, ` +
66
`month int, ` +
67
`update_time timestamp, ` +
68
`description text, ` +
69
`manifest_md5 text, ` +
70
`new_version text, ` +
71
`old_version text, ` +
72
`PRIMARY KEY ((year, month), update_time));`
73
74
createKeyspaceCQL = `CREATE KEYSPACE IF NOT EXISTS %v ` +
75
`WITH replication = { 'class' : 'SimpleStrategy', 'replication_factor' : %v};`
76
77
createKeyspaceNetworkTopologyCQL = `CREATE KEYSPACE IF NOT EXISTS %v ` +
78
`WITH replication = { 'class' : 'NetworkTopologyStrategy', '%v' : %v};`
79
)
80
81
var _ schema.DB = (*cqlClient)(nil)
82
83
// newCQLClient returns a new instance of CQLClient
84
>
func newCQLClient(cfg *CQLClientConfig, logger log.Logger) (*cqlClient, error) {
cqlclient.go ×11
85
>
var err error
86
>
87
>
cassandraConfig := cfg.toCassandraConfig()
88
>
89
>
logger.Info("Validating connection to cassandra cluster.")
90
>
session, err := commongocql.NewSession(
91
>
func() (*gocql.ClusterConfig, error) {
92
>
return commongocql.NewCassandraCluster(*cassandraConfig, resolver.NewNoopResolver())
93
>
},
94
logger,
95
metrics.NoopMetricsHandler,
96
)
98
logger.Error("Connection validation failed.", tag.Error(err))
99
return nil, err
100
}
102
>
103
>
return &cqlClient{
104
>
keyspace: cfg.Keyspace,
105
>
nReplicas: cfg.numReplicas,
106
>
datacenter: cfg.Datacenter,
107
>
timeout: time.Duration(cfg.Timeout) * time.Second,
108
>
session: session,
109
>
logger: logger,
110
>
}, nil
111
}
112
114
>
cassandraConfig := config.Cassandra{
115
>
Hosts: cfg.Hosts,
116
>
Port: cfg.Port,
117
>
User: cfg.User,
118
>
Password: cfg.Password,
119
>
AllowedAuthenticators: cfg.AllowedAuthenticators,
120
>
Keyspace: cfg.Keyspace,
121
>
TLS: cfg.TLS,
122
>
Datacenter: cfg.Datacenter,
123
>
DisableInitialHostLookup: cfg.DisableInitialHostLookup,
124
>
Consistency: &config.CassandraStoreConsistency{
125
>
Default: &config.CassandraConsistencySettings{
126
>
Consistency: cfg.Consistency,
127
>
},
128
>
},
129
>
AddressTranslator: cfg.AddressTranslator,
130
>
ConnectTimeout: time.Duration(cfg.Timeout) * time.Second,
131
>
}
132
>
133
>
return &cassandraConfig
134
>
}
135
137
>
return client.createKeyspace(name)
138
>
}
139
141
>
return client.dropKeyspace(name)
142
>
}
143
144
// createKeyspace creates a cassandra Keyspace if it doesn't exist
146
>
if client.datacenter != "" {
147
client.logger.Info(fmt.Sprintf("Creating Keyspace %v using NetworkTopologyStrategy in Datacenter %v with RF=%v.", name, client.datacenter, client.nReplicas))
148
return client.Exec(fmt.Sprintf(createKeyspaceNetworkTopologyCQL, name, client.datacenter, client.nReplicas))
149
}
150
>
client.logger.Info(fmt.Sprintf("Creating Keyspace %v using SimpleStrategy with RF=%v.", name, client.nReplicas))
cqlclient.go ×11
151
>
return client.Exec(fmt.Sprintf(createKeyspaceCQL, name, client.nReplicas))
152
}
153
154
// dropKeyspace drops a Keyspace
156
>
return client.Exec(fmt.Sprintf("DROP KEYSPACE IF EXISTS %v", name))
157
>
}
158
160
>
return client.dropAllTablesTypes()
161
>
}
162
163
// CreateSchemaVersionTables sets up the schema version tables
165
>
if err := client.Exec(createSchemaVersionTableCQL); err != nil {
166
return err
167
}
169
}
170
171
// ReadSchemaVersion returns the current schema version for the Keyspace
173
>
query := client.session.Query(readSchemaVersionCQL, client.keyspace)
174
>
175
>
iter := query.Iter()
176
>
var version string
177
>
success := iter.Scan(&version)
178
>
err := iter.Close()
179
>
if err == nil && !success {
181
>
}
183
>
return "", fmt.Errorf("unable to get current schema version from Cassandra: %w", err)
184
>
}
186
}
187
188
// UpdateShemaVersion updates the schema version for the Keyspace
189
>
func (client *cqlClient) UpdateSchemaVersion(newVersion string, minCompatibleVersion string) error {
cqlclient.go ×19
190
>
query := client.session.Query(writeSchemaVersionCQL, client.keyspace, time.Now().UTC(), newVersion, minCompatibleVersion)
191
>
return query.Exec()
192
>
}
193
194
// WriteSchemaUpdateLog adds an entry to the schema update history table
195
>
func (client *cqlClient) WriteSchemaUpdateLog(oldVersion string, newVersion string, manifestMD5 string, desc string) error {
cqlclient.go ×19
196
>
now := time.Now().UTC()
197
>
query := client.session.Query(writeSchemaUpdateHistoryCQL)
198
>
query.Bind(now.Year(), int(now.Month()), now, oldVersion, newVersion, manifestMD5, desc)
199
>
return query.Exec()
200
>
}
201
202
// Exec executes a cql statement
204
>
if err := client.session.Query(stmt, args...).Exec(); err != nil {
205
return err
206
}
208
}
209
210
// Close closes the cql client
212
>
if client.session != nil {
213
>
client.session.Close()
214
>
}
215
}
216
217
// ListTables lists the table names in a Keyspace
219
>
query := client.session.Query(listTablesCQL, client.keyspace)
220
>
iter := query.Iter()
221
>
var names []string
222
>
var name string
223
>
for iter.Scan(&name) {
225
>
}
227
return nil, err
228
}
230
}
231
232
// listTypes lists the User defined types in a Keyspace
234
>
qry := client.session.Query(listTypesCQL, client.keyspace)
235
>
iter := qry.Iter()
236
>
var names []string
237
>
var name string
238
>
for iter.Scan(&name) {
240
>
}
242
return nil, err
243
}
245
}
246
247
// dropTable drops a given table from the Keyspace
249
>
return client.Exec(fmt.Sprintf("DROP TABLE %v", name))
250
>
}
251
252
// dropType drops a given type from the Keyspace
254
>
return client.Exec(fmt.Sprintf("DROP TYPE %v", name))
255
>
}
256
257
// dropAllTablesTypes deletes all tables/types in the
258
// Keyspace without deleting the Keyspace
260
>
tables, err := client.ListTables()
261
>
if err != nil {
262
return err
263
}
265
>
for _, table := range tables {
267
>
if err1 != nil {
268
client.logger.Error(fmt.Sprintf("Error dropping table %v.", table), tag.Error(err1))
269
}
270
}
271
273
>
if err != nil {
274
return err
275
}
277
>
numOfTypes := len(types)
278
>
for i := 0; i < numOfTypes && len(types) > 0; i++ {
280
>
for _, t := range types {
281
>
err = client.dropType(t)
282
>
if err != nil {
283
client.logger.Error(fmt.Sprintf("Error dropping type %v.", t), tag.Error(err))
284
erroredTypes = append(erroredTypes, t)
285
}
286
}
288
}
290
return err
291
}
293
}
294
295
// waitSchemaAgreement wait for schema change agreements
297
>
ctx, cancel := context.WithTimeout(context.Background(), client.timeout)
298
>
defer cancel()
299
>
return client.session.AwaitSchemaAgreement(ctx)
300
>
}
301
302
// Type gives the type of db
303
func (client *cqlClient) Type() string {
304
return dbType
305
}