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);
}