modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/MeasurementAggregator.java
| 10 +
modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/MeasurementCollector.java
| 14 ++
modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/ShutdownManager.java
| 35 ++++++
modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/Simulator.java
| 52 ++++++++--
4 files changed, 102 insertions(+), 9 deletions(-)
New commits:
commit 763f6b03ba7918afcc4954768c3cefc012541e85
Author: John Sanda <jsanda(a)redhat.com>
Date: Wed Mar 6 12:32:55 2013 -0500
add error handling to terminate simulation
Since smal heap sizes are often used, it is common for a node to go down with
an OutOfMemoryError. This commit adds error handling to terminate the simulation
when an error occurs. Right now there is an assumption that the whole cluster
stays up for the duration of the simulation in order for it to be considered a
good run.
diff --git
a/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/MeasurementAggregator.java
b/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/MeasurementAggregator.java
index 483ad83..31bca0b 100644
---
a/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/MeasurementAggregator.java
+++
b/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/MeasurementAggregator.java
@@ -39,16 +39,26 @@ public class MeasurementAggregator implements Runnable {
private MetricsServer metricsServer;
+ private ShutdownManager shutdownManager;
+
public void setMetricsServer(MetricsServer metricsServer) {
this.metricsServer = metricsServer;
}
+ public void setShutdownManager(ShutdownManager shutdownManager) {
+ this.shutdownManager = shutdownManager;
+ }
+
@Override
public void run() {
log.info("Starting metrics aggregation...");
long startTime = System.currentTimeMillis();
try {
metricsServer.calculateAggregates();
+ } catch (Exception e) {
+ log.error("An error occurred while trying to perform aggregation",
e);
+ log.error("Requesting simulation shutdown...");
+ shutdownManager.shutdown(1);
} finally {
long endTime = System.currentTimeMillis();
log.info("Finished metrics aggregation in " + (endTime - startTime)
+ " ms");
diff --git
a/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/MeasurementCollector.java
b/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/MeasurementCollector.java
index 99b1d68..e265a18 100644
---
a/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/MeasurementCollector.java
+++
b/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/MeasurementCollector.java
@@ -51,6 +51,8 @@ public class MeasurementCollector implements Runnable {
private Stats stats;
+ private ShutdownManager shutdownManager;
+
private int batchSize = 500;
public void setQueue(PriorityQueue<Schedule> queue) {
@@ -69,6 +71,10 @@ public class MeasurementCollector implements Runnable {
this.stats = stats;
}
+ public void setShutdownManager(ShutdownManager shutdownManager) {
+ this.shutdownManager = shutdownManager;
+ }
+
@Override
public void run() {
long startTime = System.currentTimeMillis();
@@ -105,7 +111,13 @@ public class MeasurementCollector implements Runnable {
schedule.updateCollection();
}
metricsCollected = data.size();
- metricsServer.addNumericData(data);
+ try {
+ metricsServer.addNumericData(data);
+ } catch (Exception e) {
+ log.error("An error occurred while trying to store raw
metrics", e);
+ log.error("Requesting simulation shutdown...");
+ shutdownManager.shutdown(1);
+ }
stats.addRawInserts(metricsCollected);
try {
queueLock.lock();
diff --git
a/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/ShutdownManager.java
b/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/ShutdownManager.java
new file mode 100644
index 0000000..9e85524
--- /dev/null
+++
b/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/ShutdownManager.java
@@ -0,0 +1,35 @@
+/*
+ *
+ * * RHQ Management Platform
+ * * Copyright (C) 2005-2012 Red Hat, Inc.
+ * * All rights reserved.
+ * *
+ * * This program is free software; you can redistribute it and/or modify
+ * * it under the terms of the GNU General Public License, version 2, as
+ * * published by the Free Software Foundation, and/or the GNU Lesser
+ * * General Public License, version 2.1, also as published by the Free
+ * * Software Foundation.
+ * *
+ * * This program is distributed in the hope that it will be useful,
+ * * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * * GNU General Public License and the GNU Lesser General Public License
+ * * for more details.
+ * *
+ * * You should have received a copy of the GNU General Public License
+ * * and the GNU Lesser General Public License along with this program;
+ * * if not, write to the Free Software Foundation, Inc.,
+ * * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
+ *
+ */
+
+package org.rhq.metrics.simulator;
+
+/**
+ * @author John Sanda
+ */
+public interface ShutdownManager {
+
+ void shutdown(int status);
+
+}
diff --git
a/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/Simulator.java
b/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/Simulator.java
index bcc3a9b..ee72d4c 100644
---
a/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/Simulator.java
+++
b/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/Simulator.java
@@ -39,6 +39,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;
import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.Host;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.SimpleAuthInfoProvider;
import com.datastax.driver.core.exceptions.NoHostAvailableException;
@@ -61,10 +62,12 @@ import org.rhq.server.metrics.MetricsServer;
/**
* @author John Sanda
*/
-public class Simulator {
+public class Simulator implements ShutdownManager {
private final Log log = LogFactory.getLog(Simulator.class);
+ private boolean shutdown = false;
+
public void run(SimulationPlan plan) {
List<CassandraNode> nodes = initCluster(plan);
createSchema(nodes);
@@ -84,6 +87,7 @@ public class Simulator {
MeasurementAggregator measurementAggregator = new MeasurementAggregator();
measurementAggregator.setMetricsServer(metricsServer);
+ measurementAggregator.setShutdownManager(this);
Stats stats = new Stats();
StatsCollector statsCollector = new StatsCollector(stats);
@@ -106,6 +110,7 @@ public class Simulator {
measurementCollector.setQueue(queue);
measurementCollector.setQueueLock(queueLock);
measurementCollector.setStats(stats);
+ measurementCollector.setShutdownManager(this);
executorService.scheduleAtFixedRate(measurementCollector, 0,
plan.getCollectionInterval(),
TimeUnit.MILLISECONDS);
@@ -118,9 +123,18 @@ public class Simulator {
Thread.sleep(Minutes.minutes(plan.getSimulationTime()).toStandardDuration().getMillis());
} catch (InterruptedException e) {
}
-// shutdown(executorService);
log.info("Simulation has completed. Initiating shutdown...");
- System.exit(0);
+ shutdown(0);
+ }
+
+ @Override
+ public synchronized void shutdown(int status) {
+ if (shutdown) {
+ return;
+ }
+ shutdown = true;
+ log.info("Preparing to shutdown simulator...");
+ System.exit(status);
}
private void shutdown(ScheduledExecutorService executorService) {
@@ -135,7 +149,6 @@ public class Simulator {
executorService.shutdownNow();
}
log.info("Shut down complete");
- //System.exit(0);
}
private List<CassandraNode> initCluster(SimulationPlan plan) {
@@ -196,6 +209,11 @@ public class Simulator {
.withCompression(SNAPPY)
.build();
+ NodeFailureListener listener = new NodeFailureListener();
+ for (Host host : cluster.getMetadata().getAllHosts()) {
+ host.getMonitor().register(listener);
+ }
+
return cluster.connect("rhq");
} catch (Exception e) {
throw new RuntimeException("Failed to start simulator. Unable to create
" + Session.class, e);
@@ -214,10 +232,28 @@ public class Simulator {
return schedules;
}
- public static void main(String[] args) {
- System.out.println("Running simulator...");
- Simulator simulator = new Simulator();
- simulator.log.info("Testing logging...");
+ private static class NodeFailureListener implements Host.StateListener {
+
+ private Log log = LogFactory.getLog(NodeFailureListener.class);
+
+ @Override
+ public void onAdd(Host host) {
+ }
+
+ @Override
+ public void onUp(Host host) {
+ }
+
+ @Override
+ public void onDown(Host host) {
+ log.warn("Node " + host + " has gone down.");
+ log.warn("Preparing to shutdown simulator...");
+ System.exit(1);
+ }
+
+ @Override
+ public void onRemove(Host host) {
+ }
}
}