Add primary node discovery

Early versions of tests always first node as a primary in a cluster.
With automated leader election leader can be on any node in clustera,
so discovery of primary instance has been added.

It consists of SQL function `_LEADER()` implemented in Lua,

```
unix/:/var/run/tarantool/jepsen.control> box.execute([[SELECT _LEADER()]])
---
- metadata:
  - name: COLUMN_1
    type: string
  rows:
  - ['89.208.84.154']
...
```

`primary` function that connects to desired node and gets IP address of primary
instance using `_LEADER()` function and `primaries` function that gather
information about primaries on every node in a cluster.

Closes #34
Closes #16 (kill)
Closes #17 (partition)
Closes #21 (pause)

Author: ligurio

resources/tarantool/jepsen.lua

@@ -2,7 +2,7 @@ single_mode = %TARANTOOL_SINGLE_MODE%
 
 if single_mode then
 box.cfg {
-    listen = 3301;
+    listen = '%TARANTOOL_IP_ADDRESS%:3301';
     log_level = 6;
     log_nonblock = false;
     too_long_threshold = 0.5;
@@ -12,7 +12,7 @@ box.cfg {
 }
 else
 box.cfg {
-    listen = 3301;
+    listen = '%TARANTOOL_IP_ADDRESS%:3301';
     replication = { %TARANTOOL_REPLICATION% };
     read_only = %TARANTOOL_IS_READ_ONLY%;
     replication_synchro_quorum = 2;
@@ -76,6 +76,31 @@ box.schema.func.create('_UPSERT',
     param_list = {'integer', 'integer', 'string'},
     exports = {'LUA', 'SQL'},
     is_deterministic = true})
+
+--[[ Function returns IP address of a node where current leader
+of synchronous cluster with enabled Raft consensus protocol is started.
+Returns nil when Raft is disabled. Example: SELECT _LEADER()
+]]
+box.schema.func.create('_LEADER',
+   {language = 'LUA',
+    returns = 'string',
+    body = [[function()
+             local leader_id = box.info.election.leader
+             if leader_id == 0 or leader_id == nil then
+               return nil
+             end
+             local leader_upstream = box.info.replication[leader_id].upstream
+             if leader_upstream == nil then
+               return string.match(box.info.listen, '(.+):[0-9]+')
+             end
+             local leader_ip_address = string.match(leader_upstream.peer, '[A-z]+@(.+):[0-9]+')
+
+             return leader_ip_address
+             end]],
+    is_sandboxed = false,
+    param_list = {},
+    exports = {'LUA', 'SQL'},
+    is_deterministic = true})
 end
 
 box.once('jepsen', bootstrap)

src/tarantool/client.clj

@@ -3,6 +3,7 @@
   (:require [clojure.string :as str]
             [clojure.tools.logging :refer [info warn]]
             [next.jdbc :as j]
+            [next.jdbc.sql :as sql]
             [dom-top.core :as dt]
             [next.jdbc.connection :as connection]))
 
@@ -56,6 +57,13 @@
                      (info :caught-cause    (.cause (:rollback (ex-data e#))))
                      (throw e#))))))
 
+(defn primary
+  [node]
+  (let [conn (open node test)
+        leader (:COLUMN_1 (first (sql/query conn ["SELECT _LEADER()"])))]
+    ;(assert leader)
+    leader))
+
 (defmacro with-txn-aborts
   "Aborts body on rollbacks."
   [op & body]

src/tarantool/db.clj

@@ -4,6 +4,7 @@
             [clojure.java.io :as io]
             [next.jdbc :as j]
             [next.jdbc.connection :as connection]
+            [tarantool.client :as cl]
             [jepsen.os.debian :as debian]
             [jepsen.control.util :as cu]
             [jepsen [core :as jepsen]
@@ -203,6 +204,16 @@
       (= n 1) true
       :else false)))
 
+(defn primaries
+  "Return a seq of primaries in a cluster."
+  [test]
+  (if (= 1 (count (:nodes test)))
+    (:nodes test)
+    (->> (pmap cl/primary (:nodes test))
+         (set)
+         (into [])
+         (remove nil?))))
+
 (defn configure!
   "Configure instance"
   [test node]
@@ -212,6 +223,7 @@
     (c/exec :mkdir :-p "/etc/tarantool/instances.enabled")
     (c/exec :usermod :-a :-G :tarantool :ubuntu)
     (c/exec :echo (-> "tarantool/jepsen.lua" io/resource slurp
+                      (str/replace #"%TARANTOOL_IP_ADDRESS%" node)
                       (str/replace #"%TARANTOOL_REPLICATION%" (replica-set test))
                       (str/replace #"%TARANTOOL_IS_READ_ONLY%" (boolean-to-str read-only))
                       (str/replace #"%TARANTOOL_MVCC%" (boolean-to-str (:mvcc test)))
@@ -247,6 +259,27 @@
       (stop! test node)
       (wipe! test node))
 
+    db/Primary
+    (setup-primary! [_ test node])
+
+    (primaries [_ test]
+      (primaries test))
+
+    db/Process
+    (start! [_ test node]
+      (c/su
+        (info node :starting :tarantool)
+        (start! test node)))
+
+    (kill! [_ test node]
+      (c/su
+        (info node :stopping :tarantool)
+        (stop! test node)))
+
+    db/Pause
+      (pause!  [_ test node] (c/su (cu/grepkill! :stop "tarantool")))
+      (resume! [_ test node] (c/su (cu/grepkill! :cont "tarantool")))
+
     db/LogFiles
     (log-files [_ test node]
       [logfile])))

src/tarantool/register.clj

@@ -37,6 +37,7 @@
   (setup! [this test node]
     (let [conn (cl/open node test)]
       (assert conn)
+      (info (cl/primary node))
       (if (= node (jepsen/primary test))
         (cl/with-conn-failure-retry conn
           (j/execute! conn [(str "CREATE TABLE IF NOT EXISTS " table-name