Merge pull request nathanmarz#1 from nathanmarz/master

merge changes from nathanmarz/storm

CHANGELOG.md

@@ -5,6 +5,16 @@
  * Logs are now limited to 1GB per worker (configurable via logging configuration file)
  * Build upgraded to leiningen 2.0
  * Revamped Trident spout interfaces to support more dynamic spouts, such as a spout who reads from a changing set of brokers
+ * How tuples are serialized is now pluggable (thanks anfeng)
+ * Added blowfish encryption based tuple serialization (thanks anfeng)
+ * Have storm fall back to installed storm.yaml (thanks revans2)
+ * Improve error message when Storm detects bundled storm.yaml to show the URL's for offending resources (thanks revans2)
+ * Nimbus throws NotAliveException instead of FileNotFoundException from various query methods when topology is no longer alive (thanks revans2)
+ * Bug fix: Supervisor provides full path to workers to logging config rather than relative path (thanks revans2) 
+ * Bug fix: Call ReducerAggregator#init properly when used within persistentAggregate (thanks lorcan)
+ * Bug fix: Set component-specific configs correctly for Trident spouts
+ * Bug fix: Fix TransactionalMap and OpaqueMap to correctly do multiple updates to the same key in the same batch
+ * Bug fix: Fix race condition between supervisor and Nimbus that could lead to stormconf.ser errors and infinite crashing of supervisor
 
 ## 0.8.2
 

NOTICE

@@ -0,0 +1,4 @@
+Storm
+Copyright 2011-2013 Nathan Marz
+
+

README.markdown

@@ -65,6 +65,9 @@ You must not remove this notice, or any other, from this software.
 * Bryan Peterson ([@Lazyshot](https://github.com/Lazyshot))
 * Sam Ritchie ([@sritchie](https://github.com/sritchie))
 * Stuart Anderson ([@emblem](https://github.com/emblem))
+* Robert Evans ([@revans2](https://github.com/revans2)) 
+* Andy Feng ([@anfeng](https://github.com/anfeng))
+* Lorcan Coyle ([@lorcan](https://github.com/lorcan))
 
 ## Acknowledgements
 

bin/storm

@@ -22,11 +22,15 @@ if sys.platform == "cygwin":
 else:
     normclasspath = identity
 
-CONF_DIR = os.path.expanduser("~/.storm")
 STORM_DIR = "/".join(os.path.realpath( __file__ ).split("/")[:-2])
+USER_CONF_DIR = os.path.expanduser("~/.storm")
+CLUSTER_CONF_DIR = STORM_DIR + "/conf"
+if (not os.path.isfile(USER_CONF_DIR + "/storm.yaml")):
+    USER_CONF_DIR = CLUSTER_CONF_DIR
 CONFIG_OPTS = []
 CONFFILE = ""
 
+
 def get_config_opts():
     global CONFIG_OPTS
     return "-Dstorm.options=" + (','.join(CONFIG_OPTS)).replace(' ', "%%%%")
@@ -73,7 +77,7 @@ def print_localconfvalue(name):
     The local Storm configs are the ones in ~/.storm/storm.yaml merged 
     in with the configs in defaults.yaml.
     """
-    print name + ": " + confvalue(name, [CONF_DIR])
+    print name + ": " + confvalue(name, [USER_CONF_DIR])
 
 def print_remoteconfvalue(name):
     """Syntax: [storm remoteconfvalue conf-name]
@@ -84,7 +88,7 @@ def print_remoteconfvalue(name):
 
     This command must be run on a cluster machine.
     """
-    print name + ": " + confvalue(name, [STORM_DIR + "/conf"])
+    print name + ": " + confvalue(name, [CLUSTER_CONF_DIR])
 
 def parse_args(string):
     r"""Takes a string of whitespace-separated tokens and parses it into a list.
@@ -132,7 +136,7 @@ def jar(jarfile, klass, *args):
     exec_storm_class(
         klass,
         jvmtype="-client",
-        extrajars=[jarfile, CONF_DIR, STORM_DIR + "/bin"],
+        extrajars=[jarfile, USER_CONF_DIR, STORM_DIR + "/bin"],
         args=args,
         jvmopts=["-Dstorm.jar=" + jarfile])
 
@@ -150,7 +154,7 @@ def kill(*args):
         "backtype.storm.command.kill_topology", 
         args=args, 
         jvmtype="-client", 
-        extrajars=[CONF_DIR, STORM_DIR + "/bin"])
+        extrajars=[USER_CONF_DIR, STORM_DIR + "/bin"])
 
 def activate(*args):
     """Syntax: [storm activate topology-name]
@@ -161,7 +165,7 @@ def activate(*args):
         "backtype.storm.command.activate", 
         args=args, 
         jvmtype="-client", 
-        extrajars=[CONF_DIR, STORM_DIR + "/bin"])
+        extrajars=[USER_CONF_DIR, STORM_DIR + "/bin"])
 
 def listtopos(*args):
     """Syntax: [storm list]
@@ -172,7 +176,7 @@ def listtopos(*args):
         "backtype.storm.command.list", 
         args=args, 
         jvmtype="-client", 
-        extrajars=[CONF_DIR, STORM_DIR + "/bin"])
+        extrajars=[USER_CONF_DIR, STORM_DIR + "/bin"])
 
 def deactivate(*args):
     """Syntax: [storm deactivate topology-name]
@@ -183,7 +187,7 @@ def deactivate(*args):
         "backtype.storm.command.deactivate", 
         args=args, 
         jvmtype="-client", 
-        extrajars=[CONF_DIR, STORM_DIR + "/bin"])
+        extrajars=[USER_CONF_DIR, STORM_DIR + "/bin"])
 
 def rebalance(*args):
     """Syntax: [storm rebalance topology-name [-w wait-time-secs] [-n new-num-workers] [-e component=parallelism]*]
@@ -210,7 +214,7 @@ def rebalance(*args):
         "backtype.storm.command.rebalance", 
         args=args, 
         jvmtype="-client", 
-        extrajars=[CONF_DIR, STORM_DIR + "/bin"])
+        extrajars=[USER_CONF_DIR, STORM_DIR + "/bin"])
 
 def shell(resourcesdir, command, *args):
     tmpjarpath = "stormshell" + str(random.randint(0, 10000000)) + ".jar"
@@ -221,7 +225,7 @@ def shell(resourcesdir, command, *args):
         "backtype.storm.command.shell_submission", 
         args=runnerargs, 
         jvmtype="-client", 
-        extrajars=[CONF_DIR],
+        extrajars=[USER_CONF_DIR],
         fork=True)
     os.system("rm " + tmpjarpath)
 
@@ -231,7 +235,7 @@ def repl():
     Opens up a Clojure REPL with the storm jars and configuration 
     on the classpath. Useful for debugging.
     """
-    cppaths = [STORM_DIR + "/conf"]
+    cppaths = [CLUSTER_CONF_DIR]
     exec_storm_class("clojure.lang.Repl", jvmtype="-client", extrajars=cppaths)
 
 def nimbus(klass="backtype.storm.daemon.nimbus"):
@@ -243,7 +247,7 @@ def nimbus(klass="backtype.storm.daemon.nimbus"):
     See Setting up a Storm cluster for more information.
     (https://github.com/nathanmarz/storm/wiki/Setting-up-a-Storm-cluster)
     """
-    cppaths = [STORM_DIR + "/conf"]
+    cppaths = [CLUSTER_CONF_DIR]
     jvmopts = parse_args(confvalue("nimbus.childopts", cppaths)) + [
         "-Dlogfile.name=nimbus.log",
         "-Dlogback.configurationFile=" + STORM_DIR + "/logback/cluster.xml",
@@ -263,7 +267,7 @@ def supervisor(klass="backtype.storm.daemon.supervisor"):
     See Setting up a Storm cluster for more information.
     (https://github.com/nathanmarz/storm/wiki/Setting-up-a-Storm-cluster)
     """
-    cppaths = [STORM_DIR + "/conf"]
+    cppaths = [CLUSTER_CONF_DIR]
     jvmopts = parse_args(confvalue("supervisor.childopts", cppaths)) + [
         "-Dlogfile.name=supervisor.log",
         "-Dlogback.configurationFile=" + STORM_DIR + "/logback/cluster.xml",
@@ -284,7 +288,7 @@ def ui():
     See Setting up a Storm cluster for more information.
     (https://github.com/nathanmarz/storm/wiki/Setting-up-a-Storm-cluster)
     """
-    cppaths = [STORM_DIR + "/conf"]
+    cppaths = [CLUSTER_CONF_DIR]
     jvmopts = parse_args(confvalue("ui.childopts", cppaths)) + [
         "-Dlogfile.name=ui.log",
         "-Dlogback.configurationFile=" + STORM_DIR + "/logback/cluster.xml",
@@ -293,7 +297,7 @@ def ui():
         "backtype.storm.ui.core", 
         jvmtype="-server", 
         jvmopts=jvmopts, 
-        extrajars=[STORM_DIR, STORM_DIR + "/conf"])
+        extrajars=[STORM_DIR, CLUSTER_CONF_DIR])
 
 def drpc():
     """Syntax: [storm drpc]
@@ -312,7 +316,7 @@ def drpc():
         "backtype.storm.daemon.drpc", 
         jvmtype="-server", 
         jvmopts=jvmopts, 
-        extrajars=[STORM_DIR + "/conf"])
+        extrajars=[CLUSTER_CONF_DIR])
 
 def dev_zookeeper():
     """Syntax: [storm dev-zookeeper]
@@ -321,11 +325,11 @@ def dev_zookeeper():
     "storm.zookeeper.port" as its port. This is only intended for development/testing, the
     Zookeeper instance launched is not configured to be used in production.
     """
-    cppaths = [STORM_DIR + "/conf"]
+    cppaths = [CLUSTER_CONF_DIR]
     exec_storm_class(
         "backtype.storm.command.dev_zookeeper", 
         jvmtype="-server", 
-        extrajars=[STORM_DIR + "/conf"])
+        extrajars=[CLUSTER_CONF_DIR])
 
 def version():
   """Syntax: [storm version]

conf/storm.yaml.example

@@ -23,7 +23,7 @@
 #     - "server2"
 
 ## Metrics Consumers
-# topology.metrics.consumers.register:
+# topology.metrics.consumer.register:
 #    - class: "org.mycompany.MyMetricsConsumer"
 #      argument: 
 #         - endpoint: "metrics-collector.mycompany.org"

project.clj

@@ -1,4 +1,4 @@
-(defproject storm/storm "0.9.0-wip15"
+(defproject storm/storm "0.9.0-wip16"
   :url "http://storm-project.clj"
   :description "Distributed and fault-tolerant realtime computation"
   :license {:name "Eclipse Public License - Version 1.0" :url "https://github.com/nathanmarz/storm/blob/master/LICENSE.html"}

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

@@ -4,6 +4,7 @@
   (:import [org.apache.thrift7 TException])
   (:import [org.apache.thrift7.transport TNonblockingServerTransport TNonblockingServerSocket])
   (:import [java.nio ByteBuffer])
+  (:import [java.io FileNotFoundException])
   (:import [java.nio.channels Channels WritableByteChannel])
   (:use [backtype.storm.scheduler.DefaultScheduler])
   (:import [backtype.storm.scheduler INimbus SupervisorDetails WorkerSlot TopologyDetails
@@ -856,6 +857,22 @@
     (throw (InvalidTopologyException.
             (str "Topology name cannot contain any of the following: " (pr-str DISALLOWED-TOPOLOGY-NAME-STRS))))))
 
+(defn- try-read-storm-conf [conf storm-id]
+  (try-cause
+    (read-storm-conf conf storm-id)
+    (catch FileNotFoundException e
+       (throw (NotAliveException. storm-id)))
+  )
+)
+
+(defn- try-read-storm-topology [conf storm-id]
+  (try-cause
+    (read-storm-topology conf storm-id)
+    (catch FileNotFoundException e
+       (throw (NotAliveException. storm-id)))
+  )
+)
+
 (defserverfn service-handler [conf inimbus]
   (.prepare inimbus conf (master-inimbus-dir conf))
   (log-message "Starting Nimbus with conf " conf)
@@ -1014,13 +1031,13 @@
         (to-json (:conf nimbus)))
 
       (^String getTopologyConf [this ^String id]
-        (to-json (read-storm-conf conf id)))
+        (to-json (try-read-storm-conf conf id)))
 
       (^StormTopology getTopology [this ^String id]
-        (system-topology! (read-storm-conf conf id) (read-storm-topology conf id)))
+        (system-topology! (try-read-storm-conf conf id) (try-read-storm-topology conf id)))
 
       (^StormTopology getUserTopology [this ^String id]
-        (read-storm-topology conf id))
+        (try-read-storm-topology conf id))
 
       (^ClusterSummary getClusterInfo [this]
         (let [storm-cluster-state (:storm-cluster-state nimbus)
@@ -1063,7 +1080,7 @@
 
       (^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))
+              task->component (storm-task-info (try-read-storm-topology conf storm-id) (try-read-storm-conf conf storm-id))
               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))

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

@@ -20,11 +20,17 @@
   (shutdown-all-workers [this])
   )
 
+(defn- assignments-snapshot [storm-cluster-state callback]
+  (let [storm-ids (.assignments storm-cluster-state callback)]
+     (->> (dofor [sid storm-ids] {sid (.assignment-info storm-cluster-state sid callback)})
+          (apply merge)
+          (filter-val not-nil?)
+          )))
 
-(defn- read-my-executors [storm-cluster-state storm-id assignment-id callback]
-  (let [assignment (.assignment-info storm-cluster-state storm-id callback)
+(defn- read-my-executors [assignments-snapshot storm-id assignment-id]
+  (let [assignment (get assignments-snapshot storm-id)
         my-executors (filter (fn [[_ [node _]]] (= node assignment-id))
-                               (:executor->node+port assignment))
+                           (:executor->node+port assignment))
         port-executors (apply merge-with
                           concat
                           (for [[executor [_ port]] my-executors]
@@ -34,29 +40,18 @@
                ;; need to cast to int b/c it might be a long (due to how yaml parses things)
                ;; doall is to avoid serialization/deserialization problems with lazy seqs
                [(Integer. port) (LocalAssignment. storm-id (doall executors))]
-               ))
-    ))
+               ))))
+
 
 (defn- read-assignments
   "Returns map from port to struct containing :storm-id and :executors"
-  [storm-cluster-state assignment-id callback]
-  (let [storm-ids (.assignments storm-cluster-state callback)]
-    (apply merge-with
-           (fn [& ignored]
-             (throw (RuntimeException.
-                     "Should not have multiple topologies assigned to one port")))
-           (dofor [sid storm-ids] (read-my-executors storm-cluster-state sid assignment-id callback))
-           )))
+  [assignments-snapshot assignment-id]
+  (->> (dofor [sid (keys assignments-snapshot)] (read-my-executors assignments-snapshot sid assignment-id))
+       (apply merge-with (fn [& ignored] (throw-runtime "Should not have multiple topologies assigned to one port")))))
 
 (defn- read-storm-code-locations
-  [storm-cluster-state callback]
-  (let [storm-ids (.assignments storm-cluster-state callback)]
-    (into {}
-      (dofor [sid storm-ids]
-        [sid (:master-code-dir (.assignment-info storm-cluster-state sid callback))]
-        ))
-    ))
-
+  [assignments-snapshot]
+  (map-val :master-code-dir assignments-snapshot))
 
 (defn- read-downloaded-storm-ids [conf]
   (map #(java.net.URLDecoder/decode %) (read-dir-contents (supervisor-stormdist-root conf)))
@@ -265,12 +260,12 @@
           ^ISupervisor isupervisor (:isupervisor supervisor)
           ^LocalState local-state (:local-state supervisor)
           sync-callback (fn [& ignored] (.add event-manager this))
-          storm-code-map (read-storm-code-locations storm-cluster-state sync-callback)
+          assignments-snapshot (assignments-snapshot storm-cluster-state sync-callback)
+          storm-code-map (read-storm-code-locations assignments-snapshot)
           downloaded-storm-ids (set (read-downloaded-storm-ids conf))
           all-assignment (read-assignments
-                           storm-cluster-state
-                           (:assignment-id supervisor)
-                           sync-callback)
+                           assignments-snapshot
+                           (:assignment-id supervisor))
           new-assignment (->> all-assignment
                               (filter-key #(.confirmAssigned isupervisor %)))
           assigned-storm-ids (assigned-storm-ids-from-port-assignments new-assignment)
@@ -409,6 +404,7 @@
 (defmethod launch-worker
     :distributed [supervisor storm-id port worker-id]
     (let [conf (:conf supervisor)
+          storm-home (System/getProperty "storm.home")
           stormroot (supervisor-stormdist-root conf storm-id)
           stormjar (supervisor-stormjar-path stormroot)
           storm-conf (read-supervisor-storm-conf conf storm-id)
@@ -420,8 +416,8 @@
           command (str "java -server " childopts
                        " -Djava.library.path=" (conf JAVA-LIBRARY-PATH)
                        " -Dlogfile.name=" logfilename
-                       " -Dstorm.home=" (System/getProperty "storm.home")
-                       " -Dlogback.configurationFile=logback/cluster.xml"
+                       " -Dstorm.home=" storm-home
+                       " -Dlogback.configurationFile=" storm-home "/logback/cluster.xml"
                        " -cp " classpath " backtype.storm.daemon.worker "
                        (java.net.URLEncoder/encode storm-id) " " (:assignment-id supervisor)
                        " " port " " worker-id)]