Add Postgres (#116)

cassandra/src/cassandra/nemesis.clj

@@ -199,15 +199,15 @@
 
 (defn flush-generator
   [opts]
-  (when (contains? (:admin opts) :flush-compact)
+  (when (contains? (:admin opts) :flush)
     (->> (gen/mix [(repeat {:type :info, :f :flush})
                    (repeat {:type :info, :f :compact})])
          (gen/stagger default-interval))))
 
 (defn flush-package
   "A combined nemesis package for flush and compaction."
   [opts]
-  (when (contains? (:admin opts) :flush-compact)
+  (when (contains? (:admin opts) :flush)
     {:nemesis   (flush-nemesis)
      :generator (flush-generator opts)
      :perf      #{{:name  "flush"

cassandra/src/cassandra/runner.clj

@@ -42,21 +42,13 @@
 (def admin
   {"none" []
    "join" [:join]
-   "flush" [:flush-compact]})
+   "flush" [:flush]})
 
 (def test-opt-spec
   [(cli/repeated-opt nil "--workload NAME" "Test(s) to run" [] workload-keys)])
 
 (def cassandra-opt-spec
-  [(cli/repeated-opt nil "--nemesis NAME" "Which nemeses to use"
-                     [[]]
-                     nemeses)
-
-   (cli/repeated-opt nil "--admin NAME" "Which admin operations to use"
-                     [[]]
-                     admin)
-
-   [nil "--rf REPLICATION_FACTOR" "Replication factor"
+  [[nil "--rf REPLICATION_FACTOR" "Replication factor"
     :default 3
     :parse-fn #(Long/parseLong %)
     :validate [pos? "Must be positive"]]
@@ -66,6 +58,16 @@
 
    (cli/tarball-opt link-to-tarball)])
 
+(def nemesis-opt-spec
+  [(cli/repeated-opt nil "--nemesis NAME" "Which nemeses to use"
+                     [[]]
+                     nemeses)])
+
+(def admin-opt-spec
+  [(cli/repeated-opt nil "--admin NAME" "Which admin operations to use"
+                     [[]]
+                     admin)])
+
 (defn cassandra-test
   [opts]
   (let [target (:target opts)
@@ -108,6 +110,8 @@
   []
   {"test" {:opt-spec (->> test-opt-spec
                           (into cassandra-opt-spec)
+                          (into nemesis-opt-spec)
+                          (into admin-opt-spec)
                           (into cli/test-opt-spec))
            :opt-fn (fn [parsed] (-> parsed cli/test-opt-fn))
            :usage (cli/test-usage)

cassandra/test/cassandra/runner_test.clj

@@ -6,7 +6,7 @@
   (let [opts {:target "cassandra"
               :workload :batch
               :nemesis [:crash]
-              :admin [:flush-compact]
+              :admin [:flush]
               :time-limit 60}
         test (runner/cassandra-test opts)]
-    (is (= "cassandra-batch-crash-flush-compact" (:name test)))))
+    (is (= "cassandra-batch-crash-flush" (:name test)))))

docker/node/Dockerfile

@@ -7,6 +7,8 @@ RUN apt-get -y -q update && \
     psmisc \
     python \
     ntpdate \
+    gnupg \
+    iproute2 \
     iptables
 
 ADD ./init.sh /init.sh

scalardb/src/scalardb/db/postgres.clj

@@ -0,0 +1,136 @@
+(ns scalardb.db.postgres
+  (:require [clojure.tools.logging :refer [info]]
+            [jepsen
+             [control :as c]
+             [db :as db]
+             [util :refer [meh]]]
+            [jepsen.control.util :as cu]
+            [jepsen.os.debian :as debian]))
+
+(def ^:private ^:const DEFAULT_VERSION "15")
+(def ^:private ^:const TIMEOUT_SEC 600)
+(def ^:private ^:const INTERVAL_SEC 10)
+
+(defn- install!
+  "Installs PostgreSQL."
+  [{:keys [version] :or {version DEFAULT_VERSION}}]
+  (let [postgre (keyword (str "postgresql-" version))
+        client (keyword (str "postgresql-client-" version))]
+    (c/su
+     (c/exec :wget
+             :--quiet
+             :-O
+             :- "https://www.postgresql.org/media/keys/ACCC4CF8.asc"
+             c/| :apt-key :add :-)
+     (debian/install [:lsb-release])
+     (let [release (c/exec :lsb_release :-cs)]
+       (debian/add-repo! "postgresql"
+                         (str "deb http://apt.postgresql.org/pub/repos/apt/ "
+                              release "-pgdg main")))
+     (debian/install [postgre client])
+     (c/su (c/exec :sed :-i
+                   (c/lit "\"s/#listen_addresses = 'localhost'/listen_addresses = '*'/g\"")
+                   (str "/etc/postgresql/" version "/main/postgresql.conf")))
+     (c/su (c/exec :echo
+                   (c/lit "host all all 0.0.0.0/0 trust")
+                   c/| :tee :-a
+                   (str "/etc/postgresql/" version "/main/pg_hba.conf")
+                   :> "/dev/null"))
+     (c/su (meh (c/exec :service :postgresql :stop)))
+     (c/exec "update-rc.d" :postgresql :disable))))
+
+(defn- get-bin-dir
+  [version]
+  (str "/usr/lib/postgresql/" version "/bin"))
+
+(defn- get-main-dir
+  [version]
+  (str "/var/lib/postgresql/" version "/main"))
+
+(defn- configure!
+  [{:keys [version] :or {version DEFAULT_VERSION}}]
+  (c/sudo "postgres"
+          (c/exec (str (get-bin-dir version) "/initdb")
+                  :-D (get-main-dir version))))
+
+(defn- get-log-path
+  [{:keys [version] :or {version DEFAULT_VERSION}}]
+  (str "/var/log/postgresql/postgresql-" version "-main.log"))
+
+(defn- start!
+  []
+  (c/su (c/exec :service :postgresql :start)))
+
+(defn- stop!
+  []
+  (c/su (meh (c/exec :service :postgresql :stop))))
+
+(defn- wipe!
+  [{:keys [version] :or {version DEFAULT_VERSION}}]
+  (stop!)
+  (c/su (meh (c/exec :rm :-r (get-main-dir version))))
+  (c/su (meh (c/exec :rm (get-log-path version)))))
+
+(defn live-node?
+  [test]
+  (let [node (-> test :nodes first)]
+    (try
+      (c/on node (c/sudo "postgres" (c/exec :pg_isready)))
+      true
+      (catch Exception _
+        (info node "is down")
+        false))))
+
+(defn wait-for-recovery
+  "Wait for the node bootstrapping."
+  ([test]
+   (wait-for-recovery TIMEOUT_SEC INTERVAL_SEC test))
+  ([timeout-sec interval-sec test]
+   (when-not (live-node? test)
+     (Thread/sleep (* interval-sec 1000))
+     (if (>= timeout-sec interval-sec)
+       (wait-for-recovery (- timeout-sec interval-sec) interval-sec test)
+       (throw (ex-info "Timed out waiting for the postgres node"
+                       {:cause "The node couldn't start"}))))))
+
+(defn db
+  "Setup PostgreSQL."
+  []
+  (reify
+    db/DB
+    (setup! [_ test _]
+      (when-not (:leave-db-running? test)
+        (wipe! test))
+      (install! test)
+      (configure! test)
+      (start!))
+
+    (teardown! [_ test _]
+      (when-not (:leave-db-running? test)
+        (wipe! test)))
+
+    db/Primary
+    (primaries [_ test] (:nodes test))
+    (setup-primary! [_ _ _])
+
+    db/Pause
+    (pause! [_ _ _]
+      (c/su (c/exec :service :postgresql :stop)))
+    (resume! [_ _ _]
+      (c/su (c/exec :service :postgresql :start)))
+
+    db/Kill
+    (start! [_ _ _]
+      (c/su (c/exec :service :postgresql :restart)))
+    (kill! [_ _ _]
+      (doseq [pattern (shuffle
+                       ["postgres -D" ; Main process
+                        "main: checkpointer"
+                        "main: background writer"
+                        "main: walwriter"
+                        "main: autovacuum launcher"])]
+        (Thread/sleep (rand-int 100))
+        (info "Killing" pattern "-" (cu/grepkill! pattern))))
+
+    db/LogFiles
+    (log-files [_ test _] [(get-log-path test)])))

scalardb/src/scalardb/db_extend.clj

@@ -1,7 +1,8 @@
 (ns scalardb.db-extend
   (:require [cassandra.core :as cassandra]
             [clojure.string :as string]
-            [jepsen.db :as db])
+            [jepsen.db :as db]
+            [scalardb.db.postgres :as postgres])
   (:import (com.scalar.db.storage.cassandra CassandraAdmin
                                             CassandraAdmin$ReplicationStrategy
                                             CassandraAdmin$CompactionStrategy)
@@ -36,6 +37,7 @@
       (when (nil? nodes)
         (throw (ex-info "No living node" {:test test})))
       (doto (Properties.)
+        (.setProperty "scalar.db.storage" "cassandra")
         (.setProperty "scalar.db.contact_points" (string/join "," nodes))
         (.setProperty "scalar.db.username" "cassandra")
         (.setProperty "scalar.db.password" "cassandra")
@@ -44,8 +46,29 @@
         (.setProperty "scalar.db.consensus_commit.serializable_strategy"
                       ((:serializable-strategy test) SERIALIZABLE_STRATEGIES))))))
 
+(defrecord ExtPostgres []
+  DbExtension
+  (live-nodes [_ test] (postgres/live-node? test))
+  (wait-for-recovery [_ test] (postgres/wait-for-recovery test))
+  (create-table-opts [_ _] {})
+  (create-properties
+    [_ test]
+    (let [node (-> test :nodes first)]
+      ;; We have only one node in this test
+      (doto (Properties.)
+        (.setProperty "scalar.db.storage" "jdbc")
+        (.setProperty "scalar.db.contact_points"
+                      (str "jdbc:postgresql://" node ":5432/"))
+        (.setProperty "scalar.db.username" "postgres")
+        (.setProperty "scalar.db.password" "postgres")
+        (.setProperty "scalar.db.consensus_commit.isolation_level"
+                      ((:isolation-level test) ISOLATION_LEVELS))
+        (.setProperty "scalar.db.consensus_commit.serializable_strategy"
+                      ((:serializable-strategy test) SERIALIZABLE_STRATEGIES))))))
+
 (def ^:private ext-dbs
-  {:cassandra (->ExtCassandra)})
+  {:cassandra (->ExtCassandra)
+   :postgres (->ExtPostgres)})
 
 (defn extend-db
   [db db-type]
@@ -57,6 +80,12 @@
       db/Primary
       (primaries [_ test] (db/primaries db test))
       (setup-primary! [_ test node] (db/setup-primary! db test node))
+      db/Pause
+      (pause! [_ test node] (db/pause! db test node))
+      (resume! [_ test node] (db/resume! db test node))
+      db/Kill
+      (start! [_ test node] (db/start! db test node))
+      (kill! [_ test node] (db/kill! db test node))
       db/LogFiles
       (log-files [_ test node] (db/log-files db test node))
       DbExtension