Skip to content

Commit 392b409

Browse files
yito88yito88
committed
cluster cassandra
1 parent 9343f80 commit 392b409

File tree

3 files changed

+148
-64
lines changed

3 files changed

+148
-64
lines changed

‎scalardb/src/scalardb/core.clj‎

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -130,11 +130,6 @@
130130
[test]
131131
(prepare-service! test :2pc))
132132

133-
(defn check-storage-connection!
134-
[test]
135-
(when-not @(:storage test)
136-
(prepare-storage-service! test)))
137-
138133
(defn check-transaction-connection!
139134
[test]
140135
(when-not @(:transaction test)

‎scalardb/src/scalardb/db/cluster.clj‎

Lines changed: 124 additions & 40 deletions
Original file line numberDiff line numberDiff line change
@@ -16,10 +16,14 @@
1616
(def ^:private ^:const DEFAULT_HELM_CHART_VERSION "1.7.2")
1717
(def ^:private ^:const DEFAULT_CHAOS_MESH_VERSION "2.7.2")
1818

19+
(def ^:private ^:const DEFAULT_CLUSTER_NODE_COUNT 3)
20+
(def ^:private ^:const DEFAULT_CASSANDRA_REPLICA_COUNT 3)
21+
1922
(def ^:private ^:const TIMEOUT_SEC 600)
2023
(def ^:private ^:const INTERVAL_SEC 10)
2124

2225
(def ^:private ^:const CLUSTER_NODE_NAME "scalardb-cluster-node")
26+
(def ^:private ^:const CASSANDRA_NODE_NAME "cassandra-scalardb-cluster")
2327

2428
(def ^:private ^:const CLUSTER_VALUES
2529
{:envoy {:enabled true
@@ -30,19 +34,14 @@
3034
:tag (or (some-> (env :scalardb-cluster-version) not-empty)
3135
DEFAULT_SCALARDB_CLUSTER_VERSION)}
3236

37+
;; Storage configurations will be set later
3338
:scalardbClusterNodeProperties
3439
(str/join "\n"
3540
;; ScalarDB Cluster configurations
3641
["scalar.db.cluster.membership.type=KUBERNETES"
3742
"scalar.db.cluster.membership.kubernetes.endpoint.namespace_name=${env:SCALAR_DB_CLUSTER_MEMBERSHIP_KUBERNETES_ENDPOINT_NAMESPACE_NAME}"
3843
"scalar.db.cluster.membership.kubernetes.endpoint.name=${env:SCALAR_DB_CLUSTER_MEMBERSHIP_KUBERNETES_ENDPOINT_NAME}"
3944
""
40-
;; Storage configurations
41-
"scalar.db.storage=jdbc"
42-
"scalar.db.contact_points=jdbc:postgresql://postgresql-scalardb-cluster.default.svc.cluster.local:5432/postgres"
43-
"scalar.db.username=postgres"
44-
"scalar.db.password=postgres"
45-
""
4645
;; Set to true to include transaction metadata in the records
4746
"scalar.db.consensus_commit.include_metadata.enabled=true"])
4847

@@ -51,16 +50,39 @@
5150
(defn- update-cluster-values
5251
[test values]
5352
(let [path [:scalardbCluster :scalardbClusterNodeProperties]
53+
[storage contact-points user-pass]
54+
(case (:db-type test)
55+
:cluster ["jdbc"
56+
"jdbc:postgresql://postgresql-scalardb-cluster.default.svc.cluster.local:5432/postgres"
57+
"postgres"]
58+
:cluster-cassandra ["cassandra"
59+
(->> (map #(str "cassandra-scalardb-cluster-"
60+
%
61+
".cassandra-scalardb-cluster-headless.default.svc.cluster.local")
62+
(range DEFAULT_CASSANDRA_REPLICA_COUNT))
63+
(str/join ","))
64+
"cassandra"]
65+
(throw (ex-info "Unsupported DB type" {:db-type (:db-type test)})))
66+
isolation-level (-> test
67+
:isolation-level
68+
name
69+
str/upper-case
70+
(str/replace #"-" "_"))
5471
new-db-props (-> values
5572
(get-in path)
5673
(str
74+
;; storage
75+
"\nscalar.db.storage="
76+
storage
77+
"\nscalar.db.contact_points="
78+
contact-points
79+
"\nscalar.db.username="
80+
user-pass
81+
"\nscalar.db.password="
82+
user-pass
5783
;; isolation level
5884
"\nscalar.db.consensus_commit.isolation_level="
59-
(-> test
60-
:isolation-level
61-
name
62-
str/upper-case
63-
(str/replace #"-" "_"))
85+
isolation-level
6486
;; one phase commit
6587
(when (:enable-one-phase-commit test)
6688
"\nscalar.db.consensus_commit.one_phase_commit.enabled=true")
@@ -109,14 +131,25 @@
109131

110132
(defn- start!
111133
[test]
112-
;; postgre
113-
(c/exec
114-
:helm :install "postgresql-scalardb-cluster" "bitnami/postgresql"
115-
:--set "auth.postgresPassword=postgres"
116-
:--set "primary.persistence.enabled=true"
117-
;; Need an external IP for storage APIs
118-
:--set "service.type=LoadBalancer"
119-
:--set "primary.service.type=LoadBalancer")
134+
;; postgre or cassandra
135+
(case (:db-type test)
136+
:cluster (c/exec
137+
:helm :install "postgresql-scalardb-cluster" "bitnami/postgresql"
138+
:--set "auth.postgresPassword=postgres"
139+
:--set "primary.persistence.enabled=true"
140+
;; Need an external IP for storage APIs
141+
:--set "service.type=LoadBalancer"
142+
:--set "primary.service.type=LoadBalancer")
143+
:cluster-cassandra (c/exec
144+
:helm :install "cassandra-scalardb-cluster" "bitnami/cassandra"
145+
:--set "dbUser.user=cassandra"
146+
:--set "dbUser.user=cassandra"
147+
:--set "dbUser.password=cassandra"
148+
:--set (str "replicaCount=" DEFAULT_CASSANDRA_REPLICA_COUNT)
149+
;; Need an external IP for storage APIs
150+
:--set "service.type=LoadBalancer"
151+
:--set "primary.service.type=LoadBalancer")
152+
(throw (ex-info "Unsupported DB type" {:db-type (:db-type test)})))
120153

121154
;; ScalarDB Cluster
122155
(let [chart-version (or (some-> (env :helm-chart-version) not-empty)
@@ -139,21 +172,33 @@
139172
:--version DEFAULT_CHAOS_MESH_VERSION))
140173

141174
(defn- wipe!
142-
[]
175+
[test]
176+
;; ignore errors because these files or pods might not exist
143177
(try
144178
(info "wiping old logs...")
145179
(binding [c/*dir* (System/getProperty "user.dir")]
146180
(some->> (-> (c/exec :ls) (str/split #"\s+"))
147181
(filter #(re-matches #"scalardb-cluster-node-.*\.log" %))
148182
seq
149183
(apply c/exec :rm :-f)))
150-
(info "wiping the pods...")
151-
(c/exec :helm :uninstall :postgresql-scalardb-cluster)
184+
(catch Exception _))
185+
(info "wiping the pods...")
186+
(try
187+
(c/exec :helm :uninstall
188+
(case (:db-type test)
189+
:cluster :postgresql-scalardb-cluster
190+
:cluster-cassandra :cassandra-scalardb-cluster))
191+
(catch Exception _))
192+
(try
152193
(c/exec :kubectl :delete
153194
:pvc :-l "app.kubernetes.io/instance=postgresql-scalardb-cluster")
195+
(catch Exception _))
196+
(try
154197
(c/exec :helm :uninstall :scalardb-cluster)
198+
(catch Exception _))
199+
(try
155200
(c/exec :helm :uninstall :chaos-mesh :-n "chaos-mesh")
156-
(catch Exception _ nil)))
201+
(catch Exception _)))
157202

158203
(defn- get-pod-list
159204
[name]
@@ -182,7 +227,7 @@
182227
first))
183228

184229
(defn get-postgres-ip
185-
"Get the IP of the load balancer"
230+
"Get the IP of the Postgres"
186231
[]
187232
(->> (c/exec :kubectl :get :svc)
188233
str/split-lines
@@ -191,20 +236,29 @@
191236
(map #(nth (str/split % #"\s+") 3))
192237
first))
193238

239+
(defn get-cassandra-ip
240+
"Get one IP of the Cassandra nodes"
241+
[]
242+
(->> (c/exec :kubectl :get :svc)
243+
str/split-lines
244+
(filter #(str/includes? % "cassandra-scalardb-cluster"))
245+
(filter #(str/includes? % "LoadBalancer"))
246+
(map #(nth (str/split % #"\s+") 3))
247+
first))
248+
194249
(defn- running-pods?
195-
"Check a live node."
196-
[test]
250+
"Check if nodes are running."
251+
[test prefix num]
197252
(-> test
198253
:nodes
199254
first
200-
(c/on (get-pod-list CLUSTER_NODE_NAME))
255+
(c/on (get-pod-list prefix))
201256
count
202-
;; TODO: check the number of pods
203-
(= 3)))
257+
(= num)))
204258

205259
(defn- cluster-nodes-ready?
206260
[test]
207-
(and (running-pods? test)
261+
(and (running-pods? test CLUSTER_NODE_NAME DEFAULT_CLUSTER_NODE_COUNT)
208262
(try
209263
(c/on (-> test :nodes first)
210264
(->> (get-pod-list CLUSTER_NODE_NAME)
@@ -217,12 +271,30 @@
217271
(warn (.getMessage e))
218272
false))))
219273

274+
(defn- cassandra-nodes-ready?
275+
[test]
276+
(or (not= (:db-type test) :cluster-cassandra)
277+
(and (running-pods? test
278+
CASSANDRA_NODE_NAME
279+
DEFAULT_CASSANDRA_REPLICA_COUNT)
280+
(try
281+
(c/on (-> test :nodes first)
282+
(->> (get-pod-list CASSANDRA_NODE_NAME)
283+
(mapv #(c/exec :kubectl :wait
284+
"--for=condition=Ready"
285+
"--timeout=120s"
286+
(str "pod/" %)))))
287+
true
288+
(catch Exception e
289+
(warn (.getMessage e))
290+
false)))))
291+
220292
(defn- wait-for-recovery
221293
"Wait for the node bootstrapping."
222294
([test]
223295
(wait-for-recovery TIMEOUT_SEC INTERVAL_SEC test))
224296
([timeout-sec interval-sec test]
225-
(when-not (cluster-nodes-ready? test)
297+
(when-not (and (cassandra-nodes-ready? test) (cluster-nodes-ready? test))
226298
(Thread/sleep (* interval-sec 1000))
227299
(if (>= timeout-sec interval-sec)
228300
(wait-for-recovery (- timeout-sec interval-sec) interval-sec test)
@@ -236,7 +308,7 @@
236308
db/DB
237309
(setup! [_ test _]
238310
(when-not (:leave-db-running? test)
239-
(wipe!))
311+
(wipe! test))
240312
(install!)
241313
(configure! test)
242314
(start! test)
@@ -245,7 +317,7 @@
245317

246318
(teardown! [_ test _]
247319
(when-not (:leave-db-running? test)
248-
(wipe!)))
320+
(wipe! test)))
249321

250322
db/Primary
251323
(primaries [_ test] (:nodes test))
@@ -270,7 +342,9 @@
270342
(defrecord ExtCluster []
271343
ext/DbExtension
272344
(get-db-type [_] :cluster)
273-
(live-nodes [_ test] (running-pods? test))
345+
(live-nodes [_ test] (running-pods? test
346+
CLUSTER_NODE_NAME
347+
DEFAULT_CLUSTER_NODE_COUNT))
274348
(wait-for-recovery [_ test] (wait-for-recovery test))
275349
(create-table-opts [_ _] {})
276350
(create-properties
@@ -288,13 +362,23 @@
288362
(ext/set-common-properties test)))))
289363
(create-storage-properties [_ _]
290364
(let [node (-> test :nodes first)
291-
ip (c/on node (get-postgres-ip))]
365+
db-type (:db-type test)
366+
[storage contact-points user-pass]
367+
(c/on node (case db-type
368+
:cluster ["jdbc"
369+
(str "jdbc:postgresql://"
370+
(get-postgres-ip)
371+
":5432/postgres")
372+
"postgres"]
373+
:cluster-cassandra ["cassandra"
374+
(get-cassandra-ip)
375+
"cassandra"]
376+
(throw (ex-info "Unsupported DB type" {:db-type db-type}))))]
292377
(doto (Properties.)
293-
(.setProperty "scalar.db.storage" "jdbc")
294-
(.setProperty "scalar.db.contact_points"
295-
(str "jdbc:postgresql://" ip ":5432/postgres"))
296-
(.setProperty "scalar.db.username" "postgres")
297-
(.setProperty "scalar.db.password" "postgres")))))
378+
(.setProperty "scalar.db.storage" storage)
379+
(.setProperty "scalar.db.contact_points" contact-points)
380+
(.setProperty "scalar.db.username" user-pass)
381+
(.setProperty "scalar.db.password" user-pass)))))
298382

299383
(defn gen-db
300384
[faults admin]

‎scalardb/src/scalardb/runner.clj‎

Lines changed: 24 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -19,11 +19,12 @@
1919
[elle-write-read-2pc]]
2020
[clojure.core :as c]))
2121

22-
(def db-keys
22+
(def db-types
2323
"The map of test DBs."
2424
{"cassandra" :cassandra
2525
"postgres" :postgres
26-
"cluster" :cluster})
26+
"cluster" :cluster
27+
"cluster-cassandra" :cluster-cassandra})
2728

2829
(def workload-keys
2930
"A map of test workload keys."
@@ -58,7 +59,7 @@
5859

5960
(def test-opt-spec
6061
[(cli/repeated-opt nil "--db NAME" "DB(s) on which the test is run"
61-
[:cassandra] db-keys)
62+
[:cassandra] db-types)
6263

6364
(cli/repeated-opt nil "--workload NAME" "Test(s) to run" [] workload-keys)
6465

@@ -99,21 +100,24 @@
99100
(into (map name admin))
100101
(->> (remove nil?) (string/join "-"))))
101102

102-
(defn- load-module
103-
[db-key]
104-
(case db-key
105-
:cassandra (require 'scalardb.db.cassandra)
106-
:postgres (require 'scalardb.db.postgres)
107-
:cluster (require 'scalardb.db.cluster)
108-
(throw (ex-info "Unsupported DB" {:db db-key}))))
103+
(def ^:private db->gen-namespace
104+
{:cassandra 'scalardb.db.cassandra
105+
:postgres 'scalardb.db.postgres
106+
:cluster 'scalardb.db.cluster
107+
:cluster-cassandra 'scalardb.db.cluster})
108+
109+
(defn- load-gen-db-fn
110+
[db-type]
111+
(if-let [ns-sym (db->gen-namespace db-type)]
112+
(do
113+
(require ns-sym)
114+
(-> ns-sym str (symbol "gen-db") resolve))
115+
(throw (ex-info "Unsupported DB" {:db db-type}))))
109116

110117
(defn- gen-db
111118
"Returns [extended-db constructed-nemesis num-max-nodes]."
112-
[db-key faults admin]
113-
(load-module db-key)
114-
(let [gen-db-sym (symbol (str "scalardb.db." (name db-key)) "gen-db")
115-
gen-db-fn (resolve gen-db-sym)]
116-
(gen-db-fn faults admin)))
119+
[db-type faults admin]
120+
((load-gen-db-fn db-type) faults admin))
117121

118122
(defn- gen-test-opt-spec
119123
[]
@@ -134,8 +138,8 @@
134138
:decommissioned (atom #{})})
135139

136140
(defn scalardb-test
137-
[base-opts db-key workload-key faults admin]
138-
(let [[db nemesis max-nodes] (gen-db db-key faults admin)
141+
[base-opts db-type workload-key faults admin]
142+
(let [[db nemesis max-nodes] (gen-db db-type faults admin)
139143
consistency-model (->> base-opts :consistency-model (mapv keyword))
140144
workload-opts (merge base-opts
141145
scalardb-opts
@@ -146,6 +150,7 @@
146150
workload-opts
147151
{:name (test-name workload-key faults admin)
148152
:client (:client workload)
153+
:db-type db-type
149154
:db db
150155
:pure-generators true
151156
:generator (gen/phases
@@ -167,12 +172,12 @@
167172
:usage (cli/test-usage)
168173
:run (fn [{:keys [options]}]
169174
(doseq [_ (range (:test-count options))
170-
db-key (:db options)
175+
db-type (:db options)
171176
workload-key (:workload options)
172177
faults (:nemesis options)
173178
admin (or (:admin options) [[]])]
174179
(let [test (-> options
175-
(scalardb-test db-key
180+
(scalardb-test db-type
176181
workload-key
177182
faults
178183
admin)

0 commit comments

Comments
 (0)