modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/storage/StorageClientManagerBean.java
| 12 ++
modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/Aggregate1HourData.java
| 27 +++++-
modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/AggregateRawData.java
| 6 -
modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/Aggregator.java
| 2
modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsServer.java
| 45 +++++-----
5 files changed, 64 insertions(+), 28 deletions(-)
New commits:
commit f138443b065aad7372a44ff4a3d6d62c3f8318f5
Author: John Sanda <jsanda(a)redhat.com>
Date: Wed Oct 23 15:12:15 2013 -0400
make connection pool sizes and rate limits configurable
Connection pools per host and rate limits are configurable via system
properties. While the connection pool sizes are only set at start up, I plan to
make the rate limits self-tuning.
This commit also adds the missing method body to kick off 6 hour data
aggregation in Aggregate1HourData.java.
diff --git
a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/storage/StorageClientManagerBean.java
b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/storage/StorageClientManagerBean.java
index 98d1405..190614c 100644
---
a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/storage/StorageClientManagerBean.java
+++
b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/storage/StorageClientManagerBean.java
@@ -36,6 +36,8 @@ import javax.ejb.TransactionAttribute;
import javax.ejb.TransactionAttributeType;
import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.HostDistance;
+import com.datastax.driver.core.PoolingOptions;
import com.datastax.driver.core.ProtocolOptions;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.exceptions.NoHostAvailableException;
@@ -254,6 +256,16 @@ public class StorageClientManagerBean {
.withLoadBalancingPolicy(new RoundRobinPolicy())
.withRetryPolicy(new
LoggingRetryPolicy(DefaultRetryPolicy.INSTANCE)).withCompression(compression).build();
+ PoolingOptions poolingOptions = cluster.getConfiguration().getPoolingOptions();
+ poolingOptions.setCoreConnectionsPerHost(HostDistance.LOCAL, Integer.parseInt(
+ System.getProperty("rhq.storage.client.local-connections",
"24")));
+ poolingOptions.setCoreConnectionsPerHost(HostDistance.REMOTE, Integer.parseInt(
+ System.getProperty("rhq.storage.client.remote-connections",
"16")));
+ poolingOptions.setMaxConnectionsPerHost(HostDistance.LOCAL, Integer.parseInt(
+ System.getProperty("rhq.storage.client.max-local-connections",
"32")));
+ poolingOptions.setMaxConnectionsPerHost(HostDistance.REMOTE, Integer.parseInt(
+ System.getProperty("rhq.storage.client.max-remote-connections",
"24")));
+
return cluster.connect(RHQ_KEYSPACE);
}
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 c6584ca..bf50f26 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
@@ -1,5 +1,6 @@
package org.rhq.server.metrics;
+import java.util.ArrayList;
import java.util.List;
import java.util.Set;
@@ -54,19 +55,43 @@ public class Aggregate1HourData implements Runnable {
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()) {
- // TODO
+ 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();
+ }
+ } 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.
+ }
+ List<StorageResultSetFuture> queryFutures = new
ArrayList<StorageResultSetFuture>(scheduleIds.size());
+ for (Integer scheduleId : scheduleIds) {
+ queryFutures.add(dao.findSixHourMetricsAsync(scheduleId,
state.getTwentyFourHourTimeSlice().getMillis(),
+ state.getTwentyFourHourTimeSliceEnd().getMillis()));
+ }
+ state.getAggregationTasks().submit(new Aggregate6HourData(dao, state,
scheduleIds, queryFutures));
}
}
}
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 a7b0695..2cde763 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
@@ -38,7 +38,6 @@ public class AggregateRawData implements Runnable {
@Override
public void run() {
- log.debug("Aggregating raw data for schedule ids " + scheduleIds);
final long start = System.currentTimeMillis();
ListenableFuture<List<ResultSet>> rawDataFutures =
Futures.successfulAsList(queryFutures);
Futures.withFallback(rawDataFutures, new
FutureFallback<List<ResultSet>>() {
@@ -71,8 +70,8 @@ public class AggregateRawData implements Runnable {
}
private void start1HourDataAggregationIfNecessary() {
- log.debug("Aggregating 1 hour data for schedule ids " + scheduleIds);
if (state.is6HourTimeSliceFinished()) {
+ log.debug("Starting 1 hour data aggregation for " +
scheduleIds.size() + " schedules");
try {
state.getOneHourIndexEntriesArrival().await();
try {
@@ -86,11 +85,10 @@ public class AggregateRawData implements Runnable {
return;
} 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
+ // 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.
}
- log.debug("Fetching 1 hour data");
List<StorageResultSetFuture> oneHourDataQueryFutures = new
ArrayList<StorageResultSetFuture>(
scheduleIds.size());
for (Integer scheduleId : scheduleIds) {
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 d2d3f7d..efb7784 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
@@ -134,7 +134,7 @@ public class Aggregator {
}
public Set<AggregateNumericMetric> run() {
- log.debug("There are " + remainingIndexEntries.get() + " remaining
index entries");
+ log.info("Starting aggregation for time slice " + startTime);
readPermits.acquire();
StorageResultSetFuture rawFuture =
dao.findMetricsIndexEntriesAsync(MetricsTable.ONE_HOUR,
startTime.getMillis());
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 0741fd2..05a8fba 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
@@ -33,6 +33,7 @@ import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import com.datastax.driver.core.ResultSet;
@@ -71,17 +72,11 @@ 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")));
+ System.getProperty("rhq.storage.read-limit", "1000")), 3,
TimeUnit.MINUTES);
private RateLimiter writePermits = RateLimiter.create(Integer.parseInt(
- System.getProperty("rhq.storage.write-limit", "3000")));
+ System.getProperty("rhq.storage.write-limit", "2500")), 3,
TimeUnit.MINUTES);
private boolean pastAggregationMissed;
@@ -92,7 +87,7 @@ public class MetricsServer {
private int aggregationBatchSize;
- private boolean useAsyncAggregation;
+ private boolean useAsyncAggregation =
System.getProperty("rhq.metrics.aggregation.async") != null;
public void setDAO(MetricsDAO dao) {
this.dao = dao;
@@ -131,6 +126,9 @@ public class MetricsServer {
}
public void init() {
+ if (log.isDebugEnabled() && useAsyncAggregation) {
+ log.debug("Async aggregation is enabled");
+ }
determineMostRecentRawDataSinceLastShutdown();
}
@@ -428,24 +426,28 @@ public class MetricsServer {
* for subsequently computing baselines.
*/
public Iterable<AggregateNumericMetric> calculateAggregates() {
- DateTime theHour = currentHour();
+ long start = System.currentTimeMillis();
+ try {
+ DateTime theHour = currentHour();
- if (pastAggregationMissed) {
- theHour = roundDownToHour(mostRecentRawDataPriorToStartup).plusHours(1);
- pastAggregationMissed = false;
- }
+ if (pastAggregationMissed) {
+ theHour = roundDownToHour(mostRecentRawDataPriorToStartup).plusHours(1);
+ pastAggregationMissed = false;
+ }
- if (useAsyncAggregation) {
- DateTime timeSlice = theHour.minus(configuration.getRawTimeSliceDuration());
- return new Aggregator(aggregationWorkers, dao, configuration,
dateTimeService, timeSlice,
- aggregationBatchSize, writePermits, readPermits).run();
- } else {
- return calculateAggregates(theHour.getMillis());
+ if (useAsyncAggregation) {
+ DateTime timeSlice =
theHour.minus(configuration.getRawTimeSliceDuration());
+ return new Aggregator(aggregationWorkers, dao, configuration,
dateTimeService, timeSlice,
+ aggregationBatchSize, writePermits, readPermits).run();
+ } else {
+ return calculateAggregates(theHour.getMillis());
+ }
+ } finally {
+ log.info("Finished metrics aggregation in " +
(System.currentTimeMillis() - start) + " ms");
}
}
private Iterable<AggregateNumericMetric> calculateAggregates(long startTime) {
- long start = System.currentTimeMillis();
DateTime dt = new DateTime(startTime);
DateTime currentHour = dateTimeService.getTimeSlice(dt,
configuration.getRawTimeSliceDuration());
DateTime lastHour = currentHour.minus(configuration.getRawTimeSliceDuration());
@@ -486,7 +488,6 @@ public class MetricsServer {
dao.deleteMetricsIndexEntries(MetricsTable.TWENTY_FOUR_HOUR,
twentyFourHourTimeSlice);
}
- log.info("Finished aggregation in " + (System.currentTimeMillis() -
start) + " ms");
return newOneHourAggregates;
}