| /** |
| * 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 org.apache.storm.utils; |
| |
| import org.apache.storm.generated.*; |
| |
| import java.util.HashSet; |
| 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 { |
| 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 { |
| private long startTime = 0; |
| private long pollMs = 0; |
| |
| private Poller(long startTime, long pollMs) { |
| this.startTime = startTime; |
| this.pollMs = pollMs; |
| } |
| |
| 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) { |
| 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{ |
| HashSet<String> components = new HashSet<>(); |
| 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(); |
| GetInfoOptions getInfoOpts = new GetInfoOptions(); |
| getInfoOpts.set_num_err_choice(NumErrorsChoice.NONE); |
| TopologyInfo info = client.getTopologyInfoWithOpts(id, getInfoOpts); |
| 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"); |
| } |
| |
| if (_topology == null || _topology.isEmpty()) { |
| throw new IllegalArgumentException("topology name must be something"); |
| } |
| |
| if (_component == null || _component.isEmpty()) { |
| 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()) { |
| 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\tcomponent\tparallelism\tstream\ttime-diff ms\t" + _watch + "\tthroughput (Kt/s)"); |
| |
| long pollMs = _interval * 1000; |
| long now = System.currentTimeMillis(); |
| MetricsState state = new MetricsState(now, 0); |
| Poller poller = new Poller(now, pollMs); |
| |
| do { |
| metrics(client, now, state); |
| try { |
| now = poller.nextPoll(); |
| } catch (InterruptedException e) { |
| e.printStackTrace(); |
| break; |
| } |
| } while (true); |
| } |
| |
| public void metrics(Nimbus.Client client, long now, MetricsState state) throws Exception { |
| long totalStatted = 0; |
| |
| int componentParallelism = 0; |
| boolean streamFound = false; |
| 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(); |
| GetInfoOptions getInfoOpts = new GetInfoOptions(); |
| getInfoOpts.set_num_err_choice(NumErrorsChoice.NONE); |
| TopologyInfo info = client.getTopologyInfoWithOpts(id, getInfoOpts); |
| 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 (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 found"); |
| } |
| 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); |
| } |
| |
| 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; |
| } |
| } |