82
83
// newCQLClient returns a new instance of CQLClient
84
>
func newCQLClient(cfg *CQLClientConfig, logger log.Logger) (*cqlClient, error) {
cqlclient.go
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
}
101
>
logger.Info("Connection validation succeeded.")
cqlclient.go
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
113
>
func (cfg *CQLClientConfig) toCassandraConfig() *config.Cassandra {
cqlclient.go
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
136
>
func (client *cqlClient) CreateDatabase(name string) error {
cqlclient.go
137
>
return client.createKeyspace(name)
138
>
}
139
140
>
func (client *cqlClient) DropDatabase(name string) error {
cqlclient.go
141
>
return client.dropKeyspace(name)
142
>
}
143
144
// createKeyspace creates a cassandra Keyspace if it doesn't exist
145
>
func (client *cqlClient) createKeyspace(name string) error {
cqlclient.go
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
151
>
return client.Exec(fmt.Sprintf(createKeyspaceCQL, name, client.nReplicas))
152
}
153
154
// dropKeyspace drops a Keyspace
155
>
func (client *cqlClient) dropKeyspace(name string) error {
cqlclient.go
156
>
return client.Exec(fmt.Sprintf("DROP KEYSPACE IF EXISTS %v", name))
157
>
}
158
159
>
func (client *cqlClient) DropAllTables() error {
cqlclient.go
160
>
return client.dropAllTablesTypes()
161
>
}
162
163
// CreateSchemaVersionTables sets up the schema version tables
164
>
func (client *cqlClient) CreateSchemaVersionTables() error {
cqlclient.go
165
>
if err := client.Exec(createSchemaVersionTableCQL); err != nil {
166
return err
167
}
168
>
return client.Exec(createSchemaUpdateHistoryTableCQL)
cqlclient.go
169
}
170
171
// ReadSchemaVersion returns the current schema version for the Keyspace
172
>
func (client *cqlClient) ReadSchemaVersion() (string, error) {
cqlclient.go
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 {
180
err = fmt.Errorf("no schema version found for keyspace %q", client.keyspace)
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
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
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
203
>
func (client *cqlClient) Exec(stmt string, args ...any) error {
cqlclient.go
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
218
>
func (client *cqlClient) ListTables() ([]string, error) {
cqlclient.go
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
233
>
func (client *cqlClient) listTypes() ([]string, error) {
cqlclient.go
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) {
239
names = append(names, name)
240
}
242
return nil, err
243
}
245
}
246
247
// dropTable drops a given table from the Keyspace
248
>
func (client *cqlClient) dropTable(name string) error {
cqlclient.go
249
>
return client.Exec(fmt.Sprintf("DROP TABLE %v", name))
250
>
}
251
252
// dropType drops a given type from the Keyspace