Nimbus HA

conf/defaults.yaml

@@ -30,6 +30,9 @@ nimbus.task.launch.secs: 120
 nimbus.reassign: true
 nimbus.file.copy.expiration.secs: 600
 nimbus.topology.validator: "backtype.storm.nimbus.DefaultTopologyValidator"
+nimbus.elections.zk.path: "/storm/nimbus-leader"
+nimbus.elections.zk.session.timeout: 2000
+nimbus.elections.zk.connection.timeout: 15000
 
 ### ui.* configs are for the master
 ui.port: 8080

project.clj

@@ -3,7 +3,7 @@
   (do (println (str "ERROR: requires Leiningen 1.x but you are using " lein-version))
     (System/exit 1)))
 
-(defproject storm "0.8.2-wip15"
+(defproject storm "0.8.2"
   :source-path "src/clj"
   :test-path "test/clj"
   :java-source-path "src/jvm"
@@ -18,6 +18,7 @@
                  [clj-time "0.4.1"]
                  [log4j/log4j "1.2.16"]
                  [com.netflix.curator/curator-framework "1.0.1"]
+                 [com.netflix.curator/curator-recipes "1.0.1"]
                  [backtype/jzmq "2.1.0"]
                  [com.googlecode.json-simple/json-simple "1.1"]
                  [compojure "1.1.3"]

src/clj/backtype/storm/bootstrap.clj

@@ -8,7 +8,7 @@
      (import (quote [backtype.storm.utils Utils LocalState Time TimeCacheMap
                      TimeCacheMap$ExpiredCallback
                      RotatingMap RotatingMap$ExpiredCallback
-                     BufferFileInputStream
+                     BufferInputStream BufferFileInputStream
                      RegisteredGlobalState ThriftTopologyUtils DisruptorQueue
                      MutableObject MutableLong]))
      (import (quote [backtype.storm.serialization KryoTupleSerializer KryoTupleDeserializer]))
@@ -44,5 +44,5 @@
      (import (quote [backtype.storm.grouping CustomStreamGrouping]))
      (import (quote [java.io File FileOutputStream FileInputStream]))
      (import (quote [java.util Collection List Random Map HashMap Collections ArrayList LinkedList]))
-     (import (quote [org.apache.commons.io FileUtils]))
+     (import (quote [org.apache.commons.io FileUtils IOUtils]))
      ))

src/clj/backtype/storm/config.clj

@@ -74,12 +74,26 @@
     ret
     ))
 
+(defn nimbus-storage-local-dir [conf]
+  (let [storm-local-dir (conf STORM-LOCAL-DIR)
+        nimbus-local-dir (get conf "nimbus.local.dir")
+        local-dir (if-not (clojure.string/blank? nimbus-local-dir) nimbus-local-dir storm-local-dir)
+        ret (str local-dir "/nimbus")]
+    (FileUtils/forceMkdir (File. ret))
+    ret))
+
 (defn master-stormdist-root
   ([conf]
      (str (master-local-dir conf) "/stormdist"))
   ([conf storm-id]
      (str (master-stormdist-root conf) "/" storm-id)))
 
+(defn nimbus-storage-stormdist-root
+  ([]
+    "/stormdist")
+  ([storm-id]
+    (str (nimbus-storage-stormdist-root) "/" storm-id)))
+
 (defn master-stormjar-path [stormroot]
   (str stormroot "/stormjar.jar"))
 

src/clj/backtype/storm/daemon/nimbus.clj

@@ -5,11 +5,13 @@
   (:import [org.apache.thrift7.transport TNonblockingServerTransport TNonblockingServerSocket])
   (:import [java.nio ByteBuffer])
   (:import [java.nio.channels Channels WritableByteChannel])
+  (:import [java.io InputStream])
   (:use [backtype.storm.scheduler.DefaultScheduler])
   (:import [backtype.storm.scheduler INimbus SupervisorDetails WorkerSlot TopologyDetails
             Cluster Topologies SchedulerAssignment SchedulerAssignmentImpl DefaultScheduler ExecutorDetails])
   (:use [backtype.storm bootstrap util])
   (:use [backtype.storm.daemon common])
+  (:use [backtype.storm.nimbus elections storage])
   (:gen-class
     :methods [^{:static true} [launch [backtype.storm.scheduler.INimbus] void]]))
 
@@ -54,6 +56,7 @@
      :uploaders (file-cache-map conf)
      :uptime (uptime-computer)
      :validator (new-instance (conf NIMBUS-TOPOLOGY-VALIDATOR))
+     :storage (create-nimbus-storage conf)
      :timer (mk-timer :kill-fn (fn [t]
                                  (log-error t "Error when processing event")
                                  (halt-process! 20 "Error when processing an event")
@@ -64,13 +67,9 @@
 (defn inbox [nimbus]
   (master-inbox (:conf nimbus)))
 
-(defn- read-storm-conf [conf storm-id]
-  (let [stormroot (master-stormdist-root conf storm-id)]
-    (merge conf
-           (Utils/deserialize
-            (FileUtils/readFileToByteArray
-             (File. (master-stormconf-path stormroot))
-             )))))
+(defn- read-storm-conf [storage conf storm-id]
+  (let [stormroot (nimbus-storage-stormdist-root storm-id)]
+    (merge conf (deserialize-from-storage storage (master-stormconf-path stormroot)))))
 
 (defn set-topology-status! [nimbus storm-id status]
   (let [storm-cluster-state (:storm-cluster-state nimbus)]
@@ -87,7 +86,7 @@
   (fn [kill-time]
     (let [delay (if kill-time
                   kill-time
-                  (get (read-storm-conf (:conf nimbus) storm-id)
+                  (get (read-storm-conf (:storage nimbus) (:conf nimbus) storm-id)
                        TOPOLOGY-MESSAGE-TIMEOUT-SECS))]
       (delay-event nimbus
                    storm-id
@@ -101,7 +100,7 @@
   (fn [time num-workers executor-overrides]
     (let [delay (if time
                   time
-                  (get (read-storm-conf (:conf nimbus) storm-id)
+                  (get (read-storm-conf (:storage nimbus) (:conf nimbus) storm-id)
                        TOPOLOGY-MESSAGE-TIMEOUT-SECS))]
       (delay-event nimbus
                    storm-id
@@ -289,29 +288,25 @@
   ;; need to somehow maintain stream/component ids inside tuples
   topology)
 
-(defn- setup-storm-code [conf storm-id tmp-jar-location storm-conf topology]
-  (let [stormroot (master-stormdist-root conf storm-id)]
-   (FileUtils/forceMkdir (File. stormroot))
-   (FileUtils/cleanDirectory (File. stormroot))
-   (setup-jar conf tmp-jar-location stormroot)
-   (FileUtils/writeByteArrayToFile (File. (master-stormcode-path stormroot)) (Utils/serialize topology))
-   (FileUtils/writeByteArrayToFile (File. (master-stormconf-path stormroot)) (Utils/serialize storm-conf))
-   ))
+(defn- setup-storm-code [conf storage storm-id tmp-jar-location storm-conf topology]
+  (let [stormroot (nimbus-storage-stormdist-root storm-id)]
+   (ensure-clean-dir-in-storage storage stormroot)
+   (setup-jar conf storage tmp-jar-location stormroot)
+   (serialize-to-storage topology storage (master-stormcode-path stormroot))
+   (serialize-to-storage storm-conf storage (master-stormconf-path stormroot))))
 
-(defn- read-storm-topology [conf storm-id]
-  (let [stormroot (master-stormdist-root conf storm-id)]
-    (Utils/deserialize
-      (FileUtils/readFileToByteArray
-        (File. (master-stormcode-path stormroot))
-        ))))
+(defn- read-storm-topology [storage storm-id]
+  (let [stormroot (nimbus-storage-stormdist-root storm-id)]
+    (deserialize-from-storage storage (master-stormcode-path stormroot))))
 
 (declare compute-executor->component)
 
 (defn read-topology-details [nimbus storm-id]
   (let [conf (:conf nimbus)
+        storage (:storage nimbus)
         storm-base (.storm-base (:storm-cluster-state nimbus) storm-id nil)
-        topology-conf (read-storm-conf conf storm-id)
-        topology (read-storm-topology conf storm-id)
+        topology-conf (read-storm-conf storage conf storm-id)
+        topology (read-storm-topology storage storm-id)
         executor->component (->> (compute-executor->component nimbus storm-id)
                                  (map-key (fn [[start-task end-task]]
                                             (ExecutorDetails. (int start-task) (int end-task)))))]
@@ -401,10 +396,11 @@
 
 (defn- compute-executors [nimbus storm-id]
   (let [conf (:conf nimbus)
+        storage (:storage nimbus)
         storm-base (.storm-base (:storm-cluster-state nimbus) storm-id nil)
         component->executors (:component->executors storm-base)
-        storm-conf (read-storm-conf conf storm-id)
-        topology (read-storm-topology conf storm-id)
+        storm-conf (read-storm-conf storage conf storm-id)
+        topology (read-storm-topology storage storm-id)
         task->component (storm-task-info topology storm-conf)]
     (->> (storm-task-info topology storm-conf)
          reverse-map
@@ -417,9 +413,10 @@
 
 (defn- compute-executor->component [nimbus storm-id]
   (let [conf (:conf nimbus)
+        storage (:storage nimbus)
         executors (compute-executors nimbus storm-id)
-        topology (read-storm-topology conf storm-id)
-        storm-conf (read-storm-conf conf storm-id)
+        topology (read-storm-topology storage storm-id)
+        storm-conf (read-storm-conf storage conf storm-id)
         task->component (storm-task-info topology storm-conf)
         executor->component (into {} (for [executor executors
                                            :let [start-task (first executor)
@@ -673,7 +670,7 @@
                                                                         [id now-secs]
                                                                         )))]]
                                    {topology-id (Assignment.
-                                                 (master-stormdist-root conf topology-id)
+                                                 (nimbus-storage-stormdist-root topology-id)
                                                  (select-keys all-node->host all-nodes)
                                                  executor->node+port
                                                  start-times)}))]
@@ -702,8 +699,9 @@
   {:pre [(#{:active :inactive} topology-initial-status)]}                
   (let [storm-cluster-state (:storm-cluster-state nimbus)
         conf (:conf nimbus)
-        storm-conf (read-storm-conf conf storm-id)
-        topology (system-topology! storm-conf (read-storm-topology conf storm-id))
+        storage (:storage nimbus)
+        storm-conf (read-storm-conf storage conf storm-id)
+        topology (system-topology! storm-conf (read-storm-topology storage storm-id))
         num-executors (->> (all-components topology) (map-val num-start-executors))]
     (log-message "Activating " storm-name ": " storm-id)
     (.activate-storm! storm-cluster-state
@@ -732,17 +730,13 @@
       (throw (AlreadyAliveException. (str storm-name " is already active"))))
     ))
 
-(defn code-ids [conf]
-  (-> conf
-      master-stormdist-root
-      read-dir-contents
-      set
-      ))
+(defn code-ids [storage]
+  (set (.list storage (nimbus-storage-stormdist-root))))
 
-(defn cleanup-storm-ids [conf storm-cluster-state]
+(defn cleanup-storm-ids [storage storm-cluster-state]
   (let [heartbeat-ids (set (.heartbeat-storms storm-cluster-state))
         error-ids (set (.error-topologies storm-cluster-state))
-        code-ids (code-ids conf)
+        code-ids (code-ids storage)
         assigned-ids (set (.active-storms storm-cluster-state))]
     (set/difference (set/union heartbeat-ids error-ids code-ids) assigned-ids)
     ))
@@ -805,15 +799,16 @@
 (defn do-cleanup [nimbus]
   (let [storm-cluster-state (:storm-cluster-state nimbus)
         conf (:conf nimbus)
+        storage (:storage nimbus)
         submit-lock (:submit-lock nimbus)]
     (let [to-cleanup-ids (locking submit-lock
-                           (cleanup-storm-ids conf storm-cluster-state))]
+                           (cleanup-storm-ids storage storm-cluster-state))]
       (when-not (empty? to-cleanup-ids)
         (doseq [id to-cleanup-ids]
           (log-message "Cleaning up " id)
           (.teardown-heartbeats! storm-cluster-state id)
           (.teardown-topology-errors! storm-cluster-state id)
-          (rmr (master-stormdist-root conf id))
+          (.delete storage (nimbus-storage-stormdist-root id))
           (swap! (:heartbeats-cache nimbus) dissoc id))
         ))))
 
@@ -834,11 +829,11 @@
 
 (defn cleanup-corrupt-topologies! [nimbus]
   (let [storm-cluster-state (:storm-cluster-state nimbus)
-        code-ids (set (code-ids (:conf nimbus)))
+        code-ids (code-ids (:storage nimbus))
         active-topologies (set (.active-storms storm-cluster-state))
         corrupt-topologies (set/difference active-topologies code-ids)]
     (doseq [corrupt corrupt-topologies]
-      (log-message "Corrupt topology " corrupt " has state on zookeeper but doesn't have a local dir on Nimbus. Cleaning up...")
+      (log-message "Corrupt topology " corrupt " has state on zookeeper but doesn't have a dir in Nimbus storage. Cleaning up...")
       (.remove-storm! storm-cluster-state corrupt)
       )))
 
@@ -859,7 +854,9 @@
 (defserverfn service-handler [conf inimbus]
   (.prepare inimbus conf (master-inimbus-dir conf))
   (log-message "Starting Nimbus with conf " conf)
-  (let [nimbus (nimbus-data conf inimbus)]
+  (let [nimbus (nimbus-data conf inimbus)
+        storage (:storage nimbus)
+        leader-elections (await-leadership conf storage)]
     (cleanup-corrupt-topologies! nimbus)
     (doseq [storm-id (.active-storms (:storm-cluster-state nimbus))]
       (transition! nimbus storm-id :startup))
@@ -879,7 +876,7 @@
                         (fn []
                           (clean-inbox (inbox nimbus) (conf NIMBUS-INBOX-JAR-EXPIRATION-SECS))
                           ))    
-    (reify Nimbus$Iface
+    (reify-with-validation #(ensure-leadership leader-elections) backtype.storm.generated.Nimbus$Iface Shutdownable DaemonCommon
       (^void submitTopologyWithOpts
         [this ^String storm-name ^String uploadedJarLocation ^String serializedConf ^StormTopology topology
          ^SubmitOptions submitOptions]
@@ -911,7 +908,7 @@
             ;; lock protects against multiple topologies being submitted at once and
             ;; cleanup thread killing topology in b/w assignment and starting the topology
             (locking (:submit-lock nimbus)
-              (setup-storm-code conf storm-id uploadedJarLocation storm-conf topology)
+              (setup-storm-code conf storage storm-id uploadedJarLocation storm-conf topology)
               (.setup-heartbeats! storm-cluster-state storm-id)
               (let [thrift-status->kw-status {TopologyInitialStatus/INACTIVE :inactive
                                               TopologyInitialStatus/ACTIVE :active}]
@@ -991,15 +988,15 @@
           ))
 
       (^String beginFileDownload [this ^String file]
-        (let [is (BufferFileInputStream. file)
+        (let [is (BufferInputStream. (.open storage file))
               id (uuid)]
           (.put (:downloaders nimbus) id is)
           id
           ))
 
       (^ByteBuffer downloadChunk [this ^String id]
         (let [downloaders (:downloaders nimbus)
-              ^BufferFileInputStream is (.get downloaders id)]
+              ^BufferInputStream is (.get downloaders id)]
           (when-not is
             (throw (RuntimeException.
                     "Could not find input stream for that id")))
@@ -1014,13 +1011,13 @@
         (to-json (:conf nimbus)))
 
       (^String getTopologyConf [this ^String id]
-        (to-json (read-storm-conf conf id)))
+        (to-json (read-storm-conf storage conf id)))
 
       (^StormTopology getTopology [this ^String id]
-        (system-topology! (read-storm-conf conf id) (read-storm-topology conf id)))
+        (system-topology! (read-storm-conf storage conf id) (read-storm-topology storage id)))
 
       (^StormTopology getUserTopology [this ^String id]
-        (read-storm-topology conf id))
+        (read-storm-topology storage id))
 
       (^ClusterSummary getClusterInfo [this]
         (let [storm-cluster-state (:storm-cluster-state nimbus)
@@ -1063,7 +1060,9 @@
 
       (^TopologyInfo getTopologyInfo [this ^String storm-id]
         (let [storm-cluster-state (:storm-cluster-state nimbus)
-              task->component (storm-task-info (read-storm-topology conf storm-id) (read-storm-conf conf storm-id))
+              storm-topology (read-storm-topology storage storm-id)
+              storm-conf (read-storm-conf storage conf storm-id)
+              task->component (storm-task-info storm-topology storm-conf)
               base (.storm-base storm-cluster-state storm-id nil)
               assignment (.assignment-info storm-cluster-state storm-id nil)
               beats (.executor-beats storm-cluster-state storm-id (:executor->node+port assignment))
@@ -1095,16 +1094,17 @@
                          )
           ))
 
-      Shutdownable
+
       (shutdown [this]
         (log-message "Shutting down master")
         (cancel-timer (:timer nimbus))
         (.disconnect (:storm-cluster-state nimbus))
         (.cleanup (:downloaders nimbus))
         (.cleanup (:uploaders nimbus))
+        (.close leader-elections)
         (log-message "Shut down master")
         )
-      DaemonCommon
+
       (waiting? [this]
         (timer-waiting? (:timer nimbus))))))
 
@@ -1125,14 +1125,13 @@
 
 ;; distributed implementation
 
-(defmethod setup-jar :distributed [conf tmp-jar-location stormroot]
+(defmethod setup-jar :distributed [conf storage tmp-jar-location stormroot]
            (let [src-file (File. tmp-jar-location)]
              (if-not (.exists src-file)
                (throw
                 (IllegalArgumentException.
-                 (str tmp-jar-location " to copy to " stormroot " does not exist!"))))
-             (FileUtils/copyFile src-file (File. (master-stormjar-path stormroot)))
-             ))
+                 (str tmp-jar-location " to copy to Nimbus storage does not exist!"))))
+             (upload-file-to-storage src-file storage (master-stormjar-path stormroot))))
 
 ;; local implementation
 

src/clj/backtype/storm/daemon/supervisor.clj

@@ -439,8 +439,9 @@
 
 (defmethod download-storm-code
     :local [conf storm-id master-code-dir]
-  (let [stormroot (supervisor-stormdist-root conf storm-id)]
-      (FileUtils/copyDirectory (File. master-code-dir) (File. stormroot))
+  (let [stormroot (supervisor-stormdist-root conf storm-id)
+        master-root (nimbus-storage-local-dir conf)]
+      (FileUtils/copyDirectory (File. (str master-root master-code-dir)) (File. stormroot))
       (let [classloader (.getContextClassLoader (Thread/currentThread))
             resources-jar (resources-jar)
             url (.getResource classloader RESOURCES-SUBDIR)