modules/common/cassandra-ccm/cassandra-ccm-core/src/main/bundle/cassandra/conf/rhq.cassandra.yaml | 2
modules/common/cassandra-schema/src/main/java/org/rhq/cassandra/schema/SchemaManager.java | 2
modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MetricsManagerBean.java | 2
modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsDAO.java | 12 -
modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsServer.java | 72 ++++++----
modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsDAOTest.java | 10 -
modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsServerTest.java | 53 +++----
modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/MeasurementAggregator.java | 2
8 files changed, 93 insertions(+), 62 deletions(-)
New commits:
commit ea7e1bdcf5684f3b32353a0ceb0fa01e1b0c8faa
Author: John Sanda <jsanda(a)redhat.com>
Date: Sat May 18 14:12:22 2013 -0400
need to pass timestamp to calculateAggregates now
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 5e30683..26d580b 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
@@ -60,7 +60,7 @@ public class MetricsManagerBean implements MetricsManagerLocal {
@TransactionAttribute(TransactionAttributeType.NOT_SUPPORTED)
public Iterable<AggregateNumericMetric> calculateAggregates() {
MetricsServer metricsServer = getMetricsServer();
- return metricsServer.calculateAggregates();
+ return metricsServer.calculateAggregates(System.currentTimeMillis());
}
@Override
commit 24b2ca97916cfa9b2a87ac80ceee30ecec121887
Author: John Sanda <jsanda(a)redhat.com>
Date: Sat May 18 13:56:58 2013 -0400
updating schema for metrics_index table so that rows are not reused
The way in which the metrics_index table has been used followed a big C*
anti-pattern. There was was one row devoted for each bucket that has data to be
aggregated. Once all of the schedules in a given row had their data aggregated
all of the columns in the row were deleted. This is increasingly inefficient as
the data set grows for a couple reasons. First, reads become slower because C*
has to iterate over more and more tombstone columns. Secondly, those tombstone
columns use up memory just like live columns do.
The columns stored are the same, but we are now using a compound row/partition
key that consists of the bucket name and a timestamp. The timestamp is the
start of an interval for a given bucket. So let's say we store some raw data at
12:22 pm. We will then update the {'one_hour_metrics', 12:00} row in the
metrics_index table for that schedule id. After 13:00, that will will never be
written to again and when aggregation runs over that time interval, the entire
row will be deleted. This way, reads against the metrics_index table will
always be over live columns and not tombstones.
diff --git a/modules/common/cassandra-schema/src/main/java/org/rhq/cassandra/schema/SchemaManager.java b/modules/common/cassandra-schema/src/main/java/org/rhq/cassandra/schema/SchemaManager.java
index 068e90f..0eec414 100644
--- a/modules/common/cassandra-schema/src/main/java/org/rhq/cassandra/schema/SchemaManager.java
+++ b/modules/common/cassandra-schema/src/main/java/org/rhq/cassandra/schema/SchemaManager.java
@@ -217,7 +217,7 @@ public class SchemaManager {
"bucket varchar, " +
"time timestamp, " +
"schedule_id int, " +
- "PRIMARY KEY (bucket, time, schedule_id) " +
+ "PRIMARY KEY ((bucket, time), schedule_id) " +
") WITH COMPACT STORAGE;"
);
} catch (NoHostAvailableException e) {
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 03ee04c..dd5c44c 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
@@ -136,9 +136,9 @@ public class MetricsDAO {
MetricsTable.TWENTY_FOUR_HOUR + " WHERE schedule_id = ? AND time >= ? AND time < ?");
findIndexEntries = session.prepare("SELECT time, schedule_id FROM " + MetricsTable.INDEX +
- " WHERE bucket = ? ORDER BY time");
+ " WHERE bucket = ? AND time = ?");
- deleteIndexEntries = session.prepare("DELETE FROM " + MetricsTable.INDEX + " WHERE bucket = ?");
+ deleteIndexEntries = session.prepare("DELETE FROM " + MetricsTable.INDEX + " WHERE bucket = ? AND time = ?");
}
public ResultSet insertRawData(MeasurementDataNumeric data) {
@@ -288,8 +288,8 @@ public class MetricsDAO {
new AggregateNumericMetricMapper(), session);
}
- public Iterable<MetricsIndexEntry> findMetricsIndexEntries(final MetricsTable table) {
- BoundStatement statement = findIndexEntries.bind(table.toString());
+ public Iterable<MetricsIndexEntry> findMetricsIndexEntries(final MetricsTable table, long timestamp) {
+ BoundStatement statement = findIndexEntries.bind(table.toString(), new Date(timestamp));
return new SimplePagedResult<MetricsIndexEntry>(statement, new MetricsIndexEntryMapper(table), session);
}
@@ -301,8 +301,8 @@ public class MetricsDAO {
}
}
- public void deleteMetricsIndexEntries(MetricsTable table) {
- BoundStatement statement = deleteIndexEntries.bind(table.getTableName());
+ 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 10421d6..fd285ee 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,6 +26,8 @@
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;
@@ -67,7 +69,6 @@ public class MetricsServer {
public void setSession(Session session) {
this.session = session;
- //dao = new MetricsDAO(session);
}
public void setDAO(MetricsDAO dao) {
@@ -250,47 +251,74 @@ public class MetricsServer {
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.info("Updating one hour index wtih time slices " + dates);
dao.updateMetricsIndex(MetricsTable.ONE_HOUR, updates);
}
- public Iterable<AggregateNumericMetric> calculateAggregates() {
+ /**
+ * Computes and stores aggregate for all time slices that are ready to be aggregated.
+ * This is includes raw, 1hr, 6hr, and 24hr data.
+ *
+ * @param startTime A timestamp that the determines the period over which data will be
+ * aggregated. Aggregation is normally run hourly as part of the data
+ * purge job; so, this timestamp is rounded down to the hour and then
+ * an hour is subtracted to get the actual starting time over which
+ * data will be aggregated. Suppose the argument has a value (time) of
+ * 02:00:34. This means that raw data that has been stored between
+ * 01:00 and 02:00 will be aggregated.
+ *
+ * @return One hour aggregates. That is, any raw data that has been rolled up into onr
+ * one hour aggregates. The one hour aggregates are returned because they are needed
+ * for subsequently computing baselines.
+ */
+ public Iterable<AggregateNumericMetric> calculateAggregates(long startTime) {
+ DateTime dt = new DateTime(startTime);
+ DateTime currentHour = dateTimeService.getTimeSlice(dt, configuration.getRawTimeSliceDuration());
+ DateTime lastHour = currentHour.minusHours(1);
+
+ long hourTimeSlice = lastHour.getMillis();
+
+ long sixHourTimeSlice = dateTimeService.getTimeSlice(lastHour,
+ configuration.getOneHourTimeSliceDuration()).getMillis();
+
+ long twentyFourHourTimeSlice = dateTimeService.getTimeSlice(lastHour,
+ configuration.getSixHourTimeSliceDuration()).getMillis();
+
// We first query the metrics index table to determine which schedules have data to
// be aggregated. Then we retrieve the metric data and aggregate or compress the
// data, writing the compressed values into the next wider (i.e., longer life span
// for data) bucket/table. At this point we remove the index entries for the data
- // that has already been processed. We currently purge the entire row in the index
- // table. We can safely do this entire work flow is single threaded. It might make
- // sense to perform the deletes in a more granular fashion to avoid concurrency
- // issues in the future. The last step in the work flow is to update the metrics
+ // that has already been processed. We purge the entire row in the index table.
+ // We can safely do this because the row wi..
+ //
+ // The last step in the work flow is to update the metrics
// index for the newly persisted aggregates.
- // TODO deleteMetricsIndexEntries should take a list of schedule ids
- // MetricsDAO.deleteMetricsIndexEntries deletes the entire row, but we probably do
- // not want to delete each column unless and until we verify that the data for the
- // schedule id in that column has in fact been aggregated. It might be better for
- // deleteMetricsIndexEntries to take a list of schedule ids to purge.
-
Iterable<AggregateNumericMetric> newOneHourAggregates = null;
- List<AggregateNumericMetric> updatedSchedules = aggregateRawData();
+ List<AggregateNumericMetric> updatedSchedules = aggregateRawData(hourTimeSlice);
newOneHourAggregates = updatedSchedules;
if (!updatedSchedules.isEmpty()) {
- dao.deleteMetricsIndexEntries(MetricsTable.ONE_HOUR);
+ dao.deleteMetricsIndexEntries(MetricsTable.ONE_HOUR, hourTimeSlice);
updateMetricsIndex(MetricsTable.SIX_HOUR, updatedSchedules, configuration.getOneHourTimeSliceDuration());
}
- updatedSchedules = calculateAggregates(MetricsTable.ONE_HOUR, MetricsTable.SIX_HOUR,
+ updatedSchedules = calculateAggregates(MetricsTable.ONE_HOUR, MetricsTable.SIX_HOUR, sixHourTimeSlice,
configuration.getOneHourTimeSliceDuration());
if (!updatedSchedules.isEmpty()) {
- dao.deleteMetricsIndexEntries(MetricsTable.SIX_HOUR);
+ dao.deleteMetricsIndexEntries(MetricsTable.SIX_HOUR, sixHourTimeSlice);
updateMetricsIndex(MetricsTable.TWENTY_FOUR_HOUR, updatedSchedules,
configuration.getSixHourTimeSliceDuration());
}
updatedSchedules = calculateAggregates(MetricsTable.SIX_HOUR, MetricsTable.TWENTY_FOUR_HOUR,
- configuration.getSixHourTimeSliceDuration());
+ twentyFourHourTimeSlice, configuration.getSixHourTimeSliceDuration());
if (!updatedSchedules.isEmpty()) {
- dao.deleteMetricsIndexEntries(MetricsTable.TWENTY_FOUR_HOUR);
+ dao.deleteMetricsIndexEntries(MetricsTable.TWENTY_FOUR_HOUR, twentyFourHourTimeSlice);
}
return newOneHourAggregates;
@@ -305,8 +333,8 @@ public class MetricsServer {
dao.updateMetricsIndex(bucket, updates);
}
- private List<AggregateNumericMetric> aggregateRawData() {
- Iterable<MetricsIndexEntry> indexEntries = dao.findMetricsIndexEntries(MetricsTable.ONE_HOUR);
+ private List<AggregateNumericMetric> aggregateRawData(long theHour) {
+ Iterable<MetricsIndexEntry> indexEntries = dao.findMetricsIndexEntries(MetricsTable.ONE_HOUR, theHour);
List<AggregateNumericMetric> oneHourMetrics = new ArrayList<AggregateNumericMetric>();
for (MetricsIndexEntry indexEntry : indexEntries) {
@@ -356,9 +384,9 @@ public class MetricsServer {
}
private List<AggregateNumericMetric> calculateAggregates(MetricsTable fromTable,
- MetricsTable toTable, Duration nextDuration) {
+ MetricsTable toTable, long timeSlice, Duration nextDuration) {
- Iterable<MetricsIndexEntry> indexEntries = dao.findMetricsIndexEntries(toTable);
+ Iterable<MetricsIndexEntry> indexEntries = dao.findMetricsIndexEntries(toTable, timeSlice);
List<AggregateNumericMetric> toMetrics = new ArrayList<AggregateNumericMetric>();
DateTime currentHour = getCurrentHour();
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 261cd13..8f35486 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
@@ -271,7 +271,8 @@ public class MetricsDAOTest extends CassandraIntegrationTest {
updates.put(scheduleId2, hour0.getMillis());
dao.updateMetricsIndex(MetricsTable.ONE_HOUR, updates);
- List<MetricsIndexEntry> actual = Lists.newArrayList(dao.findMetricsIndexEntries(MetricsTable.ONE_HOUR));
+ List<MetricsIndexEntry> actual = Lists.newArrayList(dao.findMetricsIndexEntries(MetricsTable.ONE_HOUR,
+ hour0.getMillis()));
List<MetricsIndexEntry> expected = asList(new MetricsIndexEntry(MetricsTable.ONE_HOUR, hour0, scheduleId1),
new MetricsIndexEntry(MetricsTable.ONE_HOUR, hour0, scheduleId2));
@@ -285,13 +286,12 @@ public class MetricsDAOTest extends CassandraIntegrationTest {
Map<Integer, Long> updates = new HashMap<Integer, Long>();
updates.put(scheduleId1, hour0().getMillis());
updates.put(scheduleId2, hour0().getMillis());
- updates.put(scheduleId1, hour0().plusHours(1).getMillis());
- updates.put(scheduleId2, hour0().plusHours(1).getMillis());
dao.updateMetricsIndex(MetricsTable.ONE_HOUR, updates);
- dao.deleteMetricsIndexEntries(MetricsTable.ONE_HOUR);
+ dao.deleteMetricsIndexEntries(MetricsTable.ONE_HOUR, hour0().getMillis());
- List<MetricsIndexEntry> index = Lists.newArrayList(dao.findMetricsIndexEntries(MetricsTable.ONE_HOUR));
+ List<MetricsIndexEntry> index = Lists.newArrayList(dao.findMetricsIndexEntries(MetricsTable.ONE_HOUR,
+ hour0().getMillis()));
assertEquals(index.size(), 0, "Expected index for " + MetricsTable.ONE_HOUR + " to be empty but found " +
index);
}
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 7ac617f..d895753 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
@@ -152,10 +152,12 @@ public class MetricsServerTest extends CassandraIntegrationTest {
assertColumnMetadataEquals(scheduleId, hour0.plusHours(4), hour0.plusHours(5), MetricsTable.RAW.getTTL(),
timestamp);
- List<MetricsIndexEntry> expectedIndex = asList(new MetricsIndexEntry(MetricsTable.ONE_HOUR,
- hour0.plusHours(4), scheduleId));
- assertMetricsIndexEquals(MetricsTable.ONE_HOUR, expectedIndex, "Failed to update index for "
- + MetricsTable.ONE_HOUR);
+ DateTime hour4 = hour0.plusHours(4);
+
+ List<MetricsIndexEntry> expectedIndex = asList(new MetricsIndexEntry(MetricsTable.ONE_HOUR, hour4,
+ scheduleId));
+ assertMetricsIndexEquals(MetricsTable.ONE_HOUR, hour4.getMillis(), expectedIndex,
+ "Failed to update index for " + MetricsTable.ONE_HOUR);
}
@Test//(enabled = ENABLED)
@@ -176,7 +178,7 @@ public class MetricsServerTest extends CassandraIntegrationTest {
metricsServer.setCurrentHour(hour6);
metricsServer.addNumericData(data);
- metricsServer.calculateAggregates();
+ metricsServer.calculateAggregates(hour6.getMillis());
// verify that one hour metric data is updated
List<AggregateNumericMetric> expected = asList(new AggregateNumericMetric(scheduleId,
@@ -218,7 +220,7 @@ public class MetricsServerTest extends CassandraIntegrationTest {
metricsServer.updateMetricsIndex(rawMetrics);
metricsServer.setCurrentHour(hour9);
- metricsServer.calculateAggregates();
+ metricsServer.calculateAggregates(hour9.getMillis());
// verify that the 1 hour aggregates are calculated
assert1HourDataEquals(scheduleId, asList(new AggregateNumericMetric(scheduleId, divide((1.1 + 2.2 + 3.3), 3),
@@ -228,8 +230,8 @@ public class MetricsServerTest extends CassandraIntegrationTest {
List<MetricsIndexEntry> expected6HourIndex = asList(new MetricsIndexEntry(MetricsTable.SIX_HOUR,
dateTimeService.getTimeSlice(hour9, Minutes.minutes(60 * 6)), scheduleId));
- assertMetricsIndexEquals(MetricsTable.SIX_HOUR, expected6HourIndex, "Failed to update index for "
- + MetricsTable.SIX_HOUR);
+ assertMetricsIndexEquals(MetricsTable.SIX_HOUR, hour9.minusHours(3).getMillis(), expected6HourIndex,
+ "Failed to update index for " + MetricsTable.SIX_HOUR);
// The 6 hour data should not get aggregated since the current 6 hour time slice
// has not passed yet. More specifically, the aggregation job is running at 09:00
@@ -237,10 +239,10 @@ public class MetricsServerTest extends CassandraIntegrationTest {
assert6HourDataEmpty(scheduleId);
// verify that the 24 hour index is empty
- assert24HourMetricsIndexEmpty(scheduleId);
+ assert24HourMetricsIndexEmpty(scheduleId, hour0.getMillis());
// verify that the 1 hour queue has been purged
- assert1HourMetricsIndexEmpty(scheduleId);
+ assert1HourMetricsIndexEmpty(scheduleId, hour9.getMillis());
}
@Test//(enabled = ENABLED)
@@ -280,7 +282,7 @@ public class MetricsServerTest extends CassandraIntegrationTest {
// execute the system under test
metricsServer.setCurrentHour(hour12);
- metricsServer.calculateAggregates();
+ metricsServer.calculateAggregates(hour12.getMillis());
// verify the results
// verify that the one hour data has been aggregated
@@ -288,10 +290,10 @@ public class MetricsServerTest extends CassandraIntegrationTest {
max1, hour6.getMillis())));
// verify that the 6 hour queue has been updated
- assert6HourMetricsIndexEmpty(scheduleId);
+ assert6HourMetricsIndexEmpty(scheduleId, hour6.getMillis());
// verify that the 24 hour queue is updated
- assertMetricsIndexEquals(MetricsTable.TWENTY_FOUR_HOUR, asList(new MetricsIndexEntry(
+ assertMetricsIndexEquals(MetricsTable.TWENTY_FOUR_HOUR, hour0.getMillis(), asList(new MetricsIndexEntry(
MetricsTable.TWENTY_FOUR_HOUR, hour0, scheduleId)), "Failed to update index for "
+ MetricsTable.TWENTY_FOUR_HOUR);
@@ -335,7 +337,7 @@ public class MetricsServerTest extends CassandraIntegrationTest {
// execute the system under test
metricsServer.setCurrentHour(hour24);
- metricsServer.calculateAggregates();
+ metricsServer.calculateAggregates(hour24.getMillis());
// verify the results
// verify that the 6 hour data is aggregated
@@ -343,7 +345,7 @@ public class MetricsServerTest extends CassandraIntegrationTest {
min1, max2, hour0.getMillis())));
// verify that the 24 hour queue is updated
- assert24HourMetricsIndexEmpty(scheduleId);
+ assert24HourMetricsIndexEmpty(scheduleId, hour0.getMillis());
}
@Test//(enabled = ENABLED)
@@ -758,8 +760,9 @@ public class MetricsServerTest extends CassandraIntegrationTest {
}
}
- private void assertMetricsIndexEquals(MetricsTable table, List<MetricsIndexEntry> expected, String msg) {
- List<MetricsIndexEntry> actual = Lists.newArrayList(dao.findMetricsIndexEntries(table));
+ private void assertMetricsIndexEquals(MetricsTable table, long timeSlice, List<MetricsIndexEntry> expected,
+ String msg) {
+ List<MetricsIndexEntry> actual = Lists.newArrayList(dao.findMetricsIndexEntries(table, timeSlice));
assertCollectionMatchesNoOrder(msg + ": " + table + " index does not match expected values.",
expected, actual);
}
@@ -797,20 +800,20 @@ public class MetricsServerTest extends CassandraIntegrationTest {
" but found " + metrics);
}
- private void assert1HourMetricsIndexEmpty(int scheduleId) {
- assertMetricsIndexEmpty(scheduleId, MetricsTable.ONE_HOUR);
+ private void assert1HourMetricsIndexEmpty(int scheduleId, long timeSlice) {
+ assertMetricsIndexEmpty(scheduleId, MetricsTable.ONE_HOUR, timeSlice);
}
- private void assert6HourMetricsIndexEmpty(int scheduleId) {
- assertMetricsIndexEmpty(scheduleId, MetricsTable.SIX_HOUR);
+ private void assert6HourMetricsIndexEmpty(int scheduleId, long timeSlice) {
+ assertMetricsIndexEmpty(scheduleId, MetricsTable.SIX_HOUR, timeSlice);
}
- private void assert24HourMetricsIndexEmpty(int scheduleId) {
- assertMetricsIndexEmpty(scheduleId, MetricsTable.TWENTY_FOUR_HOUR);
+ private void assert24HourMetricsIndexEmpty(int scheduleId, long timeSlice) {
+ assertMetricsIndexEmpty(scheduleId, MetricsTable.TWENTY_FOUR_HOUR, timeSlice);
}
- private void assertMetricsIndexEmpty(int scheduleId, MetricsTable table) {
- List<MetricsIndexEntry> index = Lists.newArrayList(dao.findMetricsIndexEntries(table));
+ private void assertMetricsIndexEmpty(int scheduleId, MetricsTable table, long timeSlice) {
+ List<MetricsIndexEntry> index = Lists.newArrayList(dao.findMetricsIndexEntries(table, timeSlice));
assertEquals(index.size(), 0, "Expected metrics index for " + table + " to be empty but found " + index);
}
diff --git a/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/MeasurementAggregator.java b/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/MeasurementAggregator.java
index 31bca0b..06d6b6b 100644
--- a/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/MeasurementAggregator.java
+++ b/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/MeasurementAggregator.java
@@ -54,7 +54,7 @@ public class MeasurementAggregator implements Runnable {
log.info("Starting metrics aggregation...");
long startTime = System.currentTimeMillis();
try {
- metricsServer.calculateAggregates();
+ metricsServer.calculateAggregates(startTime);
} catch (Exception e) {
log.error("An error occurred while trying to perform aggregation", e);
log.error("Requesting simulation shutdown...");
commit f252084ba7f93f44cdfecef98f487817cca950bf
Author: John Sanda <jsanda(a)redhat.com>
Date: Sat May 18 11:09:18 2013 -0400
adding rhq.cassandra.start_rpc property for token replacement in cassandra.yaml
diff --git a/modules/common/cassandra-ccm/cassandra-ccm-core/src/main/bundle/cassandra/conf/rhq.cassandra.yaml b/modules/common/cassandra-ccm/cassandra-ccm-core/src/main/bundle/cassandra/conf/rhq.cassandra.yaml
index 0e0ac3d..977a108 100644
--- a/modules/common/cassandra-ccm/cassandra-ccm-core/src/main/bundle/cassandra/conf/rhq.cassandra.yaml
+++ b/modules/common/cassandra-ccm/cassandra-ccm-core/src/main/bundle/cassandra/conf/rhq.cassandra.yaml
@@ -342,7 +342,7 @@ native_transport_max_threads: ${rhq.cassandra.native_transport_max_threads}
# Whether to start the thrift rpc server.
-start_rpc: false
+start_rpc: ${rhq.cassandra.start_rpc}
# The address to bind the Thrift RPC service to -- clients connect
# here. Unlike ListenAddress above, you *can* specify 0.0.0.0 here if
# you want Thrift to listen on all interfaces.