modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/Aggregate1HourData.java
| 58 +++++---
modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/AggregateRawData.java
| 65 ++++++----
modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/AggregationState.java
| 20 +++
modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/Aggregator.java
| 6
4 files changed, 100 insertions(+), 49 deletions(-)
New commits:
commit 5180691f7af23d517bf07e0590fe03d5ab33a006
Author: John Sanda <jsanda(a)redhat.com>
Date: Wed Dec 18 17:24:01 2013 -0500
[BZ 1009945] eliminate a couple possible race condition when processing index entries
diff --git
a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/Aggregate1HourData.java
b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/Aggregate1HourData.java
index bf50f26..4a5e78b 100644
---
a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/Aggregate1HourData.java
+++
b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/Aggregate1HourData.java
@@ -14,6 +14,10 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
/**
+ * Generates 6 hour data for a batch of 1 hour data futures. After data is inserted for
the batch, aggregation of 6 hour
+ * data will start immediately for the batch if the 24 hour time slice has finished.
+ *
+ * @see Compute6HourData
* @author John Sanda
*/
public class Aggregate1HourData implements Runnable {
@@ -54,44 +58,52 @@ public class Aggregate1HourData implements Runnable {
public void onSuccess(List<ResultSet> result) {
log.debug("Finished aggregating 1 hour data for " +
result.size() + " schedules in " +
(System.currentTimeMillis() - start) + " ms");
- state.getRemaining1HourData().addAndGet(-scheduleIds.size());
start6HourDataAggregationIfNecessary();
}
@Override
public void onFailure(Throwable t) {
log.warn("Failed to aggregate 1 hour data", t);
- state.getRemaining1HourData().addAndGet(-scheduleIds.size());
start6HourDataAggregationIfNecessary();
}
});
}
private void start6HourDataAggregationIfNecessary() {
- if (state.is24HourTimeSliceFinished()) {
- log.debug("Starting 6 hour data aggregation for " +
scheduleIds.size() + " schedules");
- try {
- state.getSixHourIndexEntriesArrival().await();
- try {
- state.getSixHourIndexEntriesLock().writeLock().lock();
- state.getSixHourIndexEntries().removeAll(scheduleIds);
- } finally {
- state.getSixHourIndexEntriesLock().writeLock().unlock();
+ try {
+ if (state.is24HourTimeSliceFinished()) {
+ update6HourIndexEntries();
+ List<StorageResultSetFuture> queryFutures = new
ArrayList<StorageResultSetFuture>(scheduleIds.size());
+ for (Integer scheduleId : scheduleIds) {
+ queryFutures.add(dao.findSixHourMetricsAsync(scheduleId,
state.getTwentyFourHourTimeSlice().getMillis(),
+ state.getTwentyFourHourTimeSliceEnd().getMillis()));
}
- } catch (InterruptedException e) {
- log.warn("An interrupt occurred waiting for 6 hour index
entries", e);
- } catch (AbortedException e) {
- // This means we failed to retrieve the index entries. We can however
- // continue generating 6 hour data because we do not need the index
- // here since we already have 6 hour data to aggregate along with the
- // schedule ids.
+ state.getAggregationTasks().submit(new Aggregate6HourData(dao, state,
scheduleIds, queryFutures));
}
- List<StorageResultSetFuture> queryFutures = new
ArrayList<StorageResultSetFuture>(scheduleIds.size());
- for (Integer scheduleId : scheduleIds) {
- queryFutures.add(dao.findSixHourMetricsAsync(scheduleId,
state.getTwentyFourHourTimeSlice().getMillis(),
- state.getTwentyFourHourTimeSliceEnd().getMillis()));
+ } catch (InterruptedException e) {
+ log.debug("An interrupt occurred while waiting for 6 hour data index
entries. Aborting data aggregation",
+ e);
+ log.info("An interrupt occurred while waiting for 6 hour data index
entries. Aborting data aggregation: " +
+ e.getMessage());
+ } finally {
+ state.getRemaining1HourData().addAndGet(-scheduleIds.size());
+ }
+ }
+
+ private void update6HourIndexEntries() throws InterruptedException {
+ try {
+ state.getSixHourIndexEntriesArrival().await();
+ try {
+ state.getSixHourIndexEntriesLock().writeLock().lock();
+ state.getSixHourIndexEntries().removeAll(scheduleIds);
+ } finally {
+ state.getSixHourIndexEntriesLock().writeLock().unlock();
}
- state.getAggregationTasks().submit(new Aggregate6HourData(dao, state,
scheduleIds, queryFutures));
+ } catch (AbortedException e) {
+ // This means we failed to retrieve the index entries. We can however
+ // continue generating 6 hour data because we do not need the index
+ // here since we already have 6 hour data to aggregate along with the
+ // schedule ids.
}
}
}
diff --git
a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/AggregateRawData.java
b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/AggregateRawData.java
index 2cde763..dbfc289 100644
---
a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/AggregateRawData.java
+++
b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/AggregateRawData.java
@@ -14,6 +14,10 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
/**
+ * Generates 1 hour data for a batch of raw data futures. After data is inserted for the
batch, aggregation of 1 hour
+ * data will start immediately for the batch if the 6 hour time slice has finished.
+ *
+ * @see Compute1HourData
* @author John Sanda
*/
public class AggregateRawData implements Runnable {
@@ -55,7 +59,6 @@ public class AggregateRawData implements Runnable {
public void onSuccess(List<ResultSet> resultSets) {
log.debug("Finished aggregating raw data for " +
resultSets.size() + " schedules in " +
(System.currentTimeMillis() - start) + " ms");
- state.getRemainingRawData().addAndGet(-scheduleIds.size());
start1HourDataAggregationIfNecessary();
}
@@ -63,39 +66,51 @@ public class AggregateRawData implements Runnable {
public void onFailure(Throwable t) {
log.warn("Failed to aggregate raw data", t);
// TODO maybe add debug statement to log those schedule ids for which
aggregation failed
- state.getRemainingRawData().addAndGet(-scheduleIds.size());
start1HourDataAggregationIfNecessary();
}
}, state.getAggregationTasks());
}
private void start1HourDataAggregationIfNecessary() {
- if (state.is6HourTimeSliceFinished()) {
- log.debug("Starting 1 hour data aggregation for " +
scheduleIds.size() + " schedules");
- try {
- state.getOneHourIndexEntriesArrival().await();
- try {
- state.getOneHourIndexEntriesLock().writeLock().lock();
- state.getOneHourIndexEntries().removeAll(scheduleIds);
- } finally {
- state.getOneHourIndexEntriesLock().writeLock().unlock();
+ try {
+ if (state.is6HourTimeSliceFinished()) {
+ update1HourIndexEntries();
+ List<StorageResultSetFuture> oneHourDataQueryFutures = new
ArrayList<StorageResultSetFuture>(
+ scheduleIds.size());
+ for (Integer scheduleId : scheduleIds) {
+ oneHourDataQueryFutures.add(dao.findOneHourMetricsAsync(scheduleId,
+ state.getSixHourTimeSlice().getMillis(),
state.getSixHourTimeSliceEnd().getMillis()));
}
- } catch (InterruptedException e) {
- log.warn("An interrupt occurred waiting for one hour data index
entries", e);
- return;
- } catch (AbortedException e) {
- // This means we failed to retrieve the index entries. We can however
- // continue generating 1 hour data because we do not need the index
- // here since we already have 1 hour data to aggregate along with the
- // schedule ids.
+ state.getAggregationTasks().submit(new Aggregate1HourData(dao, state,
scheduleIds,
+ oneHourDataQueryFutures));
}
- List<StorageResultSetFuture> oneHourDataQueryFutures = new
ArrayList<StorageResultSetFuture>(
- scheduleIds.size());
- for (Integer scheduleId : scheduleIds) {
- oneHourDataQueryFutures.add(dao.findOneHourMetricsAsync(scheduleId,
- state.getSixHourTimeSlice().getMillis(),
state.getSixHourTimeSliceEnd().getMillis()));
+ } catch (InterruptedException e) {
+ log.debug("An interrupt occurred while waiting for 1 hour data index
entries. Aborting data aggregation",
+ e);
+ log.info("An interrupt occurred while waiting for 1 hour data index
entries. Aborting data aggregation: " +
+ e.getMessage());
+ } finally {
+ state.getRemainingRawData().addAndGet(-scheduleIds.size());
+ }
+ }
+
+ private void update1HourIndexEntries() throws InterruptedException {
+ try {
+ // Wait for the arrival so that we can remove the schedules ids in this
+ // batch from the one hour index entries. This will prevent duplicate tasks
+ // being submitted to process the same 1 hour data.
+ state.getOneHourIndexEntriesArrival().await();
+ try {
+ state.getOneHourIndexEntriesLock().writeLock().lock();
+ state.getOneHourIndexEntries().removeAll(scheduleIds);
+ } finally {
+ state.getOneHourIndexEntriesLock().writeLock().unlock();
}
- state.getAggregationTasks().submit(new Aggregate1HourData(dao, state,
scheduleIds, oneHourDataQueryFutures));
+ } catch (AbortedException e) {
+ // This means we failed to retrieve the index entries. We can however
+ // continue generating 1 hour data because we do not need the index
+ // here since we already have 1 hour data to aggregate along with the
+ // schedule ids.
}
}
}
diff --git
a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/AggregationState.java
b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/AggregationState.java
index 1b67350..bd1ce0d 100644
---
a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/AggregationState.java
+++
b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/AggregationState.java
@@ -62,6 +62,10 @@ public class AggregationState {
return this;
}
+ /**
+ * @return A {@link SignalingCountDownLatch} to signal the arrival of index entries
for schedules with 1 hour
+ * data to be aggregated
+ */
public SignalingCountDownLatch getOneHourIndexEntriesArrival() {
return oneHourIndexEntriesArrival;
}
@@ -71,6 +75,10 @@ public class AggregationState {
return this;
}
+ /**
+ * @return A {@link SignalingCountDownLatch} to signal the arrival of index entries
for schedules with 6 hour
+ * data to be aggregated
+ */
public SignalingCountDownLatch getSixHourIndexEntriesArrival() {
return sixHourIndexEntriesArrival;
}
@@ -80,6 +88,9 @@ public class AggregationState {
return this;
}
+ /**
+ * @return The remaining number of schedules with raw data to be aggregated
+ */
public AtomicInteger getRemainingRawData() {
return remainingRawData;
}
@@ -89,6 +100,9 @@ public class AggregationState {
return this;
}
+ /**
+ * @return The remaining number of schedules with 1 hour data to be aggregated
+ */
public AtomicInteger getRemaining1HourData() {
return remaining1HourData;
}
@@ -98,6 +112,9 @@ public class AggregationState {
return this;
}
+ /**
+ * @return The remaining number of schedules with 6 hour data to be aggregated
+ */
public AtomicInteger getRemaining6HourData() {
return remaining6HourData;
}
@@ -107,6 +124,9 @@ public class AggregationState {
return this;
}
+ /**
+ * @return The schedule ids with 1 hour data to be aggregated
+ */
public Set<Integer> getOneHourIndexEntries() {
return oneHourIndexEntries;
}
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 f9bba5f..4ff222a 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
@@ -48,6 +48,10 @@ public class Aggregator {
private DateTime startTime;
+ /**
+ * Signals when raw data index entries (in metrics_index) can be deleted. We cannot
delete the row in metrics_index
+ * until we know that it has been read, successfully or otherwise.
+ */
private SignalingCountDownLatch rawDataIndexEntriesArrival;
private RateLimiter readPermits;
@@ -289,7 +293,7 @@ public class Aggregator {
}
}
- private boolean isAggregationFinished() throws InterruptedException {
+ private boolean isAggregationFinished() {
return state.getRemainingRawData().get() <= 0 &&
state.getRemaining1HourData().get() <= 0 &&
state.getRemaining6HourData().get() <= 0 &&
remainingIndexEntries.get() <= 0;
}