You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streams.apache.org by sb...@apache.org on 2014/11/13 06:09:29 UTC
[02/43] incubator-streams git commit: STREAMS-210 | First run at
aggregating all data stored in relevant JMX beans
STREAMS-210 | First run at aggregating all data stored in relevant JMX beans
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/d48713fa
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/d48713fa
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/d48713fa
Branch: refs/heads/master
Commit: d48713fa5d3a90519beadc1c56131297254dcc28
Parents: ebfe73d 79eadea
Author: Robert Douglas <rd...@w2ogroup.com>
Authored: Thu Nov 6 18:06:45 2014 -0600
Committer: Robert Douglas <rd...@w2ogroup.com>
Committed: Thu Nov 6 18:06:45 2014 -0600
----------------------------------------------------------------------
.../local/builders/LocalStreamBuilder.java | 10 +++--
.../local/tasks/BroadcastMonitorThread.java | 45 ++++++++++++++++++--
2 files changed, 48 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/d48713fa/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/builders/LocalStreamBuilder.java
----------------------------------------------------------------------
diff --cc streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/builders/LocalStreamBuilder.java
index 10194ce,2161638..31ca808
--- a/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/builders/LocalStreamBuilder.java
+++ b/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/builders/LocalStreamBuilder.java
@@@ -20,9 -20,13 +20,10 @@@ package org.apache.streams.local.builde
import org.apache.log4j.spi.LoggerFactory;
import org.apache.streams.core.*;
+ import org.apache.streams.local.counters.StreamsTaskCounter;
import org.apache.streams.local.executors.ShutdownStreamOnUnhandleThrowableThreadPoolExecutor;
import org.apache.streams.local.queues.ThroughputQueue;
-import org.apache.streams.local.tasks.LocalStreamProcessMonitorThread;
-import org.apache.streams.local.tasks.StatusCounterMonitorThread;
-import org.apache.streams.local.tasks.StreamsProviderTask;
-import org.apache.streams.local.tasks.StreamsTask;
+import org.apache.streams.local.tasks.*;
import org.joda.time.DateTime;
import org.slf4j.Logger;
@@@ -55,7 -59,6 +56,7 @@@ public class LocalStreamBuilder impleme
private LocalStreamProcessMonitorThread monitorThread;
private Map<String, List<StreamsTask>> tasks;
private Thread shutdownHook;
- private Thread broadcastMonitor;
++ private BroadcastMonitorThread broadcastMonitor;
private int maxQueueCapacity;
/**
@@@ -185,9 -185,6 +186,7 @@@
Map<String, StreamsProviderTask> provTasks = new HashMap<String, StreamsProviderTask>();
tasks = new HashMap<String, List<StreamsTask>>();
boolean forcedShutDown = false;
+
- broadcastMonitor.start();
-
try {
monitorThread = new LocalStreamProcessMonitorThread(executor, 10);
this.monitor.submit(monitorThread);
@@@ -294,6 -295,6 +297,7 @@@
this.futures.put(task, this.executor.submit(task));
compTasks.add(task);
if( comp.isOperationCountable() ) {
++ this.monitor.submit(broadcastMonitor);
this.monitor.submit(new StatusCounterMonitorThread((DatumStatusCountable) comp.getOperation(), 10));
this.monitor.submit(new StatusCounterMonitorThread((DatumStatusCountable) task, 10));
}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/d48713fa/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/BroadcastMonitorThread.java
----------------------------------------------------------------------
diff --cc streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/BroadcastMonitorThread.java
index 3b4fc7b,0000000..c9917c2
mode 100644,000000..100644
--- a/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/BroadcastMonitorThread.java
+++ b/streams-runtimes/streams-runtime-local/src/main/java/org/apache/streams/local/tasks/BroadcastMonitorThread.java
@@@ -1,14 -1,0 +1,51 @@@
+package org.apache.streams.local.tasks;
+
++import com.google.common.collect.Lists;
++import org.apache.streams.local.counters.DatumStatusCounterPOJO;
++import org.apache.streams.local.counters.StreamsTaskCounterPOJO;
++import org.apache.streams.local.queues.ThroughputQueuePOJO;
+import org.slf4j.Logger;
+
- public class BroadcastMonitorThread extends Thread {
++import javax.management.*;
++import java.lang.management.ManagementFactory;
++import java.util.List;
++import java.util.Set;
++
++public class BroadcastMonitorThread extends NotificationBroadcasterSupport implements Runnable {
+ private static final Logger LOGGER = org.slf4j.LoggerFactory.getLogger(BroadcastMonitorThread.class);
++ private static MBeanServer server;
++ private long DEFAULT_WAIT_TIME = 30000;
+
- public BroadcastMonitorThread() { }
++ public BroadcastMonitorThread() {
++ server = ManagementFactory.getPlatformMBeanServer();
++ }
+
+ @Override
+ public void run() {
- LOGGER.debug("OMFG BROADCAST THREAD");
++ while(true) {
++ try {
++ List<String> messages = Lists.newArrayList();
++ Set<ObjectName> beans = server.queryNames(null, null);
++
++ for(ObjectName name : beans) {
++ if(name.getKeyPropertyList().get("type") != null && name.getKeyPropertyList().get("type").equals("ThroughputQueue")) {
++ MBeanInfo info = server.getMBeanInfo(name);
++ messages.add((new ThroughputQueuePOJO(server, info, name)).toString());
++ } else if(name.getKeyPropertyList().get("type") != null && name.getKeyPropertyList().get("type").equals("StreamsTaskCounter")) {
++ MBeanInfo info = server.getMBeanInfo(name);
++ messages.add((new StreamsTaskCounterPOJO(server, info, name)).toString());
++ } else if(name.getKeyPropertyList().get("type") != null && name.getKeyPropertyList().get("type").equals("DatumStatusCounter")) {
++ MBeanInfo info = server.getMBeanInfo(name);
++ messages.add((new DatumStatusCounterPOJO(server, info, name)).toString());
++ }
++ }
++
++ Thread.sleep(DEFAULT_WAIT_TIME);
++ } catch (InterruptedException e) {
++ LOGGER.error("Interrupted!: {}", e);
++ } catch (Exception e) {
++ LOGGER.error("Exception: {}", e);
++ }
++ }
+ }
- }
++}