1:58 p.m.
modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/DateTimeService.java
| 12
modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsServer.java
| 44 +-
modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/DateTimeServiceTest.java
| 12
modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsServerTest.java
| 1
modules/helpers/metrics-simulator/pom.xml
| 16
modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/MeasurementAggregator.java
| 45 +-
modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/MeasurementCollector.java
| 121 +-----
modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/Metrics.java
| 30 +
modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/Schedule.java
| 119 ------
modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/Simulator.java
| 191 ++--------
modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/SimulatorDateTimeService.java
| 27 +
modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/StatsCollector.java
| 94 ----
modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/plan/ClusterConfig.java
| 94 ----
modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/plan/ScheduleGroup.java
| 60 ---
modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/plan/SimulationPlan.java
| 63 +--
modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/plan/SimulationPlanner.java
| 190 +--------
modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/stats/Aggregate.java
| 69 ---
modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/stats/Stats.java
| 103 -----
modules/helpers/metrics-simulator/src/main/resources/conf/log4j.properties
| 2
19 files changed, 284 insertions(+), 1009 deletions(-)
New commits:
commit 45836de0bdb7f19c9b2797848ff1dcab785188de
Author: John Sanda <jsanda(a)redhat.com>
Date: Mon Sep 30 14:42:00 2013 -0400
remove extraneous logging and making reporting interval configurable
diff --git
a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsServer.java
b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsServer.java
index f77d805..886155f 100644
---
a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsServer.java
+++
b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsServer.java
@@ -345,7 +345,6 @@ public class MetricsServer {
long timeSlice = dateTimeService.getTimeSlice(new
DateTime(rawData.getTimestamp()),
configuration.getRawTimeSliceDuration()).getMillis();
- log.debug("Updating metrics_index with time " + new
DateTime(timeSlice));
StorageResultSetFuture resultSetFuture =
dao.updateMetricsIndex(MetricsTable.ONE_HOUR, rawData.getScheduleId(),
timeSlice);
Futures.addCallback(resultSetFuture, new FutureCallback<ResultSet>() {
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 2d2564e..1ec9399 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
@@ -88,7 +88,7 @@ public class Simulator implements ShutdownManager {
MeasurementAggregator measurementAggregator = new
MeasurementAggregator(metricsServer, this, metrics,
aggregationQueue);
- ConsoleReporter consoleReporter = createConsoleReporter(metrics);
+ ConsoleReporter consoleReporter = createConsoleReporter(metrics,
plan.getMetricsReportInterval());
for (int i = 0; i < plan.getNumMeasurementCollectors(); ++i) {
collectors.scheduleAtFixedRate(new MeasurementCollector(plan.getBatchSize(),
@@ -107,14 +107,14 @@ public class Simulator implements ShutdownManager {
shutdown(0);
}
- private ConsoleReporter createConsoleReporter(Metrics metrics) {
+ private ConsoleReporter createConsoleReporter(Metrics metrics, int reportInterval) {
try {
File basedir = new
File(System.getProperty("rhq.metrics.simulator.basedir"));
File logDir = new File(basedir, "log");
ConsoleReporter consoleReporter =
ConsoleReporter.forRegistry(metrics.registry)
.convertRatesTo(TimeUnit.SECONDS).convertDurationsTo(TimeUnit.MILLISECONDS)
.outputTo(new PrintStream(new FileOutputStream(new File(logDir,
"metrics.txt")))).build();
- consoleReporter.start(1, TimeUnit.MINUTES);
+ consoleReporter.start(reportInterval, TimeUnit.SECONDS);
return consoleReporter;
} catch (FileNotFoundException e) {
throw new RuntimeException("Failed to create console reporter",
e);
diff --git
a/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/plan/SimulationPlan.java
b/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/plan/SimulationPlan.java
index 1508edf..630c0ca 100644
---
a/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/plan/SimulationPlan.java
+++
b/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/plan/SimulationPlan.java
@@ -32,8 +32,6 @@ import org.rhq.server.metrics.MetricsConfiguration;
*/
public class SimulationPlan {
- private int threadPoolSize;
-
private long collectionInterval;
private long aggregationInterval;
@@ -50,6 +48,8 @@ public class SimulationPlan {
private int batchSize;
+ private int metricsReportInterval;
+
public long getCollectionInterval() {
return collectionInterval;
}
@@ -113,4 +113,12 @@ public class SimulationPlan {
public void setBatchSize(int batchSize) {
this.batchSize = batchSize;
}
+
+ public int getMetricsReportInterval() {
+ return metricsReportInterval;
+ }
+
+ public void setMetricsReportInterval(int metricsReportInterval) {
+ this.metricsReportInterval = metricsReportInterval;
+ }
}
diff --git
a/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/plan/SimulationPlanner.java
b/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/plan/SimulationPlanner.java
index cb690e5..1674766 100644
---
a/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/plan/SimulationPlanner.java
+++
b/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/plan/SimulationPlanner.java
@@ -51,6 +51,7 @@ public class SimulationPlanner {
simulation.setNumMeasurementCollectors(getInt(root.get("numMeasurementCollectors"),
5));
simulation.setSimulationTime(getInt(root.get("simulationTime"), 10));
simulation.setBatchSize(getInt(root.get("batchSize"), 5000));
+
simulation.setMetricsReportInterval(getInt(root.get("metricsReportInterval"),
180));
String[] nodes;
if (root.get("nodes") == null || root.get("nodes").size() ==
0) {
commit f050eadea380b5f5a26e51f658e6d8f293efba81
Author: John Sanda <jsanda(a)redhat.com>
Date: Mon Sep 30 13:53:14 2013 -0400
remove obsolete config setting and add nowInMills() method
The nowInMills method avoids the overhead of creating a new DateTime object
which could non-trivial since it is called continuously in MeasurementCollector.
diff --git
a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/DateTimeService.java
b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/DateTimeService.java
index 9920ad6..92499fe 100644
---
a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/DateTimeService.java
+++
b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/DateTimeService.java
@@ -50,7 +50,11 @@ public class DateTimeService {
}
public DateTime now() {
- return DateTime.now();
+ return new DateTime(nowInMillis());
+ }
+
+ public long nowInMillis() {
+ return System.currentTimeMillis();
}
public DateTime getTimeSlice(long timestamp, Minutes interval) {
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 22bf81d..eeb508e 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
@@ -66,7 +66,7 @@ public class MeasurementCollector implements Runnable {
private Set<MeasurementDataNumeric> generateData() {
Set<MeasurementDataNumeric> data = new
HashSet<MeasurementDataNumeric>(batchSize);
- long timestamp = dateTimeService.now().getMillis();
+ long timestamp = dateTimeService.nowInMillis();
ThreadLocalRandom random = ThreadLocalRandom.current();
for (int i = 0; i < batchSize; ++i) {
diff --git
a/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/plan/SimulationPlan.java
b/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/plan/SimulationPlan.java
index 70c326a..1508edf 100644
---
a/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/plan/SimulationPlan.java
+++
b/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/plan/SimulationPlan.java
@@ -50,14 +50,6 @@ public class SimulationPlan {
private int batchSize;
- public int getThreadPoolSize() {
- return threadPoolSize;
- }
-
- public void setThreadPoolSize(int threadPoolSize) {
- this.threadPoolSize = threadPoolSize;
- }
-
public long getCollectionInterval() {
return collectionInterval;
}
diff --git
a/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/plan/SimulationPlanner.java
b/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/plan/SimulationPlanner.java
index dc9142c..cb690e5 100644
---
a/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/plan/SimulationPlanner.java
+++
b/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/plan/SimulationPlanner.java
@@ -50,7 +50,6 @@ public class SimulationPlanner {
simulation.setAggregationInterval(getLong(root.get("aggregationInterval"),
150000L)); // 2.5 minutes
simulation.setNumMeasurementCollectors(getInt(root.get("numMeasurementCollectors"),
5));
simulation.setSimulationTime(getInt(root.get("simulationTime"), 10));
- simulation.setThreadPoolSize(getInt(root.get("threadPoolSize"),
simulation.getNumMeasurementCollectors() + 2));
simulation.setBatchSize(getInt(root.get("batchSize"), 5000));
String[] nodes;
commit 97e6cc2ce4d452f64a2a7f46f08b16edcec4ce61
Author: John Sanda <jsanda(a)redhat.com>
Date: Mon Sep 30 13:11:54 2013 -0400
fixing bugs and adding support for compressed time slices
The simulator now uses hard-coded TTLs and time slices. The duration of the
time slice for each table is as follows,
raw data --> 2.5 minutes
1 hr data --> 15 minutes
6 hr data --> 1 hour
Aggregation runs every 2.5 minutes. The execution time for aggregation can and
will exceed 2.5 minutes. I do not want the aggregator thread to block and wind
up kicking aggregation with the wrong start times. It now submits a task for
each aggregation so that the aggregator thread itself does not get delayed.
diff --git
a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/DateTimeService.java
b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/DateTimeService.java
index 01b6f71..9920ad6 100644
---
a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/DateTimeService.java
+++
b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/DateTimeService.java
@@ -25,8 +25,6 @@
package org.rhq.server.metrics;
-import static org.joda.time.DateTime.now;
-
import org.joda.time.Chronology;
import org.joda.time.DateTime;
import org.joda.time.DateTimeComparator;
@@ -45,12 +43,16 @@ public class DateTimeService {
private DateTimeComparator dateTimeComparator = DateTimeComparator.getInstance();
- private MetricsConfiguration configuration;
+ protected MetricsConfiguration configuration;
public void setConfiguration(MetricsConfiguration configuration) {
this.configuration = configuration;
}
+ public DateTime now() {
+ return DateTime.now();
+ }
+
public DateTime getTimeSlice(long timestamp, Minutes interval) {
return getTimeSlice(new DateTime(timestamp), interval);
}
diff --git
a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsServer.java
b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsServer.java
index f21f5c1..f77d805 100644
---
a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsServer.java
+++
b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsServer.java
@@ -98,7 +98,7 @@ public class MetricsServer {
* purged.
*/
private void determineMostRecentRawDataSinceLastShutdown() {
- DateTime previousHour =
currentInterval().minus(configuration.getRawTimeSliceDuration());
+ DateTime previousHour =
currentHour().minus(configuration.getRawTimeSliceDuration());
DateTime oldestRawTime = previousHour.minus(configuration.getRawRetention());
ResultSet resultSet = dao.setFindTimeSliceForIndex(MetricsTable.ONE_HOUR,
previousHour.getMillis());
@@ -126,17 +126,12 @@ public class MetricsServer {
}
}
- protected DateTime currentInterval() {
- return dateTimeService.getTimeSlice(DateTime.now(),
configuration.getRawTimeSliceDuration());
- }
-
protected DateTime currentHour() {
- DateTime dt = new DateTime(System.currentTimeMillis());
- return dateTimeService.getTimeSlice(dt, Duration.standardHours(1));
+ return dateTimeService.getTimeSlice(dateTimeService.now(),
configuration.getRawTimeSliceDuration());
}
protected DateTime roundDownToHour(long timestamp) {
- return dateTimeService.getTimeSlice(new DateTime(timestamp),
Duration.standardHours(1));
+ return dateTimeService.getTimeSlice(new DateTime(timestamp),
configuration.getRawTimeSliceDuration());
}
public void shutdown() {
@@ -313,7 +308,7 @@ public class MetricsServer {
log.debug("Inserting " + dataSet.size() + " raw
metrics");
}
- final long startTime = System.currentTimeMillis();
+ final long startTime = dateTimeService.now().getMillis();
final AtomicInteger remainingInserts = new AtomicInteger(dataSet.size());
for (final MeasurementDataNumeric data : dataSet) {
@@ -350,6 +345,7 @@ public class MetricsServer {
long timeSlice = dateTimeService.getTimeSlice(new
DateTime(rawData.getTimestamp()),
configuration.getRawTimeSliceDuration()).getMillis();
+ log.debug("Updating metrics_index with time " + new
DateTime(timeSlice));
StorageResultSetFuture resultSetFuture =
dao.updateMetricsIndex(MetricsTable.ONE_HOUR, rawData.getScheduleId(),
timeSlice);
Futures.addCallback(resultSetFuture, new FutureCallback<ResultSet>() {
@@ -385,7 +381,7 @@ public class MetricsServer {
* for subsequently computing baselines.
*/
public Iterable<AggregateNumericMetric> calculateAggregates() {
- DateTime theHour = currentInterval();
+ DateTime theHour = currentHour();
if (pastAggregationMissed) {
calculateAggregates(roundDownToHour(mostRecentRawDataPriorToStartup).plusHours(1).getMillis());
@@ -401,7 +397,9 @@ public class MetricsServer {
DateTime currentHour = dateTimeService.getTimeSlice(dt,
configuration.getRawTimeSliceDuration());
DateTime lastHour = currentHour.minus(configuration.getRawTimeSliceDuration());
- long hourTimeSlice = lastHour.getMillis();
+ if (log.isDebugEnabled()) {
+ log.debug("Starting aggregation for time slice " + lastHour);
+ }
long sixHourTimeSlice = dateTimeService.getTimeSlice(lastHour,
configuration.getOneHourTimeSliceDuration()).getMillis();
@@ -424,10 +422,10 @@ public class MetricsServer {
Iterable<AggregateNumericMetric> newOneHourAggregates = null;
- List<AggregateNumericMetric> updatedSchedules =
aggregateRawData(hourTimeSlice);
+ List<AggregateNumericMetric> updatedSchedules =
aggregateRawData(lastHour);
newOneHourAggregates = updatedSchedules;
if (!updatedSchedules.isEmpty()) {
- dao.deleteMetricsIndexEntries(MetricsTable.ONE_HOUR, hourTimeSlice);
+ dao.deleteMetricsIndexEntries(MetricsTable.ONE_HOUR, lastHour.getMillis());
updateMetricsIndex(MetricsTable.SIX_HOUR, updatedSchedules,
configuration.getOneHourTimeSliceDuration());
}
@@ -457,16 +455,20 @@ public class MetricsServer {
dao.updateMetricsIndex(bucket, updates);
}
- private List<AggregateNumericMetric> aggregateRawData(long theHour) {
+ private List<AggregateNumericMetric> aggregateRawData(DateTime theHour) {
long start = System.currentTimeMillis();
try {
- Iterable<MetricsIndexEntry> indexEntries =
dao.findMetricsIndexEntries(MetricsTable.ONE_HOUR, theHour);
+ if (log.isDebugEnabled()) {
+ log.debug("Preparing to aggregate raw data. Time slice start time is
[" + theHour +
+ "] and the end time is [" +
theHour.plus(configuration.getRawTimeSliceDuration()) + "]");
+ }
+ Iterable<MetricsIndexEntry> indexEntries =
dao.findMetricsIndexEntries(MetricsTable.ONE_HOUR,
+ theHour.getMillis());
List<AggregateNumericMetric> oneHourMetrics = new
ArrayList<AggregateNumericMetric>();
for (MetricsIndexEntry indexEntry : indexEntries) {
DateTime startTime = indexEntry.getTime();
DateTime endTime =
startTime.plus(configuration.getRawTimeSliceDuration());
-
Iterable<RawNumericMetric> rawMetrics =
dao.findRawMetrics(indexEntry.getScheduleId(),
startTime.getMillis(), endTime.getMillis());
AggregateNumericMetric aggregatedRaw = calculateAggregatedRaw(rawMetrics,
startTime.getMillis());
@@ -518,16 +520,14 @@ public class MetricsServer {
private List<AggregateNumericMetric> calculateAggregates(MetricsTable
fromTable,
MetricsTable toTable, long timeSlice, Duration nextDuration) {
- if (log.isDebugEnabled()) {
- log.debug("Preparing to compute aggregates for data in " +
fromTable + " table");
- }
long start = System.currentTimeMillis();
try {
DateTime startTime = new DateTime(timeSlice);
DateTime endTime = startTime.plus(nextDuration);
- DateTime currentHour = currentInterval();
+ DateTime currentHour = currentHour();
if (log.isDebugEnabled()) {
+ log.debug("Preparing to compute aggregates for data in " +
fromTable + " table");
log.debug("Time slice start time is [" + startTime + "]
and the end time is [" + endTime + "].");
}
diff --git
a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsServerTest.java
b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsServerTest.java
index a670fcf..95ada4c 100644
---
a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsServerTest.java
+++
b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsServerTest.java
@@ -102,10 +102,6 @@ public class MetricsServerTest extends CassandraIntegrationTest {
return currentHour;
}
- @Override
- protected DateTime currentInterval() {
- return currentHour;
- }
}
@BeforeMethod
diff --git a/modules/helpers/metrics-simulator/pom.xml
b/modules/helpers/metrics-simulator/pom.xml
index 7f9f6ae..885e86e 100644
--- a/modules/helpers/metrics-simulator/pom.xml
+++ b/modules/helpers/metrics-simulator/pom.xml
@@ -58,12 +58,6 @@
</dependency>
<dependency>
- <groupId>org.apache.commons</groupId>
- <artifactId>commons-math3</artifactId>
- <version>3.1.1</version>
- </dependency>
-
- <dependency>
<groupId>commons-logging</groupId>
<artifactId>commons-logging</artifactId>
<version>${commons-logging.version}</version>
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 ef223cb..03d4afe 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
@@ -25,6 +25,8 @@
package org.rhq.metrics.simulator;
+import java.util.concurrent.ExecutorService;
+
import com.codahale.metrics.Timer;
import org.apache.commons.logging.Log;
@@ -43,24 +45,34 @@ public class MeasurementAggregator implements Runnable {
private Metrics metrics;
+ private ExecutorService aggregationQueue;
+
private ShutdownManager shutdownManager;
- public MeasurementAggregator(MetricsServer metricsServer, ShutdownManager
shutdownManager, Metrics metrics) {
+ public MeasurementAggregator(MetricsServer metricsServer, ShutdownManager
shutdownManager, Metrics metrics,
+ ExecutorService aggregationQueue) {
this.metricsServer = metricsServer;
this.shutdownManager = shutdownManager;
this.metrics = metrics;
+ this.aggregationQueue = aggregationQueue;
}
public void run() {
- Timer.Context context = metrics.totalAggregationTime.time();
- 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 {
- context.stop();
- }
+ aggregationQueue.submit(new Runnable() {
+ @Override
+ public void run() {
+ Timer.Context context = metrics.totalAggregationTime.time();
+ try {
+ log.debug("Starting metrics aggregation");
+ 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 {
+ context.stop();
+ }
+ }
+ });
}
}
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 45fe409..22bf81d 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
@@ -53,16 +53,20 @@ public class MeasurementCollector implements Runnable {
private Metrics metrics;
- public MeasurementCollector(int batchSize, int startingScheduleId, Metrics metrics,
MetricsServer metricsServer) {
+ private SimulatorDateTimeService dateTimeService;
+
+ public MeasurementCollector(int batchSize, int startingScheduleId, Metrics metrics,
MetricsServer metricsServer,
+ SimulatorDateTimeService dateTimeService) {
this.batchSize = batchSize;
this.startingScheduleId = startingScheduleId;
this.metrics = metrics;
this.metricsServer = metricsServer;
+ this.dateTimeService = dateTimeService;
}
private Set<MeasurementDataNumeric> generateData() {
Set<MeasurementDataNumeric> data = new
HashSet<MeasurementDataNumeric>(batchSize);
- long timestamp = System.currentTimeMillis();
+ long timestamp = dateTimeService.now().getMillis();
ThreadLocalRandom random = ThreadLocalRandom.current();
for (int i = 0; i < batchSize; ++i) {
@@ -93,17 +97,4 @@ public class MeasurementCollector implements Runnable {
});
}
- private static class NoOpCallback implements RawDataInsertedCallback {
- @Override
- public void onFinish() {
- }
-
- @Override
- public void onSuccess(MeasurementDataNumeric measurementDataNumeric) {
- }
-
- @Override
- public void onFailure(Throwable throwable) {
- }
- }
}
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 be45017..2d2564e 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
@@ -23,6 +23,7 @@ import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.PrintStream;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
@@ -40,7 +41,6 @@ import org.joda.time.Minutes;
import org.rhq.cassandra.schema.SchemaManager;
import org.rhq.cassandra.util.ClusterBuilder;
import org.rhq.metrics.simulator.plan.SimulationPlan;
-import org.rhq.server.metrics.DateTimeService;
import org.rhq.server.metrics.MetricsDAO;
import org.rhq.server.metrics.MetricsServer;
import org.rhq.server.metrics.StorageSession;
@@ -55,13 +55,17 @@ public class Simulator implements ShutdownManager {
private boolean shutdown = false;
public void run(SimulationPlan plan) {
- final ScheduledExecutorService executorService =
Executors.newScheduledThreadPool(plan.getThreadPoolSize(),
- new SimulatorThreadFactory());
+ final ScheduledExecutorService aggregators = Executors.newScheduledThreadPool(1,
new SimulatorThreadFactory());
+ final ScheduledExecutorService collectors = Executors.newScheduledThreadPool(
+ plan.getNumMeasurementCollectors(), new SimulatorThreadFactory());
+ final ExecutorService aggregationQueue = Executors.newSingleThreadExecutor(new
SimulatorThreadFactory());
Runtime.getRuntime().addShutdownHook(new Thread() {
@Override
public void run() {
- shutdown(executorService);
+ shutdown(collectors, "collectors", 5);
+ shutdown(aggregators, "aggregators", 1);
+ shutdown(aggregationQueue, "aggregationQueue",
Integer.MAX_VALUE);
}
});
@@ -75,27 +79,25 @@ public class Simulator implements ShutdownManager {
metricsServer.setDAO(metricsDAO);
metricsServer.setConfiguration(plan.getMetricsServerConfiguration());
- DateTimeService dateTimeService = new DateTimeService();
+ SimulatorDateTimeService dateTimeService = new SimulatorDateTimeService();
dateTimeService.setConfiguration(plan.getMetricsServerConfiguration());
metricsServer.setDateTimeService(dateTimeService);
Metrics metrics = new Metrics();
- MeasurementAggregator measurementAggregator = new
MeasurementAggregator(metricsServer, this, metrics);
+ MeasurementAggregator measurementAggregator = new
MeasurementAggregator(metricsServer, this, metrics,
+ aggregationQueue);
ConsoleReporter consoleReporter = createConsoleReporter(metrics);
- int batchSize = 3;
for (int i = 0; i < plan.getNumMeasurementCollectors(); ++i) {
- MeasurementCollector measurementCollector = new
MeasurementCollector(batchSize, batchSize * i, metrics,
- metricsServer);
- executorService.scheduleAtFixedRate(measurementCollector, 0,
plan.getCollectionInterval(),
+ collectors.scheduleAtFixedRate(new MeasurementCollector(plan.getBatchSize(),
+ plan.getBatchSize() * i, metrics, metricsServer, dateTimeService), 0,
plan.getCollectionInterval(),
TimeUnit.MILLISECONDS);
}
- executorService.scheduleAtFixedRate(measurementAggregator, 0,
plan.getAggregationInterval(),
+ aggregators.scheduleAtFixedRate(measurementAggregator, 0,
plan.getAggregationInterval(),
TimeUnit.MILLISECONDS);
-
try {
Thread.sleep(Minutes.minutes(plan.getSimulationTime()).toStandardDuration().getMillis());
} catch (InterruptedException e) {
@@ -130,18 +132,18 @@ public class Simulator implements ShutdownManager {
System.exit(status);
}
- private void shutdown(ScheduledExecutorService executorService) {
- log.info("Shutting down executor service");
- executorService.shutdown();
+ private void shutdown(ExecutorService service, String serviceName, int wait) {
+ log.info("Shutting down " + serviceName);
+ service.shutdown();
try {
- executorService.awaitTermination(5, TimeUnit.SECONDS);
+ service.awaitTermination(wait, TimeUnit.SECONDS);
} catch (InterruptedException e) {
}
- if (!executorService.isTerminated()) {
- log.info("Forcing executor service shutdown.");
- executorService.shutdownNow();
+ if (!service.isTerminated()) {
+ log.info("Forcing " + serviceName + " shutdown.");
+ service.shutdownNow();
}
- log.info("Shut down complete");
+ log.info(serviceName + " shut down complete");
}
private void createSchema(String[] nodes, int cqlPort) {
diff --git
a/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/SimulatorDateTimeService.java
b/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/SimulatorDateTimeService.java
new file mode 100644
index 0000000..c62340b
--- /dev/null
+++
b/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/SimulatorDateTimeService.java
@@ -0,0 +1,27 @@
+package org.rhq.metrics.simulator;
+
+import org.joda.time.DateTime;
+import org.joda.time.Duration;
+
+import org.rhq.server.metrics.DateTimeService;
+
+/**
+ * @author John Sanda
+ */
+public class SimulatorDateTimeService extends DateTimeService {
+
+ @Override
+ public DateTime getTimeSlice(DateTime dt, Duration duration) {
+ if (duration.equals(configuration.getRawTimeSliceDuration())) {
+ int seconds = ((dt.getMinuteOfHour() * 60) + dt.getSecondOfMinute()) / 150;
+ return dt.hourOfDay().roundFloorCopy().plusSeconds(seconds * 150);
+ } else if (duration.equals(configuration.getOneHourTimeSliceDuration())) {
+ int minutes = dt.minuteOfHour().get() / 15;
+ return dt.hourOfDay().roundFloorCopy().plusMinutes(minutes * 15);
+ } else if (duration.equals(configuration.getSixHourTimeSliceDuration())) {
+ return dt.hourOfDay().roundFloorCopy();
+ } else {
+ throw new IllegalArgumentException("The duration [" + duration +
"] is not supported");
+ }
+ }
+}
diff --git
a/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/plan/ClusterConfig.java
b/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/plan/ClusterConfig.java
deleted file mode 100644
index af3ffeb..0000000
---
a/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/plan/ClusterConfig.java
+++ /dev/null
@@ -1,94 +0,0 @@
-/*
- *
- * * 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.plan;
-
-import java.io.File;
-
-/**
- * @author John Sanda
- */
-public class ClusterConfig {
-
- private boolean embedded = true;
-
- private String clusterDir = new
File(System.getProperty("rhq.metrics.simulator.basedir")).getAbsolutePath();
-
- private int numNodes = 2;
-
- private String heapSize = "256M";
-
- private String heapNewSize = "64M";
-
- private String stackSize;
-
- public boolean isEmbedded() {
- return embedded;
- }
-
- public void setEmbedded(boolean embedded) {
- this.embedded = embedded;
- }
-
- public String getClusterDir() {
- return clusterDir;
- }
-
- public void setClusterDir(String clusterDir) {
- this.clusterDir = clusterDir;
- }
-
- public int getNumNodes() {
- return numNodes;
- }
-
- public void setNumNodes(int numNodes) {
- this.numNodes = numNodes;
- }
-
- public String getHeapSize() {
- return heapSize;
- }
-
- public void setHeapSize(String heapSize) {
- this.heapSize = heapSize;
- }
-
- public String getHeapNewSize() {
- return heapNewSize;
- }
-
- public void setHeapNewSize(String heapNewSize) {
- this.heapNewSize = heapNewSize;
- }
-
- public String getStackSize() {
- return stackSize;
- }
-
- public void setStackSize(String stackSize) {
- this.stackSize = stackSize;
- }
-}
diff --git
a/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/plan/SimulationPlan.java
b/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/plan/SimulationPlan.java
index a1079db..70c326a 100644
---
a/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/plan/SimulationPlan.java
+++
b/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/plan/SimulationPlan.java
@@ -48,6 +48,8 @@ public class SimulationPlan {
private int cqlPort;
+ private int batchSize;
+
public int getThreadPoolSize() {
return threadPoolSize;
}
@@ -111,4 +113,12 @@ public class SimulationPlan {
public void setCqlPort(int cqlPort) {
this.cqlPort = cqlPort;
}
+
+ public int getBatchSize() {
+ return batchSize;
+ }
+
+ public void setBatchSize(int batchSize) {
+ this.batchSize = batchSize;
+ }
}
diff --git
a/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/plan/SimulationPlanner.java
b/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/plan/SimulationPlanner.java
index 06aaaa1..dc9142c 100644
---
a/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/plan/SimulationPlanner.java
+++
b/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/plan/SimulationPlanner.java
@@ -31,14 +31,10 @@ import java.net.InetAddress;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
-import org.joda.time.Days;
-import org.joda.time.Duration;
-import org.joda.time.Hours;
import org.joda.time.Minutes;
import org.joda.time.Seconds;
import org.rhq.server.metrics.MetricsConfiguration;
-import org.rhq.server.metrics.domain.MetricsTable;
/**
* @author John Sanda
@@ -50,11 +46,12 @@ public class SimulationPlanner {
JsonNode root = mapper.readTree(jsonFile);
SimulationPlan simulation = new SimulationPlan();
-
simulation.setCollectionInterval(getLong(root.get("collectionInterval"),
1000L));
-
simulation.setAggregationInterval(getLong(root.get("aggregationInterval"),
60000L));
+
simulation.setCollectionInterval(getLong(root.get("collectionInterval"),
1250L));
+
simulation.setAggregationInterval(getLong(root.get("aggregationInterval"),
150000L)); // 2.5 minutes
simulation.setNumMeasurementCollectors(getInt(root.get("numMeasurementCollectors"),
5));
simulation.setSimulationTime(getInt(root.get("simulationTime"), 10));
simulation.setThreadPoolSize(getInt(root.get("threadPoolSize"),
simulation.getNumMeasurementCollectors() + 2));
+ simulation.setBatchSize(getInt(root.get("batchSize"), 5000));
String[] nodes;
if (root.get("nodes") == null || root.get("nodes").size() ==
0) {
@@ -78,26 +75,18 @@ public class SimulationPlanner {
private MetricsConfiguration createDefaultMetricsConfiguration() {
- // 500 ms --> 30 sec
- // 1 sec --> 1 minute
- // 60 sec / 1 minute --> 1 hr
- // 1440 sec / 24 minutes --> 1 day
- // 168 minutes --> 1 week
- // 744 minutes / 12.4 hr --> 31 days / 1 month
- // 8928 minutes / 148.8 hr --> 1 yr
-
MetricsConfiguration configuration = new MetricsConfiguration();
configuration.setRawTTL(Minutes.minutes(168).toStandardSeconds().getSeconds());
configuration.setRawRetention(Minutes.minutes(168).toStandardDuration());
- configuration.setRawTimeSliceDuration(Minutes.ONE.toStandardDuration());
+
configuration.setRawTimeSliceDuration(Seconds.seconds(150).toStandardDuration());
configuration.setOneHourTTL(Minutes.minutes(336).toStandardSeconds().getSeconds());
configuration.setOneHourRetention(Minutes.minutes(336));
-
configuration.setOneHourTimeSliceDuration(Minutes.minutes(6).toStandardDuration());
+
configuration.setOneHourTimeSliceDuration(Minutes.minutes(15).toStandardDuration());
configuration.setSixHourTTL(Minutes.minutes(744).toStandardSeconds().getSeconds());
configuration.setSixHourRetention(Minutes.minutes(744).toStandardSeconds());
-
configuration.setSixHourTimeSliceDuration(Minutes.minutes(24).toStandardDuration());
+
configuration.setSixHourTimeSliceDuration(Minutes.minutes(60).toStandardDuration());
configuration.setTwentyFourHourTTL(Minutes.minutes(8928).toStandardSeconds().getSeconds());
configuration.setTwentyFourHourRetention(Minutes.minutes(8928).toStandardSeconds());
@@ -105,72 +94,6 @@ public class SimulationPlanner {
return configuration;
}
- private MetricsTable getTable(String name) {
- if (name.equals(MetricsTable.RAW.getTableName())) {
- return MetricsTable.RAW;
- } else if (name.equals(MetricsTable.ONE_HOUR.getTableName())) {
- return MetricsTable.ONE_HOUR;
- } else if (name.equals(MetricsTable.SIX_HOUR.getTableName())) {
- return MetricsTable.SIX_HOUR;
- } else if (name.equals(MetricsTable.TWENTY_FOUR_HOUR.getTableName())) {
- return MetricsTable.TWENTY_FOUR_HOUR;
- } else {
- throw new IllegalArgumentException(name + " is not a valid metrics table
name");
- }
- }
-
- private void setTTLAndRetention(MetricsTable table, int ttl, MetricsConfiguration
configuration) {
- switch (table) {
- case RAW:
- configuration.setRawTTL(ttl);
-
configuration.setRawRetention(Seconds.seconds(ttl).toStandardDuration());
- break;
- case ONE_HOUR:
- configuration.setOneHourTTL(ttl);
- configuration.setOneHourRetention(Seconds.seconds(ttl));
- break;
- case SIX_HOUR:
- configuration.setSixHourTTL(ttl);
- configuration.setSixHourRetention(Seconds.seconds(ttl));
- break;
- default:
- configuration.setTwentyFourHourTTL(ttl);
- configuration.setTwentyFourHourRetention(Seconds.seconds(ttl));
- break;
- }
- }
-
- private Duration getDuration(String units, int value) {
- if (units.equals("seconds")) {
- return Seconds.seconds(value).toStandardDuration();
- } else if (units.equals("minutes")) {
- return Minutes.minutes(value).toStandardDuration();
- } else if (units.equals("hours")) {
- return Hours.hours(value).toStandardDuration();
-
- } else if (units.equals("days")) {
- return Days.days(value).toStandardDuration();
- } else {
- throw new IllegalArgumentException(units + " is not a valid value for
the units property.");
- }
- }
-
- private void setTimeSliceDuration(MetricsTable table, Duration duration,
MetricsConfiguration configuration) {
- switch (table) {
- case RAW:
- configuration.setRawTimeSliceDuration(duration);
- break;
- case ONE_HOUR:
- configuration.setOneHourTimeSliceDuration(duration);
- break;
- case SIX_HOUR:
- configuration.setSixHourTimeSliceDuration(duration);
- break;
- default:
- // do nothing
- }
- }
-
private long getLong(JsonNode node, long defaultValue) {
if (node == null) {
return defaultValue;
commit cc2c6dbc98c8ae5e34fe11f577d0dfbac35a49a0
Author: John Sanda <jsanda(a)redhat.com>
Date: Fri Sep 27 10:41:55 2013 -0400
allow for configurable time slices
diff --git
a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsServer.java
b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsServer.java
index 83f7653..f21f5c1 100644
---
a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsServer.java
+++
b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsServer.java
@@ -27,6 +27,7 @@ package org.rhq.server.metrics;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -97,13 +98,13 @@ public class MetricsServer {
* purged.
*/
private void determineMostRecentRawDataSinceLastShutdown() {
- DateTime previousHour = currentHour().minusHours(1);
+ DateTime previousHour =
currentInterval().minus(configuration.getRawTimeSliceDuration());
DateTime oldestRawTime = previousHour.minus(configuration.getRawRetention());
ResultSet resultSet = dao.setFindTimeSliceForIndex(MetricsTable.ONE_HOUR,
previousHour.getMillis());
Row row = resultSet.one();
while (row == null && previousHour.compareTo(oldestRawTime) > 0) {
- previousHour = previousHour.minusHours(1);
+ previousHour = previousHour.minus(configuration.getRawTimeSliceDuration());
resultSet = dao.setFindTimeSliceForIndex(MetricsTable.ONE_HOUR,
previousHour.getMillis());
row = resultSet.one();
}
@@ -125,6 +126,10 @@ public class MetricsServer {
}
}
+ protected DateTime currentInterval() {
+ return dateTimeService.getTimeSlice(DateTime.now(),
configuration.getRawTimeSliceDuration());
+ }
+
protected DateTime currentHour() {
DateTime dt = new DateTime(System.currentTimeMillis());
return dateTimeService.getTimeSlice(dt, Duration.standardHours(1));
@@ -380,7 +385,7 @@ public class MetricsServer {
* for subsequently computing baselines.
*/
public Iterable<AggregateNumericMetric> calculateAggregates() {
- DateTime theHour = currentHour();
+ DateTime theHour = currentInterval();
if (pastAggregationMissed) {
calculateAggregates(roundDownToHour(mostRecentRawDataPriorToStartup).plusHours(1).getMillis());
@@ -394,12 +399,15 @@ public class MetricsServer {
private Iterable<AggregateNumericMetric> calculateAggregates(long startTime) {
DateTime dt = new DateTime(startTime);
DateTime currentHour = dateTimeService.getTimeSlice(dt,
configuration.getRawTimeSliceDuration());
- DateTime lastHour = currentHour.minusHours(1);
+ DateTime lastHour = currentHour.minus(configuration.getRawTimeSliceDuration());
long hourTimeSlice = lastHour.getMillis();
long sixHourTimeSlice = dateTimeService.getTimeSlice(lastHour,
configuration.getOneHourTimeSliceDuration()).getMillis();
+ if (log.isDebugEnabled()) {
+ log.debug("six hour time slice = " + new Date(sixHourTimeSlice));
+ }
long twentyFourHourTimeSlice = dateTimeService.getTimeSlice(lastHour,
configuration.getSixHourTimeSliceDuration()).getMillis();
@@ -457,7 +465,7 @@ public class MetricsServer {
for (MetricsIndexEntry indexEntry : indexEntries) {
DateTime startTime = indexEntry.getTime();
- DateTime endTime = startTime.plusMinutes(60);
+ DateTime endTime =
startTime.plus(configuration.getRawTimeSliceDuration());
Iterable<RawNumericMetric> rawMetrics =
dao.findRawMetrics(indexEntry.getScheduleId(),
startTime.getMillis(), endTime.getMillis());
@@ -517,7 +525,7 @@ public class MetricsServer {
try {
DateTime startTime = new DateTime(timeSlice);
DateTime endTime = startTime.plus(nextDuration);
- DateTime currentHour = currentHour();
+ DateTime currentHour = currentInterval();
if (log.isDebugEnabled()) {
log.debug("Time slice start time is [" + startTime + "]
and the end time is [" + endTime + "].");
@@ -554,6 +562,9 @@ public class MetricsServer {
AggregateNumericMetric aggregatedMetric = calculateAggregate(metrics,
startTime.getMillis());
aggregatedMetric.setScheduleId(indexEntry.getScheduleId());
toMetrics.add(aggregatedMetric);
+ if (toTable == MetricsTable.TWENTY_FOUR_HOUR) {
+ log.debug("Calculated 24 hour metric = " +
aggregatedMetric);
+ }
}
switch (toTable) {
diff --git
a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/DateTimeServiceTest.java
b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/DateTimeServiceTest.java
index 2e3d899..01b3a76 100644
---
a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/DateTimeServiceTest.java
+++
b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/DateTimeServiceTest.java
@@ -96,6 +96,18 @@ public class DateTimeServiceTest {
}
@Test
+ public void getMinuteTimeSliceForSixHourData() {
+ configuration = new MetricsConfiguration();
+
configuration.setSixHourTimeSliceDuration(Minutes.minutes(24).toStandardDuration());
+
+ DateTime currentHour =
dateTimeService.hour0().plusHours(9).plusMinutes(12).plusSeconds(47);
+ DateTime timeSlice = dateTimeService.getTimeSlice(currentHour,
configuration.getSixHourTimeSliceDuration());
+ DateTime expected = dateTimeService.hour0().plusHours(9);
+
+ assertEquals(timeSlice, expected, "The hour time slice for six hour data is
wrong");
+ }
+
+ @Test
public void timestampBefore7DaysShouldBeInRawDataRange() {
assertTrue(dateTimeService.isInRawDataRange(now().minusHours(1)), "1 hour
ago should be in raw data range.");
assertTrue(dateTimeService.isInRawDataRange(now().minusDays(1)), "1 day ago
should be in raw data range.");
diff --git
a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsServerTest.java
b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsServerTest.java
index 2c140f6..a670fcf 100644
---
a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsServerTest.java
+++
b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsServerTest.java
@@ -101,6 +101,11 @@ public class MetricsServerTest extends CassandraIntegrationTest {
}
return currentHour;
}
+
+ @Override
+ protected DateTime currentInterval() {
+ return currentHour;
+ }
}
@BeforeMethod
commit 3a28eb4c10139416a6ea65597b9d006ec881b313
Author: John Sanda <jsanda(a)redhat.com>
Date: Fri Sep 27 10:29:21 2013 -0400
numerous changes in metrics-simulator to simplify things
Metrics are now captured using the Metrics Core library which renders both
Stats.java and StatsCollector.java obsolete.
MetricsCollector has been simplified substantially. It is now seeded with a
starting schedule id and a batch size and generates batch size inserts each
time it runs.
A good bit of the simulator configuration that is specified in the json file
has been removed as well in an effort to make sure things are correct. As of
now intervals and time slices are fixed.
* raw data --> 1 minute
* 1 hour data --> 6 minutes
* 6 hour data --> 1 hour
This means that a day's worth of data is generated in one hour. Minor changes
have been made in MetricsServer to allow for configurable time slices.
diff --git a/modules/helpers/metrics-simulator/pom.xml
b/modules/helpers/metrics-simulator/pom.xml
index 3281f90..7f9f6ae 100644
--- a/modules/helpers/metrics-simulator/pom.xml
+++ b/modules/helpers/metrics-simulator/pom.xml
@@ -10,6 +10,10 @@
<artifactId>rhq-metrics-simulator</artifactId>
<name>RHQ Metrics Simulator</name>
+ <properties>
+ <animal.sniffer.skip>true</animal.sniffer.skip>
+ </properties>
+
<dependencies>
<dependency>
<groupId>org.rhq</groupId>
@@ -76,6 +80,12 @@
<artifactId>commons-cli</artifactId>
<version>1.2</version>
</dependency>
+
+ <dependency>
+ <groupId>com.codahale.metrics</groupId>
+ <artifactId>metrics-core</artifactId>
+ <version>3.0.1</version>
+ </dependency>
</dependencies>
<build>
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 31bca0b..ef223cb 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
@@ -25,6 +25,8 @@
package org.rhq.metrics.simulator;
+import com.codahale.metrics.Timer;
+
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -39,20 +41,18 @@ public class MeasurementAggregator implements Runnable {
private MetricsServer metricsServer;
+ private Metrics metrics;
+
private ShutdownManager shutdownManager;
- public void setMetricsServer(MetricsServer metricsServer) {
+ public MeasurementAggregator(MetricsServer metricsServer, ShutdownManager
shutdownManager, Metrics metrics) {
this.metricsServer = metricsServer;
- }
-
- public void setShutdownManager(ShutdownManager shutdownManager) {
this.shutdownManager = shutdownManager;
+ this.metrics = metrics;
}
- @Override
public void run() {
- log.info("Starting metrics aggregation...");
- long startTime = System.currentTimeMillis();
+ Timer.Context context = metrics.totalAggregationTime.time();
try {
metricsServer.calculateAggregates();
} catch (Exception e) {
@@ -60,8 +60,7 @@ public class MeasurementAggregator implements Runnable {
log.error("Requesting simulation shutdown...");
shutdownManager.shutdown(1);
} finally {
- long endTime = System.currentTimeMillis();
- log.info("Finished metrics aggregation in " + (endTime - startTime)
+ " ms");
+ context.stop();
}
}
}
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 b0a67c8..45fe409 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
@@ -26,15 +26,15 @@
package org.rhq.metrics.simulator;
import java.util.HashSet;
-import java.util.PriorityQueue;
import java.util.Set;
-import java.util.concurrent.locks.ReentrantLock;
+import java.util.concurrent.ThreadLocalRandom;
+
+import com.codahale.metrics.Timer;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.rhq.core.domain.measurement.MeasurementDataNumeric;
-import org.rhq.metrics.simulator.stats.Stats;
import org.rhq.server.metrics.MetricsServer;
import org.rhq.server.metrics.RawDataInsertedCallback;
@@ -45,98 +45,52 @@ public class MeasurementCollector implements Runnable {
private final Log log = LogFactory.getLog(MeasurementCollector.class);
- private PriorityQueue<Schedule> queue;
-
private MetricsServer metricsServer;
- private ReentrantLock queueLock;
-
- private Stats stats;
+ private int batchSize;
- private ShutdownManager shutdownManager;
+ private int startingScheduleId;
- private int batchSize = 500;
-
- private NoOpCallback rawInsertsCallback = new NoOpCallback();
-
- public void setQueue(PriorityQueue<Schedule> queue) {
- this.queue = queue;
- }
+ private Metrics metrics;
- public void setMetricsServer(MetricsServer metricsServer) {
+ public MeasurementCollector(int batchSize, int startingScheduleId, Metrics metrics,
MetricsServer metricsServer) {
+ this.batchSize = batchSize;
+ this.startingScheduleId = startingScheduleId;
+ this.metrics = metrics;
this.metricsServer = metricsServer;
}
- public void setQueueLock(ReentrantLock queueLock) {
- this.queueLock = queueLock;
- }
+ private Set<MeasurementDataNumeric> generateData() {
+ Set<MeasurementDataNumeric> data = new
HashSet<MeasurementDataNumeric>(batchSize);
+ long timestamp = System.currentTimeMillis();
+ ThreadLocalRandom random = ThreadLocalRandom.current();
- public void setStats(Stats stats) {
- this.stats = stats;
- }
+ for (int i = 0; i < batchSize; ++i) {
+ data.add(new MeasurementDataNumeric(timestamp, startingScheduleId + i,
random.nextDouble()));
+ }
- public void setShutdownManager(ShutdownManager shutdownManager) {
- this.shutdownManager = shutdownManager;
+ return data;
}
@Override
public void run() {
- long startTime = System.currentTimeMillis();
- int metricsCollected = 0;
- // TODO parameterize threshold
- try {
- log.info("Starting metrics collections...");
- Set<Schedule> schedules = new HashSet<Schedule>();
- try {
- queueLock.lock();
- Schedule first = queue.peek();
- if (first != null && first.getNextCollection() <=
System.currentTimeMillis()) {
- Schedule next = first;
- while (next != null && next.getNextCollection() ==
first.getNextCollection() &&
- schedules.size() < batchSize) {
- schedules.add(queue.poll());
- next = queue.peek();
- }
- }
- } finally {
- queueLock.unlock();
+ final Timer.Context context = metrics.batchInsertTime.time();
+ metricsServer.addNumericData(generateData(), new RawDataInsertedCallback() {
+ @Override
+ public void onFinish() {
+ context.stop();
}
- if (schedules.isEmpty()) {
- log.debug("No schedules are ready for collections.");
- return;
+ @Override
+ public void onSuccess(MeasurementDataNumeric result) {
+ metrics.rawInserts.mark();
}
- log.debug("There are " + schedules.size() + " schedules ready
for collection.");
- Set<MeasurementDataNumeric> data = new
HashSet<MeasurementDataNumeric>(schedules.size());
- for (Schedule schedule : schedules) {
- data.add(new MeasurementDataNumeric(schedule.getNextCollection(),
schedule.getId(),
- schedule.getNextValue()));
- schedule.updateCollection();
+ @Override
+ public void onFailure(Throwable t) {
+ log.warn("Failed to insert raw data", t);
}
- metricsCollected = data.size();
- try {
- metricsServer.addNumericData(data, rawInsertsCallback);
- } 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();
- for (Schedule schedule : schedules) {
- queue.offer(schedule);
- }
- } finally {
- queueLock.unlock();
- }
- } finally {
- long endTime = System.currentTimeMillis();
- long totalTime = endTime - startTime;
- stats.addRawInsertTime(totalTime);
- log.info("Finished collecting and storing " + metricsCollected +
" raw metric in " +totalTime + " ms.");
- }
+ });
}
private static class NoOpCallback implements RawDataInsertedCallback {
diff --git
a/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/Metrics.java
b/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/Metrics.java
new file mode 100644
index 0000000..f7ea87a
--- /dev/null
+++
b/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/Metrics.java
@@ -0,0 +1,30 @@
+package org.rhq.metrics.simulator;
+
+import static com.codahale.metrics.MetricRegistry.name;
+
+import com.codahale.metrics.Meter;
+import com.codahale.metrics.MetricRegistry;
+import com.codahale.metrics.Timer;
+
+/**
+ * @author John Sanda
+ */
+public class Metrics {
+
+ public final MetricRegistry registry;
+
+ public final Meter rawInserts;
+
+ public final Timer batchInsertTime;
+
+ public final Timer totalAggregationTime;
+
+ public Metrics() {
+ registry = new MetricRegistry();
+
+ rawInserts = registry.meter(name(MeasurementCollector.class,
"rawInserts"));
+ batchInsertTime = registry.timer(name(MeasurementCollector.class,
"batchInsertTime"));
+ totalAggregationTime = registry.timer(name(MeasurementAggregator.class,
"totalAggregationTime"));
+ }
+
+}
diff --git
a/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/Schedule.java
b/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/Schedule.java
deleted file mode 100644
index 0bcb4b8..0000000
---
a/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/Schedule.java
+++ /dev/null
@@ -1,119 +0,0 @@
-/*
- *
- * * 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 class Schedule implements Comparable<Schedule> {
-
- private int id;
-
- private long lastCollection;
-
- private long nextCollection;
-
- private long interval;
-
- public Schedule(int id) {
- this.id = id;
- }
-
- public int getId() {
- return id;
- }
-
- public long getLastCollection() {
- return lastCollection;
- }
-
- public void setLastCollection(long lastCollection) {
- this.lastCollection = lastCollection;
- }
-
- public long getNextCollection() {
- return nextCollection;
- }
-
- public void setNextCollection(long nextCollection) {
- this.nextCollection = nextCollection;
- }
-
- public void updateCollection() {
- nextCollection += interval;
- }
-
- public long getInterval() {
- return interval;
- }
-
- public void setInterval(long interval) {
- this.interval = interval;
- }
-
- public double getNextValue() {
- return 1.23;
- }
-
- @Override
- public int compareTo(Schedule that) {
- if (this.nextCollection < that.nextCollection) {
- return -1;
- }
-
- if (this.nextCollection > that.nextCollection) {
- return 1;
- }
-
- return 0;
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) return true;
- if (o == null || getClass() != o.getClass()) return false;
-
- Schedule schedule = (Schedule) o;
-
- if (id != schedule.id) return false;
- if (interval != schedule.interval) return false;
-
- return true;
- }
-
- @Override
- public int hashCode() {
- int result = id;
- result = 31 * result + (int) (interval ^ (interval >>> 32));
- return result;
- }
-
- @Override
- public String toString() {
- return "Schedule[id= " + id + ", lastCollection= " +
lastCollection + ", nextCollection= " + nextCollection +
- ", interval= " + interval + "]";
- }
-}
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 59bafcf..be45017 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
@@ -20,18 +20,16 @@
package org.rhq.metrics.simulator;
import java.io.File;
-import java.io.IOException;
-import java.util.HashSet;
-import java.util.PriorityQueue;
-import java.util.Set;
+import java.io.FileNotFoundException;
+import java.io.FileOutputStream;
+import java.io.PrintStream;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.locks.ReentrantLock;
+import com.codahale.metrics.ConsoleReporter;
import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.Host;
-import com.datastax.driver.core.ProtocolOptions;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.exceptions.NoHostAvailableException;
@@ -39,16 +37,9 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.joda.time.Minutes;
-import org.rhq.cassandra.CassandraClusterManager;
-import org.rhq.cassandra.ClusterInitService;
-import org.rhq.cassandra.DeploymentOptions;
-import org.rhq.cassandra.DeploymentOptionsFactory;
import org.rhq.cassandra.schema.SchemaManager;
import org.rhq.cassandra.util.ClusterBuilder;
-import org.rhq.metrics.simulator.plan.ClusterConfig;
-import org.rhq.metrics.simulator.plan.ScheduleGroup;
import org.rhq.metrics.simulator.plan.SimulationPlan;
-import org.rhq.metrics.simulator.stats.Stats;
import org.rhq.server.metrics.DateTimeService;
import org.rhq.server.metrics.MetricsDAO;
import org.rhq.server.metrics.MetricsServer;
@@ -63,8 +54,6 @@ public class Simulator implements ShutdownManager {
private boolean shutdown = false;
- private CassandraClusterManager ccm;
-
public void run(SimulationPlan plan) {
final ScheduledExecutorService executorService =
Executors.newScheduledThreadPool(plan.getThreadPoolSize(),
new SimulatorThreadFactory());
@@ -76,18 +65,9 @@ public class Simulator implements ShutdownManager {
}
});
- initCluster(plan);
- createSchema();
-
- Session session = createSession();
-// if (plan.getClientCompression() == null) {
-// session = createSession();
-// } else {
-// ProtocolOptions.Compression compression =
Enum.valueOf(ProtocolOptions.Compression.class,
-// plan.getClientCompression().toUpperCase());
-// session = createSession(compression);
-// }
+ createSchema(plan.getNodes(), plan.getCqlPort());
+ Session session = createSession(plan.getNodes(), plan.getCqlPort());
StorageSession storageSession = new StorageSession(session);
MetricsDAO metricsDAO = new MetricsDAO(storageSession,
plan.getMetricsServerConfiguration());
@@ -99,29 +79,16 @@ public class Simulator implements ShutdownManager {
dateTimeService.setConfiguration(plan.getMetricsServerConfiguration());
metricsServer.setDateTimeService(dateTimeService);
- Set<Schedule> schedules = initSchedules(plan.getScheduleSets().get(0));
- PriorityQueue<Schedule> queue = new
PriorityQueue<Schedule>(schedules);
- ReentrantLock queueLock = new ReentrantLock();
-
- MeasurementAggregator measurementAggregator = new MeasurementAggregator();
- measurementAggregator.setMetricsServer(metricsServer);
- measurementAggregator.setShutdownManager(this);
+ Metrics metrics = new Metrics();
- Stats stats = new Stats();
- StatsCollector statsCollector = new StatsCollector(stats);
-
- log.info("Starting executor service");
- executorService.scheduleAtFixedRate(statsCollector, 0, 1, TimeUnit.MINUTES);
+ MeasurementAggregator measurementAggregator = new
MeasurementAggregator(metricsServer, this, metrics);
+ ConsoleReporter consoleReporter = createConsoleReporter(metrics);
+ int batchSize = 3;
for (int i = 0; i < plan.getNumMeasurementCollectors(); ++i) {
- MeasurementCollector measurementCollector = new MeasurementCollector();
- measurementCollector.setMetricsServer(metricsServer);
- measurementCollector.setQueue(queue);
- measurementCollector.setQueueLock(queueLock);
- measurementCollector.setStats(stats);
- measurementCollector.setShutdownManager(this);
-
+ MeasurementCollector measurementCollector = new
MeasurementCollector(batchSize, batchSize * i, metrics,
+ metricsServer);
executorService.scheduleAtFixedRate(measurementCollector, 0,
plan.getCollectionInterval(),
TimeUnit.MILLISECONDS);
}
@@ -133,16 +100,31 @@ public class Simulator implements ShutdownManager {
Thread.sleep(Minutes.minutes(plan.getSimulationTime()).toStandardDuration().getMillis());
} catch (InterruptedException e) {
}
- statsCollector.reportSummaryStats();
log.info("Simulation has completed. Initiating shutdown...");
+ consoleReporter.stop();
shutdown(0);
}
+ private ConsoleReporter createConsoleReporter(Metrics metrics) {
+ try {
+ File basedir = new
File(System.getProperty("rhq.metrics.simulator.basedir"));
+ File logDir = new File(basedir, "log");
+ ConsoleReporter consoleReporter =
ConsoleReporter.forRegistry(metrics.registry)
+
.convertRatesTo(TimeUnit.SECONDS).convertDurationsTo(TimeUnit.MILLISECONDS)
+ .outputTo(new PrintStream(new FileOutputStream(new File(logDir,
"metrics.txt")))).build();
+ consoleReporter.start(1, TimeUnit.MINUTES);
+ return consoleReporter;
+ } catch (FileNotFoundException e) {
+ throw new RuntimeException("Failed to create console reporter",
e);
+ }
+ }
+
@Override
public synchronized void shutdown(int status) {
if (shutdown) {
return;
}
+
shutdown = true;
log.info("Preparing to shutdown simulator...");
System.exit(status);
@@ -159,65 +141,23 @@ public class Simulator implements ShutdownManager {
log.info("Forcing executor service shutdown.");
executorService.shutdownNow();
}
- shutdownCluster();
log.info("Shut down complete");
}
- private void initCluster(SimulationPlan plan) {
- try {
- deployCluster(plan.getClusterConfig());
- waitForClusterToInitialize();
- } catch (Exception e) {
- throw new RuntimeException("Failed to start simulator. Cluster
initialization failed.", e);
- }
- }
-
- private void deployCluster(ClusterConfig clusterConfig) throws IOException {
- File clusterDir = new File(clusterConfig.getClusterDir(),
"cassandra");
- log.info("Deploying cluster to " + clusterDir);
- clusterDir.mkdirs();
-
- DeploymentOptionsFactory factory = new DeploymentOptionsFactory();
- DeploymentOptions deploymentOptions = factory.newDeploymentOptions();
- deploymentOptions.setClusterDir(clusterDir.getAbsolutePath());
- deploymentOptions.setNumNodes(clusterConfig.getNumNodes());
- deploymentOptions.setHeapSize(clusterConfig.getHeapSize());
- deploymentOptions.setHeapNewSize(clusterConfig.getHeapNewSize());
- if (clusterConfig.getStackSize() != null) {
- deploymentOptions.setStackSize(clusterConfig.getStackSize());
- }
- deploymentOptions.setLoggingLevel("INFO");
- deploymentOptions.load();
-
- ccm = new CassandraClusterManager(deploymentOptions);
- ccm.createCluster();
- ccm.startCluster(false);
- }
-
- private void shutdownCluster() {
- log.info("Shutting down cluster");
- ccm.shutdownCluster();
- }
-
- private void waitForClusterToInitialize() {
- log.info("Waiting for cluster to initialize");
- ClusterInitService clusterInitService = new ClusterInitService();
- clusterInitService.waitForClusterToStart(ccm.getNodes(), ccm.getJmxPorts(),
ccm.getNodes().length, 2000, 20, 10);
- }
-
- private void createSchema() {
+ private void createSchema(String[] nodes, int cqlPort) {
try {
log.info("Creating schema");
- SchemaManager schemaManager = new SchemaManager("rhqadmin",
"1eeb2f255e832171df8592078de921bc", ccm.getNodes(), ccm.getCqlPort());
+ SchemaManager schemaManager = new SchemaManager("rhqadmin",
"1eeb2f255e832171df8592078de921bc",
+ new String[] {"127.0.0.1"}, 9142);
schemaManager.install();
} catch (Exception e) {
throw new RuntimeException("Failed to start simulator. An error occurred
during schema creation.", e);
}
}
- private Session createSession() throws NoHostAvailableException {
+ private Session createSession(String[] nodes, int cqlPort) throws
NoHostAvailableException {
try {
- Cluster cluster = new
ClusterBuilder().addContactPoints(ccm.getNodes()).withPort(ccm.getCqlPort())
+ Cluster cluster = new
ClusterBuilder().addContactPoints(nodes).withPort(cqlPort)
.withCredentials("rhqadmin", "rhqadmin")
.build();
@@ -231,25 +171,6 @@ public class Simulator implements ShutdownManager {
}
}
- private Session createSession(ProtocolOptions.Compression compression)
- throws NoHostAvailableException {
- try {
- log.debug("Creating session using " + compression.name() + "
compression");
-
- Cluster cluster = new
ClusterBuilder().addContactPoints(ccm.getNodes()).withPort(ccm.getCqlPort())
- .withCredentials("cassandra", "cassandra")
- .withCompression(compression)
- .build();
-
- log.debug("Created cluster object with " +
cluster.getConfiguration().getProtocolOptions().getCompression()
- + " compression.");
-
- return initSession(cluster);
- } catch (Exception e) {
- throw new RuntimeException("Failed to start simulator. Unable to create
" + Session.class, e);
- }
- }
-
@SuppressWarnings("deprecation")
private Session initSession(Cluster cluster) {
NodeFailureListener listener = new NodeFailureListener();
@@ -260,18 +181,6 @@ public class Simulator implements ShutdownManager {
return cluster.connect("rhq");
}
- private Set<Schedule> initSchedules(ScheduleGroup scheduleSet) {
- long nextCollection = System.currentTimeMillis();
- Set<Schedule> schedules = new HashSet<Schedule>();
- for (int i = 0; i < scheduleSet.getCount(); ++i) {
- Schedule schedule = new Schedule(i);
- schedule.setInterval(scheduleSet.getInterval());
- schedule.setNextCollection(nextCollection);
- schedules.add(schedule);
- }
- return schedules;
- }
-
private static class NodeFailureListener implements Host.StateListener {
private Log log = LogFactory.getLog(NodeFailureListener.class);
diff --git
a/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/StatsCollector.java
b/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/StatsCollector.java
deleted file mode 100644
index b59ca9c..0000000
---
a/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/StatsCollector.java
+++ /dev/null
@@ -1,94 +0,0 @@
-/*
- *
- * * 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;
-
-import java.text.SimpleDateFormat;
-import java.util.Date;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.joda.time.Duration;
-
-import org.rhq.metrics.simulator.stats.Stats;
-
-/**
- * @author John Sanda
- */
-public class StatsCollector implements Runnable {
-
- private final Log log = LogFactory.getLog(StatsCollector.class);
-
- private Stats stats;
-
- private long previousRawInsertTotal;
-
- private long lastRunTimestamp;
-
- private SimpleDateFormat dateFormat;
-
- public StatsCollector(Stats stats) {
- this.stats = stats;
- dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
- }
-
- @Override
- public void run() {
- long now = System.currentTimeMillis();
- long totalRawInserts = stats.getTotalRawInserts();
-
- // inserts will be null on the first run
- if (lastRunTimestamp == 0) {
- lastRunTimestamp = now;
- previousRawInsertTotal = totalRawInserts;
- return;
- }
-
- long lastRawInsertsCount = totalRawInserts - previousRawInsertTotal;
- Duration duration = new Duration(lastRunTimestamp, now);
- stats.addRawInsertsPerMinute(lastRawInsertsCount);
-
- StringBuilder data = new StringBuilder("Statistics Report\n")
-
.append("------------------------------------------------------------------------------------\n")
- .append("Sampling period start time: " + dateFormat.format(new
Date(lastRunTimestamp))).append("\n")
- .append("Sampling period length: " +
duration.toStandardSeconds().getSeconds()).append(" seconds\n")
- .append("Total raw metrics inserted:
").append(totalRawInserts).append("\n")
- .append("Raw inserts this sampling period:
").append(lastRawInsertsCount).append("\n")
- .append(stats.getRawInsertsPerMinute()).append("\n")
- .append(stats.getRawInsertTimes()).append("\n")
-
.append("------------------------------------------------------------------------------------");
-
- log.info(data);
-
- lastRunTimestamp = now;
- previousRawInsertTotal = totalRawInserts;
- }
-
- public void reportSummaryStats() {
- log.info("Reporting statistics for entire simulation run.");
- log.info(stats.getRawInsertsPerMinute());
- }
-
-}
diff --git
a/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/plan/ScheduleGroup.java
b/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/plan/ScheduleGroup.java
deleted file mode 100644
index 4dbcf29..0000000
---
a/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/plan/ScheduleGroup.java
+++ /dev/null
@@ -1,60 +0,0 @@
-/*
- *
- * * 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.plan;
-
-/**
- * @author John Sanda
- */
-public class ScheduleGroup {
-
- private int count;
-
- private long interval;
-
- public ScheduleGroup() {
- }
-
- public ScheduleGroup(int count, long interval) {
- this.count = count;
- this.interval = interval;
- }
-
- public int getCount() {
- return count;
- }
-
- public void setCount(int count) {
- this.count = count;
- }
-
- public long getInterval() {
- return interval;
- }
-
- public void setInterval(long interval) {
- this.interval = interval;
- }
-}
diff --git
a/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/plan/SimulationPlan.java
b/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/plan/SimulationPlan.java
index d2c62c0..a1079db 100644
---
a/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/plan/SimulationPlan.java
+++
b/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/plan/SimulationPlan.java
@@ -25,9 +25,6 @@
package org.rhq.metrics.simulator.plan;
-import java.util.ArrayList;
-import java.util.List;
-
import org.rhq.server.metrics.MetricsConfiguration;
/**
@@ -35,8 +32,6 @@ import org.rhq.server.metrics.MetricsConfiguration;
*/
public class SimulationPlan {
- private List<ScheduleGroup> scheduleSets = new
ArrayList<ScheduleGroup>();
-
private int threadPoolSize;
private long collectionInterval;
@@ -49,21 +44,9 @@ public class SimulationPlan {
private int simulationTime;
- private ClusterConfig clusterConfig;
-
- private String clientCompression = null;
-
- public List<ScheduleGroup> getScheduleSets() {
- return scheduleSets;
- }
-
- public void addScheduleSet(ScheduleGroup scheduleSet) {
- scheduleSets.add(scheduleSet);
- }
+ private String[] nodes;
- public void setScheduleSets(List<ScheduleGroup> scheduleSets) {
- this.scheduleSets = scheduleSets;
- }
+ private int cqlPort;
public int getThreadPoolSize() {
return threadPoolSize;
@@ -113,19 +96,19 @@ public class SimulationPlan {
this.simulationTime = simulationTime;
}
- public ClusterConfig getClusterConfig() {
- return clusterConfig;
+ public String[] getNodes() {
+ return nodes;
}
- public void setClusterConfig(ClusterConfig clusterConfig) {
- this.clusterConfig = clusterConfig;
+ public void setNodes(String[] nodes) {
+ this.nodes = nodes;
}
- public String getClientCompression() {
- return clientCompression;
+ public int getCqlPort() {
+ return cqlPort;
}
- public void setClientCompression(String clientCompression) {
- this.clientCompression = clientCompression;
+ public void setCqlPort(int cqlPort) {
+ this.cqlPort = cqlPort;
}
}
diff --git
a/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/plan/SimulationPlanner.java
b/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/plan/SimulationPlanner.java
index f0e41ed..06aaaa1 100644
---
a/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/plan/SimulationPlanner.java
+++
b/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/plan/SimulationPlanner.java
@@ -26,6 +26,7 @@
package org.rhq.metrics.simulator.plan;
import java.io.File;
+import java.net.InetAddress;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
@@ -49,117 +50,57 @@ public class SimulationPlanner {
JsonNode root = mapper.readTree(jsonFile);
SimulationPlan simulation = new SimulationPlan();
-
simulation.setCollectionInterval(getLong(root.get("collectionInterval"),
500L));
-
simulation.setAggregationInterval(getLong(root.get("aggregationInterval"),
1000L));
- simulation.setThreadPoolSize(getInt(root.get("threadPoolSize"), 7));
+
simulation.setCollectionInterval(getLong(root.get("collectionInterval"),
1000L));
+
simulation.setAggregationInterval(getLong(root.get("aggregationInterval"),
60000L));
simulation.setNumMeasurementCollectors(getInt(root.get("numMeasurementCollectors"),
5));
simulation.setSimulationTime(getInt(root.get("simulationTime"), 10));
+ simulation.setThreadPoolSize(getInt(root.get("threadPoolSize"),
simulation.getNumMeasurementCollectors() + 2));
- JsonNode clientCompressionNode = root.get("clientCompression");
- if (clientCompressionNode != null) {
- simulation.setClientCompression(clientCompressionNode.asText());
- }
-
- JsonNode schedules = root.get("schedules");
- if (schedules == null) {
- simulation.addScheduleSet(new ScheduleGroup(2500, 500L));
+ String[] nodes;
+ if (root.get("nodes") == null || root.get("nodes").size() ==
0) {
+ nodes = new String[] {InetAddress.getLocalHost().getHostAddress()};
} else {
- if (schedules.isArray()) {
- for (JsonNode node : schedules) {
- simulation.addScheduleSet(new
ScheduleGroup(getInt(node.get("count"), 2500),
- getLong(node.get("interval"), 500L)));
- }
- } else {
- simulation.addScheduleSet(new
ScheduleGroup(getInt(schedules.get("count"), 2500),
- getLong(schedules.get("interval"), 500L)));
+ nodes = new String[root.get("nodes").size()];
+ int i = 0;
+ for (JsonNode node : root.get("nodes")) {
+ nodes[i++] = node.asText();
}
}
+ simulation.setNodes(nodes);
+
+ simulation.setCqlPort(getInt(root.get("cqlPort"), 9142));
MetricsConfiguration serverConfiguration = createDefaultMetricsConfiguration();
simulation.setMetricsServerConfiguration(serverConfiguration);
- JsonNode ttlNodes = root.get("ttl");
- if (ttlNodes != null) {
- for (JsonNode node : ttlNodes) {
- String tableName = node.get("table").asText();
- if (!tableName.isEmpty()) {
- MetricsTable table = getTable(tableName);
- JsonNode ttlNode = node.get("value");
- if (ttlNode != null) {
- setTTLAndRetention(table, ttlNode.asInt(), serverConfiguration);
- }
- }
- }
- }
-
- JsonNode timeSliceNode = root.get("timeSliceDuration");
- if (timeSliceNode != null) {
- String units = timeSliceNode.get("units").asText();
- if (units.isEmpty()) {
- units = "minutes";
- }
- for (JsonNode node : timeSliceNode.get("values")) {
- JsonNode valueNode = node.get("value");
- JsonNode tableNode = node.get("table");
- if (!(tableNode == null || valueNode == null)) {
- Duration duration = getDuration(units, valueNode.asInt());
- MetricsTable table = getTable(tableNode.asText());
- setTimeSliceDuration(table, duration, serverConfiguration);
- }
- }
- }
-
- ClusterConfig clusterConfig = new ClusterConfig();
- JsonNode clusterConfigNode = root.get("cluster");
- if (clusterConfigNode != null) {
- JsonNode embeddedNode = clusterConfigNode.get("embedded");
- if (embeddedNode != null) {
- clusterConfig.setEmbedded(embeddedNode.asBoolean(true));
- }
-
- JsonNode clusterDirNode = clusterConfigNode.get("clusterDir");
- if (clusterDirNode != null) {
- clusterConfig.setClusterDir(clusterDirNode.asText());
- }
-
- JsonNode heapSizeNode = clusterConfigNode.get("heapSize");
- if (heapSizeNode != null) {
- clusterConfig.setHeapSize(heapSizeNode.asText());
- }
-
- JsonNode heapNewSizeNode = clusterConfigNode.get("heapNewSize");
- if (heapNewSizeNode != null) {
- clusterConfig.setHeapNewSize(heapNewSizeNode.asText());
- }
-
- JsonNode stackSizeNode = clusterConfigNode.get("stackSize");
- if (stackSizeNode != null) {
- clusterConfig.setStackSize(stackSizeNode.asText());
- }
-
- clusterConfig.setNumNodes(getInt(clusterConfigNode.get("numNodes"),
2));
- }
- simulation.setClusterConfig(clusterConfig);
-
return simulation;
}
private MetricsConfiguration createDefaultMetricsConfiguration() {
+
+ // 500 ms --> 30 sec
+ // 1 sec --> 1 minute
+ // 60 sec / 1 minute --> 1 hr
+ // 1440 sec / 24 minutes --> 1 day
+ // 168 minutes --> 1 week
+ // 744 minutes / 12.4 hr --> 31 days / 1 month
+ // 8928 minutes / 148.8 hr --> 1 yr
+
MetricsConfiguration configuration = new MetricsConfiguration();
- configuration.setRawTTL(180);
- configuration.setRawRetention(Seconds.seconds(180).toStandardDuration());
+ configuration.setRawTTL(Minutes.minutes(168).toStandardSeconds().getSeconds());
+ configuration.setRawRetention(Minutes.minutes(168).toStandardDuration());
configuration.setRawTimeSliceDuration(Minutes.ONE.toStandardDuration());
- configuration.setOneHourTTL(360);
- configuration.setOneHourRetention(Seconds.seconds(360));
+
configuration.setOneHourTTL(Minutes.minutes(336).toStandardSeconds().getSeconds());
+ configuration.setOneHourRetention(Minutes.minutes(336));
configuration.setOneHourTimeSliceDuration(Minutes.minutes(6).toStandardDuration());
- configuration.setSixHourTTL(540);
- configuration.setSixHourRetention(Seconds.seconds(540));
+
configuration.setSixHourTTL(Minutes.minutes(744).toStandardSeconds().getSeconds());
+ configuration.setSixHourRetention(Minutes.minutes(744).toStandardSeconds());
configuration.setSixHourTimeSliceDuration(Minutes.minutes(24).toStandardDuration());
- configuration.setTwentyFourHourTTL(720);
- configuration.setTwentyFourHourRetention(Seconds.seconds(720));
+
configuration.setTwentyFourHourTTL(Minutes.minutes(8928).toStandardSeconds().getSeconds());
+
configuration.setTwentyFourHourRetention(Minutes.minutes(8928).toStandardSeconds());
return configuration;
}
diff --git
a/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/stats/Aggregate.java
b/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/stats/Aggregate.java
deleted file mode 100644
index 61a6b51..0000000
---
a/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/stats/Aggregate.java
+++ /dev/null
@@ -1,69 +0,0 @@
-/*
- *
- * * 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.stats;
-
-/**
- * @author John Sanda
- */
-public class Aggregate {
-
- private String name;
- private double max;
- private double min;
- private double mean;
- private double standardDeviation;
-
-
- public Aggregate(String name, double max, double min, double mean, double
standardDeviation) {
- this.name = name;
- this.max = max;
- this.min = min;
- this.mean = mean;
- this.standardDeviation = standardDeviation;
- }
-
- public double getMax() {
- return max;
- }
-
- public double getMin() {
- return min;
- }
-
- public double getMean() {
- return mean;
- }
-
- public double getStandardDeviation() {
- return standardDeviation;
- }
-
- @Override
- public String toString() {
- return name + ": {min: " + getMin() + ", mean: " + getMean()
+ ", max: " + getMax() + ", standardDeviation: " +
- getStandardDeviation() + "}";
- }
-}
diff --git
a/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/stats/Stats.java
b/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/stats/Stats.java
deleted file mode 100644
index 604f1ba..0000000
---
a/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/stats/Stats.java
+++ /dev/null
@@ -1,103 +0,0 @@
-/*
- *
- * * 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.stats;
-
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicLong;
-import java.util.concurrent.locks.ReentrantLock;
-
-import org.apache.commons.math3.stat.descriptive.DescriptiveStatistics;
-
-/**
- * @author John Sanda
- */
-public class Stats {
-
- /**
- * The total number of raw inserts
- */
- private AtomicLong totalRawInserts = new AtomicLong(0);
-
- private AtomicInteger rawInsertsThisMinute = new AtomicInteger(0);
-
- private DescriptiveStatistics rawInsertsPerMinute = new DescriptiveStatistics(200);
-
- private DescriptiveStatistics rawInsertTimesPerMinute = new
DescriptiveStatistics(200);
-
- private ReentrantLock insertTimesLock = new ReentrantLock();
-
- public void addRawInserts(int count) {
- totalRawInserts.addAndGet(count);
- rawInsertsThisMinute.addAndGet(count);
- }
-
- public long getTotalRawInserts() {
- return totalRawInserts.get();
- }
-
- /**
- * Called by measurement collectors to report insertion times. This method uses an
- * internal lock to allow for concurrent access.
- *
- * @param time The time to insert a set of raw metrics
- */
- public void addRawInsertTime(long time) {
- try {
- insertTimesLock.lock();
- rawInsertTimesPerMinute.addValue(time);
- } finally {
- insertTimesLock.unlock();
- }
- }
-
- public Aggregate getRawInsertTimes() {
- try {
- insertTimesLock.lock();
- return new Aggregate("raw insertion times (milliseconds)",
rawInsertTimesPerMinute.getMax(),
- rawInsertTimesPerMinute.getMin(), rawInsertTimesPerMinute.getMean(),
- rawInsertTimesPerMinute.getStandardDeviation());
- } finally {
- insertTimesLock.unlock();
- }
- }
-
- /**
- * Called by {@link org.rhq.metrics.simulator.StatsCollector} to report the number of
raw inserts for a given
- * minute. Since there is only a single {@link
org.rhq.metrics.simulator.StatsCollector} this method does not
- * support concurrent access.
- *
- * @param value The number of raw metrics inserted in a given minute.
- */
- public void addRawInsertsPerMinute(long value) {
- rawInsertsPerMinute.addValue(value);
- }
-
- public Aggregate getRawInsertsPerMinute() {
- return new Aggregate("Raw inserts per minute",
rawInsertsPerMinute.getMax(), rawInsertsPerMinute.getMin(),
- rawInsertsPerMinute.getMean(), rawInsertsPerMinute.getStandardDeviation());
- }
-
-}
diff --git a/modules/helpers/metrics-simulator/src/main/resources/conf/log4j.properties
b/modules/helpers/metrics-simulator/src/main/resources/conf/log4j.properties
index ee270e8..58fa73c 100644
--- a/modules/helpers/metrics-simulator/src/main/resources/conf/log4j.properties
+++ b/modules/helpers/metrics-simulator/src/main/resources/conf/log4j.properties
@@ -35,5 +35,5 @@ log4j.appender.FILE.Append=false
log4j.appender.CONSOLE=org.apache.log4j.ConsoleAppender
log4j.appender.CONSOLE.layout=org.apache.log4j.PatternLayout
-log4j.appender.CONSOLE.layout.ConversionPattern=%5p %d{HH:mm:ss,SSS} %m%n
+log4j.appender.CONSOLE.layout.ConversionPattern=%d{ISO8601} %-5p [%t] (%c{5}) - %m%n
log4j.logger.org.rhq=DEBUG
\ No newline at end of file