You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ga...@apache.org on 2018/12/10 11:03:03 UTC
[flink] 01/06: [FLINK-10986][tests] Enable fine-grained
configuration of required dbs
This is an automated email from the ASF dual-hosted git repository.
gary pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 14380511d0d14d2cf05df11f0819db3a232438d2
Author: Gary Yao <ga...@data-artisans.com>
AuthorDate: Mon Nov 26 11:43:15 2018 +0100
[FLINK-10986][tests] Enable fine-grained configuration of required dbs
This enables toggling the setup of test dependencies, such as Hadooop, Mesos,
and, ZooKeeper through the --test-spec edn file. The type of the Flink cluster
can also be specified via: :flink-yarn-job :flink-yarn-session
:flink-mesos-session, and :flink-standalone-session.
Retryable operations that exhausted all attempts, now propagate the exception by
default. This is to fail fast if during the db/setup!, an operation, such as
Flink job submission, fails to complete.
---
flink-jepsen/docker/run-tests.sh | 21 +-
.../mesos-session.edn} | 3 +-
.../standalone-session.edn} | 3 +-
.../yarn-job.edn} | 3 +-
.../yarn-session.edn} | 3 +-
flink-jepsen/src/jepsen/flink/checker.clj | 2 +-
flink-jepsen/src/jepsen/flink/client.clj | 14 +-
flink-jepsen/src/jepsen/flink/db.clj | 266 +++++++++++----------
flink-jepsen/src/jepsen/flink/flink.clj | 55 +++--
flink-jepsen/src/jepsen/flink/utils.clj | 2 +-
flink-jepsen/test/jepsen/flink/utils_test.clj | 2 +-
11 files changed, 194 insertions(+), 180 deletions(-)
diff --git a/flink-jepsen/docker/run-tests.sh b/flink-jepsen/docker/run-tests.sh
index a81b873..d357201 100755
--- a/flink-jepsen/docker/run-tests.sh
+++ b/flink-jepsen/docker/run-tests.sh
@@ -29,7 +29,6 @@ n3
EOF
common_jepsen_args+=(--ha-storage-dir hdfs:///flink
---test-spec "${dockerdir}/data-stream-test-program-parallelism-1.edn"
--tarball ${2}
--ssh-private-key ~/.ssh/id_rsa
--nodes-file ${dockerdir}/nodes)
@@ -37,18 +36,18 @@ common_jepsen_args+=(--ha-storage-dir hdfs:///flink
for i in $(seq 1 ${1})
do
echo "Executing run #${i} of ${1}"
- lein run test "${common_jepsen_args[@]}" --nemesis-gen kill-task-managers --deployment-mode yarn-session
- lein run test "${common_jepsen_args[@]}" --nemesis-gen kill-job-managers --deployment-mode yarn-session
- lein run test "${common_jepsen_args[@]}" --nemesis-gen fail-name-node-during-recovery --deployment-mode yarn-session
+ lein run test "${common_jepsen_args[@]}" --nemesis-gen kill-task-managers --test-spec "${dockerdir}/test-specs/yarn-session.edn"
+ lein run test "${common_jepsen_args[@]}" --nemesis-gen kill-job-managers --test-spec "${dockerdir}/test-specs/yarn-session.edn"
+ lein run test "${common_jepsen_args[@]}" --nemesis-gen fail-name-node-during-recovery --test-spec "${dockerdir}/test-specs/yarn-session.edn"
- lein run test "${common_jepsen_args[@]}" --nemesis-gen kill-task-managers --deployment-mode yarn-job
- lein run test "${common_jepsen_args[@]}" --nemesis-gen kill-job-managers --deployment-mode yarn-job
- lein run test "${common_jepsen_args[@]}" --nemesis-gen fail-name-node-during-recovery --deployment-mode yarn-job
+ lein run test "${common_jepsen_args[@]}" --nemesis-gen kill-task-managers --test-spec "${dockerdir}/test-specs/yarn-job.edn"
+ lein run test "${common_jepsen_args[@]}" --nemesis-gen kill-job-managers --test-spec "${dockerdir}/test-specs/yarn-job.edn"
+ lein run test "${common_jepsen_args[@]}" --nemesis-gen fail-name-node-during-recovery --test-spec "${dockerdir}/test-specs/yarn-job.edn"
- lein run test "${common_jepsen_args[@]}" --nemesis-gen kill-task-managers --deployment-mode mesos-session
- lein run test "${common_jepsen_args[@]}" --nemesis-gen kill-job-managers --deployment-mode mesos-session
+ lein run test "${common_jepsen_args[@]}" --nemesis-gen kill-task-managers --test-spec "${dockerdir}/test-specs/mesos-session.edn"
+ lein run test "${common_jepsen_args[@]}" --nemesis-gen kill-job-managers --test-spec "${dockerdir}/test-specs/mesos-session.edn"
- lein run test "${common_jepsen_args[@]}" --nemesis-gen kill-job-managers --deployment-mode standalone-session
- lein run test "${common_jepsen_args[@]}" --nemesis-gen kill-job-managers --client-gen cancel-jobs --deployment-mode standalone-session
+ lein run test "${common_jepsen_args[@]}" --nemesis-gen kill-job-managers --test-spec "${dockerdir}/standalone-session.edn"
+ lein run test "${common_jepsen_args[@]}" --nemesis-gen kill-job-managers --client-gen cancel-jobs --test-spec "${dockerdir}/standalone-session.edn"
echo
done
diff --git a/flink-jepsen/docker/data-stream-test-program-parallelism-1.edn b/flink-jepsen/docker/test-specs/mesos-session.edn
similarity index 89%
copy from flink-jepsen/docker/data-stream-test-program-parallelism-1.edn
copy to flink-jepsen/docker/test-specs/mesos-session.edn
index a20903e..e4c01bf 100644
--- a/flink-jepsen/docker/data-stream-test-program-parallelism-1.edn
+++ b/flink-jepsen/docker/test-specs/mesos-session.edn
@@ -14,5 +14,6 @@
;; See the License for the specific language governing permissions and
;; limitations under the License.
-{:jobs [{:job-jar "/jepsen/bin/DataStreamAllroundTestProgram.jar"
+{:dbs [:hadoop :zookeeper :mesos :flink-mesos-session]
+ :jobs [{:job-jar "/jepsen/bin/DataStreamAllroundTestProgram.jar"
:job-args "--environment.parallelism 1 --state_backend.checkpoint_directory hdfs:///checkpoints --state_backend rocks --state_backend.rocks.incremental true"}]}
diff --git a/flink-jepsen/docker/data-stream-test-program-parallelism-1.edn b/flink-jepsen/docker/test-specs/standalone-session.edn
similarity index 89%
copy from flink-jepsen/docker/data-stream-test-program-parallelism-1.edn
copy to flink-jepsen/docker/test-specs/standalone-session.edn
index a20903e..763d592 100644
--- a/flink-jepsen/docker/data-stream-test-program-parallelism-1.edn
+++ b/flink-jepsen/docker/test-specs/standalone-session.edn
@@ -14,5 +14,6 @@
;; See the License for the specific language governing permissions and
;; limitations under the License.
-{:jobs [{:job-jar "/jepsen/bin/DataStreamAllroundTestProgram.jar"
+{:dbs [:hadoop :zookeeper :flink-standalone-session]
+ :jobs [{:job-jar "/jepsen/bin/DataStreamAllroundTestProgram.jar"
:job-args "--environment.parallelism 1 --state_backend.checkpoint_directory hdfs:///checkpoints --state_backend rocks --state_backend.rocks.incremental true"}]}
diff --git a/flink-jepsen/docker/data-stream-test-program-parallelism-1.edn b/flink-jepsen/docker/test-specs/yarn-job.edn
similarity index 90%
copy from flink-jepsen/docker/data-stream-test-program-parallelism-1.edn
copy to flink-jepsen/docker/test-specs/yarn-job.edn
index a20903e..390f21d 100644
--- a/flink-jepsen/docker/data-stream-test-program-parallelism-1.edn
+++ b/flink-jepsen/docker/test-specs/yarn-job.edn
@@ -14,5 +14,6 @@
;; See the License for the specific language governing permissions and
;; limitations under the License.
-{:jobs [{:job-jar "/jepsen/bin/DataStreamAllroundTestProgram.jar"
+{:dbs [:hadoop :zookeeper :flink-yarn-job]
+ :jobs [{:job-jar "/jepsen/bin/DataStreamAllroundTestProgram.jar"
:job-args "--environment.parallelism 1 --state_backend.checkpoint_directory hdfs:///checkpoints --state_backend rocks --state_backend.rocks.incremental true"}]}
diff --git a/flink-jepsen/docker/data-stream-test-program-parallelism-1.edn b/flink-jepsen/docker/test-specs/yarn-session.edn
similarity index 89%
rename from flink-jepsen/docker/data-stream-test-program-parallelism-1.edn
rename to flink-jepsen/docker/test-specs/yarn-session.edn
index a20903e..aa95c15 100644
--- a/flink-jepsen/docker/data-stream-test-program-parallelism-1.edn
+++ b/flink-jepsen/docker/test-specs/yarn-session.edn
@@ -14,5 +14,6 @@
;; See the License for the specific language governing permissions and
;; limitations under the License.
-{:jobs [{:job-jar "/jepsen/bin/DataStreamAllroundTestProgram.jar"
+{:dbs [:hadoop :zookeeper :flink-yarn-session]
+ :jobs [{:job-jar "/jepsen/bin/DataStreamAllroundTestProgram.jar"
:job-args "--environment.parallelism 1 --state_backend.checkpoint_directory hdfs:///checkpoints --state_backend rocks --state_backend.rocks.incremental true"}]}
diff --git a/flink-jepsen/src/jepsen/flink/checker.clj b/flink-jepsen/src/jepsen/flink/checker.clj
index b036b0a..d5ff6e0 100644
--- a/flink-jepsen/src/jepsen/flink/checker.clj
+++ b/flink-jepsen/src/jepsen/flink/checker.clj
@@ -211,7 +211,7 @@
(defn- compute-final-model
[model history]
(let [start-time (-> history first :time)]
- (reduce model/step
+ (reduce knossos.model/step
(assoc model :last-failure start-time)
history)))
diff --git a/flink-jepsen/src/jepsen/flink/client.clj b/flink-jepsen/src/jepsen/flink/client.clj
index afbfe56..17746ec 100644
--- a/flink-jepsen/src/jepsen/flink/client.clj
+++ b/flink-jepsen/src/jepsen/flink/client.clj
@@ -172,12 +172,11 @@
:cancel-jobs (dispatch-operation-or-fatal op (cancel-jobs! rest-url job-ids)))))
(defrecord Client
- [deploy-cluster! ; function that starts a non-standalone cluster and submits the jobs
- closer ; function that closes the ZK client
+ [closer ; function that closes the ZK client
rest-url ; atom storing the current rest-url
init-future ; future that completes if rest-url is set to an initial value
job-ids ; atom storing the job-ids
- job-submitted?] ; Have the jobs already been submitted? Used to avoid re-submission if the client is re-opened.
+ client-setup?] ; Has the client already been setup? Used to avoid running setup! again if the client is re-opened.
client/Client
(open! [this test _]
(info "Open client.")
@@ -186,10 +185,9 @@
:rest-url rest-url-atom
:init-future init-future)))
- (setup! [_ test]
+ (setup! [_ _]
(info "Setup client.")
- (when (compare-and-set! job-submitted? false true)
- (deploy-cluster! test)
+ (when (compare-and-set! client-setup? false true)
(deref init-future)
(let [jobs (fu/retry (fn [] (list-jobs! @rest-url))
:fallback (fn [e]
@@ -209,5 +207,5 @@
(closer)))
(defn create-client
- [deploy-cluster!]
- (Client. deploy-cluster! nil nil nil (atom nil) (atom false)))
+ []
+ (Client. nil nil nil (atom nil) (atom false)))
diff --git a/flink-jepsen/src/jepsen/flink/db.clj b/flink-jepsen/src/jepsen/flink/db.clj
index 3046776..6bac982 100644
--- a/flink-jepsen/src/jepsen/flink/db.clj
+++ b/flink-jepsen/src/jepsen/flink/db.clj
@@ -17,13 +17,11 @@
(ns jepsen.flink.db
(:require [clj-http.client :as http]
[clojure.java.io]
- [clojure.string :as str]
[clojure.tools.logging :refer :all]
[jepsen
[control :as c]
[db :as db]
- [util :refer [meh]]
- [zookeeper :as zk]]
+ [util :refer [meh]]]
[jepsen.control.util :as cu]
[jepsen.flink.hadoop :as hadoop]
[jepsen.flink.mesos :as mesos]
@@ -36,12 +34,6 @@
(def conf-file (str install-dir "/conf/flink-conf.yaml"))
(def masters-file (str install-dir "/conf/masters"))
-(def default-flink-dist-url "https://archive.apache.org/dist/flink/flink-1.6.0/flink-1.6.0-bin-hadoop28-scala_2.11.tgz")
-(def hadoop-dist-url "https://archive.apache.org/dist/hadoop/common/hadoop-2.8.3/hadoop-2.8.3.tar.gz")
-(def deb-zookeeper-package "3.4.9-3+deb8u1")
-(def deb-mesos-package "1.5.0-2.0.2")
-(def deb-marathon-package "1.6.322")
-
(def taskmanager-slots 3)
(defn flink-configuration
@@ -100,36 +92,16 @@
(meh (c/exec :rm :-rf install-dir))
(meh (c/exec :rm :-rf (c/lit "/tmp/.yarn-properties*"))))
-(defn get-log-files!
- []
- (if (cu/exists? log-dir) (cu/ls-full log-dir) []))
-
-(defn flink-db
- []
- (reify db/DB
- (setup! [_ test node]
- (c/su
- (install-flink! test node)))
-
- (teardown! [_ test node]
- (c/su
- (teardown-flink!)))
-
- db/LogFiles
- (log-files [_ test node]
- (concat
- (get-log-files!)))))
-
(defn combined-db
[dbs]
(reify db/DB
(setup! [_ test node]
(c/su
- (doall (map #(db/setup! % test node) dbs))))
+ (doseq [db dbs] (db/setup! db test node))))
(teardown! [_ test node]
(c/su
(try
- (doall (map #(db/teardown! % test node) dbs))
+ (doseq [db dbs] (db/teardown! db test node))
(finally (fu/stop-all-supervised-services!)))))
db/LogFiles
(log-files [_ test node]
@@ -138,6 +110,22 @@
(map #(db/log-files % test node))
(flatten)))))
+(defn flink-db
+ [db]
+ (let [flink-base-db (reify db/DB
+ (setup! [_ test node]
+ (c/su
+ (install-flink! test node)))
+
+ (teardown! [_ _ _]
+ (c/su
+ (teardown-flink!)))
+
+ db/LogFiles
+ (log-files [_ _ _]
+ (fu/find-files! log-dir)))]
+ (combined-db [flink-base-db db])))
+
(defn- sorted-nodes
[test]
(-> test :nodes sort))
@@ -169,9 +157,11 @@
(defn exec-flink!
[cmd args]
(c/su
- (c/exec (c/lit (str
+ (c/exec (c/lit (fu/join-space
(hadoop-env-vars)
- install-dir "/bin/flink " cmd " " (apply fu/join-space args))))))
+ (str install-dir "/bin/flink")
+ cmd
+ (apply fu/join-space args))))))
(defn flink-run-cli-args
"Returns the CLI args that should be passed to 'flink run'"
@@ -191,6 +181,14 @@
[(str install-dir "/" (file-name job-jar))
job-args])))))
+(defn- submit-job-with-retry!
+ ([test] (submit-job-with-retry! test []))
+ ([test cli-args] (fu/retry
+ (partial submit-job! test cli-args)
+ :fallback (fn [e] (do
+ (fatal e "Could not submit job.")
+ (System/exit 1))))))
+
;;; Standalone
(def standalone-master-count 2)
@@ -204,122 +202,130 @@
(select-nodes test (partial drop standalone-master-count)))
(defn- start-standalone-masters!
- [test node]
- (when (some #{node} (standalone-master-nodes test))
+ []
+ (let [jobmanager-script (str install-dir "/bin/jobmanager.sh")
+ jobmanager-log (str log-dir "/jobmanager.log")]
(fu/create-supervised-service!
"flink-master"
- (str "env " (hadoop-env-vars)
- install-dir "/bin/jobmanager.sh start-foreground "
- ">> " log-dir "/jobmanager.log"))))
+ (fu/join-space "env" (hadoop-env-vars) jobmanager-script "start-foreground" ">>" jobmanager-log))))
(defn- start-standalone-taskmanagers!
- [test node]
- (when (some #{node} (standalone-taskmanager-nodes test))
+ []
+ (let [taskmanager-script (str install-dir "/bin/taskmanager.sh")
+ taskmanager-log (str log-dir "/taskmanager.log")]
(fu/create-supervised-service!
"flink-taskmanager"
- (str "env " (hadoop-env-vars)
- install-dir "/bin/taskmanager.sh start-foreground "
- ">> " log-dir "/taskmanager.log"))))
-
-(defn- start-flink-db
- []
- (reify db/DB
- (setup! [_ test node]
- (c/su
- (start-standalone-masters! test node)
- (start-standalone-taskmanagers! test node)))
+ (fu/join-space "env" (hadoop-env-vars) taskmanager-script "start-foreground" ">>" taskmanager-log))))
- (teardown! [_ test node]
- (c/su
- (when (some #{node} (standalone-master-nodes test))
- (fu/stop-supervised-service! "flink-master"))
- (when (some #{node} (standalone-taskmanager-nodes test))
- (fu/stop-supervised-service! "flink-taskmanager"))))))
-
-(defn flink-standalone-db
+(defn start-flink-db
[]
- (let [zk (zk/db deb-zookeeper-package)
- hadoop (hadoop/db hadoop-dist-url)
- flink (flink-db)
- start-flink (start-flink-db)]
- (combined-db [hadoop zk flink start-flink])))
-
-(defn submit-job-from-first-node!
- [test]
- (c/on (first-node test)
- (submit-job! test)))
+ (flink-db
+ (reify db/DB
+ (setup! [_ test node]
+ (c/su
+ (when (some #{node} (standalone-master-nodes test))
+ (start-standalone-masters!))
+ (when (some #{node} (standalone-taskmanager-nodes test))
+ (start-standalone-taskmanagers!))
+ (when (= (first-node test) node)
+ (submit-job-with-retry! test))))
+
+ (teardown! [_ test node]
+ (c/su
+ (when (some #{node} (standalone-master-nodes test))
+ (fu/stop-supervised-service! "flink-master"))
+ (when (some #{node} (standalone-taskmanager-nodes test))
+ (fu/stop-supervised-service! "flink-taskmanager")))))))
;;; YARN
-(defn flink-yarn-db
+(defn- start-yarn-session-cmd
[]
- (let [zk (zk/db deb-zookeeper-package)
- hadoop (hadoop/db hadoop-dist-url)
- flink (flink-db)]
- (combined-db [hadoop zk flink])))
+ (fu/join-space (hadoop-env-vars)
+ (str install-dir "/bin/yarn-session.sh")
+ "-d"
+ "-jm 2048m"
+ "-tm 2048m"))
-(defn start-yarn-session!
- [test]
- (let [node (first-node test)]
- (c/on node
- (info "Starting YARN session from" node)
- (c/su
- (c/exec (c/lit (str (hadoop-env-vars)
- " " install-dir "/bin/yarn-session.sh -d -jm 2048m -tm 2048m")))
- (submit-job! test)))))
-
-(defn start-yarn-job!
+(defn- start-yarn-session!
+ []
+ (info "Starting YARN session")
+ (let [exec-start-yarn-session! #(c/su (c/exec (c/lit (start-yarn-session-cmd))))
+ log-failure! (fn [exception _] (info "Starting YARN session failed due to"
+ (.getMessage exception)
+ "Retrying..."))]
+ (fu/retry exec-start-yarn-session!
+ :delay 4000
+ :on-error log-failure!)))
+
+(defn yarn-session-db
+ []
+ (flink-db (reify db/DB
+ (setup! [_ test node]
+ (when (= node (first-node test))
+ (start-yarn-session!)
+ (submit-job! test)))
+ (teardown! [_ _ _]))))
+
+(defn- start-yarn-job!
[test]
- (c/on (first-node test)
- (c/su
- (submit-job! test ["-m yarn-cluster" "-yjm 2048m" "-ytm 2048m"]))))
-
-;;; Mesos
+ (c/su
+ (submit-job-with-retry! test ["-m yarn-cluster" "-yjm 2048m" "-ytm 2048m"])))
-(defn flink-mesos-db
+(defn yarn-job-db
[]
- (let [zk (zk/db deb-zookeeper-package)
- hadoop (hadoop/db hadoop-dist-url)
- mesos (mesos/db deb-mesos-package deb-marathon-package)
- flink (flink-db)]
- (combined-db [hadoop zk mesos flink])))
+ (flink-db (reify db/DB
+ (setup! [_ test node]
+ (when (= node (first-node test))
+ (start-yarn-job! test)))
+ (teardown! [_ _ _]))))
-(defn submit-job-with-retry!
- [test]
- (fu/retry
- (partial submit-job! test)
- :fallback (fn [e] (do
- (fatal e "Could not submit job.")
- (System/exit 1)))))
+;;; Mesos
-(defn mesos-appmaster-cmd
+(defn- mesos-appmaster-cmd
"Returns the command used by Marathon to start Flink's Mesos application master."
[test]
- (str (hadoop-env-vars)
- install-dir "/bin/mesos-appmaster.sh "
- "-Dmesos.master=" (zookeeper-uri
- test
- mesos/zk-namespace) " "
- "-Djobmanager.rpc.address=$(hostname -f) "
- "-Djobmanager.heap.mb=2048 "
- "-Djobmanager.rpc.port=6123 "
- "-Dmesos.resourcemanager.tasks.mem=2048 "
- "-Dtaskmanager.heap.mb=2048 "
- "-Dmesos.resourcemanager.tasks.cpus=1 "
- "-Drest.bind-address=$(hostname -f) "))
-
-(defn start-mesos-session!
+ (fu/join-space
+ (hadoop-env-vars)
+ (str install-dir "/bin/mesos-appmaster.sh")
+ (str "-Dmesos.master=" (zookeeper-uri test mesos/zk-namespace))
+ "-Djobmanager.rpc.address=$(hostname -f)"
+ "-Djobmanager.heap.mb=2048"
+ "-Djobmanager.rpc.port=6123"
+ "-Dmesos.resourcemanager.tasks.mem=2048"
+ "-Dtaskmanager.heap.mb=2048"
+ "-Dmesos.resourcemanager.tasks.cpus=1"
+ "-Drest.bind-address=$(hostname -f)"))
+
+(defn- start-mesos-session!
[test]
(c/su
- (let [r (fu/retry (fn []
- (http/post
- (str (mesos/marathon-base-url test) "/v2/apps")
- {:form-params {:id "flink"
- :cmd (mesos-appmaster-cmd test)
- :cpus 1.0
- :mem 2048
- :maxLaunchDelaySeconds 3}
- :content-type :json})))]
- (info "Submitted Flink Application via Marathon" r)
- (c/on (-> test :nodes sort first)
- (submit-job-with-retry! test)))))
+ (let [log-submission-failure! (fn [exception _]
+ (info "Submitting Flink Application via Marathon failed due to"
+ (.getMessage exception)
+ "Retrying..."))
+ submit-flink! (fn []
+ (http/post
+ (str (mesos/marathon-base-url test) "/v2/apps")
+ {:form-params {:id "flink"
+ :cmd (mesos-appmaster-cmd test)
+ :cpus 1.0
+ :mem 2048
+ :maxLaunchDelaySeconds 3}
+ :content-type :json}))
+ marathon-response (fu/retry submit-flink!
+ :on-retry log-submission-failure!
+ :delay 4000)]
+ (info "Submitted Flink Application via Marathon" marathon-response))))
+
+(defn flink-mesos-app-master
+ []
+ (flink-db
+ (reify
+ db/DB
+ (setup! [_ test node]
+ (when (= (first-node test) node)
+ (start-mesos-session! test)
+ (submit-job-with-retry! test)))
+
+ (teardown! [_ _ _]))))
diff --git a/flink-jepsen/src/jepsen/flink/flink.clj b/flink-jepsen/src/jepsen/flink/flink.clj
index b2b7644..f706c75 100644
--- a/flink-jepsen/src/jepsen/flink/flink.clj
+++ b/flink-jepsen/src/jepsen/flink/flink.clj
@@ -20,22 +20,31 @@
[jepsen
[cli :as cli]
[generator :as gen]
- [tests :as tests]]
+ [tests :as tests]
+ [zookeeper :as zk]]
[jepsen.os.debian :as debian]
- [jepsen.flink.client :refer :all]
- [jepsen.flink.checker :as flink-checker]
- [jepsen.flink.db :as fdb]
- [jepsen.flink.nemesis :as fn]))
+ [jepsen.flink
+ [client :refer :all]
+ [checker :as flink-checker]
+ [db :as fdb]
+ [hadoop :as hadoop]
+ [mesos :as mesos]
+ [nemesis :as fn]]))
-(def flink-test-config
- {:yarn-session {:db (fdb/flink-yarn-db)
- :deployment-strategy fdb/start-yarn-session!}
- :yarn-job {:db (fdb/flink-yarn-db)
- :deployment-strategy fdb/start-yarn-job!}
- :mesos-session {:db (fdb/flink-mesos-db)
- :deployment-strategy fdb/start-mesos-session!}
- :standalone-session {:db (fdb/flink-standalone-db)
- :deployment-strategy fdb/submit-job-from-first-node!}})
+(def default-flink-dist-url "https://archive.apache.org/dist/flink/flink-1.6.0/flink-1.6.0-bin-hadoop28-scala_2.11.tgz")
+(def hadoop-dist-url "https://archive.apache.org/dist/hadoop/common/hadoop-2.8.3/hadoop-2.8.3.tar.gz")
+(def deb-zookeeper-package "3.4.9-3+deb8u1")
+(def deb-mesos-package "1.5.0-2.0.2")
+(def deb-marathon-package "1.6.322")
+
+(def dbs
+ {:flink-yarn-job (fdb/yarn-job-db)
+ :flink-yarn-session (fdb/yarn-session-db)
+ :flink-standalone-session (fdb/start-flink-db)
+ :flink-mesos-session (fdb/flink-mesos-app-master)
+ :hadoop (hadoop/db hadoop-dist-url)
+ :mesos (mesos/db deb-mesos-package deb-marathon-package)
+ :zookeeper (zk/db deb-zookeeper-package)})
(def poll-jobs-running {:type :invoke, :f :jobs-running?, :value nil})
(def cancel-jobs {:type :invoke, :f :cancel-jobs, :value nil})
@@ -64,12 +73,12 @@
(defn flink-test
[opts]
(merge tests/noop-test
- (let [{:keys [db deployment-strategy]} (-> opts :deployment-mode flink-test-config)
+ (let [dbs (->> opts :test-spec :dbs (map dbs))
{:keys [job-running-healthy-threshold job-recovery-grace-period]} opts
client-gen ((:client-gen opts) client-gens)]
{:name "Apache Flink"
:os debian/os
- :db db
+ :db (fdb/combined-db dbs)
:nemesis (fn/nemesis)
:model (flink-checker/job-running-within-grace-period
job-running-healthy-threshold
@@ -81,7 +90,7 @@
((fn/nemesis-generator-factories (:nemesis-gen opts)) opts)
job-running-healthy-threshold
job-recovery-grace-period))))
- :client (create-client deployment-strategy)
+ :client (create-client)
:checker (flink-checker/job-running-checker)})
(assoc opts :concurrency 1)))
@@ -104,8 +113,11 @@
(merge
(cli/single-test-cmd
{:test-fn flink-test
- :tarball fdb/default-flink-dist-url
- :opt-spec [[nil "--test-spec FILE" "" :parse-fn read-test-spec]
+ :tarball default-flink-dist-url
+ :opt-spec [[nil "--test-spec FILE" "Path to a test specification (.edn)"
+ :parse-fn read-test-spec
+ :validate [#(->> % :dbs (map dbs) (every? (complement nil?)))
+ (str "Invalid :dbs specification. " (keys-as-allowed-values-help-text dbs))]]
[nil "--ha-storage-dir DIR" "high-availability.storageDir"]
[nil "--nemesis-gen GEN" (str "Which nemesis should be used?"
(keys-as-allowed-values-help-text fn/nemesis-generator-factories))
@@ -119,11 +131,6 @@
:default :poll-job-running
:validate [#(client-gens (keyword %))
(keys-as-allowed-values-help-text client-gens)]]
- [nil "--deployment-mode MODE" (keys-as-allowed-values-help-text flink-test-config)
- :parse-fn keyword
- :default :yarn-session
- :validate [#(flink-test-config (keyword %))
- (keys-as-allowed-values-help-text flink-test-config)]]
[nil "--job-running-healthy-threshold TIMES" "Number of consecutive times the job must be running to be considered healthy."
:default 5
:parse-fn #(Long/parseLong %)
diff --git a/flink-jepsen/src/jepsen/flink/utils.clj b/flink-jepsen/src/jepsen/flink/utils.clj
index 6634a7e..7ccae12 100644
--- a/flink-jepsen/src/jepsen/flink/utils.clj
+++ b/flink-jepsen/src/jepsen/flink/utils.clj
@@ -35,7 +35,7 @@
:or {on-retry (fn [exception attempt] (warn "Retryable operation failed:"
(.getMessage exception)))
success identity
- fallback :default
+ fallback #(throw %)
retries 10
delay 2000}
:as keys}]
diff --git a/flink-jepsen/test/jepsen/flink/utils_test.clj b/flink-jepsen/test/jepsen/flink/utils_test.clj
index 607f90d..17d5e5f 100644
--- a/flink-jepsen/test/jepsen/flink/utils_test.clj
+++ b/flink-jepsen/test/jepsen/flink/utils_test.clj
@@ -29,7 +29,7 @@
(testing "Exhaust all attempts."
(let [failing-always (fn [] (throw (Exception. "Expected")))]
- (is (nil? (retry failing-always :retries 1 :delay 0)))))
+ (is (nil? (retry failing-always :retries 1 :delay 0 :fallback :nil)))))
(testing "Propagate exception."
(let [failing-always (fn [] (throw (Exception. "Expected")))]