diff --git a/conf/defaults.yaml b/conf/defaults.yaml
index 0050227a343..7bd9c9eaa2e 100644
--- a/conf/defaults.yaml
+++ b/conf/defaults.yaml
@@ -39,7 +39,6 @@ storm.messaging.transport: "backtype.storm.messaging.netty.Context"
storm.meta.serialization.delegate: "backtype.storm.serialization.DefaultSerializationDelegate"
### nimbus.* configs are for the master
-nimbus.host: "localhost"
nimbus.thrift.port: 6627
nimbus.thrift.max_buffer_size: 1048576
nimbus.childopts: "-Xmx1024m"
diff --git a/pom.xml b/pom.xml
index 83620ce9003..7ae42d7bf55 100644
--- a/pom.xml
+++ b/pom.xml
@@ -363,6 +363,11 @@
+
+ org.apache.curator
+ curator-recipes
+ ${curator.version}
+
com.googlecode.json-simple
json-simple
diff --git a/storm-core/pom.xml b/storm-core/pom.xml
index 29338f0cb0a..de7fb2a669d 100644
--- a/storm-core/pom.xml
+++ b/storm-core/pom.xml
@@ -128,6 +128,10 @@
+
+ org.apache.curator
+ curator-recipes
+
com.googlecode.json-simple
json-simple
diff --git a/storm-core/src/clj/backtype/storm/config.clj b/storm-core/src/clj/backtype/storm/config.clj
index 14beb21f2bd..d0f031d6113 100644
--- a/storm-core/src/clj/backtype/storm/config.clj
+++ b/storm-core/src/clj/backtype/storm/config.clj
@@ -152,6 +152,11 @@
[stormroot]
(str stormroot file-path-separator "stormconf.ser"))
+(defn master-tmp-dir [conf]
+ (let [ret (str (master-local-dir conf) file-path-separator "tmp")]
+ (FileUtils/forceMkdir (File. ret))
+ ret ))
+
(defn master-inbox
[conf]
(let [ret (str (master-local-dir conf) file-path-separator "inbox")]
diff --git a/storm-core/src/clj/backtype/storm/daemon/nimbus.clj b/storm-core/src/clj/backtype/storm/daemon/nimbus.clj
index 9f1c82f505e..e5350614bc5 100644
--- a/storm-core/src/clj/backtype/storm/daemon/nimbus.clj
+++ b/storm-core/src/clj/backtype/storm/daemon/nimbus.clj
@@ -27,6 +27,7 @@
(:use [backtype.storm bootstrap util])
(:use [backtype.storm.config :only [validate-configs-with-schemas]])
(:use [backtype.storm.daemon common])
+ (:use [backtype.storm.nimbus leadership])
(:gen-class
:methods [^{:static true} [launch [backtype.storm.scheduler.INimbus] void]]))
@@ -892,10 +893,47 @@
)
)
+(defn- sync-storm-code-from-leader [nimbus]
+ (let [conf (:conf nimbus)
+ storm-cluster-state (:storm-cluster-state nimbus)
+ storm-ids (.assignments storm-cluster-state nil)
+ storm-code-map (->> (dofor [sid storm-ids] {sid (.assignment-info storm-cluster-state sid nil)})
+ (apply merge)
+ (filter-val not-nil?)
+ (map-val :master-code-dir)
+ )
+ downloaded-storm-ids (set (map #(java.net.URLDecoder/decode %) (read-dir-contents (master-stormdist-root conf))))
+ tmproot (str (master-tmp-dir conf) file-path-separator (uuid))]
+ (doseq [[storm-id master-code-dir] storm-code-map]
+ (when (not (downloaded-storm-ids storm-id))
+ (log-message "Downloading code for storm id " storm-id " from " master-code-dir)
+
+ (FileUtils/forceMkdir (File. tmproot))
+ (Utils/downloadFromMaster conf (master-stormjar-path master-code-dir) (master-stormjar-path tmproot))
+ (Utils/downloadFromMaster conf (master-stormcode-path master-code-dir) (master-stormcode-path tmproot))
+ (Utils/downloadFromMaster conf (master-stormconf-path master-code-dir) (master-stormconf-path tmproot))
+ (FileUtils/moveDirectory (File. tmproot) (File. (master-stormdist-root conf storm-id)))
+
+ (log-message "Finished downloading code for storm id " storm-id " from " master-code-dir)
+ )
+ )
+ )
+)
+
(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)
+ nimbus-leadership (nimbus-leadership conf)]
+ ;; Schedule synchronize storm code from leader
+ (schedule-recurring (:timer nimbus)
+ 10
+ 10
+ (fn []
+ (sync-storm-code-from-leader nimbus)
+ ))
+ ;; Compete to be nimbus leader
+ (acquire-leadership nimbus-leadership)
(.prepare ^backtype.storm.nimbus.ITopologyValidator (:validator nimbus) conf)
(cleanup-corrupt-topologies! nimbus)
(doseq [storm-id (.active-storms (:storm-cluster-state nimbus))]
@@ -1141,6 +1179,7 @@
(.disconnect (:storm-cluster-state nimbus))
(.cleanup (:downloaders nimbus))
(.cleanup (:uploaders nimbus))
+ (.close nimbus-leadership)
(log-message "Shut down master")
)
DaemonCommon
diff --git a/storm-core/src/clj/backtype/storm/nimbus/leadership.clj b/storm-core/src/clj/backtype/storm/nimbus/leadership.clj
new file mode 100644
index 00000000000..a3c31c975a0
--- /dev/null
+++ b/storm-core/src/clj/backtype/storm/nimbus/leadership.clj
@@ -0,0 +1,34 @@
+;; Licensed to the Apache Software Foundation (ASF) under one
+;; or more contributor license agreements. See the NOTICE file
+;; distributed with this work for additional information
+;; regarding copyright ownership. The ASF licenses this file
+;; to you under the Apache License, Version 2.0 (the
+;; "License"); you may not use this file except in compliance
+;; with the License. You may obtain a copy of the License at
+;;
+;; http://www.apache.org/licenses/LICENSE-2.0
+;;
+;; Unless required by applicable law or agreed to in writing, software
+;; distributed under the License is distributed on an "AS IS" BASIS,
+;; WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+;; See the License for the specific language governing permissions and
+;; limitations under the License.
+(ns backtype.storm.nimbus.leadership
+ (:import [backtype.storm.nimbus NimbusLeadership])
+ (:use [backtype.storm log]))
+
+(defn nimbus-leadership [conf]
+ (NimbusLeadership. conf))
+
+(defn get-nimbus-leader-address [conf]
+ (.getNimbusLeaderAddress (nimbus-leadership conf)))
+
+(defn get-nimbus-hosts [conf]
+ (.getNimbusHosts (nimbus-leadership conf)))
+
+(defn acquire-leadership [nimbus-leadership]
+ (when-let [nimbus-leader-address (.getNimbusLeaderAddress nimbus-leadership)]
+ (log-message "Current Nimbus Leader: " nimbus-leader-address))
+ (log-message "acquiring nimbus leadership...")
+ (.acquireLeaderShip nimbus-leadership)
+ (log-message "acuqired nimbus leadership!"))
\ No newline at end of file
diff --git a/storm-core/src/clj/backtype/storm/thrift.clj b/storm-core/src/clj/backtype/storm/thrift.clj
index f778d238c5d..e22de85fa35 100644
--- a/storm-core/src/clj/backtype/storm/thrift.clj
+++ b/storm-core/src/clj/backtype/storm/thrift.clj
@@ -28,7 +28,8 @@
(:import [backtype.storm.clojure RichShellBolt RichShellSpout])
(:import [org.apache.thrift.protocol TBinaryProtocol TProtocol])
(:import [org.apache.thrift.transport TTransport TFramedTransport TSocket])
- (:use [backtype.storm util config log]))
+ (:use [backtype.storm util config log])
+ (:use [backtype.storm.nimbus leadership]))
(defn instantiate-java-object
[^JavaObject obj]
@@ -86,7 +87,7 @@
(defmacro with-configured-nimbus-connection
[client-sym & body]
`(let [conf# (read-storm-config)
- host# (conf# NIMBUS-HOST)
+ host# (.getHostName (get-nimbus-leader-address conf#))
port# (conf# NIMBUS-THRIFT-PORT)]
(with-nimbus-connection [~client-sym host# port#]
~@body )))
diff --git a/storm-core/src/clj/backtype/storm/ui/core.clj b/storm-core/src/clj/backtype/storm/ui/core.clj
index 969b514b9f9..bebcc9de4e1 100644
--- a/storm-core/src/clj/backtype/storm/ui/core.clj
+++ b/storm-core/src/clj/backtype/storm/ui/core.clj
@@ -22,6 +22,7 @@
(:use [backtype.storm.ui helpers])
(:use [backtype.storm.daemon [common :only [ACKER-COMPONENT-ID ACKER-INIT-STREAM-ID
ACKER-ACK-STREAM-ID ACKER-FAIL-STREAM-ID system-id?]]])
+ (:use [backtype.storm.nimbus leadership])
(:use [ring.adapter.jetty :only [run-jetty]])
(:use [clojure.string :only [trim]])
(:import [backtype.storm.utils Utils])
@@ -30,6 +31,7 @@
ErrorInfo ClusterSummary SupervisorSummary TopologySummary
Nimbus$Client StormTopology GlobalStreamId RebalanceOptions
KillOptions])
+ (:import [java.net InetSocketAddress])
(:import [java.io File])
(:require [compojure.route :as route]
[compojure.handler :as handler]
@@ -43,7 +45,7 @@
(defmacro with-nimbus
[nimbus-sym & body]
`(thrift/with-nimbus-connection
- [~nimbus-sym (*STORM-CONF* NIMBUS-HOST) (*STORM-CONF* NIMBUS-THRIFT-PORT)]
+ [~nimbus-sym (.getHostName (get-nimbus-leader-address *STORM-CONF*)) (*STORM-CONF* NIMBUS-THRIFT-PORT)]
~@body))
(defn get-filled-stats
@@ -510,6 +512,15 @@
"slotsTotal" (.get_num_workers s)
"slotsUsed" (.get_num_used_workers s)})}))
+(defn nimbus-summary []
+ (let [nimbus-hosts (get-nimbus-hosts *STORM-CONF*)
+ nimbus-leader-host (get-nimbus-leader-address *STORM-CONF*)]
+ {"nimbuses"
+ (for [^InetSocketAddress nimbus-host nimbus-hosts]
+ {"nimbusHost" (str (.getHostName nimbus-host) ":" (.getPort nimbus-host))
+ "isLeader" (if (= nimbus-host nimbus-leader-host) "true" "false")
+ })}))
+
(defn all-topologies-summary
([]
(with-nimbus
@@ -845,6 +856,8 @@
(:callback m) :serialize-fn identity))
(GET "/api/v1/cluster/summary" [& m]
(json-response (cluster-summary) (:callback m)))
+ (GET "/api/v1/nimbus/summary" [& m]
+ (json-response (nimbus-summary) (:callback m)))
(GET "/api/v1/supervisor/summary" [& m]
(json-response (supervisor-summary) (:callback m)))
(GET "/api/v1/topology/summary" [& m]
diff --git a/storm-core/src/dev/resources/storm.py b/storm-core/src/dev/resources/storm.py
old mode 100755
new mode 100644
diff --git a/storm-core/src/jvm/backtype/storm/nimbus/NimbusLeadership.java b/storm-core/src/jvm/backtype/storm/nimbus/NimbusLeadership.java
new file mode 100644
index 00000000000..ae6e97db135
--- /dev/null
+++ b/storm-core/src/jvm/backtype/storm/nimbus/NimbusLeadership.java
@@ -0,0 +1,121 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package backtype.storm.nimbus;
+
+import java.io.UnsupportedEncodingException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.recipes.locks.InterProcessMutex;
+import org.apache.curator.utils.ZKPaths;
+
+import backtype.storm.Config;
+import backtype.storm.utils.Utils;
+
+@SuppressWarnings("rawtypes")
+public class NimbusLeadership {
+
+ private static final String STORM_NIMBUS_LEADERSHIP_PATH = "/nimbus/leadership";
+
+ private Map conf;
+ private CuratorFramework curator;
+ private InterProcessMutex mutex;
+ private boolean isLeader = false;
+
+ public NimbusLeadership(final Map conf) {
+ this.conf = conf;
+ }
+
+ public void acquireLeaderShip() throws Exception {
+ String nimbusHostName = InetAddress.getLocalHost().getCanonicalHostName();
+ Object nimbusPort = conf.get(Config.NIMBUS_THRIFT_PORT);
+ String nodeId = nimbusHostName + ":" + nimbusPort.toString();
+ initCurator();
+ initLeadershipMutex(nodeId);
+ mutex.acquire();
+ isLeader = true;
+ }
+
+ public InetSocketAddress getNimbusLeaderAddress() throws Exception {
+ InetSocketAddress leaderAddress = null;
+ initCurator();
+ initLeadershipMutex(null);
+ Collection nimbusNodesPath = mutex.getParticipantNodes();
+ if (nimbusNodesPath.size() > 0) {
+ leaderAddress = parseAddress(nimbusNodesPath.iterator().next());
+ }
+ close();
+ return leaderAddress;
+ }
+
+ public List getNimbusHosts() throws Exception {
+ List nimbusAddressList = new ArrayList();
+ initCurator();
+ initLeadershipMutex(null);
+ Collection nimbusNodesPath = mutex.getParticipantNodes();
+ for (String nimbusNodePath : nimbusNodesPath) {
+ nimbusAddressList.add(parseAddress(nimbusNodePath));
+ }
+ close();
+ return nimbusAddressList;
+ }
+
+ public void close() {
+ if (isLeader) {
+ try {
+ mutex.release();
+ } catch (Exception e) {
+ throw new RuntimeException("Exception while releasing mutex", e);
+ }
+ }
+ curator.close();
+ }
+
+ @SuppressWarnings("unchecked")
+ private void initCurator() throws Exception {
+ List servers = (List) conf.get(Config.STORM_ZOOKEEPER_SERVERS);
+ Object port = conf.get(Config.STORM_ZOOKEEPER_PORT);
+ this.curator = Utils.newCuratorStarted(conf, servers, port);
+ }
+
+ private void initLeadershipMutex(final String nodeId) throws Exception {
+ String path = (String) conf.get(Config.STORM_ZOOKEEPER_ROOT) + STORM_NIMBUS_LEADERSHIP_PATH;
+ ZKPaths.mkdirs(curator.getZookeeperClient().getZooKeeper(), path);
+ mutex = new InterProcessMutex(curator, path) {
+ @Override
+ protected byte[] getLockNodeBytes() {
+ try {
+ return nodeId == null ? null : nodeId.getBytes("UTF-8");
+ } catch (UnsupportedEncodingException e) {
+ throw new RuntimeException("UTF-8 isn't supported", e);
+ }
+ }
+ };
+ }
+
+ private InetSocketAddress parseAddress(String nimbusNodePath) throws Exception {
+ String nimbusNodeData = new String(curator.getData().forPath(nimbusNodePath), "UTF-8");
+ String[] split = nimbusNodeData.split(":");
+ return new InetSocketAddress(split[0], Integer.parseInt(split[1]));
+ }
+}
diff --git a/storm-core/src/jvm/backtype/storm/utils/NimbusClient.java b/storm-core/src/jvm/backtype/storm/utils/NimbusClient.java
index e93acc8acc7..8c715acf4e8 100644
--- a/storm-core/src/jvm/backtype/storm/utils/NimbusClient.java
+++ b/storm-core/src/jvm/backtype/storm/utils/NimbusClient.java
@@ -24,6 +24,7 @@
import org.apache.thrift.transport.TTransportException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import backtype.storm.nimbus.NimbusLeadership;
public class NimbusClient extends ThriftClient {
private Nimbus.Client _client;
@@ -31,12 +32,15 @@ public class NimbusClient extends ThriftClient {
public static NimbusClient getConfiguredClient(Map conf) {
try {
- String nimbusHost = (String) conf.get(Config.NIMBUS_HOST);
+ NimbusLeadership nimbusLeadership = new NimbusLeadership(conf);
+ String nimbusHost = nimbusLeadership.getNimbusLeaderAddress().getHostName();
int nimbusPort = Utils.getInt(conf.get(Config.NIMBUS_THRIFT_PORT));
return new NimbusClient(conf, nimbusHost, nimbusPort);
} catch (TTransportException ex) {
throw new RuntimeException(ex);
- }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
}
public NimbusClient(Map conf, String host, int port) throws TTransportException {
diff --git a/storm-core/src/ui/public/index.html b/storm-core/src/ui/public/index.html
index ad6f88ccc7e..f23f765cf33 100644
--- a/storm-core/src/ui/public/index.html
+++ b/storm-core/src/ui/public/index.html
@@ -38,6 +38,9 @@ Topology summary
Supervisor summary
+Nimbus summary
+
+
Nimbus Configuration
@@ -60,6 +63,7 @@ Nimbus Configuration
var clusterSummary = $("#cluster-summary");
var topologySummary = $("#topology-summary");
var supervisorSummary = $("#supervisor-summary");
+ var nimbusSummary = $("#nimbus-summary");
var config = $("#nimbus-configuration");
$.getJSON("/api/v1/cluster/summary",function(response,status,jqXHR) {
@@ -83,6 +87,14 @@ Nimbus Configuration
}
});
});
+ $.getJSON("/api/v1/nimbus/summary",function(response,status,jqXHR) {
+ $.get("/templates/index-page-template.html", function(template) {
+ nimbusSummary.append(Mustache.render($(template).filter("#nimbus-summary-template").html(),response));
+ if(response["nimbuses"].length > 0) {
+ $("#nimbus-summary-table").tablesorter({ sortList: [[0,0]], headers: {}});
+ }
+ });
+ });
$.getJSON("/api/v1/cluster/configuration",function(response,status,jqXHR) {
var formattedResponse = formatConfigData(response);
$.get("/templates/index-page-template.html", function(template) {
diff --git a/storm-core/src/ui/public/templates/index-page-template.html b/storm-core/src/ui/public/templates/index-page-template.html
index e3aab5ef19f..6e2608b5172 100644
--- a/storm-core/src/ui/public/templates/index-page-template.html
+++ b/storm-core/src/ui/public/templates/index-page-template.html
@@ -174,7 +174,32 @@
-
+