You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by bo...@apache.org on 2014/07/16 16:48:06 UTC
[01/11] git commit: Merge pull request #1 from apache/master
Repository: incubator-storm
Updated Branches:
refs/heads/master 96e81c1fb -> 69b4601f7
Merge pull request #1 from apache/master
update from apache/incubator-storm
Project: http://git-wip-us.apache.org/repos/asf/incubator-storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-storm/commit/d080e043
Tree: http://git-wip-us.apache.org/repos/asf/incubator-storm/tree/d080e043
Diff: http://git-wip-us.apache.org/repos/asf/incubator-storm/diff/d080e043
Branch: refs/heads/master
Commit: d080e043d15d710d6729963bed9cb078b7dce78e
Parents: dbcd939 254ec13
Author: Li Jiahong <Gv...@users.noreply.github.com>
Authored: Wed Mar 5 12:36:22 2014 +0800
Committer: Li Jiahong <Gv...@users.noreply.github.com>
Committed: Wed Mar 5 12:36:22 2014 +0800
----------------------------------------------------------------------
.gitignore | 2 +-
CHANGELOG.md | 14 +
NOTICE | 3 +-
bin/storm-config.cmd | 19 +-
bin/storm.cmd | 14 +-
pom.xml | 44 +-
.../maven-shade-clojure-transformer/pom.xml | 39 +
.../maven/shade/clojure/ClojureTransformer.java | 72 +
storm-core/pom.xml | 70 +-
.../src/clj/backtype/storm/daemon/drpc.clj | 8 +-
.../src/clj/backtype/storm/daemon/nimbus.clj | 8 +-
storm-core/src/clj/backtype/storm/thrift.clj | 4 +-
.../src/jvm/backtype/storm/StormSubmitter.java | 2 +-
.../storm/drpc/DRPCInvocationsClient.java | 10 +-
.../src/jvm/backtype/storm/drpc/DRPCSpout.java | 4 +-
.../jvm/backtype/storm/drpc/ReturnResults.java | 2 +-
.../storm/generated/AlreadyAliveException.java | 46 +-
.../src/jvm/backtype/storm/generated/Bolt.java | 60 +-
.../jvm/backtype/storm/generated/BoltStats.java | 182 +-
.../storm/generated/ClusterSummary.java | 86 +-
.../storm/generated/ComponentCommon.java | 100 +-
.../storm/generated/ComponentObject.java | 66 +-
.../storm/generated/DRPCExecutionException.java | 46 +-
.../backtype/storm/generated/DRPCRequest.java | 60 +-
.../storm/generated/DistributedRPC.java | 180 +-
.../generated/DistributedRPCInvocations.java | 384 +--
.../jvm/backtype/storm/generated/ErrorInfo.java | 60 +-
.../backtype/storm/generated/ExecutorInfo.java | 60 +-
.../storm/generated/ExecutorSpecificStats.java | 56 +-
.../backtype/storm/generated/ExecutorStats.java | 106 +-
.../storm/generated/ExecutorSummary.java | 114 +-
.../storm/generated/GlobalStreamId.java | 60 +-
.../jvm/backtype/storm/generated/Grouping.java | 112 +-
.../generated/InvalidTopologyException.java | 46 +-
.../backtype/storm/generated/JavaObject.java | 66 +-
.../backtype/storm/generated/JavaObjectArg.java | 90 +-
.../backtype/storm/generated/KillOptions.java | 44 +-
.../jvm/backtype/storm/generated/Nimbus.java | 2368 +++++++++---------
.../storm/generated/NotAliveException.java | 46 +-
.../backtype/storm/generated/NullStruct.java | 32 +-
.../storm/generated/RebalanceOptions.java | 76 +-
.../storm/generated/ShellComponent.java | 56 +-
.../jvm/backtype/storm/generated/SpoutSpec.java | 60 +-
.../backtype/storm/generated/SpoutStats.java | 122 +-
.../storm/generated/StateSpoutSpec.java | 60 +-
.../backtype/storm/generated/StormTopology.java | 98 +-
.../backtype/storm/generated/StreamInfo.java | 66 +-
.../backtype/storm/generated/SubmitOptions.java | 46 +-
.../storm/generated/SupervisorSummary.java | 102 +-
.../backtype/storm/generated/TopologyInfo.java | 136 +-
.../storm/generated/TopologyInitialStatus.java | 4 +-
.../storm/generated/TopologySummary.java | 130 +-
.../storm/security/auth/ITransportPlugin.java | 8 +-
.../security/auth/SaslTransportPlugin.java | 24 +-
.../security/auth/SimpleTransportPlugin.java | 24 +-
.../storm/security/auth/ThriftClient.java | 10 +-
.../storm/security/auth/ThriftServer.java | 4 +-
.../auth/digest/DigestSaslTransportPlugin.java | 10 +-
.../jvm/backtype/storm/utils/DRPCClient.java | 10 +-
.../jvm/backtype/storm/utils/NimbusClient.java | 2 +-
.../src/jvm/backtype/storm/utils/Utils.java | 2 +-
.../trident/drpc/ReturnResultsReducer.java | 2 +-
.../jvm/storm/trident/util/TridentUtils.java | 8 +-
.../storm/security/auth/ThriftClient_test.clj | 2 +-
.../storm/security/auth/ThriftServer_test.clj | 2 +-
.../backtype/storm/security/auth/auth_test.clj | 4 +-
.../test/clj/backtype/storm/utils_test.clj | 2 +-
storm-deps/libthrift/pom.xml | 87 -
storm-dist/LICENSE | 297 ---
storm-dist/NOTICE | 36 -
storm-dist/binary/LICENSE | 297 +++
storm-dist/binary/NOTICE | 36 +
storm-dist/binary/pom.xml | 78 +
storm-dist/binary/src/main/assembly/binary.xml | 105 +
storm-dist/pom.xml | 90 -
storm-dist/source/pom.xml | 77 +
storm-dist/source/src/main/assembly/source.xml | 42 +
storm-dist/src/main/assembly/binary.xml | 105 -
storm-dist/src/main/assembly/source.xml | 42 -
79 files changed, 3646 insertions(+), 3501 deletions(-)
----------------------------------------------------------------------
[05/11] git commit: STORM-312 add storm monitor command line tools
Posted by bo...@apache.org.
STORM-312 add storm monitor command line tools
Project: http://git-wip-us.apache.org/repos/asf/incubator-storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-storm/commit/edcaf71b
Tree: http://git-wip-us.apache.org/repos/asf/incubator-storm/tree/edcaf71b
Diff: http://git-wip-us.apache.org/repos/asf/incubator-storm/diff/edcaf71b
Branch: refs/heads/master
Commit: edcaf71b304cc5196eec48035b06d15aaa8df433
Parents: a45b53a
Author: jiahong.ljh <ji...@alibaba-inc.com>
Authored: Fri May 16 15:14:05 2014 +0800
Committer: jiahong.ljh <ji...@alibaba-inc.com>
Committed: Fri May 16 15:14:05 2014 +0800
----------------------------------------------------------------------
bin/storm | 14 +-
pom.xml | 6 +
storm-core/pom.xml | 4 +
.../src/jvm/backtype/storm/command/Monitor.java | 197 +++++++++++++++++++
4 files changed, 220 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/edcaf71b/bin/storm
----------------------------------------------------------------------
diff --git a/bin/storm b/bin/storm
index bb6e40a..0600d45 100755
--- a/bin/storm
+++ b/bin/storm
@@ -390,6 +390,18 @@ def print_classpath():
"""
print get_classpath([])
+def monitor(*args):
+ """Syntax: [storm monitor]
+
+ Monitor given topologies' given components' or type of components' throughput interactively.
+ """
+ exec_storm_class(
+ "backtype.storm.command.Monitor",
+ args=args,
+ jvmtype="-client",
+ extrajars=[USER_CONF_DIR, STORM_DIR + "/bin"])
+
+
def print_commands():
"""Print all client commands and link to documentation"""
print "Commands:\n\t", "\n\t".join(sorted(COMMANDS.keys()))
@@ -416,7 +428,7 @@ COMMANDS = {"jar": jar, "kill": kill, "shell": shell, "nimbus": nimbus, "ui": ui
"drpc": drpc, "supervisor": supervisor, "localconfvalue": print_localconfvalue,
"remoteconfvalue": print_remoteconfvalue, "repl": repl, "classpath": print_classpath,
"activate": activate, "deactivate": deactivate, "rebalance": rebalance, "help": print_usage,
- "list": listtopos, "dev-zookeeper": dev_zookeeper, "version": version}
+ "list": listtopos, "dev-zookeeper": dev_zookeeper, "version": version, "monitor": monitor}
def parse_config(config_list):
global CONFIG_OPTS
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/edcaf71b/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index aa93fcc..484f5f4 100644
--- a/pom.xml
+++ b/pom.xml
@@ -464,6 +464,12 @@
<version>3.8.1</version>
<scope>test</scope>
</dependency>
+
+ <dependency>
+ <groupId>args4j</groupId>
+ <artifactId>args4j</artifactId>
+ <version>2.0.16</version>
+ </dependency>
</dependencies>
</dependencyManagement>
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/edcaf71b/storm-core/pom.xml
----------------------------------------------------------------------
diff --git a/storm-core/pom.xml b/storm-core/pom.xml
index fec6218..33f6cc6 100644
--- a/storm-core/pom.xml
+++ b/storm-core/pom.xml
@@ -165,6 +165,10 @@
<groupId>io.netty</groupId>
<artifactId>netty</artifactId>
</dependency>
+ <dependency>
+ <groupId>args4j</groupId>
+ <artifactId>args4j</artifactId>
+ </dependency>
<dependency>
<groupId>org.mockito</groupId>
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/edcaf71b/storm-core/src/jvm/backtype/storm/command/Monitor.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/command/Monitor.java b/storm-core/src/jvm/backtype/storm/command/Monitor.java
new file mode 100644
index 0000000..8f39d0c
--- /dev/null
+++ b/storm-core/src/jvm/backtype/storm/command/Monitor.java
@@ -0,0 +1,197 @@
+/**
+ * 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.command;
+
+import backtype.storm.generated.*;
+import backtype.storm.utils.NimbusClient;
+import backtype.storm.utils.Utils;
+import org.kohsuke.args4j.CmdLineException;
+import org.kohsuke.args4j.CmdLineParser;
+import org.kohsuke.args4j.Option;
+
+import java.util.Map;
+
+public class Monitor {
+ @Option(name="--help", aliases={"-h"}, usage="print help message")
+ private boolean _help = false;
+
+ @Option(name="--interval", aliases={"-i"}, usage="poll frequency in seconds")
+ private int _interval = 4;
+
+ @Option(name="--name", aliases={"--topologyName"}, metaVar="NAME",
+ usage="base name of the topology (numbers may be appended to the end)")
+ private String _name;
+
+ @Option(name="--component", aliases={"--componentName"}, metaVar="NAME",
+ usage="component name of the topology")
+ private String _component;
+
+ @Option(name="--stat", aliases={"--statItem"}, metaVar="ITEM",
+ usage="stat item [emitted | transferred]")
+ private String _stat = "emitted";
+
+ private static class MetricsState {
+ long lastStatted = 0;
+ long lastTime = 0;
+ }
+
+ public void metrics(Nimbus.Client client, int poll, String name, String component, String stat) throws Exception {
+ System.out.println("status\ttopologie\tslots\tcomponent\texecutors\texecutorsWithMetrics\ttime-diff ms\t" + stat + "\tthroughput (Kt/s)");
+ MetricsState state = new MetricsState();
+ long pollMs = poll * 1000;
+ long now = System.currentTimeMillis();
+ state.lastTime = now;
+ long startTime = now;
+ long cycle, sleepTime, wakeupTime;
+
+ while (metrics(client, name, component, stat, now, state, "WAITING")) {
+ now = System.currentTimeMillis();
+ cycle = (now - startTime)/pollMs;
+ wakeupTime = startTime + (pollMs * (cycle + 1));
+ sleepTime = wakeupTime - now;
+ if (sleepTime > 0) {
+ Thread.sleep(sleepTime);
+ }
+ now = System.currentTimeMillis();
+ }
+
+ now = System.currentTimeMillis();
+ cycle = (now - startTime)/pollMs;
+ wakeupTime = startTime + (pollMs * (cycle + 1));
+ sleepTime = wakeupTime - now;
+ if (sleepTime > 0) {
+ Thread.sleep(sleepTime);
+ }
+ now = System.currentTimeMillis();
+ do {
+ metrics(client, name, component, stat, now, state, "RUNNING");
+ now = System.currentTimeMillis();
+ cycle = (now - startTime)/pollMs;
+ wakeupTime = startTime + (pollMs * (cycle + 1));
+ sleepTime = wakeupTime - now;
+ if (sleepTime > 0) {
+ Thread.sleep(sleepTime);
+ }
+ now = System.currentTimeMillis();
+ } while (true);
+ }
+
+ public boolean metrics(Nimbus.Client client, String name, String component, String stat, long now, MetricsState state, String message) throws Exception {
+ long totalStatted = 0;
+
+ boolean topologyFound = false;
+ boolean componentFound = false;
+ int slotsUsed = 0;
+ int executors = 0;
+ int executorsWithMetrics = 0;
+ ClusterSummary summary = client.getClusterInfo();
+ for (TopologySummary ts: summary.get_topologies()) {
+ if (name.equals(ts.get_name())) {
+ topologyFound = true;
+ slotsUsed = ts.get_num_workers();
+ String id = ts.get_id();
+ TopologyInfo info = client.getTopologyInfo(id);
+ for (ExecutorSummary es: info.get_executors()) {
+ if (component.equals(es.get_component_id())) {
+ componentFound = true;
+ executors ++;
+ ExecutorStats stats = es.get_stats();
+ if (stats != null) {
+ Map<String,Map<String,Long>> statted =
+ "emitted".equals(stat) ? stats.get_emitted() : stats.get_transferred();
+ if ( statted != null) {
+ Map<String, Long> e2 = statted.get(":all-time");
+ if (e2 != null) {
+ executorsWithMetrics++;
+ //topology messages are always on the default stream, so just count those
+ Long dflt = e2.get("default");
+ if (dflt != null){
+ totalStatted += dflt;
+ }
+ }
+ }
+ }
+ }
+
+
+ }
+ }
+ }
+
+ if (!topologyFound) {
+ throw new IllegalArgumentException("topology: " + name + " not found");
+ }
+ if (!componentFound) {
+ throw new IllegalArgumentException("component: " + component + " not fouond");
+ }
+ long timeDelta = now - state.lastTime;
+ long stattedDelta = totalStatted - state.lastStatted;
+ state.lastTime = now;
+ state.lastStatted = totalStatted;
+ double throughput = (stattedDelta == 0 || timeDelta == 0) ? 0.0 : ((double)stattedDelta/(double)timeDelta);
+ System.out.println(message+"\t"+name+"\t"+slotsUsed+"\t"+component+"\t"+executors+"\t"+executorsWithMetrics+"\t"+timeDelta+"\t"+stattedDelta+"\t"+throughput);
+
+ return !(executors > 0 && executorsWithMetrics >= executors);
+ }
+
+ public void realMain(String[] args) {
+ /*** parse command line ***/
+ CmdLineParser parser = new CmdLineParser(this);
+ parser.setUsageWidth(80);
+ try {
+ parser.parseArgument(args);
+ } catch( CmdLineException e ) {
+ System.err.println(e.getMessage());
+ _help = true;
+ }
+ if(_help) {
+ parser.printUsage(System.err);
+ System.err.println();
+ return;
+ }
+
+ if (_name == null || _name.isEmpty()) {
+ throw new IllegalArgumentException("topology name must be something");
+ }
+
+ if (_component == null || _component.isEmpty()) {
+ throw new IllegalArgumentException("component name must be something");
+ }
+
+ if (_interval <= 0) {
+ throw new IllegalArgumentException("poll interval must be positive");
+ }
+
+ if ( !"transferred".equals(_stat) && !"emitted".equals(_stat)) {
+ throw new IllegalArgumentException("stat item must either be transferred or emitted");
+ }
+
+ Map clusterConf = Utils.readStormConfig();
+ clusterConf.putAll(Utils.readCommandLineOpts());
+ Nimbus.Client client = NimbusClient.getConfiguredClient(clusterConf).getClient();
+ try {
+ metrics(client, _interval, _name, _component, _stat);
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+
+ public static void main(String[] args) {
+ new Monitor().realMain(args);
+ }
+}
[06/11] git commit: seperate out argument parsing in Monitor class;
remove 'waiting' status in mertics; add argument -s stream to specify stream
Posted by bo...@apache.org.
seperate out argument parsing in Monitor class; remove 'waiting' status in mertics; add argument -s stream to specify stream
Project: http://git-wip-us.apache.org/repos/asf/incubator-storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-storm/commit/81cf56ed
Tree: http://git-wip-us.apache.org/repos/asf/incubator-storm/tree/81cf56ed
Diff: http://git-wip-us.apache.org/repos/asf/incubator-storm/diff/81cf56ed
Branch: refs/heads/master
Commit: 81cf56edd053ae6e392a81577e0c4cb7a305cc36
Parents: edcaf71
Author: jiahong.ljh <ji...@alibaba-inc.com>
Authored: Mon Jun 30 18:06:38 2014 +0800
Committer: jiahong.ljh <ji...@alibaba-inc.com>
Committed: Mon Jun 30 18:06:38 2014 +0800
----------------------------------------------------------------------
bin/storm | 7 +-
pom.xml | 6 -
storm-core/pom.xml | 4 -
.../src/clj/backtype/storm/command/monitor.clj | 38 ++++
.../src/jvm/backtype/storm/command/Monitor.java | 197 -------------------
.../src/jvm/backtype/storm/utils/Monitor.java | 178 +++++++++++++++++
6 files changed, 220 insertions(+), 210 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/81cf56ed/bin/storm
----------------------------------------------------------------------
diff --git a/bin/storm b/bin/storm
index 0600d45..9ec4ef4 100755
--- a/bin/storm
+++ b/bin/storm
@@ -391,12 +391,13 @@ def print_classpath():
print get_classpath([])
def monitor(*args):
- """Syntax: [storm monitor]
+ """Syntax: [storm monitor [-i interval-secs] -t topology-name -m component-id [-s stream-id] [-w [emitted | transferred]]*]
- Monitor given topologies' given components' or type of components' throughput interactively.
+ Monitor topology's throughput interactively.
+ One can specify poll-interval, topology's name, component-id, stream-id, watch-item[emitted | transferred]
"""
exec_storm_class(
- "backtype.storm.command.Monitor",
+ "backtype.storm.command.monitor",
args=args,
jvmtype="-client",
extrajars=[USER_CONF_DIR, STORM_DIR + "/bin"])
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/81cf56ed/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 484f5f4..aa93fcc 100644
--- a/pom.xml
+++ b/pom.xml
@@ -464,12 +464,6 @@
<version>3.8.1</version>
<scope>test</scope>
</dependency>
-
- <dependency>
- <groupId>args4j</groupId>
- <artifactId>args4j</artifactId>
- <version>2.0.16</version>
- </dependency>
</dependencies>
</dependencyManagement>
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/81cf56ed/storm-core/pom.xml
----------------------------------------------------------------------
diff --git a/storm-core/pom.xml b/storm-core/pom.xml
index 33f6cc6..fec6218 100644
--- a/storm-core/pom.xml
+++ b/storm-core/pom.xml
@@ -165,10 +165,6 @@
<groupId>io.netty</groupId>
<artifactId>netty</artifactId>
</dependency>
- <dependency>
- <groupId>args4j</groupId>
- <artifactId>args4j</artifactId>
- </dependency>
<dependency>
<groupId>org.mockito</groupId>
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/81cf56ed/storm-core/src/clj/backtype/storm/command/monitor.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/backtype/storm/command/monitor.clj b/storm-core/src/clj/backtype/storm/command/monitor.clj
new file mode 100644
index 0000000..0869032
--- /dev/null
+++ b/storm-core/src/clj/backtype/storm/command/monitor.clj
@@ -0,0 +1,38 @@
+;; 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.command.monitor
+ (:use [clojure.tools.cli :only [cli]])
+ (:use [backtype.storm thrift config log])
+ (:import [backtype.storm.utils Monitor])
+ (:gen-class)
+ )
+
+(defn -main [& args]
+ (let [[{interval :interval topology :topology component :component stream :stream watch :watch} [_] _]
+ (cli args ["-i" "--interval" :default 4 :parse-fn #(Integer/parseInt %)]
+ ["-t" "--topology" :default nil]
+ ["-m" "--component" :default nil]
+ ["-s" "--stream" :default "default"]
+ ["-w" "--watch" :default "emitted"])
+ mon (Monitor.)]
+ (if interval (.set_interval mon interval))
+ (if topology (.set_topology mon topology))
+ (if component (.set_component mon component))
+ (if stream (.set_stream mon stream))
+ (if watch (.set_watch mon watch))
+ (with-configured-nimbus-connection nimbus
+ (.metrics mon nimbus)
+ )))
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/81cf56ed/storm-core/src/jvm/backtype/storm/command/Monitor.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/command/Monitor.java b/storm-core/src/jvm/backtype/storm/command/Monitor.java
deleted file mode 100644
index 8f39d0c..0000000
--- a/storm-core/src/jvm/backtype/storm/command/Monitor.java
+++ /dev/null
@@ -1,197 +0,0 @@
-/**
- * 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.command;
-
-import backtype.storm.generated.*;
-import backtype.storm.utils.NimbusClient;
-import backtype.storm.utils.Utils;
-import org.kohsuke.args4j.CmdLineException;
-import org.kohsuke.args4j.CmdLineParser;
-import org.kohsuke.args4j.Option;
-
-import java.util.Map;
-
-public class Monitor {
- @Option(name="--help", aliases={"-h"}, usage="print help message")
- private boolean _help = false;
-
- @Option(name="--interval", aliases={"-i"}, usage="poll frequency in seconds")
- private int _interval = 4;
-
- @Option(name="--name", aliases={"--topologyName"}, metaVar="NAME",
- usage="base name of the topology (numbers may be appended to the end)")
- private String _name;
-
- @Option(name="--component", aliases={"--componentName"}, metaVar="NAME",
- usage="component name of the topology")
- private String _component;
-
- @Option(name="--stat", aliases={"--statItem"}, metaVar="ITEM",
- usage="stat item [emitted | transferred]")
- private String _stat = "emitted";
-
- private static class MetricsState {
- long lastStatted = 0;
- long lastTime = 0;
- }
-
- public void metrics(Nimbus.Client client, int poll, String name, String component, String stat) throws Exception {
- System.out.println("status\ttopologie\tslots\tcomponent\texecutors\texecutorsWithMetrics\ttime-diff ms\t" + stat + "\tthroughput (Kt/s)");
- MetricsState state = new MetricsState();
- long pollMs = poll * 1000;
- long now = System.currentTimeMillis();
- state.lastTime = now;
- long startTime = now;
- long cycle, sleepTime, wakeupTime;
-
- while (metrics(client, name, component, stat, now, state, "WAITING")) {
- now = System.currentTimeMillis();
- cycle = (now - startTime)/pollMs;
- wakeupTime = startTime + (pollMs * (cycle + 1));
- sleepTime = wakeupTime - now;
- if (sleepTime > 0) {
- Thread.sleep(sleepTime);
- }
- now = System.currentTimeMillis();
- }
-
- now = System.currentTimeMillis();
- cycle = (now - startTime)/pollMs;
- wakeupTime = startTime + (pollMs * (cycle + 1));
- sleepTime = wakeupTime - now;
- if (sleepTime > 0) {
- Thread.sleep(sleepTime);
- }
- now = System.currentTimeMillis();
- do {
- metrics(client, name, component, stat, now, state, "RUNNING");
- now = System.currentTimeMillis();
- cycle = (now - startTime)/pollMs;
- wakeupTime = startTime + (pollMs * (cycle + 1));
- sleepTime = wakeupTime - now;
- if (sleepTime > 0) {
- Thread.sleep(sleepTime);
- }
- now = System.currentTimeMillis();
- } while (true);
- }
-
- public boolean metrics(Nimbus.Client client, String name, String component, String stat, long now, MetricsState state, String message) throws Exception {
- long totalStatted = 0;
-
- boolean topologyFound = false;
- boolean componentFound = false;
- int slotsUsed = 0;
- int executors = 0;
- int executorsWithMetrics = 0;
- ClusterSummary summary = client.getClusterInfo();
- for (TopologySummary ts: summary.get_topologies()) {
- if (name.equals(ts.get_name())) {
- topologyFound = true;
- slotsUsed = ts.get_num_workers();
- String id = ts.get_id();
- TopologyInfo info = client.getTopologyInfo(id);
- for (ExecutorSummary es: info.get_executors()) {
- if (component.equals(es.get_component_id())) {
- componentFound = true;
- executors ++;
- ExecutorStats stats = es.get_stats();
- if (stats != null) {
- Map<String,Map<String,Long>> statted =
- "emitted".equals(stat) ? stats.get_emitted() : stats.get_transferred();
- if ( statted != null) {
- Map<String, Long> e2 = statted.get(":all-time");
- if (e2 != null) {
- executorsWithMetrics++;
- //topology messages are always on the default stream, so just count those
- Long dflt = e2.get("default");
- if (dflt != null){
- totalStatted += dflt;
- }
- }
- }
- }
- }
-
-
- }
- }
- }
-
- if (!topologyFound) {
- throw new IllegalArgumentException("topology: " + name + " not found");
- }
- if (!componentFound) {
- throw new IllegalArgumentException("component: " + component + " not fouond");
- }
- long timeDelta = now - state.lastTime;
- long stattedDelta = totalStatted - state.lastStatted;
- state.lastTime = now;
- state.lastStatted = totalStatted;
- double throughput = (stattedDelta == 0 || timeDelta == 0) ? 0.0 : ((double)stattedDelta/(double)timeDelta);
- System.out.println(message+"\t"+name+"\t"+slotsUsed+"\t"+component+"\t"+executors+"\t"+executorsWithMetrics+"\t"+timeDelta+"\t"+stattedDelta+"\t"+throughput);
-
- return !(executors > 0 && executorsWithMetrics >= executors);
- }
-
- public void realMain(String[] args) {
- /*** parse command line ***/
- CmdLineParser parser = new CmdLineParser(this);
- parser.setUsageWidth(80);
- try {
- parser.parseArgument(args);
- } catch( CmdLineException e ) {
- System.err.println(e.getMessage());
- _help = true;
- }
- if(_help) {
- parser.printUsage(System.err);
- System.err.println();
- return;
- }
-
- if (_name == null || _name.isEmpty()) {
- throw new IllegalArgumentException("topology name must be something");
- }
-
- if (_component == null || _component.isEmpty()) {
- throw new IllegalArgumentException("component name must be something");
- }
-
- if (_interval <= 0) {
- throw new IllegalArgumentException("poll interval must be positive");
- }
-
- if ( !"transferred".equals(_stat) && !"emitted".equals(_stat)) {
- throw new IllegalArgumentException("stat item must either be transferred or emitted");
- }
-
- Map clusterConf = Utils.readStormConfig();
- clusterConf.putAll(Utils.readCommandLineOpts());
- Nimbus.Client client = NimbusClient.getConfiguredClient(clusterConf).getClient();
- try {
- metrics(client, _interval, _name, _component, _stat);
- } catch (Exception e) {
- e.printStackTrace();
- }
- }
-
- public static void main(String[] args) {
- new Monitor().realMain(args);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/81cf56ed/storm-core/src/jvm/backtype/storm/utils/Monitor.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/utils/Monitor.java b/storm-core/src/jvm/backtype/storm/utils/Monitor.java
new file mode 100644
index 0000000..c517068
--- /dev/null
+++ b/storm-core/src/jvm/backtype/storm/utils/Monitor.java
@@ -0,0 +1,178 @@
+/**
+ * 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.utils;
+
+import backtype.storm.generated.*;
+import java.util.Map;
+
+public class Monitor {
+ private static final String WATCH_TRANSFERRED = "transferred";
+ private static final String WATCH_EMITTED = "emitted";
+
+ private int _interval = 4;
+ private String _topology;
+ private String _component;
+ private String _stream;
+ private String _watch;
+
+ private static class MetricsState {
+ long lastStatted = 0;
+ long lastTime = 0;
+ }
+
+ private static class Poller {
+ long startTime = 0;
+ long pollMs = 0;
+
+ public long nextPoll() {
+ long now = System.currentTimeMillis();
+ long cycle = (now - startTime) / pollMs;
+ long wakeupTime = startTime + (pollMs * (cycle + 1));
+ long sleepTime = wakeupTime - now;
+ if (sleepTime > 0) {
+ try {
+ Thread.sleep(sleepTime);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ }
+ now = System.currentTimeMillis();
+ return now;
+ }
+ }
+
+ public void metrics(Nimbus.Client client) throws Exception {
+ if (_interval <= 0) {
+ throw new IllegalArgumentException("poll interval must be positive");
+ }
+
+ if (_topology == null || _topology.isEmpty()) {
+ throw new IllegalArgumentException("topology name must be something");
+ }
+
+ if (_component == null || _component.isEmpty()) {
+ throw new IllegalArgumentException("component name must be something");
+ }
+
+ if (_stream == null || _stream.isEmpty()) {
+ throw new IllegalArgumentException("stream name must be something");
+ }
+
+ if ( !WATCH_TRANSFERRED.equals(_watch) && !WATCH_EMITTED.equals(_watch)) {
+ throw new IllegalArgumentException("watch item must either be transferred or emitted");
+ }
+ System.out.println("topology\tslots\texecutors\texecutorsWithMetrics\tcomponent\tstream\ttime-diff ms\t" + _watch + "\tthroughput (Kt/s)");
+
+ long pollMs = _interval * 1000;
+
+ MetricsState state = new MetricsState();
+ Poller poller = new Poller();
+ long now = System.currentTimeMillis();
+ state.lastTime = now;
+ state.lastStatted = 0;
+ poller.startTime = now;
+ poller.pollMs = pollMs;
+
+ do {
+ metrics(client, now, state);
+ now = poller.nextPoll();
+ } while (true);
+ }
+
+ public void metrics(Nimbus.Client client, long now, MetricsState state) throws Exception {
+ long totalStatted = 0;
+
+ boolean topologyFound = false;
+ boolean componentFound = false;
+ boolean streamFound = false;
+ int slotsUsed = 0;
+ int executors = 0;
+ int executorsWithMetrics = 0;
+ ClusterSummary summary = client.getClusterInfo();
+ for (TopologySummary ts: summary.get_topologies()) {
+ if (_topology.equals(ts.get_name())) {
+ topologyFound = true;
+ slotsUsed = ts.get_num_workers();
+ String id = ts.get_id();
+ TopologyInfo info = client.getTopologyInfo(id);
+ for (ExecutorSummary es: info.get_executors()) {
+ if (_component.equals(es.get_component_id())) {
+ componentFound = true;
+ executors ++;
+ ExecutorStats stats = es.get_stats();
+ if (stats != null) {
+ Map<String,Map<String,Long>> statted =
+ WATCH_EMITTED.equals(_watch) ? stats.get_emitted() : stats.get_transferred();
+ if ( statted != null) {
+ Map<String, Long> e2 = statted.get(":all-time");
+ if (e2 != null) {
+ Long stream = e2.get(_stream);
+ if (stream != null){
+ streamFound = true;
+ executorsWithMetrics ++;
+ totalStatted += stream;
+ }
+ }
+ }
+ }
+ }
+
+
+ }
+ }
+ }
+
+ if (!topologyFound) {
+ throw new IllegalArgumentException("topology: " + _topology + " not found");
+ }
+
+ if (!componentFound) {
+ throw new IllegalArgumentException("component: " + _component + " not fouond");
+ }
+
+ if (!streamFound) {
+ throw new IllegalArgumentException("stream: " + _stream + " not fouond");
+ }
+ long timeDelta = now - state.lastTime;
+ long stattedDelta = totalStatted - state.lastStatted;
+ state.lastTime = now;
+ state.lastStatted = totalStatted;
+ double throughput = (stattedDelta == 0 || timeDelta == 0) ? 0.0 : ((double)stattedDelta/(double)timeDelta);
+ System.out.println(_topology+"\t"+slotsUsed+"\t"+executors+"\t"+executorsWithMetrics+"\t"+_component+"\t"+_stream+"\t"+timeDelta+"\t"+stattedDelta+"\t"+throughput);
+ }
+
+ public void set_interval(int _interval) {
+ this._interval = _interval;
+ }
+
+ public void set_topology(String _topology) {
+ this._topology = _topology;
+ }
+
+ public void set_component(String _component) {
+ this._component = _component;
+ }
+
+ public void set_stream(String _stream) {
+ this._stream = _stream;
+ }
+
+ public void set_watch(String _watch) {
+ this._watch = _watch;
+ }
+}
[07/11] git commit: make topology-name to be an argument;
make component-id optional
Posted by bo...@apache.org.
make topology-name to be an argument; make component-id optional
Project: http://git-wip-us.apache.org/repos/asf/incubator-storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-storm/commit/b68c33a3
Tree: http://git-wip-us.apache.org/repos/asf/incubator-storm/tree/b68c33a3
Diff: http://git-wip-us.apache.org/repos/asf/incubator-storm/diff/b68c33a3
Branch: refs/heads/master
Commit: b68c33a31927c45f52b61007ba155d89590bf26c
Parents: 81cf56e
Author: jiahong.ljh <ji...@alibaba-inc.com>
Authored: Wed Jul 2 15:55:21 2014 +0800
Committer: jiahong.ljh <ji...@alibaba-inc.com>
Committed: Wed Jul 2 15:55:21 2014 +0800
----------------------------------------------------------------------
bin/storm | 13 ++-
.../src/clj/backtype/storm/command/monitor.clj | 5 +-
.../src/jvm/backtype/storm/utils/Monitor.java | 113 ++++++++++++-------
3 files changed, 84 insertions(+), 47 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/b68c33a3/bin/storm
----------------------------------------------------------------------
diff --git a/bin/storm b/bin/storm
index 9ec4ef4..9d7b133 100755
--- a/bin/storm
+++ b/bin/storm
@@ -391,10 +391,15 @@ def print_classpath():
print get_classpath([])
def monitor(*args):
- """Syntax: [storm monitor [-i interval-secs] -t topology-name -m component-id [-s stream-id] [-w [emitted | transferred]]*]
-
- Monitor topology's throughput interactively.
- One can specify poll-interval, topology's name, component-id, stream-id, watch-item[emitted | transferred]
+ """Syntax: [storm monitor topology-name [-i interval-secs] [-m component-id] [-s stream-id] [-w [emitted | transferred]]]
+
+ Monitor given topology's throughput interactively.
+ One can specify poll-interval, component-id, stream-id, watch-item[emitted | transferred]
+ By default,
+ poll-interval is 4 seconds;
+ all component-ids will be list;
+ stream-id is 'default';
+ watch-item is 'emitted';
"""
exec_storm_class(
"backtype.storm.command.monitor",
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/b68c33a3/storm-core/src/clj/backtype/storm/command/monitor.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/backtype/storm/command/monitor.clj b/storm-core/src/clj/backtype/storm/command/monitor.clj
index 0869032..17498e9 100644
--- a/storm-core/src/clj/backtype/storm/command/monitor.clj
+++ b/storm-core/src/clj/backtype/storm/command/monitor.clj
@@ -21,15 +21,14 @@
)
(defn -main [& args]
- (let [[{interval :interval topology :topology component :component stream :stream watch :watch} [_] _]
+ (let [[{interval :interval component :component stream :stream watch :watch} [name] _]
(cli args ["-i" "--interval" :default 4 :parse-fn #(Integer/parseInt %)]
- ["-t" "--topology" :default nil]
["-m" "--component" :default nil]
["-s" "--stream" :default "default"]
["-w" "--watch" :default "emitted"])
mon (Monitor.)]
(if interval (.set_interval mon interval))
- (if topology (.set_topology mon topology))
+ (if name (.set_topology mon name))
(if component (.set_component mon component))
(if stream (.set_stream mon stream))
(if watch (.set_watch mon watch))
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/b68c33a3/storm-core/src/jvm/backtype/storm/utils/Monitor.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/utils/Monitor.java b/storm-core/src/jvm/backtype/storm/utils/Monitor.java
index c517068..0c87d9b 100644
--- a/storm-core/src/jvm/backtype/storm/utils/Monitor.java
+++ b/storm-core/src/jvm/backtype/storm/utils/Monitor.java
@@ -18,6 +18,9 @@
package backtype.storm.utils;
import backtype.storm.generated.*;
+
+import java.util.HashSet;
+import java.util.List;
import java.util.Map;
public class Monitor {
@@ -56,6 +59,28 @@ public class Monitor {
}
}
+ private HashSet<String> getComponents(Nimbus.Client client, String topology) throws Exception{
+ HashSet<String> components = new HashSet<String>();
+ ClusterSummary clusterSummary = client.getClusterInfo();
+ TopologySummary topologySummary = null;
+ for (TopologySummary ts: clusterSummary.get_topologies()) {
+ if (topology.equals(ts.get_name())) {
+ topologySummary = ts;
+ break;
+ }
+ }
+ if (topologySummary == null) {
+ throw new IllegalArgumentException("topology: " + topology + " not found");
+ } else {
+ String id = topologySummary.get_id();
+ TopologyInfo info = client.getTopologyInfo(id);
+ for (ExecutorSummary es: info.get_executors()) {
+ components.add(es.get_component_id());
+ }
+ }
+ return components;
+ }
+
public void metrics(Nimbus.Client client) throws Exception {
if (_interval <= 0) {
throw new IllegalArgumentException("poll interval must be positive");
@@ -66,7 +91,15 @@ public class Monitor {
}
if (_component == null || _component.isEmpty()) {
- throw new IllegalArgumentException("component name must be something");
+ HashSet<String> components = getComponents(client, _topology);
+ System.out.println("Available components for " + _topology + " :");
+ System.out.println("------------------");
+ for (String comp : components) {
+ System.out.println(comp);
+ }
+ System.out.println("------------------");
+ System.out.println("Please use -m to specify one component");
+ return;
}
if (_stream == null || _stream.isEmpty()) {
@@ -76,7 +109,7 @@ public class Monitor {
if ( !WATCH_TRANSFERRED.equals(_watch) && !WATCH_EMITTED.equals(_watch)) {
throw new IllegalArgumentException("watch item must either be transferred or emitted");
}
- System.out.println("topology\tslots\texecutors\texecutorsWithMetrics\tcomponent\tstream\ttime-diff ms\t" + _watch + "\tthroughput (Kt/s)");
+ System.out.println("topology\tcomponent\tparallelism\tstream\ttime-diff ms\t" + _watch + "\tthroughput (Kt/s)");
long pollMs = _interval * 1000;
@@ -97,63 +130,63 @@ public class Monitor {
public void metrics(Nimbus.Client client, long now, MetricsState state) throws Exception {
long totalStatted = 0;
- boolean topologyFound = false;
- boolean componentFound = false;
+ int componentParallelism = 0;
boolean streamFound = false;
- int slotsUsed = 0;
- int executors = 0;
- int executorsWithMetrics = 0;
- ClusterSummary summary = client.getClusterInfo();
- for (TopologySummary ts: summary.get_topologies()) {
+ ClusterSummary clusterSummary = client.getClusterInfo();
+ TopologySummary topologySummary = null;
+ for (TopologySummary ts: clusterSummary.get_topologies()) {
if (_topology.equals(ts.get_name())) {
- topologyFound = true;
- slotsUsed = ts.get_num_workers();
- String id = ts.get_id();
- TopologyInfo info = client.getTopologyInfo(id);
- for (ExecutorSummary es: info.get_executors()) {
- if (_component.equals(es.get_component_id())) {
- componentFound = true;
- executors ++;
- ExecutorStats stats = es.get_stats();
- if (stats != null) {
- Map<String,Map<String,Long>> statted =
- WATCH_EMITTED.equals(_watch) ? stats.get_emitted() : stats.get_transferred();
- if ( statted != null) {
- Map<String, Long> e2 = statted.get(":all-time");
- if (e2 != null) {
- Long stream = e2.get(_stream);
- if (stream != null){
- streamFound = true;
- executorsWithMetrics ++;
- totalStatted += stream;
- }
+ topologySummary = ts;
+ break;
+ }
+ }
+ if (topologySummary == null) {
+ throw new IllegalArgumentException("topology: " + _topology + " not found");
+ } else {
+ String id = topologySummary.get_id();
+ TopologyInfo info = client.getTopologyInfo(id);
+ for (ExecutorSummary es: info.get_executors()) {
+ if (_component.equals(es.get_component_id())) {
+ componentParallelism ++;
+ ExecutorStats stats = es.get_stats();
+ if (stats != null) {
+ Map<String,Map<String,Long>> statted =
+ WATCH_EMITTED.equals(_watch) ? stats.get_emitted() : stats.get_transferred();
+ if ( statted != null) {
+ Map<String, Long> e2 = statted.get(":all-time");
+ if (e2 != null) {
+ Long stream = e2.get(_stream);
+ if (stream != null){
+ streamFound = true;
+ totalStatted += stream;
}
}
}
}
-
-
}
}
}
- if (!topologyFound) {
- throw new IllegalArgumentException("topology: " + _topology + " not found");
- }
-
- if (!componentFound) {
- throw new IllegalArgumentException("component: " + _component + " not fouond");
+ if (componentParallelism <= 0) {
+ HashSet<String> components = getComponents(client, _topology);
+ System.out.println("Available components for " + _topology + " :");
+ System.out.println("------------------");
+ for (String comp : components) {
+ System.out.println(comp);
+ }
+ System.out.println("------------------");
+ throw new IllegalArgumentException("component: " + _component + " not found");
}
if (!streamFound) {
- throw new IllegalArgumentException("stream: " + _stream + " not fouond");
+ throw new IllegalArgumentException("stream: " + _stream + " not found");
}
long timeDelta = now - state.lastTime;
long stattedDelta = totalStatted - state.lastStatted;
state.lastTime = now;
state.lastStatted = totalStatted;
double throughput = (stattedDelta == 0 || timeDelta == 0) ? 0.0 : ((double)stattedDelta/(double)timeDelta);
- System.out.println(_topology+"\t"+slotsUsed+"\t"+executors+"\t"+executorsWithMetrics+"\t"+_component+"\t"+_stream+"\t"+timeDelta+"\t"+stattedDelta+"\t"+throughput);
+ System.out.println(_topology+"\t"+_component+"\t"+componentParallelism+"\t"+_stream+"\t"+timeDelta+"\t"+stattedDelta+"\t"+throughput);
}
public void set_interval(int _interval) {
[04/11] git commit: Merge pull request #4 from apache/master
Posted by bo...@apache.org.
Merge pull request #4 from apache/master
update from apache/incubator-storm
Project: http://git-wip-us.apache.org/repos/asf/incubator-storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-storm/commit/a45b53a1
Tree: http://git-wip-us.apache.org/repos/asf/incubator-storm/tree/a45b53a1
Diff: http://git-wip-us.apache.org/repos/asf/incubator-storm/diff/a45b53a1
Branch: refs/heads/master
Commit: a45b53a167cd48a275bbb21279bcaa9e9437b97b
Parents: b2a5102 5b4bb31
Author: Li Jiahong <Gv...@users.noreply.github.com>
Authored: Wed May 7 20:09:34 2014 +0800
Committer: Li Jiahong <Gv...@users.noreply.github.com>
Committed: Wed May 7 20:09:34 2014 +0800
----------------------------------------------------------------------
CHANGELOG.md | 19 ++
README.markdown | 2 +-
bin/storm | 3 +-
bin/storm-config.cmd | 11 +-
bin/storm.cmd | 9 +-
external/storm-kafka/CHANGELOG.md | 13 +
external/storm-kafka/README.md | 25 ++
external/storm-kafka/pom.xml | 138 ++++++++++
.../storm-kafka/src/jvm/storm/kafka/Broker.java | 80 ++++++
.../src/jvm/storm/kafka/BrokerHosts.java | 25 ++
.../jvm/storm/kafka/DynamicBrokersReader.java | 145 ++++++++++
.../kafka/DynamicPartitionConnections.java | 94 +++++++
.../jvm/storm/kafka/FailedFetchException.java | 29 ++
.../src/jvm/storm/kafka/KafkaConfig.java | 50 ++++
.../src/jvm/storm/kafka/KafkaError.java | 43 +++
.../src/jvm/storm/kafka/KafkaSpout.java | 190 +++++++++++++
.../src/jvm/storm/kafka/KafkaUtils.java | 235 ++++++++++++++++
.../src/jvm/storm/kafka/KeyValueScheme.java | 28 ++
.../kafka/KeyValueSchemeAsMultiScheme.java | 36 +++
.../src/jvm/storm/kafka/Partition.java | 64 +++++
.../jvm/storm/kafka/PartitionCoordinator.java | 26 ++
.../src/jvm/storm/kafka/PartitionManager.java | 241 +++++++++++++++++
.../src/jvm/storm/kafka/SpoutConfig.java | 36 +++
.../src/jvm/storm/kafka/StaticCoordinator.java | 48 ++++
.../src/jvm/storm/kafka/StaticHosts.java | 38 +++
.../storm/kafka/StaticPartitionConnections.java | 52 ++++
.../jvm/storm/kafka/StringKeyValueScheme.java | 37 +++
.../src/jvm/storm/kafka/StringScheme.java | 46 ++++
.../src/jvm/storm/kafka/ZkCoordinator.java | 112 ++++++++
.../src/jvm/storm/kafka/ZkHosts.java | 36 +++
.../src/jvm/storm/kafka/ZkState.java | 116 ++++++++
.../src/jvm/storm/kafka/bolt/KafkaBolt.java | 89 ++++++
.../jvm/storm/kafka/trident/Coordinator.java | 50 ++++
.../storm/kafka/trident/DefaultCoordinator.java | 31 +++
.../trident/GlobalPartitionInformation.java | 99 +++++++
.../storm/kafka/trident/IBatchCoordinator.java | 26 ++
.../jvm/storm/kafka/trident/IBrokerReader.java | 25 ++
.../src/jvm/storm/kafka/trident/MaxMetric.java | 40 +++
.../kafka/trident/OpaqueTridentKafkaSpout.java | 59 ++++
.../storm/kafka/trident/StaticBrokerReader.java | 36 +++
.../trident/TransactionalTridentKafkaSpout.java | 58 ++++
.../storm/kafka/trident/TridentKafkaConfig.java | 37 +++
.../kafka/trident/TridentKafkaEmitter.java | 269 +++++++++++++++++++
.../jvm/storm/kafka/trident/ZkBrokerReader.java | 62 +++++
.../storm/kafka/DynamicBrokersReaderTest.java | 155 +++++++++++
.../src/test/storm/kafka/KafkaErrorTest.java | 39 +++
.../src/test/storm/kafka/KafkaTestBroker.java | 75 ++++++
.../src/test/storm/kafka/KafkaUtilsTest.java | 225 ++++++++++++++++
.../storm/kafka/StringKeyValueSchemeTest.java | 38 +++
.../src/test/storm/kafka/TestUtils.java | 20 ++
.../src/test/storm/kafka/ZkCoordinatorTest.java | 130 +++++++++
.../test/storm/kafka/bolt/KafkaBoltTest.java | 171 ++++++++++++
pom.xml | 18 +-
storm-core/pom.xml | 42 ++-
storm-core/src/clj/backtype/storm/config.clj | 2 +-
.../src/clj/backtype/storm/daemon/common.clj | 6 +-
.../src/clj/backtype/storm/daemon/logviewer.clj | 29 +-
.../clj/backtype/storm/daemon/supervisor.clj | 48 ++--
.../src/clj/backtype/storm/daemon/worker.clj | 2 +-
storm-core/src/clj/backtype/storm/ui/core.clj | 16 +-
.../src/clj/backtype/storm/ui/helpers.clj | 4 +-
storm-core/src/clj/backtype/storm/util.clj | 9 +-
storm-core/src/clj/backtype/storm/zookeeper.clj | 6 +-
storm-core/src/jvm/backtype/storm/Config.java | 5 +-
.../jvm/backtype/storm/ConfigValidation.java | 17 ++
.../src/jvm/backtype/storm/StormSubmitter.java | 20 +-
.../backtype/storm/task/TopologyContext.java | 6 +-
.../transactional/state/TransactionalState.java | 2 +-
.../src/jvm/backtype/storm/utils/Utils.java | 36 ++-
.../trident/spout/RichSpoutBatchTriggerer.java | 1 +
.../topology/state/TransactionalState.java | 2 +-
.../jvm/storm/trident/util/TridentUtils.java | 14 +-
.../test/clj/backtype/storm/config_test.clj | 20 ++
.../test/clj/backtype/storm/supervisor_test.clj | 64 ++++-
.../test/clj/backtype/storm/utils_test.clj | 2 +-
storm-dist/binary/pom.xml | 40 +--
storm-dist/binary/src/main/assembly/binary.xml | 19 ++
storm-dist/source/pom.xml | 38 +--
78 files changed, 4088 insertions(+), 141 deletions(-)
----------------------------------------------------------------------
[08/11] git commit: rm unused :use class;
better handling InterruptedException;
change MetricState and Poller better fits for Java
Posted by bo...@apache.org.
rm unused :use class; better handling InterruptedException; change MetricState and Poller better fits for Java
Project: http://git-wip-us.apache.org/repos/asf/incubator-storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-storm/commit/e6b6e3ca
Tree: http://git-wip-us.apache.org/repos/asf/incubator-storm/tree/e6b6e3ca
Diff: http://git-wip-us.apache.org/repos/asf/incubator-storm/diff/e6b6e3ca
Branch: refs/heads/master
Commit: e6b6e3ca049f89222f2aad96e324aa47409383f1
Parents: b68c33a
Author: jiahong.ljh <ji...@alibaba-inc.com>
Authored: Tue Jul 8 11:56:53 2014 +0800
Committer: jiahong.ljh <ji...@alibaba-inc.com>
Committed: Tue Jul 8 11:56:53 2014 +0800
----------------------------------------------------------------------
.../src/clj/backtype/storm/command/monitor.clj | 2 +-
.../src/jvm/backtype/storm/utils/Monitor.java | 82 ++++++++++++++------
2 files changed, 61 insertions(+), 23 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/e6b6e3ca/storm-core/src/clj/backtype/storm/command/monitor.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/backtype/storm/command/monitor.clj b/storm-core/src/clj/backtype/storm/command/monitor.clj
index 17498e9..36ccbc9 100644
--- a/storm-core/src/clj/backtype/storm/command/monitor.clj
+++ b/storm-core/src/clj/backtype/storm/command/monitor.clj
@@ -15,7 +15,7 @@
;; limitations under the License.
(ns backtype.storm.command.monitor
(:use [clojure.tools.cli :only [cli]])
- (:use [backtype.storm thrift config log])
+ (:use [backtype.storm.thrift :only [with-configured-nimbus-connection]])
(:import [backtype.storm.utils Monitor])
(:gen-class)
)
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/e6b6e3ca/storm-core/src/jvm/backtype/storm/utils/Monitor.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/utils/Monitor.java b/storm-core/src/jvm/backtype/storm/utils/Monitor.java
index 0c87d9b..43ddcf8 100644
--- a/storm-core/src/jvm/backtype/storm/utils/Monitor.java
+++ b/storm-core/src/jvm/backtype/storm/utils/Monitor.java
@@ -34,29 +34,67 @@ public class Monitor {
private String _watch;
private static class MetricsState {
- long lastStatted = 0;
- long lastTime = 0;
+ private long lastTime = 0;
+ private long lastStatted = 0;
+
+ private MetricsState(long lastTime, long lastStatted) {
+ this.lastTime = lastTime;
+ this.lastStatted = lastStatted;
+ }
+
+ public long getLastStatted() {
+ return lastStatted;
+ }
+
+ public void setLastStatted(long lastStatted) {
+ this.lastStatted = lastStatted;
+ }
+
+ public long getLastTime() {
+ return lastTime;
+ }
+
+ public void setLastTime(long lastTime) {
+ this.lastTime = lastTime;
+ }
}
private static class Poller {
- long startTime = 0;
- long pollMs = 0;
+ private long startTime = 0;
+ private long pollMs = 0;
+
+ private Poller(long startTime, long pollMs) {
+ this.startTime = startTime;
+ this.pollMs = pollMs;
+ }
- public long nextPoll() {
+ public long nextPoll() throws InterruptedException {
long now = System.currentTimeMillis();
long cycle = (now - startTime) / pollMs;
long wakeupTime = startTime + (pollMs * (cycle + 1));
long sleepTime = wakeupTime - now;
if (sleepTime > 0) {
- try {
- Thread.sleep(sleepTime);
- } catch (InterruptedException e) {
- e.printStackTrace();
- }
+ Thread.sleep(sleepTime);
}
now = System.currentTimeMillis();
return now;
}
+
+ public long getStartTime() {
+ return startTime;
+ }
+
+ public void setStartTime(long startTime) {
+ this.startTime = startTime;
+ }
+
+ public long getPollMs() {
+ return pollMs;
+ }
+
+ public void setPollMs(long pollMs) {
+ this.pollMs = pollMs;
+ }
}
private HashSet<String> getComponents(Nimbus.Client client, String topology) throws Exception{
@@ -112,18 +150,18 @@ public class Monitor {
System.out.println("topology\tcomponent\tparallelism\tstream\ttime-diff ms\t" + _watch + "\tthroughput (Kt/s)");
long pollMs = _interval * 1000;
-
- MetricsState state = new MetricsState();
- Poller poller = new Poller();
long now = System.currentTimeMillis();
- state.lastTime = now;
- state.lastStatted = 0;
- poller.startTime = now;
- poller.pollMs = pollMs;
+ MetricsState state = new MetricsState(now, 0);
+ Poller poller = new Poller(now, pollMs);
do {
metrics(client, now, state);
- now = poller.nextPoll();
+ try {
+ now = poller.nextPoll();
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ break;
+ }
} while (true);
}
@@ -181,10 +219,10 @@ public class Monitor {
if (!streamFound) {
throw new IllegalArgumentException("stream: " + _stream + " not found");
}
- long timeDelta = now - state.lastTime;
- long stattedDelta = totalStatted - state.lastStatted;
- state.lastTime = now;
- state.lastStatted = totalStatted;
+ long timeDelta = now - state.getLastTime();
+ long stattedDelta = totalStatted - state.getLastStatted();
+ state.setLastTime(now);
+ state.setLastStatted(totalStatted);
double throughput = (stattedDelta == 0 || timeDelta == 0) ? 0.0 : ((double)stattedDelta/(double)timeDelta);
System.out.println(_topology+"\t"+_component+"\t"+componentParallelism+"\t"+_stream+"\t"+timeDelta+"\t"+stattedDelta+"\t"+throughput);
}
[11/11] git commit: Merge branch 'STORM-312'
Posted by bo...@apache.org.
Merge branch 'STORM-312'
Conflicts:
CHANGELOG.md
Project: http://git-wip-us.apache.org/repos/asf/incubator-storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-storm/commit/69b4601f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-storm/tree/69b4601f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-storm/diff/69b4601f
Branch: refs/heads/master
Commit: 69b4601f71bad51605813fca10a18ccb82859dd2
Parents: 96e81c1 ae383e3
Author: Robert (Bobby) Evans <bo...@apache.org>
Authored: Wed Jul 16 09:20:20 2014 -0500
Committer: Robert (Bobby) Evans <bo...@apache.org>
Committed: Wed Jul 16 09:20:20 2014 -0500
----------------------------------------------------------------------
CHANGELOG.md | 1 +
README.markdown | 1 +
bin/storm | 20 +-
.../src/clj/backtype/storm/command/monitor.clj | 37 +++
.../src/jvm/backtype/storm/utils/Monitor.java | 249 +++++++++++++++++++
5 files changed, 307 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/69b4601f/CHANGELOG.md
----------------------------------------------------------------------
diff --cc CHANGELOG.md
index f198eea,efd21c3..2817e5d
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@@ -17,7 -17,7 +17,8 @@@
* STORM-351: multilang python process fall into endless loop
* STORM-375: Smarter downloading of assignments by supervisors and workers
* STORM-328: More restrictive Config checks, strict range check within Utils.getInt()
+ * STORM-381: Replace broken jquery.tablesorter.min.js to latest
+ * STORM-312: add storm monitor tools to monitor throughtput interactively
## 0.9.2-incubating
* STORM-66: send taskid on initial handshake
[02/11] git commit: Merge pull request #2 from apache/master
Posted by bo...@apache.org.
Merge pull request #2 from apache/master
update from apache/incubator-storm
Project: http://git-wip-us.apache.org/repos/asf/incubator-storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-storm/commit/33715a2b
Tree: http://git-wip-us.apache.org/repos/asf/incubator-storm/tree/33715a2b
Diff: http://git-wip-us.apache.org/repos/asf/incubator-storm/diff/33715a2b
Branch: refs/heads/master
Commit: 33715a2b4613742dfb3294a097874312d4e6c3b6
Parents: d080e04 c621a6c
Author: Li Jiahong <Gv...@users.noreply.github.com>
Authored: Thu Apr 3 15:46:40 2014 +0800
Committer: Li Jiahong <Gv...@users.noreply.github.com>
Committed: Thu Apr 3 15:46:40 2014 +0800
----------------------------------------------------------------------
CHANGELOG.md | 5 +
NOTICE | 6 +-
README.markdown | 5 +-
examples/storm-starter/README.markdown | 118 ++++++
.../multilang/resources/splitsentence.py | 24 ++
.../multilang/resources/splitsentence.rb | 26 ++
.../storm-starter/multilang/resources/storm.py | 221 +++++++++++
.../storm-starter/multilang/resources/storm.rb | 200 ++++++++++
examples/storm-starter/pom.xml | 161 ++++++++
.../src/clj/storm/starter/clj/word_count.clj | 95 +++++
.../jvm/storm/starter/BasicDRPCTopology.java | 78 ++++
.../jvm/storm/starter/ExclamationTopology.java | 87 +++++
.../src/jvm/storm/starter/ManualDRPC.java | 68 ++++
.../jvm/storm/starter/PrintSampleStream.java | 54 +++
.../src/jvm/storm/starter/ReachTopology.java | 196 ++++++++++
.../src/jvm/storm/starter/RollingTopWords.java | 78 ++++
.../jvm/storm/starter/SingleJoinExample.java | 64 ++++
.../storm/starter/TransactionalGlobalCount.java | 173 +++++++++
.../jvm/storm/starter/TransactionalWords.java | 246 +++++++++++++
.../jvm/storm/starter/WordCountTopology.java | 107 ++++++
.../storm/starter/bolt/AbstractRankerBolt.java | 110 ++++++
.../starter/bolt/IntermediateRankingsBolt.java | 58 +++
.../src/jvm/storm/starter/bolt/PrinterBolt.java | 37 ++
.../storm/starter/bolt/RollingCountBolt.java | 142 +++++++
.../jvm/storm/starter/bolt/SingleJoinBolt.java | 114 ++++++
.../storm/starter/bolt/TotalRankingsBolt.java | 59 +++
.../starter/spout/RandomSentenceSpout.java | 64 ++++
.../storm/starter/spout/TwitterSampleSpout.java | 122 ++++++
.../tools/NthLastModifiedTimeTracker.java | 70 ++++
.../src/jvm/storm/starter/tools/Rankable.java | 32 ++
.../starter/tools/RankableObjectWithFields.java | 148 ++++++++
.../src/jvm/storm/starter/tools/Rankings.java | 156 ++++++++
.../starter/tools/SlidingWindowCounter.java | 119 ++++++
.../storm/starter/tools/SlotBasedCounter.java | 118 ++++++
.../jvm/storm/starter/trident/TridentReach.java | 156 ++++++++
.../storm/starter/trident/TridentWordCount.java | 85 +++++
.../src/jvm/storm/starter/util/StormRunner.java | 39 ++
.../jvm/storm/starter/util/TupleHelpers.java | 33 ++
.../bolt/IntermediateRankingsBoltTest.java | 146 ++++++++
.../starter/bolt/RollingCountBoltTest.java | 113 ++++++
.../starter/bolt/TotalRankingsBoltTest.java | 147 ++++++++
.../storm/starter/tools/MockTupleHelpers.java | 40 ++
.../tools/NthLastModifiedTimeTrackerTest.java | 125 +++++++
.../tools/RankableObjectWithFieldsTest.java | 252 +++++++++++++
.../jvm/storm/starter/tools/RankingsTest.java | 368 +++++++++++++++++++
.../starter/tools/SlidingWindowCounterTest.java | 106 ++++++
.../starter/tools/SlotBasedCounterTest.java | 181 +++++++++
pom.xml | 45 ++-
storm-core/pom.xml | 3 +
storm-core/src/clj/backtype/storm/timer.clj | 2 +-
storm-core/src/clj/backtype/storm/util.clj | 3 +
storm-core/src/clj/backtype/storm/zookeeper.clj | 4 +-
.../src/jvm/backtype/storm/StormSubmitter.java | 4 +-
.../backtype/storm/messaging/netty/Client.java | 11 +-
.../src/jvm/backtype/storm/utils/Utils.java | 26 +-
.../jvm/storm/trident/state/map/OpaqueMap.java | 6 +-
.../trident/state/map/RemovableMapState.java | 8 +
.../storm/trident/testing/MemoryMapState.java | 27 +-
.../test/clj/backtype/storm/utils_test.clj | 11 +
.../test/clj/storm/trident/state_test.clj | 33 +-
storm-dist/binary/NOTICE | 9 +-
storm-dist/binary/src/main/assembly/binary.xml | 12 +
62 files changed, 5313 insertions(+), 43 deletions(-)
----------------------------------------------------------------------
[03/11] git commit: Merge pull request #3 from apache/master
Posted by bo...@apache.org.
Merge pull request #3 from apache/master
update from apache/incubator-storm
Project: http://git-wip-us.apache.org/repos/asf/incubator-storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-storm/commit/b2a51029
Tree: http://git-wip-us.apache.org/repos/asf/incubator-storm/tree/b2a51029
Diff: http://git-wip-us.apache.org/repos/asf/incubator-storm/diff/b2a51029
Branch: refs/heads/master
Commit: b2a5102917ebb4b32d42b9b722427500f939e8b5
Parents: 33715a2 1a0b46e
Author: Li Jiahong <Gv...@users.noreply.github.com>
Authored: Tue Apr 15 16:05:55 2014 +0800
Committer: Li Jiahong <Gv...@users.noreply.github.com>
Committed: Tue Apr 15 16:05:55 2014 +0800
----------------------------------------------------------------------
CHANGELOG.md | 10 +
bin/storm | 19 +-
examples/storm-starter/README.markdown | 14 +-
pom.xml | 6 +-
storm-core/pom.xml | 4 +
storm-core/src/clj/backtype/storm/ui/core.clj | 25 ++-
storm-core/src/clj/backtype/storm/util.clj | 2 +-
.../backtype/storm/messaging/netty/Client.java | 184 +++++++++++++------
.../backtype/storm/messaging/netty/Context.java | 30 ++-
.../storm/messaging/netty/MessageBatch.java | 55 +-----
.../backtype/storm/messaging/netty/Server.java | 2 +-
.../messaging/netty/StormClientHandler.java | 80 +++-----
.../src/jvm/backtype/storm/utils/Time.java | 18 +-
.../src/jvm/backtype/storm/utils/Utils.java | 9 +-
14 files changed, 249 insertions(+), 209 deletions(-)
----------------------------------------------------------------------
[09/11] git commit: Merge branch 'storm-monitor-cmd' of
https://github.com/Gvain/incubator-storm into STORM-312
Posted by bo...@apache.org.
Merge branch 'storm-monitor-cmd' of https://github.com/Gvain/incubator-storm into STORM-312
STORM-312: add storm monitor tools to monitor throughtput interactively
Project: http://git-wip-us.apache.org/repos/asf/incubator-storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-storm/commit/bbb3e749
Tree: http://git-wip-us.apache.org/repos/asf/incubator-storm/tree/bbb3e749
Diff: http://git-wip-us.apache.org/repos/asf/incubator-storm/diff/bbb3e749
Branch: refs/heads/master
Commit: bbb3e749ea3a5ccb5d04916656bba670477fa696
Parents: e5226db e6b6e3c
Author: Robert (Bobby) Evans <bo...@apache.org>
Authored: Tue Jul 15 12:52:29 2014 -0500
Committer: Robert (Bobby) Evans <bo...@apache.org>
Committed: Tue Jul 15 12:52:29 2014 -0500
----------------------------------------------------------------------
bin/storm | 20 +-
.../src/clj/backtype/storm/command/monitor.clj | 37 +++
.../src/jvm/backtype/storm/utils/Monitor.java | 249 +++++++++++++++++++
3 files changed, 305 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/bbb3e749/bin/storm
----------------------------------------------------------------------
diff --cc bin/storm
index 2cdf2be,9d7b133..a4aadb1
--- a/bin/storm
+++ b/bin/storm
@@@ -422,14 -388,32 +422,32 @@@ def print_classpath()
Prints the classpath used by the storm client when running commands.
"""
- print get_classpath([])
+ print(get_classpath([]))
+ def monitor(*args):
+ """Syntax: [storm monitor topology-name [-i interval-secs] [-m component-id] [-s stream-id] [-w [emitted | transferred]]]
+
+ Monitor given topology's throughput interactively.
+ One can specify poll-interval, component-id, stream-id, watch-item[emitted | transferred]
+ By default,
+ poll-interval is 4 seconds;
+ all component-ids will be list;
+ stream-id is 'default';
+ watch-item is 'emitted';
+ """
+ exec_storm_class(
+ "backtype.storm.command.monitor",
+ args=args,
+ jvmtype="-client",
+ extrajars=[USER_CONF_DIR, STORM_DIR + "/bin"])
+
+
def print_commands():
"""Print all client commands and link to documentation"""
- print "Commands:\n\t", "\n\t".join(sorted(COMMANDS.keys()))
- print "\nHelp:", "\n\thelp", "\n\thelp <command>"
- print "\nDocumentation for the storm client can be found at http://storm.incubator.apache.org/documentation/Command-line-client.html\n"
- print "Configs can be overridden using one or more -c flags, e.g. \"storm list -c nimbus.host=nimbus.mycompany.com\"\n"
+ print("Commands:\n\t" + "\n\t".join(sorted(COMMANDS.keys())))
+ print("\nHelp: \n\thelp \n\thelp <command>")
+ print("\nDocumentation for the storm client can be found at http://storm.incubator.apache.org/documentation/Command-line-client.html\n")
+ print("Configs can be overridden using one or more -c flags, e.g. \"storm list -c nimbus.host=nimbus.mycompany.com\"\n")
def print_usage(command=None):
"""Print one help message or list of available commands"""
[10/11] git commit: Added STORM-312 to changelog and readme.
Posted by bo...@apache.org.
Added STORM-312 to changelog and readme.
Project: http://git-wip-us.apache.org/repos/asf/incubator-storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-storm/commit/ae383e37
Tree: http://git-wip-us.apache.org/repos/asf/incubator-storm/tree/ae383e37
Diff: http://git-wip-us.apache.org/repos/asf/incubator-storm/diff/ae383e37
Branch: refs/heads/master
Commit: ae383e37e2a57c4dcc9a11c17192e3ff44429376
Parents: bbb3e74
Author: Robert (Bobby) Evans <bo...@apache.org>
Authored: Tue Jul 15 12:55:09 2014 -0500
Committer: Robert (Bobby) Evans <bo...@apache.org>
Committed: Tue Jul 15 12:55:09 2014 -0500
----------------------------------------------------------------------
CHANGELOG.md | 1 +
README.markdown | 1 +
2 files changed, 2 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/ae383e37/CHANGELOG.md
----------------------------------------------------------------------
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 6a77bdf..efd21c3 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -17,6 +17,7 @@
* STORM-351: multilang python process fall into endless loop
* STORM-375: Smarter downloading of assignments by supervisors and workers
* STORM-328: More restrictive Config checks, strict range check within Utils.getInt()
+ * STORM-312: add storm monitor tools to monitor throughtput interactively
## 0.9.2-incubating
* STORM-66: send taskid on initial handshake
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/ae383e37/README.markdown
----------------------------------------------------------------------
diff --git a/README.markdown b/README.markdown
index 25cc9ae..03b4ccd 100644
--- a/README.markdown
+++ b/README.markdown
@@ -150,6 +150,7 @@ under the License.
* Adrian Petrescu ([@apetresc](https://github.com/apetresc))
* DashengJu ([@dashengju](https://github.com/dashengju))
* Jungtaek Lim ([@HeartSaVioR](https://github.com/HeartSaVioR))
+* Li Jiahong ([@Gvain](https://github.com/Gvain))
## Acknowledgements