modules/common/cassandra-util/src/main/java/org/rhq/cassandra/util/ClusterBuilder.java
| 8
modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/cassandra/SessionManagerBean.java
| 5
modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/cloud/LoadInfo.java
| 75 ++++++
modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementDataManagerBean.java
| 38 ++-
modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementDataManagerLocal.java
| 2
modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MetricsManagerBean.java
| 4
modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MetricsManagerLocal.java
| 2
modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsDAO.java
| 10
modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsServer.java
| 119 +++++++--
modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/RawDataInsertedCallback.java
| 39 +++
modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/CassandraIntegrationTest.java
| 40 +++
modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsDAOTest.java
| 59 ++++
modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsServerTest.java
| 113 +++++++--
modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/TimeoutTest.java
| 123 ++++++++++
modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/MeasurementCollector.java
| 19 +
modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/Simulator.java
| 3
16 files changed, 604 insertions(+), 55 deletions(-)
New commits:
commit cc2d6c042251ff58a1ebc8d0f4df091d02fae0eb
Merge: 39ad031 f0a6431
Author: John Sanda <jsanda(a)redhat.com>
Date: Fri May 24 21:51:16 2013 -0400
Merge branch 'async-cql' into feature/cassandra-backend
Conflicts:
modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsBaselineCalculator.java
modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsServer.java
commit f0a643191697efc5183450b8a35603aaec08a64d
Author: John Sanda <jsanda(a)redhat.com>
Date: Fri May 24 21:39:03 2013 -0400
removing sync version of MetricsServer.updateIndex
This commit also updates tsts in MetricsServerTest to block on async calls in
order to make the test logic simpler.
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 880f21b..66d9a4b 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
@@ -26,8 +26,6 @@
package org.rhq.server.metrics;
import java.util.ArrayList;
-import java.util.Date;
-import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -309,29 +307,7 @@ public class MetricsServer {
}
}
- private void persistRawData(final Set<MeasurementDataNumeric> dataSet, final
RawDataInsertedCallback callback) {
- final long startTime = System.currentTimeMillis();
- final AtomicInteger remainingInserts = new AtomicInteger(dataSet.size());
-
- for (final MeasurementDataNumeric data : dataSet) {
- ResultSetFuture resultSetFuture = dao.insertRawData(data);
- Futures.addCallback(resultSetFuture, new FutureCallback<ResultSet>() {
- @Override
- public void onSuccess(ResultSet rows) {
- updateMetricsIndex(data, dataSet.size(), remainingInserts, startTime,
callback);
- }
-
- @Override
- public void onFailure(Throwable throwable) {
- log.error("An error occurred while inserting raw data " +
data, throwable);
- callback.onFailure(throwable);
- semaphore.release();
- }
- });
- }
- }
-
- private void updateMetricsIndex(final MeasurementDataNumeric rawData, final int
total,
+ void updateMetricsIndex(final MeasurementDataNumeric rawData, final int total,
final AtomicInteger remainingInserts, final long startTime, final
RawDataInsertedCallback callback) {
long timeSlice = dateTimeService.getTimeSlice(new
DateTime(rawData.getTimestamp()),
@@ -362,20 +338,6 @@ public class MetricsServer {
});
}
- void updateMetricsIndex(Set<MeasurementDataNumeric> rawMetrics) {
- Map<Integer, Long> updates = new TreeMap<Integer, Long>();
- for (MeasurementDataNumeric rawMetric : rawMetrics) {
- updates.put(rawMetric.getScheduleId(), dateTimeService.getTimeSlice(
- new DateTime(rawMetric.getTimestamp()),
configuration.getRawTimeSliceDuration()).getMillis());
- }
- Set<Date> dates = new HashSet<Date>();
- for (Long ts : updates.values()) {
- dates.add(new Date(ts));
- }
- log.debug("Updating one hour index wtih time slices " + dates);
- dao.updateMetricsIndex(MetricsTable.ONE_HOUR, updates);
- }
-
/**
* Computes and stores aggregate for all time slices that are ready to be
aggregated.
* This is includes raw, 1hr, 6hr, and 24hr data.
diff --git
a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/CassandraIntegrationTest.java
b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/CassandraIntegrationTest.java
index 9bec034..8e311e1 100644
---
a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/CassandraIntegrationTest.java
+++
b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/CassandraIntegrationTest.java
@@ -25,14 +25,21 @@
package org.rhq.server.metrics;
+import static org.testng.Assert.fail;
+
import java.util.Date;
+import java.util.concurrent.CountDownLatch;
import com.datastax.driver.core.BoundStatement;
import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.exceptions.NoHostAvailableException;
+import com.google.common.base.Throwables;
+import com.google.common.util.concurrent.FutureCallback;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
import org.joda.time.DateTime;
import org.testng.annotations.AfterSuite;
import org.testng.annotations.BeforeSuite;
@@ -58,7 +65,7 @@ public class CassandraIntegrationTest {
private static DateTimeService dateTimeService;
@BeforeSuite
- @DeployCluster(numNodes = 2, username = "cassandra", password =
"cassandra")
+ @DeployCluster(numNodes = 1, username = "cassandra", password =
"cassandra")
public void deployCluster() throws Exception {
dateTimeService = new DateTimeService();
@@ -108,4 +115,37 @@ public class CassandraIntegrationTest {
throw new CQLException(e);
}
}
+
+ protected static class WaitForResults<ResultSetFuture> implements
FutureCallback<ResultSetFuture> {
+
+ private final Log log = LogFactory.getLog(WaitForResults.class);
+
+ private CountDownLatch latch;
+
+ private Throwable throwable;
+
+ public WaitForResults(int numResults) {
+ latch = new CountDownLatch(numResults);
+ }
+
+ @Override
+ public void onSuccess(ResultSetFuture resultSetFuture) {
+ latch.countDown();
+ }
+
+ @Override
+ public void onFailure(Throwable throwable) {
+ latch.countDown();
+ this.throwable = throwable;
+ log.error("An async operation failed", throwable);
+ }
+
+ public void await(String errorMsg) throws InterruptedException {
+ latch.await();
+ if (throwable != null) {
+ fail(errorMsg, Throwables.getRootCause(throwable));
+ }
+ }
+
+ }
}
diff --git
a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsDAOTest.java
b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsDAOTest.java
index c522316..6195016 100644
---
a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsDAOTest.java
+++
b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsDAOTest.java
@@ -29,7 +29,6 @@ import static java.util.Arrays.asList;
import static org.rhq.test.AssertUtils.assertCollectionMatchesNoOrder;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotNull;
-import static org.testng.Assert.fail;
import java.util.ArrayList;
import java.util.HashMap;
@@ -38,12 +37,9 @@ import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;
-import java.util.concurrent.CountDownLatch;
import com.datastax.driver.core.ResultSetFuture;
-import com.google.common.base.Throwables;
import com.google.common.collect.Lists;
-import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import org.apache.commons.logging.Log;
@@ -672,37 +668,4 @@ public class MetricsDAOTest extends CassandraIntegrationTest {
}
}
- private static class WaitForResults<ResultSetFuture> implements
FutureCallback<ResultSetFuture> {
-
- private final Log log = LogFactory.getLog(WaitForResults.class);
-
- private CountDownLatch latch;
-
- private Throwable throwable;
-
- public WaitForResults(int numResults) {
- latch = new CountDownLatch(numResults);
- }
-
- @Override
- public void onSuccess(ResultSetFuture resultSetFuture) {
- latch.countDown();
- }
-
- @Override
- public void onFailure(Throwable throwable) {
- latch.countDown();
- this.throwable = throwable;
- log.error("An async operation failed", throwable);
- }
-
- public void await(String errorMsg) throws InterruptedException {
- latch.await();
- if (throwable != null) {
- fail(errorMsg, Throwables.getRootCause(throwable));
- }
- }
-
- }
-
}
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 cf74fcd..d89b2d9 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
@@ -41,9 +41,12 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicInteger;
+import com.datastax.driver.core.ResultSetFuture;
import com.google.common.base.Throwables;
import com.google.common.collect.Lists;
+import com.google.common.util.concurrent.Futures;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -187,10 +190,10 @@ public class MetricsServerTest extends CassandraIntegrationTest {
metricsServer.setCurrentHour(hour6);
metricsServer.addNumericData(data, waitForRawInserts);
- metricsServer.calculateAggregates(hour6.getMillis());
-
waitForRawInserts.await("Failed to insert raw data");
+ metricsServer.calculateAggregates(hour6.getMillis());
+
// verify that one hour metric data is updated
List<AggregateNumericMetric> expected = asList(new
AggregateNumericMetric(scheduleId,
divide((3.9 + 3.2 + 2.6), 3), 2.6, 3.9, lastHour.getMillis()));
@@ -205,7 +208,7 @@ public class MetricsServerTest extends CassandraIntegrationTest {
}
@Test//(enabled = ENABLED)
- public void aggregateRawDataDuring9thHour() {
+ public void aggregateRawDataDuring9thHour() throws Exception {
int scheduleId = 123;
DateTime hour0 = hour0();
@@ -225,10 +228,21 @@ public class MetricsServerTest extends CassandraIntegrationTest {
rawMetrics.add(new MeasurementDataNumeric(secondMetricTime.getMillis(),
scheduleId, secondValue));
rawMetrics.add(new MeasurementDataNumeric(thirdMetricTime.getMillis(),
scheduleId, thirdValue));
+ WaitForResults waitForRawInserts = new WaitForResults(rawMetrics.size());
+
+ for (MeasurementDataNumeric raw : rawMetrics) {
+ ResultSetFuture resultSetFuture = dao.insertRawData(raw);
+ Futures.addCallback(resultSetFuture, waitForRawInserts);
+ }
+ waitForRawInserts.await("Failed to insert raw data");
+
+ WaitForRawInserts waitForIndexUpdates = new
WaitForRawInserts(rawMetrics.size());
+ AtomicInteger remainingUpdates = new AtomicInteger(rawMetrics.size());
for (MeasurementDataNumeric raw : rawMetrics) {
- dao.insertRawData(raw);
+ metricsServer.updateMetricsIndex(raw, rawMetrics.size(), remainingUpdates,
hour8.getMillis(),
+ waitForIndexUpdates);
}
- metricsServer.updateMetricsIndex(rawMetrics);
+ waitForIndexUpdates.await("Failed to update metrics index for raw
data");
metricsServer.setCurrentHour(hour9);
metricsServer.calculateAggregates(hour9.getMillis());
@@ -382,11 +396,11 @@ public class MetricsServerTest extends CassandraIntegrationTest {
WaitForRawInserts waitForRawInserts = new WaitForRawInserts(data.size());
metricsServer.addNumericData(data, waitForRawInserts);
+ waitForRawInserts.await("Failed to insert raw data");
+
List<MeasurementDataNumericHighLowComposite> actualData =
Lists.newArrayList(metricsServer.findDataForResource(
scheduleId, beginTime.getMillis(), endTime.getMillis()));
- waitForRawInserts.await("Faile to insert raw data");
-
assertEquals(actualData.size(), buckets.getNumDataPoints(), "Expected to get
back 60 data points.");
MeasurementDataNumericHighLowComposite expectedBucket0Data = new
MeasurementDataNumericHighLowComposite(
diff --git
a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/TimeoutTest.java
b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/TimeoutTest.java
index 2f960fe..b3c30a5 100644
---
a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/TimeoutTest.java
+++
b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/TimeoutTest.java
@@ -60,25 +60,25 @@ public class TimeoutTest extends CassandraIntegrationTest {
long time = hour0().getMillis();
Set<MeasurementDataNumeric> data1 = new
HashSet<MeasurementDataNumeric>();
- for (int i = 0; i < 500000; ++i) {
+ for (int i = 0; i < 50000; ++i) {
data1.add(new MeasurementDataNumeric(time, i, (double) i));
}
- Set<MeasurementDataNumeric> data2 = new
HashSet<MeasurementDataNumeric>();
- for (int i = 0; i < 500000; ++i) {
- data2.add(new MeasurementDataNumeric(time, i + data1.size(), (double) i));
- }
+// Set<MeasurementDataNumeric> data2 = new
HashSet<MeasurementDataNumeric>();
+// for (int i = 0; i < 50000; ++i) {
+// data2.add(new MeasurementDataNumeric(time, i + data1.size(), (double) i));
+// }
WaitForRawInserts waitForRawInserts1 = new WaitForRawInserts(data1.size());
- WaitForRawInserts waitForRawInserts2 = new WaitForRawInserts(data2.size());
+// WaitForRawInserts waitForRawInserts2 = new WaitForRawInserts(data2.size());
long start = System.currentTimeMillis();
metricsServer.addNumericData(data1, waitForRawInserts1);
waitForRawInserts1.await("Failed to insert raw metrics");
- log.info("Sleep before second round of inserts...");
- Thread.sleep(3000);
-
- metricsServer.addNumericData(data2, waitForRawInserts2);
+// log.info("Sleep before second round of inserts...");
+// Thread.sleep(3000);
+//
+// metricsServer.addNumericData(data2, waitForRawInserts2);
long end = System.currentTimeMillis();
log.info("Inserted " + data1.size() + " raw metrics in " +
(end - start) + " ms");
commit 3ddfe5272d9ad38ffc3006ec7f8c2a5afe46a4f8
Author: John Sanda <jsanda(a)redhat.com>
Date: Fri May 24 13:20:20 2013 -0400
remove extra call to updateMetricsIndex and break up test into 2 large inserts
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 6f1187c..880f21b 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
@@ -303,7 +303,6 @@ public class MetricsServer {
}
});
}
- updateMetricsIndex(dataSet);
} catch (Exception e) {
log.error("An error occurred while inserting raw numeric data",
e);
throw new RuntimeException(e);
diff --git
a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/CassandraIntegrationTest.java
b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/CassandraIntegrationTest.java
index f59324d..9bec034 100644
---
a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/CassandraIntegrationTest.java
+++
b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/CassandraIntegrationTest.java
@@ -58,7 +58,7 @@ public class CassandraIntegrationTest {
private static DateTimeService dateTimeService;
@BeforeSuite
- @DeployCluster(numNodes = 1, username = "cassandra", password =
"cassandra")
+ @DeployCluster(numNodes = 2, username = "cassandra", password =
"cassandra")
public void deployCluster() throws Exception {
dateTimeService = new DateTimeService();
diff --git
a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/TimeoutTest.java
b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/TimeoutTest.java
index 4496d33..2f960fe 100644
---
a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/TimeoutTest.java
+++
b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/TimeoutTest.java
@@ -59,19 +59,29 @@ public class TimeoutTest extends CassandraIntegrationTest {
metricsServer.setDAO(dao);
long time = hour0().getMillis();
- Set<MeasurementDataNumeric> data = new
HashSet<MeasurementDataNumeric>();
- for (int i = 0; i < 10000; ++i) {
- data.add(new MeasurementDataNumeric(time, i, (double) i));
+ Set<MeasurementDataNumeric> data1 = new
HashSet<MeasurementDataNumeric>();
+ for (int i = 0; i < 500000; ++i) {
+ data1.add(new MeasurementDataNumeric(time, i, (double) i));
}
- WaitForRawInserts waitForRawInserts = new WaitForRawInserts(data.size());
+ Set<MeasurementDataNumeric> data2 = new
HashSet<MeasurementDataNumeric>();
+ for (int i = 0; i < 500000; ++i) {
+ data2.add(new MeasurementDataNumeric(time, i + data1.size(), (double) i));
+ }
+
+ WaitForRawInserts waitForRawInserts1 = new WaitForRawInserts(data1.size());
+ WaitForRawInserts waitForRawInserts2 = new WaitForRawInserts(data2.size());
long start = System.currentTimeMillis();
- metricsServer.addNumericData(data, waitForRawInserts);
+ metricsServer.addNumericData(data1, waitForRawInserts1);
+ waitForRawInserts1.await("Failed to insert raw metrics");
+
+ log.info("Sleep before second round of inserts...");
+ Thread.sleep(3000);
- waitForRawInserts.await("Failed to insert raw metrics");
+ metricsServer.addNumericData(data2, waitForRawInserts2);
long end = System.currentTimeMillis();
- log.info("Inserted " + data.size() + " raw metrics in " +
(end - start) + " ms");
+ log.info("Inserted " + data1.size() + " raw metrics in " +
(end - start) + " ms");
}
private static class WaitForRawInserts implements RawDataInsertedCallback {
commit 70f3fa68b0f8ab9aafe520df938de0e277365fde
Author: John Sanda <jsanda(a)redhat.com>
Date: Fri May 24 09:00:45 2013 -0400
throttle async inserts with a semaphore
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 5e49d45..6f1187c 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
@@ -32,7 +32,9 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executors;
+import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicInteger;
import com.datastax.driver.core.ResultSet;
@@ -78,6 +80,32 @@ public class MetricsServer {
private ListeningExecutorService listeningExecutorService =
MoreExecutors.listeningDecorator(
Executors.newFixedThreadPool(5));
+ private ConcurrentLinkedQueue<MeasurementDataNumeric> rawDataQueue =
+ new ConcurrentLinkedQueue<MeasurementDataNumeric>();
+
+ private Semaphore semaphore = new Semaphore(100);
+
+ private boolean shutdown = false;
+
+ public MetricsServer() {
+// listeningExecutorService.submit(new Runnable() {
+// @Override
+// public void run() {
+// while (!shutdown) {
+// try {
+// if (!rawDataQueue.isEmpty()) {
+// semaphore.acquire();
+// MeasurementDataNumeric rawData = rawDataQueue.poll();
+// persistRawData();
+// }
+// } catch (InterruptedException e) {
+//
+// }
+// }
+// }
+// });
+ }
+
public void setSession(Session session) {
this.session = session;
}
@@ -94,6 +122,10 @@ public class MetricsServer {
this.dateTimeService = dateTimeService;
}
+ public void shutdown() {
+ shutdown = true;
+ }
+
public RawNumericMetric findLatestValueForResource(int scheduleId) {
return dao.findLatestRawMetric(scheduleId);
}
@@ -246,33 +278,58 @@ public class MetricsServer {
public void addNumericData(final Set<MeasurementDataNumeric> dataSet,
final RawDataInsertedCallback callback) {
-// try {
+ try {
if (log.isDebugEnabled()) {
log.debug("Inserting " + dataSet.size() + " raw
metrics");
}
- final long startTime = System.currentTimeMillis();
- final AtomicInteger remainingInserts = new AtomicInteger(dataSet.size());
-
- for (final MeasurementDataNumeric data : dataSet) {
- ResultSetFuture resultSetFuture = dao.insertRawData(data);
- Futures.addCallback(resultSetFuture, new
FutureCallback<ResultSet>() {
- @Override
- public void onSuccess(ResultSet rows) {
- updateMetricsIndex(data, dataSet.size(), remainingInserts,
startTime, callback);
- }
- @Override
- public void onFailure(Throwable throwable) {
- log.error("An error occurred while inserting raw data "
+ data, throwable);
- callback.onFailure(throwable);
- }
- });
- }
-// updateMetricsIndex(dataSet);
-// } catch (Exception e) {
-// log.error("An error occurred while inserting raw numeric data",
e);
-// throw new RuntimeException(e);
-// }
+ final long startTime = System.currentTimeMillis();
+ final AtomicInteger remainingInserts = new AtomicInteger(dataSet.size());
+
+ for (final MeasurementDataNumeric data : dataSet) {
+ semaphore.acquire();
+ ResultSetFuture resultSetFuture = dao.insertRawData(data);
+ Futures.addCallback(resultSetFuture, new FutureCallback<ResultSet>() {
+ @Override
+ public void onSuccess(ResultSet rows) {
+ updateMetricsIndex(data, dataSet.size(), remainingInserts, startTime,
callback);
+ }
+
+ @Override
+ public void onFailure(Throwable throwable) {
+ log.error("An error occurred while inserting raw data " +
data, throwable);
+ callback.onFailure(throwable);
+ semaphore.release();
+ }
+ });
+ }
+ updateMetricsIndex(dataSet);
+ } catch (Exception e) {
+ log.error("An error occurred while inserting raw numeric data",
e);
+ throw new RuntimeException(e);
+ }
+ }
+
+ private void persistRawData(final Set<MeasurementDataNumeric> dataSet, final
RawDataInsertedCallback callback) {
+ final long startTime = System.currentTimeMillis();
+ final AtomicInteger remainingInserts = new AtomicInteger(dataSet.size());
+
+ for (final MeasurementDataNumeric data : dataSet) {
+ ResultSetFuture resultSetFuture = dao.insertRawData(data);
+ Futures.addCallback(resultSetFuture, new FutureCallback<ResultSet>() {
+ @Override
+ public void onSuccess(ResultSet rows) {
+ updateMetricsIndex(data, dataSet.size(), remainingInserts, startTime,
callback);
+ }
+
+ @Override
+ public void onFailure(Throwable throwable) {
+ log.error("An error occurred while inserting raw data " +
data, throwable);
+ callback.onFailure(throwable);
+ semaphore.release();
+ }
+ });
+ }
}
private void updateMetricsIndex(final MeasurementDataNumeric rawData, final int
total,
@@ -293,6 +350,7 @@ public class MetricsServer {
}
callback.onFinish();
}
+ semaphore.release();
}
@Override
@@ -300,6 +358,7 @@ public class MetricsServer {
log.error("An error occurred while trying to update " +
MetricsTable.INDEX + " for raw data " +
rawData);
callback.onFailure(throwable);
+ semaphore.release();
}
});
}
commit 0c4801d681161aa6edf16bf998a22bb8319d254c
Author: John Sanda <jsanda(a)redhat.com>
Date: Fri May 24 08:59:01 2013 -0400
log total insertion time
diff --git
a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/TimeoutTest.java
b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/TimeoutTest.java
index f32fd66..4496d33 100644
---
a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/TimeoutTest.java
+++
b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/TimeoutTest.java
@@ -41,6 +41,8 @@ import org.rhq.core.domain.measurement.MeasurementDataNumeric;
public class TimeoutTest extends CassandraIntegrationTest {
+ private final Log log = LogFactory.getLog(TimeoutTest.class);
+
@Test
public void generateTimeout() throws Exception {
MetricsConfiguration configuration = new MetricsConfiguration();
@@ -58,14 +60,18 @@ public class TimeoutTest extends CassandraIntegrationTest {
long time = hour0().getMillis();
Set<MeasurementDataNumeric> data = new
HashSet<MeasurementDataNumeric>();
- for (int i = 0; i < 5000; ++i) {
+ for (int i = 0; i < 10000; ++i) {
data.add(new MeasurementDataNumeric(time, i, (double) i));
}
WaitForRawInserts waitForRawInserts = new WaitForRawInserts(data.size());
+ long start = System.currentTimeMillis();
metricsServer.addNumericData(data, waitForRawInserts);
waitForRawInserts.await("Failed to insert raw metrics");
+ long end = System.currentTimeMillis();
+
+ log.info("Inserted " + data.size() + " raw metrics in " +
(end - start) + " ms");
}
private static class WaitForRawInserts implements RawDataInsertedCallback {
commit bd117a2bb465a879f57223958d8430c914be7b0d
Author: John Sanda <jsanda(a)redhat.com>
Date: Fri May 24 07:37:35 2013 -0400
adding test to generate timeout exception in driver due to large number of inserts
diff --git
a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/TimeoutTest.java
b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/TimeoutTest.java
new file mode 100644
index 0000000..f32fd66
--- /dev/null
+++
b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/TimeoutTest.java
@@ -0,0 +1,107 @@
+/*
+ *
+ * * 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.server.metrics;
+
+import static org.testng.Assert.fail;
+
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+
+import com.google.common.base.Throwables;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.testng.annotations.Test;
+
+import org.rhq.core.domain.measurement.MeasurementDataNumeric;
+
+public class TimeoutTest extends CassandraIntegrationTest {
+
+ @Test
+ public void generateTimeout() throws Exception {
+ MetricsConfiguration configuration = new MetricsConfiguration();
+
+ MetricsServer metricsServer = new MetricsServer();
+ metricsServer.setSession(session);
+ metricsServer.setConfiguration(configuration);
+
+ DateTimeService dateTimeService = new DateTimeService();
+ dateTimeService.setConfiguration(configuration);
+ metricsServer.setDateTimeService(dateTimeService);
+
+ MetricsDAO dao = new MetricsDAO(session, configuration);
+ metricsServer.setDAO(dao);
+
+ long time = hour0().getMillis();
+ Set<MeasurementDataNumeric> data = new
HashSet<MeasurementDataNumeric>();
+ for (int i = 0; i < 5000; ++i) {
+ data.add(new MeasurementDataNumeric(time, i, (double) i));
+ }
+
+ WaitForRawInserts waitForRawInserts = new WaitForRawInserts(data.size());
+ metricsServer.addNumericData(data, waitForRawInserts);
+
+ waitForRawInserts.await("Failed to insert raw metrics");
+ }
+
+ private static class WaitForRawInserts implements RawDataInsertedCallback {
+
+ private final Log log = LogFactory.getLog(WaitForRawInserts.class);
+
+ private CountDownLatch latch;
+
+ private Throwable throwable;
+
+ public WaitForRawInserts(int numInserts) {
+ latch = new CountDownLatch(numInserts);
+ }
+
+ @Override
+ public void onFinish() {
+ }
+
+ @Override
+ public void onSuccess(MeasurementDataNumeric measurementDataNumeric) {
+ latch.countDown();
+ }
+
+ @Override
+ public void onFailure(Throwable throwable) {
+ latch.countDown();
+ this.throwable = throwable;
+ log.error("An async operation failed", throwable);
+ }
+
+ public void await(String errorMsg) throws InterruptedException {
+ latch.await();
+ if (throwable != null) {
+ fail(errorMsg, Throwables.getRootCause(throwable));
+ }
+ }
+ }
+
+}
commit e03eb8443c1dad9628452505295813ad3f3bb074
Author: John Sanda <jsanda(a)redhat.com>
Date: Thu May 23 20:13:59 2013 -0400
initial implementation for using the CQL driver's async API for inserting raw
metrics
diff --git
a/modules/common/cassandra-util/src/main/java/org/rhq/cassandra/util/ClusterBuilder.java
b/modules/common/cassandra-util/src/main/java/org/rhq/cassandra/util/ClusterBuilder.java
index 6587c71..4687b4b 100644
---
a/modules/common/cassandra-util/src/main/java/org/rhq/cassandra/util/ClusterBuilder.java
+++
b/modules/common/cassandra-util/src/main/java/org/rhq/cassandra/util/ClusterBuilder.java
@@ -26,6 +26,7 @@
package org.rhq.cassandra.util;
import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.PoolingOptions;
import com.datastax.driver.core.ProtocolOptions;
/**
@@ -50,6 +51,13 @@ public class ClusterBuilder {
}
/**
+ * @see com.datastax.driver.core.Cluster.Builder#poolingOptions()
+ */
+ public PoolingOptions poolingOptions() {
+ return builder.poolingOptions();
+ }
+
+ /**
* @see Cluster.Builder#withCredentials(String, String)
*/
public ClusterBuilder withCredentials(String username, String password) {
diff --git
a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/cassandra/SessionManagerBean.java
b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/cassandra/SessionManagerBean.java
index 6af6ea8..911261e 100644
---
a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/cassandra/SessionManagerBean.java
+++
b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/cassandra/SessionManagerBean.java
@@ -33,6 +33,8 @@ import javax.ejb.EJB;
import javax.ejb.Singleton;
import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.HostDistance;
+import com.datastax.driver.core.PoolingOptions;
import com.datastax.driver.core.Session;
import org.rhq.cassandra.util.ClusterBuilder;
@@ -90,6 +92,9 @@ public class SessionManagerBean {
.withCredentials(username, password)
.withPort(port)
.build();
+ PoolingOptions poolingOptions =
cluster.getConfiguration().getPoolingOptions();
+ poolingOptions.setMaxConnectionsPerHost(HostDistance.LOCAL, 16);
+ poolingOptions.setMaxConnectionsPerHost(HostDistance.REMOTE, 16);
session = cluster.connect("rhq");
metricsDAO = new MetricsDAO(session, metricsConfiguration);
diff --git
a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/cloud/LoadInfo.java
b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/cloud/LoadInfo.java
new file mode 100644
index 0000000..5cf05c2
--- /dev/null
+++
b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/cloud/LoadInfo.java
@@ -0,0 +1,75 @@
+/*
+ *
+ * * 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.enterprise.server.cloud;
+
+import org.rhq.enterprise.server.measurement.MeasurementAggregate;
+
+/**
+ * @author John Sanda
+ */
+public class LoadInfo {
+
+ private String diskSpaceUsed;
+
+ private String ringOwnership;
+
+ private MeasurementAggregate heapCommitted;
+
+ private MeasurementAggregate heapUsage;
+
+ public String getDiskSpaceUsed() {
+ return diskSpaceUsed;
+ }
+
+ public void setDiskSpaceUsed(String diskSpaceUsed) {
+ this.diskSpaceUsed = diskSpaceUsed;
+ }
+
+ public String getRingOwnership() {
+ return ringOwnership;
+ }
+
+ public void setRingOwnership(String ringOwnership) {
+ this.ringOwnership = ringOwnership;
+ }
+
+ public MeasurementAggregate getHeapCommitted() {
+ return heapCommitted;
+ }
+
+ public void setHeapCommitted(MeasurementAggregate heapCommitted) {
+ this.heapCommitted = heapCommitted;
+ }
+
+ public MeasurementAggregate getHeapUsage() {
+ return heapUsage;
+ }
+
+ public void setHeapUsage(MeasurementAggregate heapUsage) {
+ this.heapUsage = heapUsage;
+ }
+
+}
diff --git
a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementDataManagerBean.java
b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementDataManagerBean.java
index 2f300c3..9686dea 100644
---
a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementDataManagerBean.java
+++
b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementDataManagerBean.java
@@ -34,6 +34,7 @@ import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeoutException;
+import javax.ejb.Asynchronous;
import javax.ejb.EJB;
import javax.ejb.Stateless;
import javax.ejb.TransactionAttribute;
@@ -87,12 +88,15 @@ import
org.rhq.enterprise.server.alert.engine.AlertConditionCacheManagerLocal;
import org.rhq.enterprise.server.alert.engine.AlertConditionCacheStats;
import org.rhq.enterprise.server.authz.AuthorizationManagerLocal;
import org.rhq.enterprise.server.authz.PermissionException;
+import org.rhq.enterprise.server.cassandra.SessionManagerBean;
import org.rhq.enterprise.server.core.AgentManagerLocal;
import org.rhq.enterprise.server.measurement.instrumentation.MeasurementMonitor;
import org.rhq.enterprise.server.measurement.util.MeasurementDataManagerUtility;
import org.rhq.enterprise.server.resource.group.ResourceGroupManagerLocal;
import org.rhq.enterprise.server.util.CriteriaQueryGenerator;
import org.rhq.enterprise.server.util.CriteriaQueryRunner;
+import org.rhq.server.metrics.MetricsServer;
+import org.rhq.server.metrics.RawDataInsertedCallback;
/**
* A manager for {@link MeasurementData}s.
@@ -152,6 +156,9 @@ public class MeasurementDataManagerBean implements
MeasurementDataManagerLocal,
private MetricsManagerLocal metricsManager;
@EJB
+ private SessionManagerBean sessionManagerBean;
+
+ @EJB
private MeasurementScheduleManagerLocal measurementScheduleManager;
// doing a bulk delete in here, need to be in its own tx
@@ -214,13 +221,31 @@ public class MeasurementDataManagerBean implements
MeasurementDataManagerLocal,
* @param data the actual data points
*/
@TransactionAttribute(TransactionAttributeType.NOT_SUPPORTED)
- public void addNumericData(Set<MeasurementDataNumeric> data) {
+ public void addNumericData(final Set<MeasurementDataNumeric> data) {
if ((data == null) || (data.isEmpty())) {
return;
}
- metricsManager.addNumericData(data);
- notifyAlertConditionCacheManager("mergeMeasurementReport",
data.toArray(new MeasurementData[data.size()]));
+ MetricsServer metricsServer = sessionManagerBean.getMetricsServer();
+ metricsServer.addNumericData(data, new RawDataInsertedCallback() {
+
+ private List<MeasurementData> insertedData = new
ArrayList<MeasurementData>();
+
+ @Override
+ public void onFinish() {
+
measurementDataManager.updateAlertConditionCache("mergeMeasurementReport",
+ insertedData.toArray(new MeasurementData[insertedData.size()]));
+ }
+
+ @Override
+ public void onSuccess(MeasurementDataNumeric measurementDataNumeric) {
+ insertedData.add(measurementDataNumeric);
+ }
+
+ @Override
+ public void onFailure(Throwable throwable) {
+ }
+ });
}
@TransactionAttribute(TransactionAttributeType.REQUIRES_NEW)
@@ -548,6 +573,13 @@ public class MeasurementDataManagerBean implements
MeasurementDataManagerLocal,
return metricsManager.findLatestValueForResource(scheduleId);
}
+ @Asynchronous
+ @Override
+ public void updateAlertConditionCache(String callingMethod, MeasurementData[] data)
{
+ AlertConditionCacheStats stats =
alertConditionCacheManager.checkConditions(data);
+ log.debug(callingMethod + ": " + stats.toString());
+ }
+
private void notifyAlertConditionCacheManager(String callingMethod, MeasurementData[]
data) {
AlertConditionCacheStats stats =
alertConditionCacheManager.checkConditions(data);
diff --git
a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementDataManagerLocal.java
b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementDataManagerLocal.java
index 644107d..cc4a86e 100644
---
a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementDataManagerLocal.java
+++
b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementDataManagerLocal.java
@@ -140,4 +140,6 @@ public interface MeasurementDataManagerLocal extends
MeasurementDataManagerRemot
MeasurementAggregate getAggregate(Subject subject, int groupId, int definitionId,
long startTime, long endTime);
+ void updateAlertConditionCache(String callingMethod, MeasurementData[] data);
+
}
\ No newline at end of file
diff --git
a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MetricsManagerBean.java
b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MetricsManagerBean.java
index 26d580b..c77698b 100644
---
a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MetricsManagerBean.java
+++
b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MetricsManagerBean.java
@@ -51,9 +51,9 @@ public class MetricsManagerBean implements MetricsManagerLocal {
@Override
@TransactionAttribute(TransactionAttributeType.NOT_SUPPORTED)
- public void addNumericData(Set<MeasurementDataNumeric> data) {
+ public void addNumericData(Set<MeasurementDataNumeric> data, Runnable callback)
{
MetricsServer metricsServer = getMetricsServer();
- metricsServer.addNumericData(data);
+ metricsServer.addNumericData(data, null);
}
@Override
diff --git
a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MetricsManagerLocal.java
b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MetricsManagerLocal.java
index 9880d1f..c49dd75 100644
---
a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MetricsManagerLocal.java
+++
b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MetricsManagerLocal.java
@@ -45,7 +45,7 @@ import org.rhq.server.metrics.domain.AggregateNumericMetric;
@Local
public interface MetricsManagerLocal {
- void addNumericData(Set<MeasurementDataNumeric> data);
+ void addNumericData(Set<MeasurementDataNumeric> data, Runnable callback);
Iterable<AggregateNumericMetric> calculateAggregates();
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 dd5c44c..6ad9755 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
@@ -37,6 +37,7 @@ import java.util.Set;
import com.datastax.driver.core.BoundStatement;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.ResultSet;
+import com.datastax.driver.core.ResultSetFuture;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.Statement;
import com.datastax.driver.core.exceptions.NoHostAvailableException;
@@ -141,10 +142,10 @@ public class MetricsDAO {
deleteIndexEntries = session.prepare("DELETE FROM " +
MetricsTable.INDEX + " WHERE bucket = ? AND time = ?");
}
- public ResultSet insertRawData(MeasurementDataNumeric data) {
+ public ResultSetFuture insertRawData(MeasurementDataNumeric data) {
BoundStatement statement = insertRawData.bind(data.getScheduleId(), new
Date(data.getTimestamp()),
data.getValue());
- return session.execute(statement);
+ return session.executeAsync(statement);
}
public List<MetricResultFuture<MeasurementDataNumeric>>
insertRawMetricsAsync(Set<MeasurementDataNumeric> dataSet,
@@ -301,6 +302,11 @@ public class MetricsDAO {
}
}
+ public ResultSetFuture updateMetricsIndex(MetricsTable table, int scheduleId, long
timestamp) {
+ BoundStatement statement = updateMetricsIndex.bind(table.getTableName(), new
Date(timestamp), scheduleId);
+ return session.executeAsync(statement);
+ }
+
public void deleteMetricsIndexEntries(MetricsTable table, long timestamp) {
BoundStatement statement = deleteIndexEntries.bind(table.getTableName(), new
Date(timestamp));
session.execute(statement);
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 fd285ee..5e49d45 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
@@ -32,8 +32,16 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicInteger;
+import com.datastax.driver.core.ResultSet;
+import com.datastax.driver.core.ResultSetFuture;
import com.datastax.driver.core.Session;
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -67,6 +75,9 @@ public class MetricsServer {
private MetricsConfiguration configuration;
+ private ListeningExecutorService listeningExecutorService =
MoreExecutors.listeningDecorator(
+ Executors.newFixedThreadPool(5));
+
public void setSession(Session session) {
this.session = session;
}
@@ -233,16 +244,64 @@ public class MetricsServer {
return data;
}
- public void addNumericData(Set<MeasurementDataNumeric> dataSet) {
- try {
- for (MeasurementDataNumeric data : dataSet) {
- dao.insertRawData(data);
+ public void addNumericData(final Set<MeasurementDataNumeric> dataSet,
+ final RawDataInsertedCallback callback) {
+// try {
+ if (log.isDebugEnabled()) {
+ log.debug("Inserting " + dataSet.size() + " raw
metrics");
}
- updateMetricsIndex(dataSet);
- } catch (Exception e) {
- log.error("An error occurred while inserting raw numeric data",
e);
- throw new RuntimeException(e);
- }
+ final long startTime = System.currentTimeMillis();
+ final AtomicInteger remainingInserts = new AtomicInteger(dataSet.size());
+
+ for (final MeasurementDataNumeric data : dataSet) {
+ ResultSetFuture resultSetFuture = dao.insertRawData(data);
+ Futures.addCallback(resultSetFuture, new
FutureCallback<ResultSet>() {
+ @Override
+ public void onSuccess(ResultSet rows) {
+ updateMetricsIndex(data, dataSet.size(), remainingInserts,
startTime, callback);
+ }
+
+ @Override
+ public void onFailure(Throwable throwable) {
+ log.error("An error occurred while inserting raw data "
+ data, throwable);
+ callback.onFailure(throwable);
+ }
+ });
+ }
+// updateMetricsIndex(dataSet);
+// } catch (Exception e) {
+// log.error("An error occurred while inserting raw numeric data",
e);
+// throw new RuntimeException(e);
+// }
+ }
+
+ private void updateMetricsIndex(final MeasurementDataNumeric rawData, final int
total,
+ final AtomicInteger remainingInserts, final long startTime, final
RawDataInsertedCallback callback) {
+
+ long timeSlice = dateTimeService.getTimeSlice(new
DateTime(rawData.getTimestamp()),
+ configuration.getRawTimeSliceDuration()).getMillis();
+ ResultSetFuture resultSetFuture = dao.updateMetricsIndex(MetricsTable.ONE_HOUR,
rawData.getScheduleId(),
+ timeSlice);
+ Futures.addCallback(resultSetFuture, new FutureCallback<ResultSet>() {
+ @Override
+ public void onSuccess(ResultSet rows) {
+ callback.onSuccess(rawData);
+ if (remainingInserts.decrementAndGet() == 0) {
+ long endTime = System.currentTimeMillis();
+ if (log.isDebugEnabled()) {
+ log.debug("Finished inserting " + total + " raw
metrics in " + (endTime - startTime) + " ms");
+ }
+ callback.onFinish();
+ }
+ }
+
+ @Override
+ public void onFailure(Throwable throwable) {
+ log.error("An error occurred while trying to update " +
MetricsTable.INDEX + " for raw data " +
+ rawData);
+ callback.onFailure(throwable);
+ }
+ });
}
void updateMetricsIndex(Set<MeasurementDataNumeric> rawMetrics) {
@@ -255,7 +314,7 @@ public class MetricsServer {
for (Long ts : updates.values()) {
dates.add(new Date(ts));
}
- log.info("Updating one hour index wtih time slices " + dates);
+ log.debug("Updating one hour index wtih time slices " + dates);
dao.updateMetricsIndex(MetricsTable.ONE_HOUR, updates);
}
diff --git
a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/RawDataInsertedCallback.java
b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/RawDataInsertedCallback.java
new file mode 100644
index 0000000..ab9e50a
--- /dev/null
+++
b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/RawDataInsertedCallback.java
@@ -0,0 +1,39 @@
+/*
+ *
+ * * 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.server.metrics;
+
+import com.google.common.util.concurrent.FutureCallback;
+
+import org.rhq.core.domain.measurement.MeasurementDataNumeric;
+
+/**
+ * @author John Sanda
+ */
+public interface RawDataInsertedCallback extends
FutureCallback<MeasurementDataNumeric> {
+
+ void onFinish();
+
+}
diff --git
a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsDAOTest.java
b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsDAOTest.java
index 8f35486..c522316 100644
---
a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsDAOTest.java
+++
b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsDAOTest.java
@@ -29,6 +29,7 @@ import static java.util.Arrays.asList;
import static org.rhq.test.AssertUtils.assertCollectionMatchesNoOrder;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.fail;
import java.util.ArrayList;
import java.util.HashMap;
@@ -37,9 +38,16 @@ import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import com.datastax.driver.core.ResultSetFuture;
+import com.google.common.base.Throwables;
import com.google.common.collect.Lists;
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
import org.joda.time.DateTime;
import org.joda.time.Hours;
import org.testng.annotations.BeforeClass;
@@ -59,6 +67,8 @@ import org.rhq.server.metrics.domain.RawNumericMetric;
*/
public class MetricsDAOTest extends CassandraIntegrationTest {
+ private final Log log = LogFactory.getLog(MetricsDAOTest.class);
+
private static final boolean ENABLED = true;
private final long SECOND = 1000;
@@ -84,16 +94,19 @@ public class MetricsDAOTest extends CassandraIntegrationTest {
}
@Test(enabled = ENABLED)
- public void insertAndFindRawData() {
+ public void insertAndFindRawData() throws Exception {
DateTime hour0 = hour0();
DateTime currentTime = hour0.plusHours(4).plusMinutes(44);
- DateTime currentHour = currentTime.hourOfDay().roundFloorCopy();
DateTime threeMinutesAgo = currentTime.minusMinutes(3);
int scheduleId = 1;
MeasurementDataNumeric expected = new
MeasurementDataNumeric(threeMinutesAgo.getMillis(), scheduleId, 1.23);
- dao.insertRawData(expected);
+ WaitForResults waitForResults = new WaitForResults(1);
+
+ ResultSetFuture resultSetFuture = dao.insertRawData(expected);
+ Futures.addCallback(resultSetFuture, waitForResults);
+ waitForResults.await("Failed to insert raw data");
List<RawNumericMetric> actualMetrics =
Lists.newArrayList(dao.findRawMetrics(scheduleId,
threeMinutesAgo.minusSeconds(1).getMillis(),
threeMinutesAgo.plusSeconds(1).getMillis()));
@@ -104,7 +117,7 @@ public class MetricsDAOTest extends CassandraIntegrationTest {
}
@Test(enabled = ENABLED)
- public void findLatestRawMetric() {
+ public void findLatestRawMetric() throws Exception {
DateTime hour0 = hour0();
DateTime currentTime = hour0.plusHours(4).plusMinutes(44);
DateTime threeMinutesAgo = currentTime.minusMinutes(3);
@@ -118,9 +131,13 @@ public class MetricsDAOTest extends CassandraIntegrationTest {
data.add(new MeasurementDataNumeric(twoMinutesAgo.getMillis(), scheduleId,
3.9));
data.add(new MeasurementDataNumeric(oneMinuteAgo.getMillis(), scheduleId, 2.6));
+ WaitForResults waitForResults = new WaitForResults(data.size());
+
for (MeasurementDataNumeric raw : data) {
- dao.insertRawData(raw);
+ ResultSetFuture resultSetFuture = dao.insertRawData(raw);
+ Futures.addCallback(resultSetFuture, waitForResults);
}
+ waitForResults.await("Failed to insert raw data");
RawNumericMetric actual = dao.findLatestRawMetric(scheduleId);
RawNumericMetric expected = new RawNumericMetric(scheduleId,
oneMinuteAgo.getMillis(), 2.6);
@@ -129,7 +146,7 @@ public class MetricsDAOTest extends CassandraIntegrationTest {
}
@Test(enabled = ENABLED)
- public void findRawMetricsForMultipleSchedules() {
+ public void findRawMetricsForMultipleSchedules() throws Exception {
DateTime currentTime = hour0().plusHours(4).plusMinutes(44);
DateTime currentHour = currentTime.hourOfDay().roundFloorCopy();
DateTime threeMinutesAgo = currentTime.minusMinutes(3);
@@ -147,9 +164,13 @@ public class MetricsDAOTest extends CassandraIntegrationTest {
data.add(new MeasurementDataNumeric(oneMinuteAgo.getMillis(), scheduleId1,
3.1));
data.add(new MeasurementDataNumeric(oneMinuteAgo.getMillis(), scheduleId2,
3.2));
+ WaitForResults waitForResults = new WaitForResults(data.size());
+
for (MeasurementDataNumeric raw : data) {
- dao.insertRawData(raw);
+ ResultSetFuture resultSetFuture = dao.insertRawData(raw);
+ Futures.addCallback(resultSetFuture, waitForResults);
}
+ waitForResults.await("Failed to insert raw data");
List<RawNumericMetric> actualMetrics =
Lists.newArrayList(dao.findRawMetrics(asList(scheduleId1, scheduleId2),
currentHour.getMillis(), currentHour.plusHours(1).getMillis()));
@@ -261,6 +282,34 @@ public class MetricsDAOTest extends CassandraIntegrationTest {
}
@Test(enabled = ENABLED)
+ public void updateSixHourMetricsIndex() throws Exception {
+ int scheduleId1 = 1;
+ int scheduleId2= 2;
+
+ WaitForResults waitForResults = new WaitForResults(2);
+
+ ResultSetFuture resultSetFuture1 =
dao.updateMetricsIndex(MetricsTable.TWENTY_FOUR_HOUR, scheduleId1,
+ hour0().getMillis());
+ ResultSetFuture resultSetFuture2 =
dao.updateMetricsIndex(MetricsTable.TWENTY_FOUR_HOUR, scheduleId2,
+ hour0().getMillis());
+
+ Futures.addCallback(resultSetFuture1, waitForResults);
+ Futures.addCallback(resultSetFuture2, waitForResults);
+
+ waitForResults.await("Failed to update metrics index");
+
+ List<MetricsIndexEntry> expected = asList(
+ new MetricsIndexEntry(MetricsTable.TWENTY_FOUR_HOUR, hour0(), scheduleId1),
+ new MetricsIndexEntry(MetricsTable.TWENTY_FOUR_HOUR, hour0(), scheduleId2)
+ );
+
+ List<MetricsIndexEntry> actual =
Lists.newArrayList(dao.findMetricsIndexEntries(MetricsTable.TWENTY_FOUR_HOUR,
+ hour0().getMillis()));
+
+ assertCollectionMatchesNoOrder(expected, actual, "Failed to update or
retrieve metrics index entries");
+ }
+
+ @Test(enabled = ENABLED)
public void updateAndFindOneHourIndexEntries() {
DateTime hour0 = hour0();
int scheduleId1 = 1;
@@ -623,4 +672,37 @@ public class MetricsDAOTest extends CassandraIntegrationTest {
}
}
+ private static class WaitForResults<ResultSetFuture> implements
FutureCallback<ResultSetFuture> {
+
+ private final Log log = LogFactory.getLog(WaitForResults.class);
+
+ private CountDownLatch latch;
+
+ private Throwable throwable;
+
+ public WaitForResults(int numResults) {
+ latch = new CountDownLatch(numResults);
+ }
+
+ @Override
+ public void onSuccess(ResultSetFuture resultSetFuture) {
+ latch.countDown();
+ }
+
+ @Override
+ public void onFailure(Throwable throwable) {
+ latch.countDown();
+ this.throwable = throwable;
+ log.error("An async operation failed", throwable);
+ }
+
+ public void await(String errorMsg) throws InterruptedException {
+ latch.await();
+ if (throwable != null) {
+ fail(errorMsg, Throwables.getRootCause(throwable));
+ }
+ }
+
+ }
+
}
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 d895753..cf74fcd 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
@@ -31,6 +31,7 @@ import static org.rhq.test.AssertUtils.assertCollectionMatchesNoOrder;
import static org.rhq.test.AssertUtils.assertPropertiesMatch;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.fail;
import java.math.BigDecimal;
import java.math.MathContext;
@@ -39,7 +40,9 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import com.google.common.base.Throwables;
import com.google.common.collect.Lists;
import org.apache.commons.logging.Log;
@@ -137,8 +140,12 @@ public class MetricsServerTest extends CassandraIntegrationTest {
data.add(new MeasurementDataNumeric(twoMinutesAgo.getMillis(), scheduleId,
3.9));
data.add(new MeasurementDataNumeric(oneMinuteAgo.getMillis(), scheduleId, 2.6));
+ WaitForRawInserts waitForRawInserts = new WaitForRawInserts(data.size());
+
long timestamp = System.currentTimeMillis();
- metricsServer.addNumericData(data);
+ metricsServer.addNumericData(data, waitForRawInserts);
+
+ waitForRawInserts.await("Failed to insert raw data");
List<RawNumericMetric> actual =
Lists.newArrayList(dao.findRawMetrics(scheduleId, hour0.plusHours(4)
.getMillis(), hour0.plusHours(5).getMillis()));
@@ -161,7 +168,7 @@ public class MetricsServerTest extends CassandraIntegrationTest {
}
@Test//(enabled = ENABLED)
- public void calculateAggregatesForOneScheduleWhenDBIsEmpty() {
+ public void calculateAggregatesForOneScheduleWhenDBIsEmpty() throws Exception {
int scheduleId = 123;
DateTime hour0 = hour0();
@@ -176,10 +183,14 @@ public class MetricsServerTest extends CassandraIntegrationTest {
data.add(new MeasurementDataNumeric(secondMetricTime.getMillis(), scheduleId,
3.9));
data.add(new MeasurementDataNumeric(thirdMetricTime.getMillis(), scheduleId,
2.6));
+ WaitForRawInserts waitForRawInserts = new WaitForRawInserts(data.size());
+
metricsServer.setCurrentHour(hour6);
- metricsServer.addNumericData(data);
+ metricsServer.addNumericData(data, waitForRawInserts);
metricsServer.calculateAggregates(hour6.getMillis());
+ waitForRawInserts.await("Failed to insert raw data");
+
// verify that one hour metric data is updated
List<AggregateNumericMetric> expected = asList(new
AggregateNumericMetric(scheduleId,
divide((3.9 + 3.2 + 2.6), 3), 2.6, 3.9, lastHour.getMillis()));
@@ -349,7 +360,7 @@ public class MetricsServerTest extends CassandraIntegrationTest {
}
@Test//(enabled = ENABLED)
- public void findRawDataCompositesForResource() {
+ public void findRawDataCompositesForResource() throws Exception {
DateTime beginTime = now().minusHours(4);
DateTime endTime = now();
Buckets buckets = new Buckets(beginTime, endTime);
@@ -368,10 +379,14 @@ public class MetricsServerTest extends CassandraIntegrationTest {
data.add(new MeasurementDataNumeric(buckets.get(59).getStartTime() +
buckets.getInterval() + 50, scheduleId,
4.56));
- metricsServer.addNumericData(data);
+ WaitForRawInserts waitForRawInserts = new WaitForRawInserts(data.size());
+
+ metricsServer.addNumericData(data, waitForRawInserts);
List<MeasurementDataNumericHighLowComposite> actualData =
Lists.newArrayList(metricsServer.findDataForResource(
scheduleId, beginTime.getMillis(), endTime.getMillis()));
+ waitForRawInserts.await("Faile to insert raw data");
+
assertEquals(actualData.size(), buckets.getNumDataPoints(), "Expected to get
back 60 data points.");
MeasurementDataNumericHighLowComposite expectedBucket0Data = new
MeasurementDataNumericHighLowComposite(
@@ -390,7 +405,7 @@ public class MetricsServerTest extends CassandraIntegrationTest {
}
@Test
- public void findLatestValueForResource() {
+ public void findLatestValueForResource() throws Exception {
int scheduleId = 123;
DateTime fifteenMinutesAgo = now().minusMinutes(15);
@@ -402,7 +417,11 @@ public class MetricsServerTest extends CassandraIntegrationTest {
data.add(new MeasurementDataNumeric(tenMinutesAgo.getMillis(), scheduleId,
2.2));
data.add(new MeasurementDataNumeric(fiveMinutesAgo.getMillis(), scheduleId,
3.3));
- metricsServer.addNumericData(data);
+ WaitForRawInserts waitForRawInserts = new WaitForRawInserts(data.size());
+
+ metricsServer.addNumericData(data, waitForRawInserts);
+
+ waitForRawInserts.await("Failed to insert raw data");
RawNumericMetric actual = metricsServer.findLatestValueForResource(scheduleId);
RawNumericMetric expected = new RawNumericMetric(scheduleId,
fiveMinutesAgo.getMillis(), 3.3);
@@ -411,7 +430,7 @@ public class MetricsServerTest extends CassandraIntegrationTest {
}
@Test
- public void getSummaryRawAggregateForResource() {
+ public void getSummaryRawAggregateForResource() throws Exception {
DateTime beginTime = now().minusHours(4);
DateTime endTime = now();
Buckets buckets = new Buckets(beginTime, endTime);
@@ -430,7 +449,11 @@ public class MetricsServerTest extends CassandraIntegrationTest {
data.add(new MeasurementDataNumeric(buckets.get(59).getStartTime() +
buckets.getInterval() + 50, scheduleId,
4.56));
- metricsServer.addNumericData(data);
+ WaitForRawInserts waitForRawInserts = new WaitForRawInserts(data.size());
+
+ metricsServer.addNumericData(data, waitForRawInserts);
+
+ waitForRawInserts.await("Failed to insert raw data");
AggregateNumericMetric actual = metricsServer.getSummaryAggregate(scheduleId,
beginTime.getMillis(),
endTime.getMillis());
@@ -511,7 +534,7 @@ public class MetricsServerTest extends CassandraIntegrationTest {
}
@Test
- public void getSummaryRawAggregateForGroup() {
+ public void getSummaryRawAggregateForGroup() throws Exception {
DateTime beginTime = now().minusHours(4);
DateTime endTime = now();
Buckets buckets = new Buckets(beginTime, endTime);
@@ -545,7 +568,11 @@ public class MetricsServerTest extends CassandraIntegrationTest {
data.add(new MeasurementDataNumeric(buckets.get(59).getStartTime() +
buckets.getInterval() + 50, scheduleId2,
4.56));
- metricsServer.addNumericData(data);
+ WaitForRawInserts waitForRawInserts = new WaitForRawInserts(data.size());
+
+ metricsServer.addNumericData(data, waitForRawInserts);
+
+ waitForRawInserts.await("Failed to insert raw data");
AggregateNumericMetric actual =
metricsServer.getSummaryAggregate(asList(scheduleId1, scheduleId2),
beginTime.getMillis(), endTime.getMillis());
@@ -559,7 +586,7 @@ public class MetricsServerTest extends CassandraIntegrationTest {
}
@Test
- public void findRawDataCompositesForGroup() {
+ public void findRawDataCompositesForGroup() throws Exception {
DateTime beginTime = now().minusHours(4);
DateTime endTime = now();
Buckets buckets = new Buckets(beginTime, endTime);
@@ -593,7 +620,11 @@ public class MetricsServerTest extends CassandraIntegrationTest {
data.add(new MeasurementDataNumeric(buckets.get(59).getStartTime() +
buckets.getInterval() + 50, scheduleId2,
4.56));
- metricsServer.addNumericData(data);
+ WaitForRawInserts waitForRawInserts = new WaitForRawInserts(data.size());
+
+ metricsServer.addNumericData(data, waitForRawInserts);
+
+ waitForRawInserts.await("Failed to insert raw data");
List<MeasurementDataNumericHighLowComposite> actualData =
metricsServer.findDataForGroup(
asList(scheduleId1, scheduleId2), beginTime.getMillis(),
endTime.getMillis());
@@ -829,4 +860,40 @@ public class MetricsServerTest extends CassandraIntegrationTest {
"WHERE schedule_id = " + scheduleId + " AND time >= "
+ startTime + " AND time < " + endTime;
return new SimplePagedResult<RawNumericMetric>(cql, new
RawNumericMetricMapper(true), session);
}
+
+ private static class WaitForRawInserts implements RawDataInsertedCallback {
+
+ private final Log log = LogFactory.getLog(WaitForRawInserts.class);
+
+ private CountDownLatch latch;
+
+ private Throwable throwable;
+
+ public WaitForRawInserts(int numInserts) {
+ latch = new CountDownLatch(numInserts);
+ }
+
+ @Override
+ public void onFinish() {
+ }
+
+ @Override
+ public void onSuccess(MeasurementDataNumeric measurementDataNumeric) {
+ latch.countDown();
+ }
+
+ @Override
+ public void onFailure(Throwable throwable) {
+ latch.countDown();
+ this.throwable = throwable;
+ log.error("An async operation failed", throwable);
+ }
+
+ public void await(String errorMsg) throws InterruptedException {
+ latch.await();
+ if (throwable != null) {
+ fail(errorMsg, Throwables.getRootCause(throwable));
+ }
+ }
+ }
}
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 35f575f..b0a67c8 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
@@ -36,6 +36,7 @@ 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;
/**
* @author John Sanda
@@ -56,6 +57,8 @@ public class MeasurementCollector implements Runnable {
private int batchSize = 500;
+ private NoOpCallback rawInsertsCallback = new NoOpCallback();
+
public void setQueue(PriorityQueue<Schedule> queue) {
this.queue = queue;
}
@@ -113,7 +116,7 @@ public class MeasurementCollector implements Runnable {
}
metricsCollected = data.size();
try {
- metricsServer.addNumericData(data);
+ metricsServer.addNumericData(data, rawInsertsCallback);
} catch (Exception e) {
log.error("An error occurred while trying to store raw
metrics", e);
log.error("Requesting simulation shutdown...");
@@ -135,4 +138,18 @@ public class MeasurementCollector implements Runnable {
log.info("Finished collecting and storing " + metricsCollected +
" raw metric in " +totalTime + " ms.");
}
}
+
+ 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 681eb16..3dacf23 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
@@ -58,6 +58,7 @@ 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;
/**
@@ -95,8 +96,10 @@ public class Simulator implements ShutdownManager {
session = createSession(nodes, compression);
}
+ MetricsDAO metricsDAO = new MetricsDAO(session,
plan.getMetricsServerConfiguration());
MetricsServer metricsServer = new MetricsServer();
metricsServer.setSession(session);
+ metricsServer.setDAO(metricsDAO);
metricsServer.setConfiguration(plan.getMetricsServerConfiguration());
DateTimeService dateTimeService = new DateTimeService();
commit c7d8e5d6bb4d2921761a06ac1fe225e100c6bab5
Author: John Sanda <jsanda(a)redhat.com>
Date: Wed May 22 20:20:35 2013 -0400
changing transaction demarcation for basline and OOB calculations
The queries and methods in MeasurementBaselineManagerBean and in
MeasurementOOBManagerBean were too coarse-grained. In a test environment, I
was seeing regular timeouts while querying for schedules that do not yet have
baselines. And the method that computes OOBs was experiencing frequent
transaction timeouts. The transaction demarcation has been changed so that we
are doing a small, fixed size amount of working for each transaction when
calculating baselines and OOBs. Calls to Cassandra have been moved to
non-transactional methods.
Lastly, fixing a bug in MetricsBaselineCalculator to make sure we always return
a baseline for each schedule that it receives.
diff --git
a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementBaselineManagerBean.java
b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementBaselineManagerBean.java
index 8fb615c..37af9de 100644
---
a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementBaselineManagerBean.java
+++
b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementBaselineManagerBean.java
@@ -20,6 +20,8 @@ package org.rhq.enterprise.server.measurement;
import java.util.Arrays;
import java.util.Date;
+import java.util.Iterator;
+import java.util.LinkedList;
import java.util.List;
import java.util.Properties;
@@ -88,7 +90,8 @@ public class MeasurementBaselineManagerBean implements
MeasurementBaselineManage
private SessionManagerBean sessionManager;
private final Log log = LogFactory.getLog(MeasurementBaselineManagerBean.class);
- private static final int BASELINE_PROCESSING_LIMIT = 50000;
+
+ private static final int BASELINE_PROCESSING_LIMIT = 100;
@TransactionAttribute(TransactionAttributeType.NEVER)
public void calculateAutoBaselines() {
@@ -185,11 +188,14 @@ public class MeasurementBaselineManagerBean implements
MeasurementBaselineManage
* In any event, an appropriate chunking solution needs to be found, and
that partitioning strategy
* needs to replace the limits in the query today.
*/
- int schedulesWithoutBaselines = measurementBaselineManager
- ._calculateAutoBaselinesINSERT(amountOfData);
- totalProcessed += schedulesWithoutBaselines;
-
- if (schedulesWithoutBaselines < BASELINE_PROCESSING_LIMIT) {
+// int schedulesWithoutBaselines = measurementBaselineManager
+// ._calculateAutoBaselinesINSERT(amountOfData);
+ List<MeasurementSchedule> schedulesWithoutBaselines =
+ measurementBaselineManager.getSchedulesWithoutBaselines();
+ measurementBaselineManager.calculateBaselines(schedulesWithoutBaselines,
now, amountOfData);
+ totalProcessed += schedulesWithoutBaselines.size();
+
+ if (schedulesWithoutBaselines.size() < BASELINE_PROCESSING_LIMIT) {
break;
}
}
@@ -217,6 +223,62 @@ public class MeasurementBaselineManagerBean implements
MeasurementBaselineManage
@SuppressWarnings("unchecked")
@TransactionAttribute(TransactionAttributeType.REQUIRES_NEW)
+ public List<MeasurementSchedule> getSchedulesWithoutBaselines() {
+ Query query = this.entityManager
+
.createNamedQuery(MeasurementBaseline.QUERY_FIND_MEASUREMENT_SCHEDULES_WITHOUT_AUTOBASELINES);
+ query.setMaxResults(BASELINE_PROCESSING_LIMIT);
+ List<MeasurementSchedule> scheduleIdsWithoutBaselines =
query.getResultList();
+
+ return scheduleIdsWithoutBaselines;
+ }
+
+ @TransactionAttribute(TransactionAttributeType.NOT_SUPPORTED)
+ public void calculateBaselines(List<MeasurementSchedule> schedules, long
olderThan, long amountOfData) {
+ long endTime = olderThan;
+ long startTime = endTime - amountOfData;
+
+ log.debug("Computing baselines for " + schedules.size() + "
schedules");
+ MetricsBaselineCalculator baselineCalculator = new
MetricsBaselineCalculator(sessionManager.getMetricsDAO());
+ long calcStartTime = System.currentTimeMillis();
+ List<MeasurementBaseline> results =
baselineCalculator.calculateBaselines(schedules, startTime, endTime);
+ long calcEndTime = System.currentTimeMillis();
+ int count = results.size();
+
+ if (log.isDebugEnabled()) {
+ log.debug("Finished computing " + count + " new baselines in
" + (calcEndTime - calcStartTime) + " ms");
+ }
+
+ log.debug("Persisting baselines calculations");
+ long saveStartTime = System.currentTimeMillis();
+ Iterator<MeasurementBaseline> iterator = results.iterator();
+ List<MeasurementBaseline> queue = new
LinkedList<MeasurementBaseline>();
+ while (iterator.hasNext()) {
+ if (queue.size() == 10) {
+ measurementBaselineManager.saveNewBaselines(queue);
+ queue = new LinkedList<MeasurementBaseline>();
+ }
+ queue.add(iterator.next());
+ }
+ if (!queue.isEmpty()) {
+ measurementBaselineManager.saveNewBaselines(queue);
+ }
+
+ long saveEndTime = System.currentTimeMillis();
+ if (log.isDebugEnabled()) {
+ log.debug("Finished persisting " + count + " baselines in
" + (saveEndTime - saveStartTime) + " ms");
+ }
+ }
+
+ @Override
+ @TransactionAttribute(TransactionAttributeType.REQUIRES_NEW)
+ public void saveNewBaselines(List<MeasurementBaseline> baselines) {
+ for (MeasurementBaseline baseline : baselines) {
+ entityManager.merge(baseline);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ @TransactionAttribute(TransactionAttributeType.REQUIRES_NEW)
public int _calculateAutoBaselinesINSERT(long amountOfData) throws Exception {
long endTime = System.currentTimeMillis();
long startTime = endTime - amountOfData;
diff --git
a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementBaselineManagerLocal.java
b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementBaselineManagerLocal.java
index 7f7655a..8781fc5 100644
---
a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementBaselineManagerLocal.java
+++
b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementBaselineManagerLocal.java
@@ -24,6 +24,7 @@ import javax.ejb.Local;
import org.rhq.core.domain.auth.Subject;
import org.rhq.core.domain.measurement.MeasurementBaseline;
+import org.rhq.core.domain.measurement.MeasurementSchedule;
import org.rhq.core.domain.resource.Resource;
/**
@@ -68,10 +69,47 @@ public interface MeasurementBaselineManagerLocal {
* Inserts baselines "as appropriate" for measurements that have at least
amountOfData
* @param amountOfData will use amountOfData to compute new min/max/mean for
baselines as appropriate
* @return number of rows inserted
+ * @deprecated This method is pending removal after peer review. Because 1hr data is
+ * now stored in Cassandra, transaction boundaries needed to change. This method has
+ * been replaced by {@link #getSchedulesWithoutBaselines()}, {@link
#calculateBaselines(java.util.List, long)},
+ * and {@link #saveNewBaselines(java.util.List)}.
* @throws Exception
*/
int _calculateAutoBaselinesINSERT(long amountOfData) throws Exception;
+ /**
+ * <strong>Note</strong> This method exists only for transaction
demarcation.
+ *
+ * @return A list of schedules that do not have baselines. This list is not assumed
+ * to be an exhaustive list of schedules that lack a baseline. As such, this method
+ * will be called repeatedly during baseline calculations to get all of the
necessary
+ * schedules.
+ */
+ List<MeasurementSchedule> getSchedulesWithoutBaselines();
+
+ /**
+ * Given a list of schedules, this method calculates and stores baselines using the
+ * amount of 1 hr data specified and older than the time specified.
+ * <br/><br/>
+ * <strong>Note</strong> This method exists only for transaction
demarcation.
+ *
+ * @param schedules The schedules that do not yet have baselines
+ * @param olderThan Use 1 hr data prior to this time
+ * @param amountOfData The amount of data to use for calculating baselines. This
value
+ * is treated as a duration. For example, a value of 259200000
+ * would be treated as 3 days.
+ */
+ void calculateBaselines(List<MeasurementSchedule> schedules, long olderThan,
long amountOfData);
+
+ /**
+ * Persists the newly calculated baselines.
+ * <br/><br/>
+ * <strong>Note</strong> This method exists only for transaction
demarcation.
+ *
+ * @param baselines The baselines to persist.
+ */
+ void saveNewBaselines(List<MeasurementBaseline> baselines);
+
MeasurementBaseline getBaselineIfEqual(Subject subject, int groupId, int
definitionId);
/**
diff --git
a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementOOBManagerBean.java
b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementOOBManagerBean.java
index eecbbca..f742c7f 100644
---
a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementOOBManagerBean.java
+++
b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementOOBManagerBean.java
@@ -82,6 +82,9 @@ public class MeasurementOOBManagerBean implements
MeasurementOOBManagerLocal {
@EJB
AuthorizationManagerLocal authMangager;
+ @EJB
+ MeasurementOOBManagerLocal oobManager;
+
/**
* Compute oobs from the values in the 1h measurement table that just got added.
* For the total result, this is an incremental computation. The idea is that
@@ -191,56 +194,15 @@ public class MeasurementOOBManagerBean implements
MeasurementOOBManagerLocal {
}
@SuppressWarnings("unchecked")
+ @TransactionAttribute(value = TransactionAttributeType.NOT_SUPPORTED)
public void computeOOBsForLastHour(Subject subject,
Iterable<AggregateNumericMetric> metrics) {
+ log.info("Computing OOBs");
int count = 0;
long startTime = System.currentTimeMillis();
try {
for (AggregateNumericMetric metric : metrics) {
try {
- List<MeasurementBaseline> baselines =
entityManager.createQuery(
- "select baseline from MeasurementBaseline baseline where
baseline.schedule.id = :scheduleId")
- .setParameter("scheduleId", metric.getScheduleId())
- .getResultList();
- if (baselines.isEmpty()) {
- continue;
- }
- MeasurementBaseline baseline = baselines.get(0);
- Long upperDelta = null;
- Long lowerDelta = null;
-
- if (isPastUpperBound(baseline, metric)) {
- upperDelta =
- Math.round(((metric.getMax() - baseline.getMax()) /
(baseline.getMax() - baseline.getMin())) * 100);
- }
-
- if (isPastLowerBound(baseline, metric)) {
- lowerDelta =
- Math.round(((baseline.getMin() - metric.getMin()) /
(baseline.getMax() - baseline.getMin())) * 100);
- }
-
- Integer oobFactor;
- if (upperDelta != null && lowerDelta == null) {
- oobFactor = upperDelta.intValue();
- } else if (upperDelta == null && lowerDelta != null) {
- oobFactor = lowerDelta.intValue();
- } else if (upperDelta != null && lowerDelta != null) {
- if (upperDelta > lowerDelta) {
- oobFactor = upperDelta.intValue();
- } else {
- oobFactor = lowerDelta.intValue();
- }
- } else { // both are null
- oobFactor = null;
- }
-
- if (oobFactor != null) {
- MeasurementOOB oob = new MeasurementOOB();
- oob.setScheduleId(metric.getScheduleId());
- oob.setTimestamp(metric.getTimestamp());
- oob.setOobFactor(oobFactor);
- entityManager.merge(oob);
- ++count;
- }
+ count += oobManager.calculateOOB(metric);
} catch (Exception e) {
log.error("An error occurred while calculating OOBs for " +
metric, e);
throw new RuntimeException(e);
@@ -248,10 +210,66 @@ public class MeasurementOOBManagerBean implements
MeasurementOOBManagerLocal {
}
} finally {
long endTime = System.currentTimeMillis();
+ if (log.isInfoEnabled()) {
+ log.info("Finished calculating " + count + " OOBs in
" + (endTime - startTime) + " ms");
+ }
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ @TransactionAttribute(value = TransactionAttributeType.REQUIRES_NEW)
+ public int calculateOOB(AggregateNumericMetric metric) {
+ List<MeasurementBaseline> baselines = entityManager.createQuery(
+ "select baseline from MeasurementBaseline baseline where
baseline.schedule.id = :scheduleId")
+ .setParameter("scheduleId", metric.getScheduleId())
+ .getResultList();
+ if (baselines.isEmpty()) {
+ return 0;
+ }
+ MeasurementBaseline baseline = baselines.get(0);
+ Long upperDelta = null;
+ Long lowerDelta = null;
+
+ if (isPastUpperBound(baseline, metric)) {
+ upperDelta =
+ Math.round(((metric.getMax() - baseline.getMax()) / (baseline.getMax() -
baseline.getMin())) * 100);
+ }
+
+ if (isPastLowerBound(baseline, metric)) {
+ lowerDelta =
+ Math.round(((baseline.getMin() - metric.getMin()) / (baseline.getMax() -
baseline.getMin())) * 100);
+ }
+
+ Integer oobFactor;
+ if (upperDelta != null && lowerDelta == null) {
+ oobFactor = upperDelta.intValue();
+ } else if (upperDelta == null && lowerDelta != null) {
+ oobFactor = lowerDelta.intValue();
+ } else if (upperDelta != null && lowerDelta != null) {
+ if (upperDelta > lowerDelta) {
+ oobFactor = upperDelta.intValue();
+ } else {
+ oobFactor = lowerDelta.intValue();
+ }
+ } else { // both are null
+ oobFactor = null;
+ }
+
+ if (oobFactor != null) {
+ MeasurementOOB oob = new MeasurementOOB();
+ oob.setScheduleId(metric.getScheduleId());
+ oob.setTimestamp(metric.getTimestamp());
+ oob.setOobFactor(oobFactor);
+
if (log.isDebugEnabled()) {
- log.debug("Finished calculating " + count + " OOBs in
" + (endTime - startTime) + " ms");
+ log.debug("Generated OOB " + oob + " for 1 hr metric
" + metric + " with baseline " + baseline);
}
+
+ entityManager.merge(oob);
+ return 1;
}
+
+ return 0;
}
private boolean isPastUpperBound(MeasurementBaseline baseline, AggregateNumericMetric
metric) {
diff --git
a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementOOBManagerLocal.java
b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementOOBManagerLocal.java
index b452a28..853c5af 100644
---
a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementOOBManagerLocal.java
+++
b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementOOBManagerLocal.java
@@ -45,9 +45,30 @@ public interface MeasurementOOBManagerLocal {
*/
void computeOOBsFromHourBeginingAt(Subject subject, long begin);
+ /**
+ * Computes OOBs using the provided 1 hr data which should be the most recent 1 hr
+ * aggregates. These metrics are provided as an argument as opposed to querying for
+ * them because we already have the 1 hr aggregates load in memory when metrics
+ * aggregation runs prior to calculating OOBs.
+ *
+ * @param subject
+ * @param metrics The most recent 1 hr aggregates
+ */
void computeOOBsForLastHour(Subject subject, Iterable<AggregateNumericMetric>
metrics);
/**
+ * Determines and calculates an OOB if necessary, If an OOB is generated, this
method
+ * saves it to the database.
+ * <br/><br/>
+ * <strong>Note</strong> This method exists only for transaction
demarcation.
+ *
+ * @param metric The 1 hr metric that is used to determine whether or not an OOB
should
+ * be generated
+ * @return 1 if an OOB is generated, 0 otherwise
+ */
+ int calculateOOB(AggregateNumericMetric metric);
+
+ /**
* Return OOB Composites that contain all information about the OOBs in a given time
as aggregates.
* @param subject The caller
* @param metricNameFilter
diff --git
a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsBaselineCalculator.java
b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsBaselineCalculator.java
index 235e157..ef8eb44 100644
---
a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsBaselineCalculator.java
+++
b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsBaselineCalculator.java
@@ -39,10 +39,6 @@ public class MetricsBaselineCalculator {
private MetricsDAO metricsDAO;
-// public MetricsBaselineCalculator(Session session) {
-// this.metricsDAO = new MetricsDAO(session);
-// }
-
public MetricsBaselineCalculator(MetricsDAO metricsDAO) {
this.metricsDAO = metricsDAO;
}
@@ -108,6 +104,12 @@ public class MetricsBaselineCalculator {
return baseline;
}
- return null;
+ MeasurementBaseline baseline = new MeasurementBaseline();
+ baseline.setMax(Double.NaN);
+ baseline.setMin(Double.NaN);
+ baseline.setMean(Double.NaN);
+ baseline.setSchedule(schedule);
+
+ return baseline;
}
}