modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/AggregateIndexEntriesHandler.java
| 62 +++
modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/Aggregator.java
| 168 +++-------
modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsServer.java
| 8
3 files changed, 125 insertions(+), 113 deletions(-)
New commits:
commit 7228eddf1ad6ef5305bcf9f357888ff86efdea20
Author: John Sanda <jsanda(a)redhat.com>
Date: Tue Oct 22 17:01:50 2013 -0400
refactoring some duplicate code and fixing when tasks are scheduled
Aggregation tasks for 1 and 6 hour data were getting schedule too early. They
cannot get scheduled until both their respective index entries have arrived and
aggregation for the previous buckets has completed.
diff --git
a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/AggregateIndexEntriesHandler.java
b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/AggregateIndexEntriesHandler.java
new file mode 100644
index 0000000..1eb9e53
--- /dev/null
+++
b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/AggregateIndexEntriesHandler.java
@@ -0,0 +1,62 @@
+package org.rhq.server.metrics;
+
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import com.datastax.driver.core.ResultSet;
+import com.datastax.driver.core.Row;
+import com.google.common.util.concurrent.FutureCallback;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+/**
+* @author John Sanda
+*/
+class AggregateIndexEntriesHandler implements FutureCallback<ResultSet> {
+
+ private final Log log = LogFactory.getLog(AggregateIndexEntriesHandler.class);
+
+ private Set<Integer> indexEntries;
+
+ private AtomicInteger remainingData;
+
+ private SignalingCountDownLatch indexEntriesArrival;
+
+ private long startTime;
+
+ private String src;
+
+ private String dest;
+
+ public AggregateIndexEntriesHandler(Set<Integer> indexEntries, AtomicInteger
remainingData,
+ SignalingCountDownLatch indexEntriesArrival, long startTime, String src, String
dest) {
+ this.indexEntries = indexEntries;
+ this.remainingData = remainingData;
+ this.indexEntriesArrival = indexEntriesArrival;
+ this.startTime = startTime;
+ this.src = src;
+ this.dest = dest;
+ }
+
+ @Override
+ public void onSuccess(ResultSet resultSet) {
+ for (Row row : resultSet) {
+ indexEntries.add(row.getInt(1));
+ }
+ remainingData.set(indexEntries.size());
+ indexEntriesArrival.countDown();
+ if (log.isDebugEnabled()) {
+ log.debug("Finished loading " + indexEntries.size() + " "
+ src + " index entries in " +
+ (System.currentTimeMillis() - startTime) + " ms");
+ }
+ }
+
+ @Override
+ public void onFailure(Throwable t) {
+ log.warn("Failed to retrieve " + src + " index entries. Some
" + dest + " aggregates may not get generated.",
+ t);
+ remainingData.set(0);
+ indexEntriesArrival.abort();
+ }
+}
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 d0812e2..d2d3f7d 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
@@ -91,6 +91,9 @@ public class Aggregator {
.set6HourTimeSliceFinished(hasTimeSliceEnded(sixHourTimeSlice,
configuration.getOneHourTimeSliceDuration()))
.set24HourTimeSliceFinished(hasTimeSliceEnded(twentyFourHourTimeSlice,
configuration.getSixHourTimeSliceDuration()))
+ .setRemainingRawData(new AtomicInteger(0))
+ .setRemaining1HourData(new AtomicInteger(0))
+ .setRemaining6HourData(new AtomicInteger(0))
.setOneHourIndexEntries(new TreeSet<Integer>())
.setSixHourIndexEntries(new TreeSet<Integer>())
.setOneHourIndexEntriesLock(new ReentrantReadWriteLock())
@@ -139,36 +142,32 @@ public class Aggregator {
@Override
public void onSuccess(ResultSet result) {
List<Row> rows = result.all();
- state.setRemainingRawData(new AtomicInteger(rows.size()));
+ state.getRemainingRawData().set(rows.size());
rawDataIndexEntriesArrival.countDown();
- if (rows.isEmpty()) {
- deleteIndexEntries(MetricsTable.ONE_HOUR);
- } else {
- deleteIndexEntries(MetricsTable.ONE_HOUR);
- log.debug("Starting raw data aggregation for " +
rows.size() + " schedules");
- long start = System.currentTimeMillis();
- final DateTime endTime =
startTime.plus(configuration.getRawTimeSliceDuration());
- Set<Integer> scheduleIds = new TreeSet<Integer>();
- List<StorageResultSetFuture> rawDataFutures = new
ArrayList<StorageResultSetFuture>(batchSize);
- for (final Row row : rows) {
- scheduleIds.add(row.getInt(1));
- readPermits.acquire();
- rawDataFutures.add(dao.findRawMetricsAsync(row.getInt(1),
startTime.getMillis(),
- endTime.getMillis()));
- if (rawDataFutures.size() == batchSize) {
- state.getAggregationTasks().submit(new AggregateRawData(dao,
state, scheduleIds,
- rawDataFutures));
- rawDataFutures = new
ArrayList<StorageResultSetFuture>();
- scheduleIds = new TreeSet<Integer>();
- }
- }
- if (!rawDataFutures.isEmpty()) {
+
+ log.debug("Starting raw data aggregation for " + rows.size() +
" schedules");
+ long start = System.currentTimeMillis();
+ final DateTime endTime =
startTime.plus(configuration.getRawTimeSliceDuration());
+ Set<Integer> scheduleIds = new TreeSet<Integer>();
+ List<StorageResultSetFuture> rawDataFutures = new
ArrayList<StorageResultSetFuture>(batchSize);
+ for (final Row row : rows) {
+ scheduleIds.add(row.getInt(1));
+ readPermits.acquire();
+ rawDataFutures.add(dao.findRawMetricsAsync(row.getInt(1),
startTime.getMillis(),
+ endTime.getMillis()));
+ if (rawDataFutures.size() == batchSize) {
state.getAggregationTasks().submit(new AggregateRawData(dao,
state, scheduleIds,
rawDataFutures));
+ rawDataFutures = new ArrayList<StorageResultSetFuture>();
+ scheduleIds = new TreeSet<Integer>();
}
- log.debug("Finished processing one hour index entries in "
+ (System.currentTimeMillis() - start) +
- " ms");
}
+ if (!rawDataFutures.isEmpty()) {
+ state.getAggregationTasks().submit(new AggregateRawData(dao, state,
scheduleIds,
+ rawDataFutures));
+ }
+ log.debug("Finished processing one hour index entries in " +
(System.currentTimeMillis() - start) +
+ " ms");
}
@Override
@@ -182,75 +181,38 @@ public class Aggregator {
}, state.getAggregationTasks());
if (state.is6HourTimeSliceFinished()) {
+ long start = System.currentTimeMillis();
log.debug("Fetching 1 hour index entries");
StorageResultSetFuture oneHourFuture =
dao.findMetricsIndexEntriesAsync(MetricsTable.SIX_HOUR,
state.getSixHourTimeSlice().getMillis());
- Futures.addCallback(oneHourFuture, new FutureCallback<ResultSet>() {
- @Override
- public void onSuccess(ResultSet resultSet) {
- for (Row row : resultSet) {
- // No need to get the write lock here because there won't be
concurrent writes until after
- // state.getOneHourIndexEntriesArrival().countDown() is called
- state.getOneHourIndexEntries().add(row.getInt(1));
- }
- state.setRemaining1HourData(new
AtomicInteger(state.getOneHourIndexEntries().size()));
- state.getOneHourIndexEntriesArrival().countDown();
- log.debug("Finished loading 1 hour index entries");
- }
-
- @Override
- public void onFailure(Throwable t) {
- log.warn("Failed to retrieve one hour aggregate index entries
for time slice [" +
- get6HourTimeSlice() + "]. Some six hour aggregates may not
get generated.");
- state.getOneHourIndexEntriesArrival().abort();
- state.setRemaining1HourData(new AtomicInteger(0));
- }
- }, state.getAggregationTasks());
+ Futures.addCallback(oneHourFuture, new
AggregateIndexEntriesHandler(state.getOneHourIndexEntries(),
+ state.getRemaining1HourData(), state.getOneHourIndexEntriesArrival(),
start, "1 hour", "6 hour"),
+ state.getAggregationTasks());
}
if (state.is24HourTimeSliceFinished()) {
+ long start = System.currentTimeMillis();
log.debug("Fetching 6 hour index entries");
StorageResultSetFuture sixHourFuture =
dao.findMetricsIndexEntriesAsync(MetricsTable.TWENTY_FOUR_HOUR,
state.getTwentyFourHourTimeSlice().getMillis());
- Futures.addCallback(sixHourFuture, new FutureCallback<ResultSet>() {
- @Override
- public void onSuccess(ResultSet resultSet) {
- for (Row row : resultSet) {
- // No need to get the write lock here because there won't be
concurrent writes until after
- // state.getSixHourIndexEntriesArrival().countDown() is called
- state.getSixHourIndexEntries().add(row.getInt(1));
- }
- state.setRemaining6HourData(new
AtomicInteger(state.getSixHourIndexEntries().size()));
- state.getSixHourIndexEntriesArrival().countDown();
- log.debug("Finished loading 6 hour index entrie");
- }
-
- @Override
- public void onFailure(Throwable t) {
- log.warn("Failed to retrieve 6 hour index entries for time slice
" +
- state.getTwentyFourHourTimeSlice() + ". Some 24 hour
aggregates may not get generated.", t);
- state.getSixHourIndexEntriesArrival().abort();
- state.setRemaining6HourData(new AtomicInteger(0));
- }
- }, state.getAggregationTasks());
+ Futures.addCallback(sixHourFuture, new
AggregateIndexEntriesHandler(state.getSixHourIndexEntries(),
+ state.getRemaining6HourData(), state.getSixHourIndexEntriesArrival(),
start, "6 hour", "24 hour"),
+ state.getAggregationTasks());
}
try {
+ try {
+ rawDataIndexEntriesArrival.await();
+ deleteIndexEntries(MetricsTable.ONE_HOUR);
+ } catch (AbortedException e) {
+ }
+
if (state.is6HourTimeSliceFinished()) {
- // TODO do we need to wait for raw index entries arrival?
+ waitFor(state.getRemainingRawData());
try {
- log.debug("Waiting for arrival of raw data index
entries");
- rawDataIndexEntriesArrival.await();
- } catch (AbortedException e) {
- log.warn("Failed to load raw data index entries. Proceeding with
1 hour data aggregation.");
- log.warn("Remaining raw data: " +
state.getRemainingRawData());
- }
- log.debug("Waiting for raw aggregation to complete");
- waitForRawAggregationToComplete();
- deleteIndexEntries(MetricsTable.SIX_HOUR);
- // Now queue up remaining 1 hour data aggregation
- try {
- log.debug("Remaining schedule ids for 1 hour aggregation: "
+ state.getOneHourIndexEntries());
+ state.getOneHourIndexEntriesArrival().await();
+ deleteIndexEntries(MetricsTable.SIX_HOUR);
+
List<StorageResultSetFuture> queryFutures = new
ArrayList<StorageResultSetFuture>(batchSize);
Set<Integer> scheduleIds = new TreeSet<Integer>();
state.getOneHourIndexEntriesLock().writeLock().lock();
@@ -259,7 +221,6 @@ public class Aggregator {
state.getSixHourTimeSliceEnd().getMillis()));
scheduleIds.add(scheduleId);
if (queryFutures.size() == batchSize) {
- log.debug("Submitting 1 hour aggregation task for
schedule ids " + scheduleIds);
state.getAggregationTasks().submit(new
Aggregate1HourData(dao, state, scheduleIds,
queryFutures));
queryFutures = new
ArrayList<StorageResultSetFuture>(batchSize);
@@ -267,22 +228,24 @@ public class Aggregator {
}
}
if (!queryFutures.isEmpty()) {
- log.debug("Submitting 1 hour aggregation task for schedule
ids " + scheduleIds);
state.getAggregationTasks().submit(new Aggregate1HourData(dao,
state, scheduleIds,
queryFutures));
queryFutures = null;
scheduleIds = null;
}
+ } catch (AbortedException e) {
+ log.warn("Failed to load 1 hour index entries. Some 6 hour
aggregates may not get generated.", e);
} finally {
state.getOneHourIndexEntriesLock().writeLock().unlock();
}
}
if (state.is24HourTimeSliceFinished()) {
- waitFor1HourAggregationToComplete();
- deleteIndexEntries(MetricsTable.TWENTY_FOUR_HOUR);
+ waitFor(state.getRemaining1HourData());
try {
- log.debug("Remaining scheulde ids for 6 hour aggregation: "
+ state.getSixHourIndexEntries());
+ state.getSixHourIndexEntriesArrival().await();
+ deleteIndexEntries(MetricsTable.TWENTY_FOUR_HOUR);
+
List<StorageResultSetFuture> queryFutures = new
ArrayList<StorageResultSetFuture>(batchSize);
Set<Integer> scheduleIds = new TreeSet<Integer>();
state.getSixHourIndexEntriesLock().writeLock().lock();
@@ -291,7 +254,6 @@ public class Aggregator {
state.getTwentyFourHourTimeSliceEnd().getMillis()));
scheduleIds.add(scheduleId);
if (queryFutures.size() == batchSize) {
- log.debug("Submitting 6 hour aggregation task for
schedule ids " + scheduleIds);
state.getAggregationTasks().submit(new
Aggregate6HourData(dao, state, scheduleIds,
queryFutures));
queryFutures = new
ArrayList<StorageResultSetFuture>(batchSize);
@@ -305,6 +267,8 @@ public class Aggregator {
queryFutures = null;
scheduleIds = null;
}
+ } catch (AbortedException e) {
+ log.warn("Failed to load 6 hour index entries. Some 24 hour
aggregates may not get generated.", e);
} finally {
state.getSixHourIndexEntriesLock().writeLock().unlock();
}
@@ -313,54 +277,34 @@ public class Aggregator {
while (!isAggregationFinished()) {
Thread.sleep(50);
}
- log.debug("Aggregation is done. There are " +
remainingIndexEntries.get() + " remaining index entries");
} catch (InterruptedException e) {
log.warn("An interrupt occurred while waiting for aggregation to
finish", e);
}
return oneHourData;
}
- private void waitForRawAggregationToComplete() throws InterruptedException {
- while (state.getRemainingRawData().get() > 0) {
- Thread.sleep(50);
- }
- }
-
- private void waitFor1HourAggregationToComplete() throws InterruptedException {
- while (state.getRemaining1HourData().get() > 0) {
+ private void waitFor(AtomicInteger remainingData) throws InterruptedException {
+ while (remainingData.get() > 0) {
Thread.sleep(50);
}
}
private boolean isAggregationFinished() throws InterruptedException {
- try {
- rawDataIndexEntriesArrival.await();
- } catch (AbortedException e) {
- }
- try {
- state.getOneHourIndexEntriesArrival().await();
- } catch (AbortedException e) {
-
- }
- try {
- state.getSixHourIndexEntriesArrival().await();
- } catch (AbortedException e) {
- }
return state.getRemainingRawData().get() <= 0 &&
state.getRemaining1HourData().get() <= 0 &&
state.getRemaining6HourData().get() <= 0 &&
remainingIndexEntries.get() <= 0;
}
private void deleteIndexEntries(final MetricsTable table) {
- DateTime time;
+ final DateTime time;
switch (table) {
case ONE_HOUR:
time = startTime;
break;
case SIX_HOUR:
- time = get6HourTimeSlice();
+ time = state.getSixHourTimeSlice();
break;
default:
- time = get24HourTimeSlice();
+ time = state.getTwentyFourHourTimeSlice();
break;
}
log.debug("Deleting " + table + " index entries for time slice
" + time);
@@ -374,7 +318,7 @@ public class Aggregator {
@Override
public void onFailure(Throwable t) {
- log.warn("Failed to delete index entries for table " + table +
" at time [" + startTime + "]");
+ log.warn("Failed to delete index entries for table " + table +
" at time [" + time + "]");
remainingIndexEntries.decrementAndGet();
}
});
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 34cc2a9..0741fd2 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
@@ -71,11 +71,17 @@ public class MetricsServer {
private MetricsConfiguration configuration;
+// private RateLimiter readPermits = RateLimiter.create(Integer.parseInt(
+// System.getProperty("rhq.storage.read-limit", "1000")));
+//
+// private RateLimiter writePermits = RateLimiter.create(Integer.parseInt(
+// System.getProperty("rhq.storage.write-limit", "2500")));
+
private RateLimiter readPermits = RateLimiter.create(Integer.parseInt(
System.getProperty("rhq.storage.read-limit", "1200")));
private RateLimiter writePermits = RateLimiter.create(Integer.parseInt(
- System.getProperty("rhq.storage.write-limit", "3100")));
+ System.getProperty("rhq.storage.write-limit", "3000")));
private boolean pastAggregationMissed;