[rhq] Branch 'feature/cassandra-backend' - 3 commits - modules/common modules/enterprise
by John Sanda
modules/common/cassandra-common/src/main/cassandra/cql/dbsetup.cql | 10
modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsIndexEntry.java | 111 ++++++++++
modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsServer.java | 78 +++++--
modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsServerTest.java | 93 ++++++--
4 files changed, 247 insertions(+), 45 deletions(-)
New commits:
commit 5bc4b9ee30a97e81d660af0468b392555187716f
Author: John Sanda <jsanda(a)redhat.com>
Date: Thu Nov 1 14:16:36 2012 -0400
(cql port) Adding back logic to purge schema between test runs
Purging tables is now easily done with the TRUNCATE statement.
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 83425cc..cf7e9de 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
@@ -108,11 +108,11 @@ public class MetricsServerTest {
private final String RAW_METRIC_DATA_CF = "raw_metrics";
- private final String ONE_HOUR_METRIC_DATA_CF = "one_hour_metric_data";
+ private final String ONE_HOUR_METRIC_DATA_CF = "one_hour_metrics";
- private final String SIX_HOUR_METRIC_DATA_CF = "six_hour_metric_data";
+ private final String SIX_HOUR_METRIC_DATA_CF = "six_hour_metrics";
- private final String TWENTY_FOUR_HOUR_METRIC_DATA_CF = "twenty_four_hour_metric_data";
+ private final String TWENTY_FOUR_HOUR_METRIC_DATA_CF = "twenty_four_hour_metrics";
private final String METRICS_INDEX = "metrics_index";
@@ -182,17 +182,23 @@ public class MetricsServerTest {
metricsServer.setTraitsCF(TRAITS_CF);
metricsServer.setResourceTraitsCF(RESOURCE_TRAITS_CF);
metricsServer.setCassandraDS(dataSource);
- //purgeDB();
+ purgeDB();
}
- private void purgeDB() {
- deleteAllRows(METRICS_INDEX, StringSerializer.get());
- deleteAllRows(RAW_METRIC_DATA_CF, IntegerSerializer.get());
- deleteAllRows(ONE_HOUR_METRIC_DATA_CF, IntegerSerializer.get());
- deleteAllRows(SIX_HOUR_METRIC_DATA_CF, IntegerSerializer.get());
- deleteAllRows(TWENTY_FOUR_HOUR_METRIC_DATA_CF, IntegerSerializer.get());
- deleteAllRows(TRAITS_CF, IntegerSerializer.get());
- deleteAllRows(RESOURCE_TRAITS_CF, IntegerSerializer.get());
+ private void purgeDB() throws SQLException {
+// deleteAllRows(METRICS_INDEX, StringSerializer.get());
+// deleteAllRows(RAW_METRIC_DATA_CF, IntegerSerializer.get());
+// deleteAllRows(ONE_HOUR_METRIC_DATA_CF, IntegerSerializer.get());
+// deleteAllRows(SIX_HOUR_METRIC_DATA_CF, IntegerSerializer.get());
+// deleteAllRows(TWENTY_FOUR_HOUR_METRIC_DATA_CF, IntegerSerializer.get());
+// deleteAllRows(TRAITS_CF, IntegerSerializer.get());
+// deleteAllRows(RESOURCE_TRAITS_CF, IntegerSerializer.get());
+ Statement statement = connection.createStatement();
+ statement.executeUpdate("TRUNCATE " + RAW_METRIC_DATA_CF);
+ statement.executeUpdate("TRUNCATE " + ONE_HOUR_METRIC_DATA_CF);
+ statement.executeUpdate("TRUNCATE " + METRICS_INDEX);
+
+ statement.close();
}
private <K> MutationResult deleteAllRows(String columnFamily, Serializer<K> keySerializer) {
commit 54d9960e990aaa69d13eb824d8341c6744310ec6
Author: John Sanda <jsanda(a)redhat.com>
Date: Thu Nov 1 13:50:48 2012 -0400
renaming table from metrics_work_queue to metrics_index
metrics_index is a more accurate name because the table is in fact an index
table/CF used for querying which schedule ids in each bucket have data to be
aggregated.
diff --git a/modules/common/cassandra-common/src/main/cassandra/cql/dbsetup.cql b/modules/common/cassandra-common/src/main/cassandra/cql/dbsetup.cql
index 13b6d80..8dcfd16 100644
--- a/modules/common/cassandra-common/src/main/cassandra/cql/dbsetup.cql
+++ b/modules/common/cassandra-common/src/main/cassandra/cql/dbsetup.cql
@@ -17,7 +17,7 @@ CREATE TABLE one_hour_metrics (
PRIMARY KEY (schedule_id, time, type)
);
-CREATE TABLE metrics_work_queue (
+CREATE TABLE metrics_index (
bucket varchar,
time timestamp,
schedule_id int,
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 bc89df6..0894339 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
@@ -89,7 +89,7 @@ public class MetricsServer {
private String twentyFourHourMetricsDataCF;
- private String metricsQueueCF;
+ private String metricsIndex;
private String traitsCF;
@@ -152,12 +152,12 @@ public class MetricsServer {
this.twentyFourHourMetricsDataCF = twentyFourHourMetricsDataCF;
}
- public String getMetricsQueueCF() {
- return metricsQueueCF;
+ public String getMetricsIndex() {
+ return metricsIndex;
}
- public void setMetricsQueueCF(String metricsQueueCF) {
- this.metricsQueueCF = metricsQueueCF;
+ public void setMetricsIndex(String metricsIndex) {
+ this.metricsIndex = metricsIndex;
}
public String getTraitsCF() {
@@ -335,7 +335,7 @@ public class MetricsServer {
SliceQuery<String, Composite, Integer> queueQuery = HFactory.createSliceQuery(keyspace, StringSerializer.get(),
new CompositeSerializer().get(), IntegerSerializer.get());
- queueQuery.setColumnFamily(metricsQueueCF);
+ queueQuery.setColumnFamily(metricsIndex);
queueQuery.setKey(oneHourMetricsDataCF);
ColumnSliceIterator<String, Composite, Integer> queueIterator = new ColumnSliceIterator<String, Composite, Integer>(
@@ -388,7 +388,7 @@ public class MetricsServer {
updatedSchedules.put(scheduleId, dateTimeService.getTimeSlice(startTime, Minutes.minutes(60 * 6)));
- queueMutator.addDeletion(oneHourMetricsDataCF, metricsQueueCF, queueColumn.getName(),
+ queueMutator.addDeletion(oneHourMetricsDataCF, metricsIndex, queueColumn.getName(),
CompositeSerializer.get());
}
mutator.execute();
@@ -406,7 +406,7 @@ public class MetricsServer {
SliceQuery<String, Composite, Integer> queueQuery = HFactory.createSliceQuery(keyspace, StringSerializer.get(),
new CompositeSerializer().get(), IntegerSerializer.get());
- queueQuery.setColumnFamily(metricsQueueCF);
+ queueQuery.setColumnFamily(metricsIndex);
queueQuery.setKey(toColumnFamily);
ColumnSliceIterator<String, Composite, Integer> queueIterator = new ColumnSliceIterator<String, Composite, Integer>(
@@ -485,7 +485,7 @@ public class MetricsServer {
updatedSchedules.put(scheduleId, dateTimeService.getTimeSlice(startTime, nextInterval));
- queueMutator.addDeletion(toColumnFamily, metricsQueueCF, queueColumn.getName(), CompositeSerializer.get());
+ queueMutator.addDeletion(toColumnFamily, metricsIndex, queueColumn.getName(), CompositeSerializer.get());
}
mutator.execute();
queueMutator.execute();
@@ -545,7 +545,7 @@ public class MetricsServer {
log.warn(errorMsg);
return null;
}
- String sql = "INSERT INTO " + metricsQueueCF + " (bucket, time, schedule_id, null_col) VALUES (?, ?, ?, ?)";
+ String sql = "INSERT INTO " + metricsIndex + " (bucket, time, schedule_id, null_col) VALUES (?, ?, ?, ?)";
PreparedStatement statement = null;
try {
statement = connection.prepareStatement(sql);
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 0ff4358..83425cc 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
@@ -114,7 +114,7 @@ public class MetricsServerTest {
private final String TWENTY_FOUR_HOUR_METRIC_DATA_CF = "twenty_four_hour_metric_data";
- private final String METRICS_WORK_QUEUE_CF = "metrics_work_queue";
+ private final String METRICS_INDEX = "metrics_index";
private final String TRAITS_CF = "traits";
@@ -178,7 +178,7 @@ public class MetricsServerTest {
metricsServer.setOneHourMetricsDataCF(ONE_HOUR_METRIC_DATA_CF);
metricsServer.setSixHourMetricsDataCF(SIX_HOUR_METRIC_DATA_CF);
metricsServer.setTwentyFourHourMetricsDataCF(TWENTY_FOUR_HOUR_METRIC_DATA_CF);
- metricsServer.setMetricsQueueCF(METRICS_WORK_QUEUE_CF);
+ metricsServer.setMetricsIndex(METRICS_INDEX);
metricsServer.setTraitsCF(TRAITS_CF);
metricsServer.setResourceTraitsCF(RESOURCE_TRAITS_CF);
metricsServer.setCassandraDS(dataSource);
@@ -186,7 +186,7 @@ public class MetricsServerTest {
}
private void purgeDB() {
- deleteAllRows(METRICS_WORK_QUEUE_CF, StringSerializer.get());
+ deleteAllRows(METRICS_INDEX, StringSerializer.get());
deleteAllRows(RAW_METRIC_DATA_CF, IntegerSerializer.get());
deleteAllRows(ONE_HOUR_METRIC_DATA_CF, IntegerSerializer.get());
deleteAllRows(SIX_HOUR_METRIC_DATA_CF, IntegerSerializer.get());
@@ -373,7 +373,7 @@ public class MetricsServerTest {
Composite key = createQueueColumnName(hour8, scheduleId);
HColumn<Composite, Integer> oneHourQueueColumn = HFactory.createColumn(key, 0, CompositeSerializer.get(),
IntegerSerializer.get());
- queueMutator.addInsertion(ONE_HOUR_METRIC_DATA_CF, METRICS_WORK_QUEUE_CF, oneHourQueueColumn);
+ queueMutator.addInsertion(ONE_HOUR_METRIC_DATA_CF, METRICS_INDEX, oneHourQueueColumn);
queueMutator.execute();
@@ -452,7 +452,7 @@ public class MetricsServerTest {
Composite key = createQueueColumnName(hour6, scheduleId);
HColumn<Composite, Integer> sixHourQueueColumn = HFactory.createColumn(key, 0, CompositeSerializer.get(),
IntegerSerializer.get());
- queueMutator.addInsertion(SIX_HOUR_METRIC_DATA_CF, METRICS_WORK_QUEUE_CF, sixHourQueueColumn);
+ queueMutator.addInsertion(SIX_HOUR_METRIC_DATA_CF, METRICS_INDEX, sixHourQueueColumn);
queueMutator.execute();
@@ -519,7 +519,7 @@ public class MetricsServerTest {
Composite key = createQueueColumnName(hour0, scheduleId);
HColumn<Composite, Integer> twentyFourHourQueueColumn = HFactory.createColumn(key, 0, CompositeSerializer.get(),
IntegerSerializer.get());
- queueMutator.addInsertion(TWENTY_FOUR_HOUR_METRIC_DATA_CF, METRICS_WORK_QUEUE_CF, twentyFourHourQueueColumn);
+ queueMutator.addInsertion(TWENTY_FOUR_HOUR_METRIC_DATA_CF, METRICS_INDEX, twentyFourHourQueueColumn);
queueMutator.execute();
@@ -785,7 +785,7 @@ public class MetricsServerTest {
private void assertMetricsQueueEquals(String columnFamily, List<HColumn<Composite, Integer>> expected) {
SliceQuery<String,Composite, Integer> sliceQuery = HFactory.createSliceQuery(keyspace, StringSerializer.get(),
new CompositeSerializer().get(), IntegerSerializer.get());
- sliceQuery.setColumnFamily(METRICS_WORK_QUEUE_CF);
+ sliceQuery.setColumnFamily(METRICS_INDEX);
sliceQuery.setKey(columnFamily);
ColumnSliceIterator<String, Composite, Integer> iterator = new ColumnSliceIterator<String, Composite, Integer>(
@@ -813,7 +813,7 @@ public class MetricsServerTest {
try {
List<MetricsIndexEntry> actual = new ArrayList<MetricsIndexEntry>();
statement = connection.createStatement();
- resultSet = statement.executeQuery("SELECT bucket, time, schedule_id FROM " + METRICS_WORK_QUEUE_CF +
+ resultSet = statement.executeQuery("SELECT bucket, time, schedule_id FROM " + METRICS_INDEX +
" WHERE bucket = '" + columnFamily + "'");
while (resultSet.next()) {
actual.add(new MetricsIndexEntry(resultSet.getString(1), resultSet.getDate(2), resultSet.getInt(3)));
@@ -927,7 +927,7 @@ public class MetricsServerTest {
private void assertMetricsQueueEmpty(int scheduleId, String columnFamily) {
SliceQuery<String,Composite, Integer> sliceQuery = HFactory.createSliceQuery(keyspace, StringSerializer.get(),
new CompositeSerializer().get(), IntegerSerializer.get());
- sliceQuery.setColumnFamily(METRICS_WORK_QUEUE_CF);
+ sliceQuery.setColumnFamily(METRICS_INDEX);
sliceQuery.setKey(columnFamily);
ColumnSliceIterator<String, Composite, Integer> iterator = new ColumnSliceIterator<String, Composite, Integer>(
commit 3926a6c2d14d07d77a69c3404063b21e230f5ab3
Author: John Sanda <jsanda(a)redhat.com>
Date: Thu Nov 1 13:39:37 2012 -0400
(cql port) update metrics index when inserting raw metrics
Adding one_hour_metrics table/CF back to schema. Adding back logic to update the
metrics index when inserting raw metrics. Some of MetricsServerTest is updated
as well to verify the index update.
diff --git a/modules/common/cassandra-common/src/main/cassandra/cql/dbsetup.cql b/modules/common/cassandra-common/src/main/cassandra/cql/dbsetup.cql
index 52cba95..13b6d80 100644
--- a/modules/common/cassandra-common/src/main/cassandra/cql/dbsetup.cql
+++ b/modules/common/cassandra-common/src/main/cassandra/cql/dbsetup.cql
@@ -9,6 +9,14 @@ CREATE TABLE raw_metrics (
PRIMARY KEY (schedule_id, time)
);
+CREATE TABLE one_hour_metrics (
+ schedule_id int,
+ time timestamp,
+ type int,
+ value double,
+ PRIMARY KEY (schedule_id, time, type)
+);
+
CREATE TABLE metrics_work_queue (
bucket varchar,
time timestamp,
diff --git a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsIndexEntry.java b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsIndexEntry.java
new file mode 100644
index 0000000..63c93d0
--- /dev/null
+++ b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsIndexEntry.java
@@ -0,0 +1,111 @@
+/*
+ *
+ * * 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 java.util.Date;
+
+import org.joda.time.DateTime;
+import org.joda.time.format.DateTimeFormat;
+
+/**
+ * @author John Sanda
+ */
+public class MetricsIndexEntry {
+
+ private String bucket;
+
+ private int scheduleId;
+
+ private DateTime time;
+
+ public MetricsIndexEntry () {
+ }
+
+ public MetricsIndexEntry(String bucket, Date time, int scheduleId) {
+ this.bucket = bucket;
+ this.scheduleId = scheduleId;
+ this.time = new DateTime(time);
+ }
+
+ public MetricsIndexEntry(String bucket, DateTime time, int scheduleId) {
+ this.bucket = bucket;
+ this.scheduleId = scheduleId;
+ this.time = time;
+ }
+
+ public String getBucket() {
+
+ return bucket;
+ }
+
+ public void setBucket(String bucket) {
+ this.bucket = bucket;
+ }
+
+ public int getScheduleId() {
+ return scheduleId;
+ }
+
+ public void setScheduleId(int scheduleId) {
+ this.scheduleId = scheduleId;
+ }
+
+ public DateTime getTime() {
+ return time;
+ }
+
+ public void setTime(DateTime time) {
+ this.time = time;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ MetricsIndexEntry that = (MetricsIndexEntry) o;
+
+ if (scheduleId != that.scheduleId) return false;
+ if (!bucket.equals(that.bucket)) return false;
+ if (!time.equals(that.time)) return false;
+
+ return true;
+ }
+
+ @Override
+ public int hashCode() {
+ int result = bucket.hashCode();
+ result = 31 * result + scheduleId;
+ result = 31 * result + time.hashCode();
+ return result;
+ }
+
+ @Override
+ public String toString() {
+ return "MetricsIndexEntry[bucket=" + bucket + ", scheduleId=" + scheduleId + ", time=" +
+ DateTimeFormat.mediumDateTime().print(time) + "]";
+ }
+}
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 bf137b7..bc89df6 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
@@ -39,6 +39,8 @@ import java.util.TreeMap;
import javax.sql.DataSource;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
import org.joda.time.DateTime;
import org.joda.time.DateTimeComparator;
import org.joda.time.Hours;
@@ -73,6 +75,8 @@ public class MetricsServer {
private static final int DEFAULT_PAGE_SIZE = 200;
+ private final Log log = LogFactory.getLog(MetricsServer.class);
+
private Cluster cluster;
private String keyspaceName;
@@ -284,9 +288,11 @@ public class MetricsServer {
statement.setDouble(3, data.getValue());
statement.executeUpdate();
+ updates.put(data.getScheduleId(), new DateTime(data.getTimestamp()).hourOfDay().roundFloorCopy());
//statement.addBatch();
}
// statement.executeUpdate();
+ updateMetricsQueue(oneHourMetricsDataCF, updates);
} catch (SQLException e) {
throw new RuntimeException(e);
} finally {
@@ -517,19 +523,51 @@ public class MetricsServer {
// }
private MutationResult updateMetricsQueue(String columnFamily, Map<Integer, DateTime> updates) {
- Mutator<String> mutator = HFactory.createMutator(keyspace, StringSerializer.get());
-
- for (Integer scheduleId : updates.keySet()) {
- DateTime collectionTime = new DateTime(updates.get(scheduleId));
- Composite composite = new Composite();
- composite.addComponent(collectionTime.getMillis(), LongSerializer.get());
- composite.addComponent(scheduleId, IntegerSerializer.get());
- HColumn<Composite, Integer> column = HFactory.createColumn(composite, 0, CompositeSerializer.get(),
- IntegerSerializer.get());
- mutator.addInsertion(columnFamily, metricsQueueCF, column);
+// Mutator<String> mutator = HFactory.createMutator(keyspace, StringSerializer.get());
+//
+// for (Integer scheduleId : updates.keySet()) {
+// DateTime collectionTime = new DateTime(updates.get(scheduleId));
+// Composite composite = new Composite();
+// composite.addComponent(collectionTime.getMillis(), LongSerializer.get());
+// composite.addComponent(scheduleId, IntegerSerializer.get());
+// HColumn<Composite, Integer> column = HFactory.createColumn(composite, 0, CompositeSerializer.get(),
+// IntegerSerializer.get());
+// mutator.addInsertion(columnFamily, metricsQueueCF, column);
+// }
+//
+// return mutator.execute();
+ Connection connection = null;
+ String errorMsg = "Cannot update " + columnFamily + " index for schedule ids " + updates.keySet();
+ try {
+ connection = cassandraDS.getConnection();
+ } catch (SQLException e) {
+ log.error("Failed to obtain connection", e);
+ log.warn(errorMsg);
+ return null;
}
+ String sql = "INSERT INTO " + metricsQueueCF + " (bucket, time, schedule_id, null_col) VALUES (?, ?, ?, ?)";
+ PreparedStatement statement = null;
+ try {
+ statement = connection.prepareStatement(sql);
+ } catch (SQLException e) {
+ log.error("Failed to create prepared statement [" + sql + "]", e);
+ log.warn(errorMsg);
+ return null;
+ }
+ for (Integer scheduleId : updates.keySet()) {
+ try {
+ statement.setString(1, columnFamily);
+ statement.setDate(2, new java.sql.Date(updates.get(scheduleId).getMillis()));
+ statement.setInt(3, scheduleId);
+ statement.setBoolean(4, false);
- return mutator.execute();
+ statement.executeUpdate();
+ } catch (SQLException e) {
+ log.warn("Failed to update " + columnFamily + " index for " + scheduleId + " at time slice " +
+ updates.get(scheduleId));
+ }
+ }
+ return null;
}
protected DateTime getCurrentHour() {
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 e086627..0ff4358 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
@@ -37,6 +37,7 @@ import static org.testng.Assert.assertEquals;
import java.sql.Connection;
import java.sql.ResultSet;
+import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.HashSet;
@@ -54,6 +55,8 @@ import org.joda.time.Days;
import org.joda.time.Duration;
import org.joda.time.chrono.GregorianChronology;
import org.joda.time.field.DividedDateTimeField;
+import org.joda.time.format.DateTimeFormat;
+import org.joda.time.format.DateTimeFormatter;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.BeforeMethod;
@@ -123,6 +126,8 @@ public class MetricsServerTest {
private CassandraDataSource dataSource;
+ private Connection connection;
+
private static class MetricsServerStub extends MetricsServer {
private DateTime currentHour;
@@ -149,6 +154,11 @@ public class MetricsServerTest {
initService.waitForSchemaAgreement("rhq", hosts);
dataSource = new CassandraDataSource("127.0.0.1", 9160, "rhq", null, null, "3.0.0");
+ try {
+ connection = dataSource.getConnection();
+ } catch (SQLException e) {
+ throw new CassandraException("Unable to get JDBC connection.", e);
+ }
}
@AfterClass
@@ -193,14 +203,21 @@ public class MetricsServerTest {
return rowMutator.execute();
}
+ @Test
+ public void formatDate() {
+ DateTime dateTime = now();
+ DateTimeFormatter format = DateTimeFormat.mediumDateTime();
+ }
+
@Test//(enabled = ENABLED)
public void insertMultipleRawNumericDataForOneSchedule() throws Exception {
int scheduleId = 123;
- //DateTime hour0 = now.hourOfDay().roundFloorCopy().minusHours(now.hourOfDay().get());
- DateTime threeMinutesAgo = now().minusMinutes(3);
- DateTime twoMinutesAgo = now().minusMinutes(2);
- DateTime oneMinuteAgo = now().minusMinutes(1);
+ DateTime hour0 = now().hourOfDay().roundFloorCopy().minusHours(now().hourOfDay().get());
+ DateTime currentTime = hour0.plusHours(4).plusMinutes(44);
+ DateTime threeMinutesAgo = currentTime.minusMinutes(3);
+ DateTime twoMinutesAgo = currentTime.minusMinutes(2);
+ DateTime oneMinuteAgo = currentTime.minusMinutes(1);
int sevenDays = Duration.standardDays(7).toStandardSeconds().getSeconds();
@@ -249,7 +266,6 @@ public class MetricsServerTest {
// assert1HourMetricsQueueEquals(asList(HFactory.createColumn(expectedComposite, 0, CompositeSerializer.get(),
// IntegerSerializer.get())));
- Connection connection = dataSource.getConnection();
Statement statement = connection.createStatement();
ResultSet resultSet = statement.executeQuery("SELECT * FROM raw_metrics WHERE schedule_id = " + scheduleId);
@@ -261,6 +277,11 @@ public class MetricsServerTest {
resultSet.close();
assertCollectionMatchesNoOrder("Failed to retrieve raw metric data", data, actual, "name");
+
+ List<MetricsIndexEntry> expectedIndex = asList(new MetricsIndexEntry(ONE_HOUR_METRIC_DATA_CF,
+ hour0.plusHours(4), scheduleId));
+ assertMetricsQueueEquals(ONE_HOUR_METRIC_DATA_CF, expectedIndex, "Failed to update index for " +
+ ONE_HOUR_METRIC_DATA_CF);
}
@Test(enabled = ENABLED)
@@ -786,6 +807,24 @@ public class MetricsServerTest {
}
}
+ private void assertMetricsQueueEquals(String columnFamily, List<MetricsIndexEntry> expected, String msg) {
+ Statement statement = null;
+ ResultSet resultSet = null;
+ try {
+ List<MetricsIndexEntry> actual = new ArrayList<MetricsIndexEntry>();
+ statement = connection.createStatement();
+ resultSet = statement.executeQuery("SELECT bucket, time, schedule_id FROM " + METRICS_WORK_QUEUE_CF +
+ " WHERE bucket = '" + columnFamily + "'");
+ while (resultSet.next()) {
+ actual.add(new MetricsIndexEntry(resultSet.getString(1), resultSet.getDate(2), resultSet.getInt(3)));
+ }
+ assertCollectionMatchesNoOrder("Failed to retrieve raw metric data", expected, actual, msg + ": " +
+ columnFamily + " index not match expected values.");
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
private void assert1HourDataEquals(int scheduleId, List<HColumn<Composite, Double>> expected) {
assertMetricDataEquals(scheduleId, ONE_HOUR_METRIC_DATA_CF, expected);
}
11 years, 6 months
[rhq] modules/core
by Jiri Kremser
modules/core/plugin-container/src/main/java/org/rhq/core/pc/measurement/MeasurementCollectorRunner.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
New commits:
commit ce0f7f28d85bd4fbd23b6969499715ae8a28a6e6
Author: Jirka Kremser <jkremser(a)redhat.com>
Date: Thu Nov 1 15:54:18 2012 +0100
[BZ 872185 - Traceability of particular measurements in agent's log] If TRACE log lvl is enabled, each measurement is logged into the agent's log
diff --git a/modules/core/plugin-container/src/main/java/org/rhq/core/pc/measurement/MeasurementCollectorRunner.java b/modules/core/plugin-container/src/main/java/org/rhq/core/pc/measurement/MeasurementCollectorRunner.java
index e9303b4..d3d366f 100644
--- a/modules/core/plugin-container/src/main/java/org/rhq/core/pc/measurement/MeasurementCollectorRunner.java
+++ b/modules/core/plugin-container/src/main/java/org/rhq/core/pc/measurement/MeasurementCollectorRunner.java
@@ -133,7 +133,7 @@ public class MeasurementCollectorRunner implements Callable<MeasurementReport>,
long start = System.currentTimeMillis();
measurementComponent.getValues(report, Collections.unmodifiableSet(requests));
long duration = (System.currentTimeMillis() - start);
- if (duration > 2000L) {
+ if (duration > 2000L || log.isTraceEnabled()) {
String message = "[PERF] Collection of measurements for [" + resource + "] (component=["
+ measurementComponent + "]) took [" + duration + "]ms";
if (log.isDebugEnabled()) {
11 years, 6 months
[rhq] Branch 'feature/cassandra-backend' - modules/common
by John Sanda
modules/common/cassandra-common-itests/src/main/java/org/rhq/cassandra/CassandraClusterManager.java | 6 ++-
modules/common/cassandra-common-itests/src/test/java/org/rhq/cassandra/common/BootstrapDeployerTest.java | 18 +++++-----
modules/common/cassandra-common/src/main/java/org/rhq/cassandra/DeploymentOptions.java | 18 +++++-----
3 files changed, 23 insertions(+), 19 deletions(-)
New commits:
commit fa4b609c4c15456fd28a89e6d842b92dd4b3de95
Author: John Sanda <jsanda(a)redhat.com>
Date: Wed Oct 31 22:12:03 2012 -0400
Moving logic out of DeploymentOptions constructor and into new load method
The default constructor of DeploymentOptions initialized its properties first
looking at system properties then at values in cassandra.properties. This was
kind of awkward for tests as they had to populate a Properties object and pass
that into the DeploymentOptions(Properties) constructor. The logic in the
default constructor has been moved to a new load method. This makes it easier
for tests to simply call setter methods on properties then invoke the load
method to get default values for other, remaining properties.
diff --git a/modules/common/cassandra-common-itests/src/main/java/org/rhq/cassandra/CassandraClusterManager.java b/modules/common/cassandra-common-itests/src/main/java/org/rhq/cassandra/CassandraClusterManager.java
index 669f5ba..0fa49b8 100644
--- a/modules/common/cassandra-common-itests/src/main/java/org/rhq/cassandra/CassandraClusterManager.java
+++ b/modules/common/cassandra-common-itests/src/main/java/org/rhq/cassandra/CassandraClusterManager.java
@@ -82,7 +82,11 @@ public class CassandraClusterManager implements IInvokedMethodListener {
DeploymentOptions deploymentOptions = new DeploymentOptions();
deploymentOptions.setClusterDir(clusterDir.getAbsolutePath());
deploymentOptions.setNumNodes(numNodes);
- deploymentOptions.setLoggingLevel("DEBUG");
+ try {
+ deploymentOptions.load();
+ } catch (IOException e) {
+ throw new RuntimeException("Unable to load deployment options.", e);
+ }
BootstrapDeployer deployer = new BootstrapDeployer();
deployer.setDeploymentOptions(deploymentOptions);
diff --git a/modules/common/cassandra-common-itests/src/test/java/org/rhq/cassandra/common/BootstrapDeployerTest.java b/modules/common/cassandra-common-itests/src/test/java/org/rhq/cassandra/common/BootstrapDeployerTest.java
index 0819a51..e1eef5c 100644
--- a/modules/common/cassandra-common-itests/src/test/java/org/rhq/cassandra/common/BootstrapDeployerTest.java
+++ b/modules/common/cassandra-common-itests/src/test/java/org/rhq/cassandra/common/BootstrapDeployerTest.java
@@ -30,9 +30,9 @@ import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertTrue;
import java.io.File;
+import java.io.IOException;
import java.util.List;
import java.util.Map;
-import java.util.Properties;
import org.testng.annotations.Listeners;
import org.testng.annotations.Test;
@@ -61,14 +61,14 @@ public class BootstrapDeployerTest {
File clusterDir = new File(basedir, "cassandra");
int numNodes = 2;
- Properties properties = new Properties();
- properties.setProperty("rhq.cassandra.cluster.dir", clusterDir.getAbsolutePath());
- properties.setProperty("rhq.cassandra.logging.level", "DEBUG");
- properties.setProperty("rhq.cassandra.cluster.num-nodes", "2");
- properties.setProperty("rhq.cassandra.ring.delay", "1000");
- properties.setProperty("rhq.cassandra.node.num-tokens", "256");
-
- DeploymentOptions deploymentOptions = new DeploymentOptions(properties);
+ DeploymentOptions deploymentOptions = new DeploymentOptions();
+ try {
+ deploymentOptions.setClusterDir(clusterDir.getAbsolutePath());
+ deploymentOptions.setRingDelay(1000L);
+ deploymentOptions.load();
+ } catch (IOException e) {
+ throw new CassandraException(e);
+ }
BootstrapDeployer deployer = new BootstrapDeployer();
deployer.setDeploymentOptions(deploymentOptions);
diff --git a/modules/common/cassandra-common/src/main/java/org/rhq/cassandra/DeploymentOptions.java b/modules/common/cassandra-common/src/main/java/org/rhq/cassandra/DeploymentOptions.java
index 60cb9a3..831ee3f 100644
--- a/modules/common/cassandra-common/src/main/java/org/rhq/cassandra/DeploymentOptions.java
+++ b/modules/common/cassandra-common/src/main/java/org/rhq/cassandra/DeploymentOptions.java
@@ -51,20 +51,24 @@ public class DeploymentOptions {
private Integer numTokens;
public DeploymentOptions() {
- init(loadProperties());
}
- private Properties loadProperties() {
+ public DeploymentOptions(Properties properties) {
+ init(properties);
+ }
+
+ public void load() throws IOException {
InputStream stream = null;
try {
stream = getClass().getResourceAsStream("/cassandra.properties");
Properties props = new Properties();
props.load(stream);
- return props;
+ init(props);
} catch (IOException e) {
- throw new RuntimeException("Unable to create DeploymentOptions. An error occurred while loading " +
- "cassandra.properties", e);
+ log.warn("Unable to load deployment options from cassandra.properties.");
+ log.info("The following error occurred while trying to load options.", e);
+ throw e;
} finally {
if (stream != null) {
try {
@@ -77,10 +81,6 @@ public class DeploymentOptions {
}
}
- public DeploymentOptions(Properties properties) {
- init(properties);
- }
-
private void init(Properties properties) {
setBundleFileName(properties.getProperty("rhq.cassandra.bundle.filename"));
setBundleName(properties.getProperty("rhq.cassandra.bundle.name"));
11 years, 6 months