modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/Aggregator.java
| 215 +++++++++-
modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsDAO.java
| 23 +
modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/AggregatorTest.java
| 15
3 files changed, 227 insertions(+), 26 deletions(-)
New commits:
commit aaa337e9044fa3e81087a748f2590c836eeed183
Author: John Sanda <jsanda(a)redhat.com>
Date: Wed Oct 2 22:53:11 2013 -0400
initial support for generating 6 hr and 24 hr data
diff --git
a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/Aggregator.java
b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/Aggregator.java
index d362c97..ef95516 100644
---
a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/Aggregator.java
+++
b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/Aggregator.java
@@ -3,6 +3,7 @@ package org.rhq.server.metrics;
import java.util.Comparator;
import java.util.List;
import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;
@@ -21,6 +22,7 @@ import org.joda.time.DateTimeComparator;
import org.joda.time.Duration;
import org.rhq.server.metrics.domain.AggregateNumericMetric;
+import org.rhq.server.metrics.domain.AggregateNumericMetricMapper;
import org.rhq.server.metrics.domain.AggregateType;
import org.rhq.server.metrics.domain.MetricsIndexEntry;
import org.rhq.server.metrics.domain.MetricsIndexEntryMapper;
@@ -57,9 +59,15 @@ public class Aggregator {
private boolean sixHourDataReady;
private AtomicInteger remainingRawData;
-// private AtomicInteger remainingOneHourData
+ private AtomicInteger remainingOneHourData;
+ private AtomicInteger remainingSixHourData;
- private CountDownLatch countDownLatch;
+ private CountDownLatch allAggregationFinished;
+ private CountDownLatch oneHourIndexEntriesArrival;
+ private CountDownLatch sixHourIndexEntriesArrival;
+
+ private ConcurrentHashMap<Integer, MetricsIndexEntry> oneHourIndexEntries;
+ private ConcurrentHashMap<Integer, MetricsIndexEntry> sixHourIndexEntries;
public Aggregator(ListeningExecutorService workers, MetricsDAO dao,
MetricsConfiguration configuration,
DateTimeService dtService, DateTime startTime) {
@@ -69,17 +77,31 @@ public class Aggregator {
this.dtService = dtService;
this.startTime = startTime;
- DateTime oneHourTimeSlice = dtService.getTimeSlice(startTime,
configuration.getOneHourTimeSliceDuration());
- DateTime sixHourTimeSlice = dtService.getTimeSlice(startTime,
configuration.getSixHourTimeSliceDuration());
+ DateTime oneHourTimeSlice = getOneHourTimeSlice();
+ DateTime sixHourTimeSlice = getSixHourTimeSlice();
oneHourDataReady = hasTimeSliceEnded(oneHourTimeSlice,
configuration.getOneHourTimeSliceDuration());
sixHourDataReady = hasTimeSliceEnded(sixHourTimeSlice,
configuration.getSixHourTimeSliceDuration());
int count = 1;
-// count += oneHourDataReady ? 1 : 0;
-// count += sixHourDataReady ? 1 : 0;
+ if (oneHourDataReady) {
+ count++;
+ oneHourIndexEntriesArrival = new CountDownLatch(1);
+ }
+ if (sixHourDataReady) {
+ count++;
+ sixHourIndexEntriesArrival = new CountDownLatch(1);
+ }
+
+ allAggregationFinished = new CountDownLatch(count);
+ }
+
+ private DateTime getSixHourTimeSlice() {
+ return dtService.getTimeSlice(startTime,
configuration.getSixHourTimeSliceDuration());
+ }
- countDownLatch = new CountDownLatch(count);
+ private DateTime getOneHourTimeSlice() {
+ return dtService.getTimeSlice(startTime,
configuration.getOneHourTimeSliceDuration());
}
private boolean hasTimeSliceEnded(DateTime startTime, Duration duration) {
@@ -107,11 +129,38 @@ public class Aggregator {
@Override
public void onFailure(Throwable t) {
- log.warn("Failed to retrieve raw data index entries for time slice
" + startTime +
- ". Data aggregation cannot proceed.", t);
+ log.warn("Failed to retrieve raw data index entries for time slice
[" + startTime +
+ "]. Data aggregation cannot proceed.", t);
}
}, workers);
- countDownLatch.await();
+
+ if (oneHourDataReady) {
+ StorageResultSetFuture oneHourFuture =
dao.findMetricsIndexEntriesAsync(MetricsTable.SIX_HOUR,
+ startTime.getMillis());
+ Futures.addCallback(oneHourFuture, new FutureCallback<ResultSet>() {
+ @Override
+ public void onSuccess(ResultSet result) {
+ List<Row> rows = result.all();
+ oneHourIndexEntries = new ConcurrentHashMap<Integer,
MetricsIndexEntry>(rows.size());
+ MetricsIndexEntryMapper mapper = new
MetricsIndexEntryMapper(MetricsTable.SIX_HOUR);
+
+ for (Row row : result) {
+ MetricsIndexEntry indexEntry = mapper.map(row);
+ oneHourIndexEntries.put(indexEntry.getScheduleId(), indexEntry);
+ }
+ remainingOneHourData = new
AtomicInteger(oneHourIndexEntries.size());
+ oneHourIndexEntriesArrival.countDown();
+ }
+
+ @Override
+ public void onFailure(Throwable t) {
+ log.warn("Failed to retrieve one hour aggregate index entries
for time slice [" + startTime +
+ "]. Some six hour aggregates may not get generated.");
+ }
+ });
+ }
+
+ allAggregationFinished.await();
return oneHourAggregates;
}
@@ -157,6 +206,16 @@ public class Aggregator {
@Override
public void onSuccess(ResultSet result) {
updateRemainingRawDataCount();
+ try {
+ if (oneHourDataReady) {
+ oneHourIndexEntriesArrival.await();
+ calculateAggregate(indexEntry.getScheduleId(),
MetricsTable.SIX_HOUR);
+ }
+ } catch (InterruptedException e) {
+ log.warn("An interrupt occurred while waiting
one hour data index entries. " +
+ "No 6 hour data will be generated for "
+ oneHourAggregate.getScheduleId() +
+ " for time slice starting at " +
getOneHourTimeSlice());
+ }
}
@Override
@@ -166,8 +225,6 @@ public class Aggregator {
updateRemainingRawDataCount();
}
});
-
- // TODO compute six hour aggregate
}
@Override
@@ -185,6 +242,95 @@ public class Aggregator {
}, workers);
}
+ private void calculateAggregate(final int scheduleId, final MetricsTable toTable) {
+ StorageResultSetFuture future;
+ final DateTime startTime;
+ final DateTime endTime;
+
+ if (toTable == MetricsTable.SIX_HOUR) {
+ startTime = getOneHourTimeSlice();
+ endTime = startTime.plus(configuration.getOneHourTimeSliceDuration());
+ future = dao.findOneHourMetricsAsync(scheduleId, startTime.getMillis(),
endTime.getMillis());
+ } else { // toTable == 24 hr table
+ startTime = getSixHourTimeSlice();
+ endTime = startTime.plus(configuration.getSixHourTimeSliceDuration());
+ future = dao.findSixHourMetricsAsync(scheduleId, startTime.getMillis(),
endTime.getMillis());
+ }
+
+ Futures.addCallback(future, new FutureCallback<ResultSet>() {
+ @Override
+ public void onSuccess(ResultSet result) {
+ AggregateNumericMetricMapper mapper = new
AggregateNumericMetricMapper();
+ List<AggregateNumericMetric> metrics = mapper.mapAll(result);
+ final AggregateNumericMetric aggregate = calculateAggregate(metrics,
startTime.getMillis());
+
+ if (toTable == MetricsTable.SIX_HOUR) {
+ ListenableFuture<List<ResultSet>> insertFuture =
Futures.allAsList(
+ dao.insertSixHourDataAsync(aggregate.getScheduleId(),
startTime.getMillis(), AggregateType.MIN,
+ aggregate.getMin()),
+ dao.insertSixHourDataAsync(aggregate.getScheduleId(),
startTime.getMillis(), AggregateType.MAX,
+ aggregate.getMax()),
+ dao.insertSixHourDataAsync(aggregate.getScheduleId(),
startTime.getMillis(), AggregateType.AVG,
+ aggregate.getAvg())
+ );
+ Futures.addCallback(insertFuture, new
FutureCallback<List<ResultSet>>() {
+ @Override
+ public void onSuccess(List<ResultSet> result) {
+ if (result.get(0) == null) {
+ log.warn("Failed to store the minimum of the 6 hour
aggregate " + aggregate);
+ }
+ if (result.get(1) == null) {
+ log.warn("Failed to store the maximum of the 6 hour
aggregate " + aggregate);
+ }
+ if (result.get(2) == null) {
+ log.warn("Failed to store the average of the 6 hour
aggregate " + aggregate);
+ }
+ DateTime twentyFourHourTimeSlice = dtService.getTimeSlice(new
DateTime(
+ aggregate.getTimestamp()),
configuration.getSixHourTimeSliceDuration());
+ StorageResultSetFuture indexFuture =
dao.updateMetricsIndex(MetricsTable.TWENTY_FOUR_HOUR,
+ aggregate.getScheduleId(),
twentyFourHourTimeSlice.getDayOfMonth());
+ Futures.addCallback(indexFuture, new
FutureCallback<ResultSet>() {
+ @Override
+ public void onSuccess(ResultSet result) {
+
oneHourIndexEntries.remove(aggregate.getScheduleId());
+ updateRemainingOneHourDataCount();
+ // TODO compute 24 hr aggregate if necessary
+ }
+
+ @Override
+ public void onFailure(Throwable t) {
+ log.warn("Failed to update 24 hour data index
for 6 hour aggregate " + aggregate);
+
oneHourIndexEntries.remove(aggregate.getScheduleId());
+ updateRemainingOneHourDataCount();
+ }
+ });
+ }
+
+ @Override
+ public void onFailure(Throwable t) {
+ log.warn("Failed to store six hour aggregate " +
aggregate);
+ oneHourIndexEntries.remove(aggregate.getScheduleId());
+ updateRemainingOneHourDataCount();
+ }
+ });
+ } else {
+ // TODO handle 6 hr data
+ }
+ }
+
+ @Override
+ public void onFailure(Throwable t) {
+ if (toTable == MetricsTable.SIX_HOUR) {
+ log.warn("Failed to retrieve one hour data for scheduleId "
+ scheduleId + ". 6 hour aggregates " +
+ "for time slice [startTime: " + startTime + ",
endTime: " + endTime + "] will not be generated.");
+ } else {
+ log.warn("Failed to retrieve six hour data for scheduleId "
+ scheduleId + ". 24 hour aggregates " +
+ "for time slice [startTime: " + startTime + ",
endTime: " + endTime + "] will not be generated.");
+ }
+ }
+ });
+ }
+
private AggregateNumericMetric calculateAggregatedRaw(List<RawNumericMetric>
rawMetrics, long timestamp) {
double min = Double.NaN;
double max = min;
@@ -207,15 +353,54 @@ public class Aggregator {
++count;
}
- // We let the caller handle setting the schedule id because in some cases we do
- // not care about it.
return new AggregateNumericMetric(rawMetrics.get(0).getScheduleId(),
mean.getArithmeticMean(), min, max,
timestamp);
}
+ private AggregateNumericMetric calculateAggregate(List<AggregateNumericMetric>
metrics, long timestamp) {
+ double min = Double.NaN;
+ double max = min;
+ int count = 0;
+ ArithmeticMeanCalculator mean = new ArithmeticMeanCalculator();
+
+ for (AggregateNumericMetric metric : metrics) {
+ if (count == 0) {
+ min = metric.getMin();
+ max = metric.getMax();
+ }
+ if (metric.getMin() < min) {
+ min = metric.getMin();
+ } else if (metric.getMax() > max) {
+ max = metric.getMax();
+ }
+ mean.add(metric.getAvg());
+ ++count;
+ }
+
+ return new AggregateNumericMetric(metrics.get(0).getScheduleId(),
mean.getArithmeticMean(), min, max,
+ timestamp);
+ }
+
private void updateRemainingRawDataCount() {
if (remainingRawData.decrementAndGet() == 0) {
- countDownLatch.countDown();
+ if (log.isDebugEnabled()) {
+ log.debug("Finished aggregating raw data for time slice [" +
startTime + "]");
+ }
+ allAggregationFinished.countDown();
+ // TODO delete the row from metrics_index
+ }
+ }
+
+ private void updateRemainingOneHourDataCount() {
+ if (remainingOneHourData.decrementAndGet() == 0) {
+ if (log.isDebugEnabled()) {
+ DateTime start = getOneHourTimeSlice();
+ DateTime end = start.plus(configuration.getOneHourTimeSliceDuration());
+ log.debug("Finished aggregating one hour data for time slice
[startTime: " + start +
+ ", endTime: " + end + "]");
+ }
+ allAggregationFinished.countDown();
+ // TODO delete the row from metrics_index
}
}
diff --git
a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsDAO.java
b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsDAO.java
index 45a949d..40450b9 100644
---
a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsDAO.java
+++
b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsDAO.java
@@ -165,12 +165,25 @@ public class MetricsDAO {
return storageSession.execute(statement);
}
+ public StorageResultSetFuture insertSixHourDataAsync(int scheduleId, long timestamp,
AggregateType type,
+ double value) {
+ BoundStatement statement = insertSixHourData.bind(scheduleId, new
Date(timestamp), type.ordinal(), value);
+ return storageSession.executeAsync(statement);
+ }
+
public ResultSet insertTwentyFourHourData(int scheduleId, long timestamp,
AggregateType type, double value) {
BoundStatement statement = insertTwentyFourHourData.bind(scheduleId, new
Date(timestamp), type.ordinal(),
value);
return storageSession.execute(statement);
}
+ public StorageResultSetFuture insertTwentyFourHourDataAsync(int scheduleId, long
timestamp, AggregateType type,
+ double value) {
+ BoundStatement statement = insertTwentyFourHourData.bind(scheduleId, new
Date(timestamp), type.ordinal(),
+ value);
+ return storageSession.executeAsync(statement);
+ }
+
public Iterable<RawNumericMetric> findRawMetrics(int scheduleId, long
startTime, long endTime) {
try {
BoundStatement boundStatement = rawMetricsQuery.bind(scheduleId, new
Date(startTime), new Date(endTime));
@@ -205,12 +218,22 @@ public class MetricsDAO {
storageSession);
}
+ public StorageResultSetFuture findOneHourMetricsAsync(int scheduleId, long startTime,
long endTime) {
+ BoundStatement statement = findOneHourMetricsByDateRange.bind(scheduleId, new
Date(startTime), new Date(endTime));
+ return storageSession.executeAsync(statement);
+ }
+
public Iterable<AggregateNumericMetric> findSixHourMetrics(int scheduleId, long
startTime, long endTime) {
BoundStatement statement = findSixHourMetricsByDateRange.bind(scheduleId, new
Date(startTime), new Date(endTime));
return new SimplePagedResult<AggregateNumericMetric>(statement, new
AggregateNumericMetricMapper(),
storageSession);
}
+ public StorageResultSetFuture findSixHourMetricsAsync(int scheduleId, long startTime,
long endTime) {
+ BoundStatement statement = findSixHourMetricsByDateRange.bind(scheduleId, new
Date(startTime), new Date(endTime));
+ return storageSession.executeAsync(statement);
+ }
+
public Iterable<AggregateNumericMetric> findTwentyFourHourMetrics(int
scheduleId, long startTime, long endTime) {
BoundStatement statement = findTwentyFourHourMetricsByDateRange.bind(scheduleId,
new Date(startTime), new Date(endTime));
return new SimplePagedResult<AggregateNumericMetric>(statement, new
AggregateNumericMetricMapper(),
diff --git
a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/AggregatorTest.java
b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/AggregatorTest.java
index 4c65728..283d6fb 100644
---
a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/AggregatorTest.java
+++
b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/AggregatorTest.java
@@ -34,28 +34,21 @@ public class AggregatorTest extends CassandraIntegrationTest {
private static final double TEST_PRECISION = Math.pow(10, -9);
- private static class AggregatorTestStub extends Aggregator {
+ private class AggregatorTestStub extends Aggregator {
public AggregatorTestStub(ListeningExecutorService workers, MetricsDAO dao,
MetricsConfiguration configuration,
DateTimeService dtService, DateTime startTime) {
super(workers, dao, configuration, dtService, startTime);
}
- private DateTime currentHour;
-
- public void setCurrentHour(DateTime currentHour) {
- this.currentHour = currentHour;
- }
-
@Override
protected DateTime currentHour() {
- if (currentHour == null) {
- return super.currentHour();
- }
return currentHour;
}
}
+ private DateTime currentHour;
+
private ListeningExecutorService workers;
private MetricsDAO dao;
@@ -97,8 +90,8 @@ public class AggregatorTest extends CassandraIntegrationTest {
DateTime hour9 = hour0.plusHours(9);
DateTime hour8 = hour9.minusHours(1);
+ currentHour = hour9;
AggregatorTestStub aggregator = new AggregatorTestStub(workers, dao,
configuration, dateTimeService, hour8);
- aggregator.setCurrentHour(hour9);
DateTime firstMetricTime = hour8.plusMinutes(5);
DateTime secondMetricTime = hour8.plusMinutes(10);