modules/enterprise/server/itests-2/src/test/java/org/rhq/enterprise/server/measurement/MeasurementDataManagerBeanTest.java
| 6
modules/enterprise/server/itests-2/src/test/java/org/rhq/enterprise/server/measurement/test/MeasurementBaselineManagerTest.java
| 6
modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/core/StartupBean.java
| 1
modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/storage/StorageClientManagerBean.java
| 18 -
modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/storage/StorageClusterMonitor.java
| 69 ++++
modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsDAO.java
| 153 ++--------
modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsServer.java
| 5
modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/StorageResultSetFuture.java
| 76 ++++
modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/StorageSession.java
| 110 +++++++
modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/StorageStateListener.java
| 20 +
modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/domain/ListPagedResult.java
| 6
modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/domain/SimplePagedResult.java
| 11
modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/CassandraIntegrationTest.java
| 35 ++
modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/ClusterMonitorTest.java
| 92 ++++++
modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsBaselineCalculatorTest.java
| 14
modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsDAOTest.java
| 17 -
modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsServerTest.java
| 7
modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/TimeoutTest.java
| 5
modules/enterprise/server/server-metrics/src/test/resources/log4j.properties
| 2
modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/Simulator.java
| 5
pom.xml
| 2
21 files changed, 497 insertions(+), 163 deletions(-)
New commits:
commit 9fbd3da4446f026dd4095f23345a5acaa0b956c0
Author: John Sanda <jsanda(a)redhat.com>
Date: Thu Jul 25 22:08:20 2013 -0400
check cluster status using cql driver instead of using jmx
The server had been polling storage cluster nodes with a jmx call to make sure
that at least one node is up for client requests. If no nodes are up, then the
server goes into maintenance mode.
That check is no longer done with a quartz job using jmx. It is now performed
with event notifications we receive from the cql driver. This means that the
server no longer has to care about storage node jmx ports. This has a couple
benefits. First, it reduces the complexity involved with changing the jmx port.
Secondly, since the server no longer is making jmx calls to storage nodes, the
port can be locked down to localhost access.
Any server side code that needs access to the driver's Session should instead
use the new StorageSession class. StorageSession provides the same API, and it
also encapsulates the event handling/propagation logic that is necesary for
monitoring the cluster availability.
diff --git
a/modules/enterprise/server/itests-2/src/test/java/org/rhq/enterprise/server/measurement/MeasurementDataManagerBeanTest.java
b/modules/enterprise/server/itests-2/src/test/java/org/rhq/enterprise/server/measurement/MeasurementDataManagerBeanTest.java
index 4cc67c0..9f49143 100644
---
a/modules/enterprise/server/itests-2/src/test/java/org/rhq/enterprise/server/measurement/MeasurementDataManagerBeanTest.java
+++
b/modules/enterprise/server/itests-2/src/test/java/org/rhq/enterprise/server/measurement/MeasurementDataManagerBeanTest.java
@@ -39,7 +39,6 @@ import java.util.List;
import javax.ejb.EJB;
-import com.datastax.driver.core.Session;
import com.datastax.driver.core.exceptions.NoHostAvailableException;
import org.joda.time.DateTime;
@@ -60,13 +59,14 @@ import org.rhq.core.domain.resource.Agent;
import org.rhq.core.domain.resource.Resource;
import org.rhq.core.domain.resource.ResourceType;
import org.rhq.enterprise.server.auth.SubjectManagerLocal;
-import org.rhq.enterprise.server.storage.StorageClientManagerBean;
import org.rhq.enterprise.server.drift.DriftServerPluginService;
import org.rhq.enterprise.server.resource.ResourceManagerLocal;
+import org.rhq.enterprise.server.storage.StorageClientManagerBean;
import org.rhq.enterprise.server.test.AbstractEJB3Test;
import org.rhq.enterprise.server.test.TransactionCallback;
import org.rhq.enterprise.server.util.ResourceTreeHelper;
import org.rhq.server.metrics.MetricsDAO;
+import org.rhq.server.metrics.StorageSession;
import org.rhq.server.metrics.domain.AggregateNumericMetric;
import org.rhq.server.metrics.domain.AggregateType;
import org.rhq.server.metrics.domain.MetricsTable;
@@ -360,7 +360,7 @@ public class MeasurementDataManagerBeanTest extends AbstractEJB3Test
{
private void purgeMetricsTables() {
try {
- Session session = storageClientManager.getSession();
+ StorageSession session = storageClientManager.getSession();
session.execute("TRUNCATE " + MetricsTable.RAW);
session.execute("TRUNCATE " + MetricsTable.ONE_HOUR);
diff --git
a/modules/enterprise/server/itests-2/src/test/java/org/rhq/enterprise/server/measurement/test/MeasurementBaselineManagerTest.java
b/modules/enterprise/server/itests-2/src/test/java/org/rhq/enterprise/server/measurement/test/MeasurementBaselineManagerTest.java
index 49048ee..fdf2b1a 100644
---
a/modules/enterprise/server/itests-2/src/test/java/org/rhq/enterprise/server/measurement/test/MeasurementBaselineManagerTest.java
+++
b/modules/enterprise/server/itests-2/src/test/java/org/rhq/enterprise/server/measurement/test/MeasurementBaselineManagerTest.java
@@ -28,7 +28,6 @@ import java.util.Random;
import javax.inject.Inject;
import javax.persistence.Query;
-import com.datastax.driver.core.Session;
import com.datastax.driver.core.exceptions.NoHostAvailableException;
import org.testng.annotations.Test;
@@ -45,13 +44,14 @@ import org.rhq.core.domain.resource.Resource;
import org.rhq.core.domain.resource.ResourceCategory;
import org.rhq.core.domain.resource.ResourceType;
import org.rhq.core.domain.util.PageControl;
-import org.rhq.enterprise.server.storage.StorageClientManagerBean;
import org.rhq.enterprise.server.measurement.MeasurementBaselineManagerLocal;
import org.rhq.enterprise.server.measurement.MeasurementOOBManagerLocal;
import org.rhq.enterprise.server.resource.ResourceManagerLocal;
+import org.rhq.enterprise.server.storage.StorageClientManagerBean;
import org.rhq.enterprise.server.test.AbstractEJB3Test;
import org.rhq.enterprise.server.util.LookupUtil;
import org.rhq.server.metrics.MetricsDAO;
+import org.rhq.server.metrics.StorageSession;
import org.rhq.server.metrics.domain.AggregateNumericMetric;
import org.rhq.server.metrics.domain.AggregateType;
import org.rhq.server.metrics.domain.MetricsTable;
@@ -645,7 +645,7 @@ public class MeasurementBaselineManagerTest extends AbstractEJB3Test
{
// Query q = em.createNativeQuery(sql);
// q.executeUpdate();
try {
- Session session = storageClientManager.getSession();
+ StorageSession session = storageClientManager.getSession();
session.execute("DELETE FROM " +
MetricsTable.ONE_HOUR.getTableName() + " WHERE schedule_id = " +
schedule.getId());
} catch (NoHostAvailableException e) {
diff --git
a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/core/StartupBean.java
b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/core/StartupBean.java
index 7ce77d8..fa88263 100644
---
a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/core/StartupBean.java
+++
b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/core/StartupBean.java
@@ -450,7 +450,6 @@ public class StartupBean implements StartupLocal {
}
storageClientManager.init();
- storageClusterHeartBeatJob.scheduleJob();
}
/**
diff --git
a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/storage/StorageClientManagerBean.java
b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/storage/StorageClientManagerBean.java
index 2d48092..ae80e50 100644
---
a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/storage/StorageClientManagerBean.java
+++
b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/storage/StorageClientManagerBean.java
@@ -50,6 +50,7 @@ import org.rhq.server.metrics.DateTimeService;
import org.rhq.server.metrics.MetricsConfiguration;
import org.rhq.server.metrics.MetricsDAO;
import org.rhq.server.metrics.MetricsServer;
+import org.rhq.server.metrics.StorageSession;
/**
* @author John Sanda
@@ -70,7 +71,7 @@ public class StorageClientManagerBean {
@EJB
private StorageNodeManagerLocal storageNodeManager;
- private Session session;
+ private StorageSession session;
private MetricsConfiguration metricsConfiguration;
private MetricsDAO metricsDAO;
private MetricsServer metricsServer;
@@ -95,7 +96,12 @@ public class StorageClientManagerBean {
String password = getRequiredStorageProperty(PASSWORD_PROP);
metricsConfiguration = new MetricsConfiguration();
- session = createSession(username, password,
storageNodeManager.getStorageNodes());
+
+ Session wrappedSession = createSession(username, password,
storageNodeManager.getStorageNodes());
+ session = new StorageSession(wrappedSession);
+
+ session.addStorageStateListener(new StorageClusterMonitor());
+
metricsDAO = new MetricsDAO(session, metricsConfiguration);
Server server = serverManager.getServer();
@@ -121,15 +127,15 @@ public class StorageClientManagerBean {
}
public MetricsDAO getMetricsDAO() {
- return this.metricsDAO;
+ return metricsDAO;
}
public MetricsServer getMetricsServer() {
- return this.metricsServer;
+ return metricsServer;
}
- public Session getSession() {
- return this.session;
+ public StorageSession getSession() {
+ return session;
}
public MetricsConfiguration getMetricsConfiguration() {
diff --git
a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/storage/StorageClusterMonitor.java
b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/storage/StorageClusterMonitor.java
new file mode 100644
index 0000000..ec28888
--- /dev/null
+++
b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/storage/StorageClusterMonitor.java
@@ -0,0 +1,69 @@
+package org.rhq.enterprise.server.storage;
+
+import java.net.InetAddress;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import com.datastax.driver.core.exceptions.NoHostAvailableException;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.rhq.core.domain.cloud.Server;
+import org.rhq.enterprise.server.auth.SubjectManagerLocal;
+import org.rhq.enterprise.server.cloud.TopologyManagerLocal;
+import org.rhq.enterprise.server.cloud.instance.ServerManagerLocal;
+import org.rhq.enterprise.server.util.LookupUtil;
+import org.rhq.server.metrics.StorageStateListener;
+
+/**
+ * @author John Sanda
+ */
+public class StorageClusterMonitor implements StorageStateListener {
+
+ private Log log = LogFactory.getLog(StorageClusterMonitor.class);
+
+ private AtomicBoolean isClusterDown = new AtomicBoolean(false);
+
+ public boolean isClusterDown() {
+ return isClusterDown.get();
+ }
+
+ @Override
+ public void onStorageNodeUp(InetAddress address) {
+ log.info("Storage node at " + address.getHostAddress() + " is
up");
+
+ if (isClusterDown.compareAndSet(true, false)) {
+ log.info("Taking server out of maintenance mode");
+ updateServerMode(Server.OperationMode.NORMAL);
+ }
+ }
+
+ @Override
+ public void onStorageNodeDown(InetAddress address) {
+ log.info("Storage node at " + address.getHostAddress() + " is
down");
+ }
+
+ @Override
+ public void onStorageNodeRemoved(InetAddress address) {
+ log.info("Storage node at " + address.getHostAddress() + " has
been removed from the cluster");
+ }
+
+ @Override
+ public void onStorageClusterDown(NoHostAvailableException e) {
+ if (isClusterDown.compareAndSet(false, true)) {
+ log.error("The server cannot connect to any storage nodes. The server
will now go into maintenance mode.");
+ updateServerMode(Server.OperationMode.MAINTENANCE);
+ }
+ }
+
+ private void updateServerMode(Server.OperationMode mode) {
+ ServerManagerLocal serverManager = LookupUtil.getServerManager();
+ TopologyManagerLocal topologyManager = LookupUtil.getTopologyManager();
+ SubjectManagerLocal subjectManager = LookupUtil.getSubjectManager();
+
+ Server server = serverManager.getServer();
+
+ topologyManager.updateServerMode(subjectManager.getOverlord(), new Integer[]
{server.getId()},
+ Server.OperationMode.MAINTENANCE);
+ }
+}
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 3ea47e8..63c11f1 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
@@ -26,20 +26,13 @@
package org.rhq.server.metrics;
-import static com.datastax.driver.core.querybuilder.QueryBuilder.insertInto;
-
-import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
-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;
import org.apache.commons.logging.Log;
@@ -67,7 +60,7 @@ public class MetricsDAO {
private final Log log = LogFactory.getLog(MetricsDAO.class);
- private Session session;
+ private StorageSession storageSession;
private MetricsConfiguration configuration;
@@ -86,8 +79,8 @@ public class MetricsDAO {
private PreparedStatement findTimeSliceForIndex;
private PreparedStatement deleteIndexEntries;
- public MetricsDAO(Session session, MetricsConfiguration configuration) {
- this.session = session;
+ public MetricsDAO(StorageSession session, MetricsConfiguration configuration) {
+ this.storageSession = session;
this.configuration = configuration;
initPreparedStatements();
}
@@ -104,176 +97,118 @@ public class MetricsDAO {
// re-initialized and re-prepared with the new TTLs. None of this would be
necessary
// if the TTL value could be a bound value.
- insertRawData = session.prepare(
+ insertRawData = storageSession.prepare(
"INSERT INTO " + MetricsTable.RAW + " (schedule_id, time,
value) VALUES (?, ?, ?) USING TTL " +
configuration.getRawTTL());
- rawMetricsQuery = session.prepare("SELECT schedule_id, time, value FROM
" + MetricsTable.RAW +
+ rawMetricsQuery = storageSession.prepare("SELECT schedule_id, time, value
FROM " + MetricsTable.RAW +
" WHERE schedule_id = ? AND time >= ? AND time < ? ORDER BY
time");
- insertOneHourData = session.prepare("INSERT INTO " +
MetricsTable.ONE_HOUR + "(schedule_id, time, " +
+ insertOneHourData = storageSession.prepare("INSERT INTO " +
MetricsTable.ONE_HOUR + "(schedule_id, time, " +
"type, value) VALUES (?, ?, ?, ?) USING TTL " +
configuration.getOneHourTTL());
- insertSixHourData = session.prepare("INSERT INTO " +
MetricsTable.SIX_HOUR + "(schedule_id, time, " +
+ insertSixHourData = storageSession.prepare("INSERT INTO " +
MetricsTable.SIX_HOUR + "(schedule_id, time, " +
"type, value) VALUES (?, ?, ?, ?) USING TTL " +
configuration.getOneHourTTL());
- insertTwentyFourHourData = session.prepare("INSERT INTO " +
MetricsTable.TWENTY_FOUR_HOUR + "(schedule_id, " +
+ insertTwentyFourHourData = storageSession.prepare("INSERT INTO " +
MetricsTable.TWENTY_FOUR_HOUR + "(schedule_id, " +
"time, type, value) VALUES (?, ?, ?, ?) USING TTL " +
configuration.getOneHourTTL());
- updateMetricsIndex = session.prepare("INSERT INTO " +
MetricsTable.INDEX + " (bucket, time, schedule_id) " +
+ updateMetricsIndex = storageSession.prepare("INSERT INTO " +
MetricsTable.INDEX + " (bucket, time, schedule_id) " +
"VALUES (?, ?, ?)");
- findLatestRawMetric = session.prepare("SELECT schedule_id, time, value FROM
" + MetricsTable.RAW +
+ findLatestRawMetric = storageSession.prepare("SELECT schedule_id, time,
value FROM " + MetricsTable.RAW +
" WHERE schedule_id = ? ORDER BY time DESC LIMIT 1");
- findRawMetrics = session.prepare("SELECT schedule_id, time, value FROM
" + MetricsTable.RAW +
+ findRawMetrics = storageSession.prepare("SELECT schedule_id, time, value
FROM " + MetricsTable.RAW +
" WHERE schedule_id = ? AND time >= ? AND time <= ?");
- findOneHourMetricsByDateRange = session.prepare("SELECT schedule_id, time,
type, value FROM " +
+ findOneHourMetricsByDateRange = storageSession.prepare("SELECT schedule_id,
time, type, value FROM " +
MetricsTable.ONE_HOUR + " WHERE schedule_id = ? AND time >= ? AND
time < ?");
- findSixHourMetricsByDateRange = session.prepare("SELECT schedule_id, time,
type, value FROM "
+ findSixHourMetricsByDateRange = storageSession.prepare("SELECT schedule_id,
time, type, value FROM "
+ MetricsTable.SIX_HOUR + " WHERE schedule_id = ? AND time >= ? AND
time < ?");
- findTwentyFourHourMetricsByDateRange = session.prepare("SELECT schedule_id,
time, type, value FROM " +
+ findTwentyFourHourMetricsByDateRange = storageSession.prepare("SELECT
schedule_id, time, type, value FROM " +
MetricsTable.TWENTY_FOUR_HOUR + " WHERE schedule_id = ? AND time >= ?
AND time < ?");
- findIndexEntries = session.prepare("SELECT time, schedule_id FROM " +
MetricsTable.INDEX +
+ findIndexEntries = storageSession.prepare("SELECT time, schedule_id FROM
" + MetricsTable.INDEX +
" WHERE bucket = ? AND time = ?");
- findTimeSliceForIndex = session.prepare("SELECT time FROM " +
MetricsTable.INDEX +
+ findTimeSliceForIndex = storageSession.prepare("SELECT time FROM " +
MetricsTable.INDEX +
" WHERE bucket = ? AND time = ?");
- deleteIndexEntries = session.prepare("DELETE FROM " +
MetricsTable.INDEX + " WHERE bucket = ? AND time = ?");
+ deleteIndexEntries = storageSession.prepare("DELETE FROM " +
MetricsTable.INDEX + " WHERE bucket = ? AND time = ?");
long endTime = System.currentTimeMillis();
log.info("Finished initializing prepared statements in " + (endTime -
startTime) + " ms");
}
- public ResultSetFuture insertRawData(MeasurementDataNumeric data) {
+ public StorageResultSetFuture insertRawData(MeasurementDataNumeric data) {
BoundStatement statement = insertRawData.bind(data.getScheduleId(), new
Date(data.getTimestamp()),
data.getValue());
- return session.executeAsync(statement);
- }
-
- public List<MetricResultFuture<MeasurementDataNumeric>>
insertRawMetricsAsync(Set<MeasurementDataNumeric> dataSet,
- int ttl) {
- try {
- List<MetricResultFuture<MeasurementDataNumeric>> resultFutures =
new ArrayList<MetricResultFuture<MeasurementDataNumeric>>();
-
- String cql = "INSERT INTO raw_metrics (schedule_id, time, value) VALUES
(?, ?, ?) " + "USING TTL " + ttl;
- PreparedStatement statement = session.prepare(cql);
-
- for (MeasurementDataNumeric data : dataSet) {
- BoundStatement boundStatement = statement.bind(data.getScheduleId(), new
Date(data.getTimestamp()),
- data.getValue());
-
- resultFutures.add(new
MetricResultFuture<MeasurementDataNumeric>(session.executeAsync(boundStatement),
- data));
- }
-
- return resultFutures;
- } catch (NoHostAvailableException e) {
- throw new CQLException(e);
- }
+ return storageSession.executeAsync(statement);
}
public ResultSet insertOneHourData(int scheduleId, long timestamp, AggregateType
type, double value) {
BoundStatement statement = insertOneHourData.bind(scheduleId, new
Date(timestamp), type.ordinal(), value);
- return session.execute(statement);
+ return storageSession.execute(statement);
}
public ResultSet insertSixHourData(int scheduleId, long timestamp, AggregateType
type, double value) {
BoundStatement statement = insertSixHourData.bind(scheduleId, new
Date(timestamp), type.ordinal(), value);
- return session.execute(statement);
+ return storageSession.execute(statement);
}
public ResultSet insertTwentyFourHourData(int scheduleId, long timestamp,
AggregateType type, double value) {
BoundStatement statement = insertTwentyFourHourData.bind(scheduleId, new
Date(timestamp), type.ordinal(),
value);
- return session.execute(statement);
- }
-
- public List<MetricResultFuture<AggregateNumericMetric>>
insertAggregatesAsync(MetricsTable table,
- List<AggregateNumericMetric> metrics, int ttl) {
- List<MetricResultFuture<AggregateNumericMetric>> updates = new
ArrayList<MetricResultFuture<AggregateNumericMetric>>();
-
- if (metrics.isEmpty()) {
- return updates;
- }
-
- try {
- Statement statement = null;
-
- for (AggregateNumericMetric metric : metrics) {
- statement = insertInto(table.getTableName())
- .value("schedule_id", metric.getScheduleId())
- .value("time", new Date(metric.getTimestamp()))
- .value("type", AggregateType.MIN.ordinal())
- .value("value", metric.getMin());
- updates.add(new
MetricResultFuture<AggregateNumericMetric>(session.executeAsync(statement),
metric));
-
- statement = insertInto(table.getTableName())
- .value("schedule_id", metric.getScheduleId())
- .value("time", new Date(metric.getTimestamp()))
- .value("type", AggregateType.MAX.ordinal())
- .value("value", metric.getMax());
- updates.add(new
MetricResultFuture<AggregateNumericMetric>(session.executeAsync(statement),
metric));
-
- statement = insertInto(table.getTableName())
- .value("schedule_id", metric.getScheduleId())
- .value("time", new Date(metric.getTimestamp()))
- .value("type", AggregateType.AVG.ordinal())
- .value("value", metric.getAvg());
- updates.add(new
MetricResultFuture<AggregateNumericMetric>(session.executeAsync(statement),
metric));
- }
-
- return updates;
- } catch (Exception e) {
- throw new CQLException(e);
- }
+ return storageSession.execute(statement);
}
public Iterable<RawNumericMetric> findRawMetrics(int scheduleId, long
startTime, long endTime) {
try {
BoundStatement boundStatement = rawMetricsQuery.bind(scheduleId, new
Date(startTime), new Date(endTime));
- return new SimplePagedResult<RawNumericMetric>(boundStatement, new
RawNumericMetricMapper(false), session);
+ return new SimplePagedResult<RawNumericMetric>(boundStatement, new
RawNumericMetricMapper(false),
+ storageSession);
} catch (NoHostAvailableException e) {
throw new CQLException(e);
}
}
- public ResultSetFuture findRawMetricsAsync(int scheduleId, long startTime, long
endTime) {
+ public StorageResultSetFuture findRawMetricsAsync(int scheduleId, long startTime,
long endTime) {
BoundStatement boundStatement = rawMetricsQuery.bind(scheduleId, new
Date(startTime), new Date(endTime));
- return session.executeAsync(boundStatement);
+ return storageSession.executeAsync(boundStatement);
}
public RawNumericMetric findLatestRawMetric(int scheduleId) {
RawNumericMetricMapper mapper = new RawNumericMetricMapper(false);
BoundStatement boundStatement = findLatestRawMetric.bind(scheduleId);
- ResultSet resultSet = session.execute(boundStatement);
+ ResultSet resultSet = storageSession.execute(boundStatement);
return mapper.mapOne(resultSet);
}
public Iterable<RawNumericMetric> findRawMetrics(List<Integer>
scheduleIds, long startTime, long endTime) {
return new ListPagedResult<RawNumericMetric>(findRawMetrics, scheduleIds,
startTime, endTime,
- new RawNumericMetricMapper(), session);
+ new RawNumericMetricMapper(), storageSession);
}
public Iterable<AggregateNumericMetric> findOneHourMetrics(int scheduleId, long
startTime, long endTime) {
BoundStatement statement = findOneHourMetricsByDateRange.bind(scheduleId, new
Date(startTime), new Date(endTime));
- return new SimplePagedResult<AggregateNumericMetric>(statement, new
AggregateNumericMetricMapper(), session);
+ return new SimplePagedResult<AggregateNumericMetric>(statement, new
AggregateNumericMetricMapper(),
+ storageSession);
}
public Iterable<AggregateNumericMetric> findSixHourMetrics(int scheduleId, long
startTime, long endTime) {
BoundStatement statement = findSixHourMetricsByDateRange.bind(scheduleId, new
Date(startTime), new Date(endTime));
- return new SimplePagedResult<AggregateNumericMetric>(statement, new
AggregateNumericMetricMapper(), session);
+ return new SimplePagedResult<AggregateNumericMetric>(statement, new
AggregateNumericMetricMapper(),
+ storageSession);
}
public Iterable<AggregateNumericMetric> findTwentyFourHourMetrics(int
scheduleId, long startTime, long endTime) {
BoundStatement statement = findTwentyFourHourMetricsByDateRange.bind(scheduleId,
new Date(startTime), new Date(endTime));
- return new SimplePagedResult<AggregateNumericMetric>(statement, new
AggregateNumericMetricMapper(), session);
+ return new SimplePagedResult<AggregateNumericMetric>(statement, new
AggregateNumericMetricMapper(),
+ storageSession);
}
public Iterable<AggregateSimpleNumericMetric>
findAggregatedSimpleOneHourMetric(int scheduleId, long startTime,
@@ -281,52 +216,52 @@ public class MetricsDAO {
BoundStatement statement = findOneHourMetricsByDateRange.bind(scheduleId, new
Date(startTime),
new Date(endTime));
return new SimplePagedResult<AggregateSimpleNumericMetric>(statement, new
AggregateSimpleNumericMetricMapper(),
- session);
+ storageSession);
}
public Iterable<AggregateNumericMetric> findOneHourMetrics(List<Integer>
scheduleIds, long startTime,
long endTime) {
return new
ListPagedResult<AggregateNumericMetric>(findOneHourMetricsByDateRange, scheduleIds,
startTime, endTime,
- new AggregateNumericMetricMapper(), session);
+ new AggregateNumericMetricMapper(), storageSession);
}
public Iterable<AggregateNumericMetric> findSixHourMetrics(List<Integer>
scheduleIds, long startTime,
long endTime) {
return new
ListPagedResult<AggregateNumericMetric>(findSixHourMetricsByDateRange, scheduleIds,
startTime, endTime,
- new AggregateNumericMetricMapper(), session);
+ new AggregateNumericMetricMapper(), storageSession);
}
public Iterable<AggregateNumericMetric>
findTwentyFourHourMetrics(List<Integer> scheduleIds, long startTime,
long endTime) {
return new
ListPagedResult<AggregateNumericMetric>(findTwentyFourHourMetricsByDateRange,
scheduleIds, startTime, endTime,
- new AggregateNumericMetricMapper(), session);
+ new AggregateNumericMetricMapper(), storageSession);
}
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);
+ return new SimplePagedResult<MetricsIndexEntry>(statement, new
MetricsIndexEntryMapper(table), storageSession);
}
public ResultSet setFindTimeSliceForIndex(MetricsTable table, long timestamp) {
BoundStatement statement = findTimeSliceForIndex.bind(table.toString(), new
Date(timestamp));
- return session.execute(statement);
+ return storageSession.execute(statement);
}
public void updateMetricsIndex(MetricsTable table, Map<Integer, Long> updates)
{
for (Integer scheduleId : updates.keySet()) {
BoundStatement statement = updateMetricsIndex.bind(table.getTableName(),
new Date(updates.get(scheduleId)), scheduleId);
- session.execute(statement);
+ storageSession.execute(statement);
}
}
- public ResultSetFuture updateMetricsIndex(MetricsTable table, int scheduleId, long
timestamp) {
+ public StorageResultSetFuture updateMetricsIndex(MetricsTable table, int scheduleId,
long timestamp) {
BoundStatement statement = updateMetricsIndex.bind(table.getTableName(), new
Date(timestamp), scheduleId);
- return session.executeAsync(statement);
+ return storageSession.executeAsync(statement);
}
public void deleteMetricsIndexEntries(MetricsTable table, long timestamp) {
BoundStatement statement = deleteIndexEntries.bind(table.getTableName(), new
Date(timestamp));
- session.execute(statement);
+ storageSession.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 7395953..9756006 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
@@ -35,7 +35,6 @@ import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicInteger;
import com.datastax.driver.core.ResultSet;
-import com.datastax.driver.core.ResultSetFuture;
import com.datastax.driver.core.Row;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
@@ -299,7 +298,7 @@ public class MetricsServer {
for (final MeasurementDataNumeric data : dataSet) {
semaphore.acquire();
- ResultSetFuture resultSetFuture = dao.insertRawData(data);
+ StorageResultSetFuture resultSetFuture = dao.insertRawData(data);
Futures.addCallback(resultSetFuture, new FutureCallback<ResultSet>() {
@Override
public void onSuccess(ResultSet rows) {
@@ -325,7 +324,7 @@ public class MetricsServer {
long timeSlice = dateTimeService.getTimeSlice(new
DateTime(rawData.getTimestamp()),
configuration.getRawTimeSliceDuration()).getMillis();
- ResultSetFuture resultSetFuture = dao.updateMetricsIndex(MetricsTable.ONE_HOUR,
rawData.getScheduleId(),
+ StorageResultSetFuture resultSetFuture =
dao.updateMetricsIndex(MetricsTable.ONE_HOUR, rawData.getScheduleId(),
timeSlice);
Futures.addCallback(resultSetFuture, new FutureCallback<ResultSet>() {
@Override
diff --git
a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/StorageResultSetFuture.java
b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/StorageResultSetFuture.java
new file mode 100644
index 0000000..902bf4a
--- /dev/null
+++
b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/StorageResultSetFuture.java
@@ -0,0 +1,76 @@
+package org.rhq.server.metrics;
+
+import java.util.List;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import com.datastax.driver.core.ResultSet;
+import com.datastax.driver.core.ResultSetFuture;
+import com.datastax.driver.core.exceptions.NoHostAvailableException;
+import com.google.common.util.concurrent.ListenableFuture;
+
+/**
+ * @author John Sanda
+ */
+public class StorageResultSetFuture implements ListenableFuture<ResultSet> {
+
+ private ResultSetFuture wrapperFuture;
+
+ private List<StorageStateListener> listeners;
+
+ public StorageResultSetFuture(ResultSetFuture resultSetFuture,
List<StorageStateListener> listeners) {
+ wrapperFuture = resultSetFuture;
+ this.listeners = listeners;
+ }
+
+ @Override
+ public void addListener(Runnable listener, Executor executor) {
+ wrapperFuture.addListener(listener, executor);
+ }
+
+ @Override
+ public boolean cancel(boolean mayInterruptIfRunning) {
+ return wrapperFuture.cancel(mayInterruptIfRunning);
+ }
+
+ @Override
+ public boolean isCancelled() {
+ return wrapperFuture.isCancelled();
+ }
+
+ @Override
+ public boolean isDone() {
+ return wrapperFuture.isDone();
+ }
+
+ @Override
+ public ResultSet get() throws InterruptedException, ExecutionException {
+ try {
+ return wrapperFuture.get();
+ } catch (ExecutionException e) {
+ return handleException(e);
+ }
+ }
+
+ @Override
+ public ResultSet get(long timeout, TimeUnit unit) throws InterruptedException,
ExecutionException,
+ TimeoutException {
+ try {
+ return wrapperFuture.get(timeout, unit);
+ } catch (ExecutionException e) {
+ return handleException(e);
+ }
+ }
+
+ private ResultSet handleException(ExecutionException e) throws ExecutionException {
+ if (e.getCause() instanceof NoHostAvailableException) {
+ NoHostAvailableException cause = (NoHostAvailableException) e.getCause();
+ for (StorageStateListener listener : listeners) {
+ listener.onStorageClusterDown(cause);
+ }
+ }
+ throw e;
+ }
+}
diff --git
a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/StorageSession.java
b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/StorageSession.java
new file mode 100644
index 0000000..3f7af3f
--- /dev/null
+++
b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/StorageSession.java
@@ -0,0 +1,110 @@
+package org.rhq.server.metrics;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.Host;
+import com.datastax.driver.core.PreparedStatement;
+import com.datastax.driver.core.Query;
+import com.datastax.driver.core.ResultSet;
+import com.datastax.driver.core.ResultSetFuture;
+import com.datastax.driver.core.Session;
+import com.datastax.driver.core.exceptions.NoHostAvailableException;
+
+/**
+ * @author John Sanda
+ */
+public class StorageSession implements Host.StateListener {
+
+ private Session wrappedSession;
+
+ private List<StorageStateListener> listeners = new
ArrayList<StorageStateListener>();
+
+ public StorageSession(Session wrappedSession) {
+ this.wrappedSession = wrappedSession;
+ this.wrappedSession.getCluster().register(this);
+ }
+
+ public void addStorageStateListener(StorageStateListener listener) {
+ listeners.add(listener);
+ }
+
+ public ResultSet execute(String query) {
+ try {
+ return wrappedSession.execute(query);
+ } catch (NoHostAvailableException e) {
+ return handleException(e);
+ }
+ }
+
+ public ResultSet execute(Query query) {
+ try {
+ return wrappedSession.execute(query);
+ } catch (NoHostAvailableException e) {
+ return handleException(e);
+ }
+ }
+
+ public StorageResultSetFuture executeAsync(String query) {
+ ResultSetFuture future = wrappedSession.executeAsync(query);
+ return new StorageResultSetFuture(future, listeners);
+ }
+
+ public StorageResultSetFuture executeAsync(Query query) {
+ ResultSetFuture future = wrappedSession.executeAsync(query);
+ return new StorageResultSetFuture(future, listeners);
+ }
+
+ public PreparedStatement prepare(String query) {
+ return wrappedSession.prepare(query);
+ }
+
+ public void shutdown() {
+ wrappedSession.shutdown();
+ }
+
+ public boolean shutdown(long timeout, TimeUnit unit) {
+ return wrappedSession.shutdown(timeout, unit);
+ }
+
+ public Cluster getCluster() {
+ return wrappedSession.getCluster();
+ }
+
+ @Override
+ public void onAdd(Host host) {
+ for (StorageStateListener listener : listeners) {
+ listener.onStorageNodeUp(host.getAddress());
+ }
+ }
+
+ @Override
+ public void onUp(Host host) {
+ for (StorageStateListener listener : listeners) {
+ listener.onStorageNodeUp(host.getAddress());
+ }
+ }
+
+ @Override
+ public void onDown(Host host) {
+ for (StorageStateListener listener : listeners) {
+ listener.onStorageNodeDown(host.getAddress());
+ }
+ }
+
+ @Override
+ public void onRemove(Host host) {
+ for (StorageStateListener listener : listeners) {
+ listener.onStorageNodeRemoved(host.getAddress());
+ }
+ }
+
+ private ResultSet handleException(NoHostAvailableException e) {
+ for (StorageStateListener listener : listeners) {
+ listener.onStorageClusterDown(e);
+ }
+ throw e;
+ }
+}
diff --git
a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/StorageStateListener.java
b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/StorageStateListener.java
new file mode 100644
index 0000000..6a3a216
--- /dev/null
+++
b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/StorageStateListener.java
@@ -0,0 +1,20 @@
+package org.rhq.server.metrics;
+
+import java.net.InetAddress;
+
+import com.datastax.driver.core.exceptions.NoHostAvailableException;
+
+/**
+ * @author John Sanda
+ */
+public interface StorageStateListener {
+
+ void onStorageNodeUp(InetAddress address);
+
+ void onStorageNodeDown(InetAddress address);
+
+ void onStorageNodeRemoved(InetAddress address);
+
+ void onStorageClusterDown(NoHostAvailableException e);
+
+}
diff --git
a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/domain/ListPagedResult.java
b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/domain/ListPagedResult.java
index 1c1fdd2..f3abd20 100644
---
a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/domain/ListPagedResult.java
+++
b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/domain/ListPagedResult.java
@@ -33,10 +33,10 @@ import java.util.List;
import com.datastax.driver.core.BoundStatement;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.ResultSet;
-import com.datastax.driver.core.Session;
import com.datastax.driver.core.exceptions.NoHostAvailableException;
import org.rhq.server.metrics.CQLException;
+import org.rhq.server.metrics.StorageSession;
/**
* This class helps paginate Cassandra results that have a list in the matching clause.
Instead of running
@@ -52,12 +52,12 @@ public class ListPagedResult<T> implements Iterable<T> {
private final long startTime;
private final long endTime;
private final ResultSetMapper<T> mapper;
- private final Session session;
+ private final StorageSession session;
private final PreparedStatement preparedStatement;
public ListPagedResult(PreparedStatement preparedStatement, List<Integer>
scheduleIds, long startTime, long endTime,
- ResultSetMapper<T> mapper, Session session) {
+ ResultSetMapper<T> mapper, StorageSession session) {
this.preparedStatement = preparedStatement;
this.scheduleIds = new LinkedList<Integer>(scheduleIds);
this.startTime = startTime;
diff --git
a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/domain/SimplePagedResult.java
b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/domain/SimplePagedResult.java
index d14195c..e31db95 100644
---
a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/domain/SimplePagedResult.java
+++
b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/domain/SimplePagedResult.java
@@ -34,6 +34,7 @@ import com.datastax.driver.core.SimpleStatement;
import com.datastax.driver.core.exceptions.NoHostAvailableException;
import org.rhq.server.metrics.CQLException;
+import org.rhq.server.metrics.StorageSession;
/**
* This class is just a placeholder for future pagination implementations once Cassandra
gets native support for paging results.
@@ -51,7 +52,7 @@ public class SimplePagedResult<T> implements Iterable<T> {
private final ResultSetMapper<T> mapper;
private final Query query;
- private final Session session;
+ private final StorageSession session;
private final int pageSize;
/**
@@ -60,7 +61,7 @@ public class SimplePagedResult<T> implements Iterable<T> {
* @param session Cassandra session
* @param pageSize page size
*/
- public SimplePagedResult(Query query, ResultSetMapper<T> mapper, Session
session, int pageSize) {
+ public SimplePagedResult(Query query, ResultSetMapper<T> mapper, StorageSession
session, int pageSize) {
this.query = query;
this.mapper = mapper;
this.session = session;
@@ -73,7 +74,7 @@ public class SimplePagedResult<T> implements Iterable<T> {
* @param session Cassandra session
* @param pageSize page size
*/
- public SimplePagedResult(String query, ResultSetMapper<T> mapper, Session
session, int pageSize) {
+ public SimplePagedResult(String query, ResultSetMapper<T> mapper,
StorageSession session, int pageSize) {
this(new SimpleStatement(query), mapper, session, pageSize);
}
@@ -82,7 +83,7 @@ public class SimplePagedResult<T> implements Iterable<T> {
* @param mapper result set mapper
* @param session Cassandra session
*/
- public SimplePagedResult(Query query, ResultSetMapper<T> mapper, Session
session) {
+ public SimplePagedResult(Query query, ResultSetMapper<T> mapper, StorageSession
session) {
this(query, mapper, session, DEFAULT_PAGE_SIZE);
}
@@ -92,7 +93,7 @@ public class SimplePagedResult<T> implements Iterable<T> {
* @param session Cassandra session
* @param pageSize page size
*/
- public SimplePagedResult(String query, ResultSetMapper<T> mapper, Session
session) {
+ public SimplePagedResult(String query, ResultSetMapper<T> mapper,
StorageSession session) {
this(new SimpleStatement(query), mapper, session);
}
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 c4c48c8..b227b2d 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
@@ -33,6 +33,7 @@ import java.util.concurrent.CountDownLatch;
import com.datastax.driver.core.BoundStatement;
import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.Host;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Session;
@@ -65,8 +66,12 @@ public class CassandraIntegrationTest {
protected static Session session;
+ protected static StorageSession storageSession;
+
private static DateTimeService dateTimeService;
+ private final Log log = LogFactory.getLog(CassandraIntegrationTest.class);
+
@BeforeSuite
@DeployCluster(numNodes = 2, username = "rhqadmin", password =
"rhqadmin", waitForSchemaAgreement = true)
public void deployCluster() throws Exception {
@@ -76,7 +81,31 @@ public class CassandraIntegrationTest {
.addContactPoints("127.0.0.1", "127.0.02")
.withCredentials("rhqadmin", "rhqadmin")
.build();
+
+ cluster.register(new Host.StateListener() {
+ @Override
+ public void onAdd(Host host) {
+ log.info("host " + host + " added");
+ }
+
+ @Override
+ public void onUp(Host host) {
+ log.info("host " + host + " up");
+ }
+
+ @Override
+ public void onDown(Host host) {
+ log.info("host " + host + " down");
+ }
+
+ @Override
+ public void onRemove(Host host) {
+ log.info("host " + host + " removed");
+ }
+ });
+
session = cluster.connect("rhq");
+ storageSession = new StorageSession(session);
}
@AfterSuite(alwaysRun = true)
@@ -98,7 +127,7 @@ public class CassandraIntegrationTest {
BoundStatement boundStatement = statement.bind(scheduleId);
return new SimplePagedResult<AggregateNumericMetric>(boundStatement, new
AggregateNumericMetricMapper(),
- session);
+ storageSession);
}
protected Iterable<AggregateNumericMetric>
findAggregateMetricsWithMetadata(MetricsTable table, int scheduleId,
@@ -109,11 +138,11 @@ public class CassandraIntegrationTest {
"SELECT schedule_id, time, type, value, ttl(value), writetime(value)
" +
"FROM " + table + " " +
"WHERE schedule_id = ? AND time >= ? AND time < ?";
- PreparedStatement statement = session.prepare(cql);
+ PreparedStatement statement = storageSession.prepare(cql);
BoundStatement boundStatement = statement.bind(scheduleId, new
Date(startTime), new Date(endTime));
return new SimplePagedResult<AggregateNumericMetric>(boundStatement,
new AggregateNumericMetricMapper(true),
- session);
+ storageSession);
} catch (NoHostAvailableException e) {
throw new CQLException(e);
}
diff --git
a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/ClusterMonitorTest.java
b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/ClusterMonitorTest.java
new file mode 100644
index 0000000..871b7c5
--- /dev/null
+++
b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/ClusterMonitorTest.java
@@ -0,0 +1,92 @@
+package org.rhq.server.metrics;
+
+import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.ConsistencyLevel;
+import com.datastax.driver.core.Host;
+import com.datastax.driver.core.ResultSet;
+import com.datastax.driver.core.Session;
+import com.datastax.driver.core.querybuilder.QueryBuilder;
+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.rhq.cassandra.util.ClusterBuilder;
+import org.rhq.core.domain.measurement.MeasurementDataNumeric;
+
+/**
+ * @author John Sanda
+ */
+public class ClusterMonitorTest {
+
+ private final Log log = LogFactory.getLog(ClusterMonitorTest.class);
+
+// @Test
+ public void monitorCluster() {
+ Cluster cluster = new ClusterBuilder()
+ .addContactPoints("127.0.0.1")
+// .withCredentials("cassandra", "cassandra")
+ .withCredentials("rhqadmin", "rhqadmin")
+ .build();
+
+ cluster.register(new Host.StateListener() {
+ @Override
+ public void onAdd(Host host) {
+ log.info("host " + host + " ADDED");
+ }
+
+ @Override
+ public void onUp(Host host) {
+ log.info("host " + host + " UP");
+ }
+
+ @Override
+ public void onDown(Host host) {
+ log.info("host " + host + " DOWN");
+ }
+
+ @Override
+ public void onRemove(Host host) {
+ log.info("host " + host + " REMOVED");
+ }
+ });
+
+ Session session = cluster.connect("rhq");
+ StorageSession storageSession = new StorageSession(session);
+
+ MetricsDAO dao = new MetricsDAO(storageSession, new MetricsConfiguration());
+
+ while (true) {
+ try {
+ Thread.sleep(10000);
+ try {
+// session.execute("select * from
system.schema_keyspaces");
+ com.datastax.driver.core.Query query =
QueryBuilder.select().from("rhq", "raw_metrics").setConsistencyLevel(
+ ConsistencyLevel.ALL);
+// session.execute(query);
+// session.execute("select * from rhq.raw_metrics");
+// log.info("query succeeded");
+ StorageResultSetFuture future = dao.insertRawData
+ (new MeasurementDataNumeric(System.currentTimeMillis(), 123,
1.1));
+
+ Futures.addCallback(future, new FutureCallback<ResultSet>() {
+ @Override
+ public void onSuccess(ResultSet rows) {
+ log.info("insert succeeded");
+ }
+
+ @Override
+ public void onFailure(Throwable throwable) {
+ log.error("insert failed", throwable);
+ }
+ });
+ } catch (Exception e) {
+ log.error("query failed", e);
+ }
+ } catch (InterruptedException e) {
+ }
+ }
+ }
+
+}
diff --git
a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsBaselineCalculatorTest.java
b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsBaselineCalculatorTest.java
index ab4bd03..2b58222 100644
---
a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsBaselineCalculatorTest.java
+++
b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsBaselineCalculatorTest.java
@@ -35,8 +35,6 @@ import java.util.Arrays;
import java.util.List;
import java.util.Random;
-import com.datastax.driver.core.Session;
-
import org.powermock.api.mockito.PowerMockito;
import org.powermock.core.classloader.annotations.PrepareForTest;
import org.powermock.modules.testng.PowerMockObjectFactory;
@@ -71,9 +69,9 @@ public class MetricsBaselineCalculatorTest {
//tell the method story as it happens: mock dependencies and configure
//those dependencies to get the method under test to completion.
- Session mockSession = mock(Session.class);
+ StorageSession mockSession = mock(StorageSession.class);
MetricsDAO mockMetricsDAO = mock(MetricsDAO.class);
- PowerMockito.whenNew(MetricsDAO.class).withParameterTypes(Session.class,
MetricsConfiguration.class)
+ PowerMockito.whenNew(MetricsDAO.class).withParameterTypes(StorageSession.class,
MetricsConfiguration.class)
.withArguments(eq(mockSession),
eq(metricsConfiguration)).thenReturn(mockMetricsDAO);
when(mockMetricsDAO.findAggregatedSimpleOneHourMetric(eq(1), eq(0),
@@ -147,9 +145,9 @@ public class MetricsBaselineCalculatorTest {
//tell the method story as it happens: mock dependencies and configure
//those dependencies to get the method under test to completion.
- Session mockSession = mock(Session.class);
+ StorageSession mockSession = mock(StorageSession.class);
MetricsDAO mockMetricsDAO = mock(MetricsDAO.class);
- PowerMockito.whenNew(MetricsDAO.class).withParameterTypes(Session.class,
MetricsConfiguration.class)
+ PowerMockito.whenNew(MetricsDAO.class).withParameterTypes(StorageSession.class,
MetricsConfiguration.class)
.withArguments(eq(mockSession),
eq(metricsConfiguration)).thenReturn(mockMetricsDAO);
when(
@@ -214,9 +212,9 @@ public class MetricsBaselineCalculatorTest {
//tell the method story as it happens: mock dependencies and configure
//those dependencies to get the method under test to completion.
- Session mockSession = mock(Session.class);
+ StorageSession mockSession = mock(StorageSession.class);
MetricsDAO mockMetricsDAO = mock(MetricsDAO.class);
- PowerMockito.whenNew(MetricsDAO.class).withParameterTypes(Session.class,
MetricsConfiguration.class)
+ PowerMockito.whenNew(MetricsDAO.class).withParameterTypes(StorageSession.class,
MetricsConfiguration.class)
.withArguments(eq(mockSession),
eq(metricsConfiguration)).thenReturn(mockMetricsDAO);
when(
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 ee65292..b06dfda 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
@@ -38,7 +38,6 @@ import java.util.Map;
import java.util.Random;
import java.util.Set;
-import com.datastax.driver.core.ResultSetFuture;
import com.google.common.collect.Lists;
import com.google.common.util.concurrent.Futures;
@@ -78,7 +77,7 @@ public class MetricsDAOTest extends CassandraIntegrationTest {
@BeforeClass
public void initDAO() throws Exception {
- dao = new MetricsDAO(session, new MetricsConfiguration());
+ dao = new MetricsDAO(storageSession, new MetricsConfiguration());
}
@BeforeMethod
@@ -101,7 +100,7 @@ public class MetricsDAOTest extends CassandraIntegrationTest {
WaitForWrite waitForResults = new WaitForWrite(1);
- ResultSetFuture resultSetFuture = dao.insertRawData(expected);
+ StorageResultSetFuture resultSetFuture = dao.insertRawData(expected);
Futures.addCallback(resultSetFuture, waitForResults);
waitForResults.await("Failed to insert raw data");
@@ -131,7 +130,7 @@ public class MetricsDAOTest extends CassandraIntegrationTest {
WaitForWrite waitForWrite = new WaitForWrite(data.size());
for (MeasurementDataNumeric raw : data) {
- ResultSetFuture resultSetFuture = dao.insertRawData(raw);
+ StorageResultSetFuture resultSetFuture = dao.insertRawData(raw);
Futures.addCallback(resultSetFuture, waitForWrite);
}
waitForWrite.await("Failed to insert raw data");
@@ -160,14 +159,14 @@ public class MetricsDAOTest extends CassandraIntegrationTest {
WaitForWrite waitForWrite = new WaitForWrite(data.size());
for (MeasurementDataNumeric raw : data) {
- ResultSetFuture resultSetFuture = dao.insertRawData(raw);
+ StorageResultSetFuture resultSetFuture = dao.insertRawData(raw);
Futures.addCallback(resultSetFuture, waitForWrite);
}
waitForWrite.await("Failed to insert raw data");
RawNumericMetricMapper mapper = new RawNumericMetricMapper();
WaitForRead<RawNumericMetric> waitForRead = new
WaitForRead<RawNumericMetric>(mapper);
- ResultSetFuture resultSetFuture = dao.findRawMetricsAsync(scheduleId,
+ StorageResultSetFuture resultSetFuture = dao.findRawMetricsAsync(scheduleId,
threeMinutesAgo.minusSeconds(5).getMillis(),
oneMinuteAgo.plusSeconds(5).getMillis());
Futures.addCallback(resultSetFuture, waitForRead);
@@ -200,7 +199,7 @@ public class MetricsDAOTest extends CassandraIntegrationTest {
WaitForWrite waitForWrite = new WaitForWrite(data.size());
for (MeasurementDataNumeric raw : data) {
- ResultSetFuture resultSetFuture = dao.insertRawData(raw);
+ StorageResultSetFuture resultSetFuture = dao.insertRawData(raw);
Futures.addCallback(resultSetFuture, waitForWrite);
}
waitForWrite.await("Failed to insert raw data");
@@ -321,9 +320,9 @@ public class MetricsDAOTest extends CassandraIntegrationTest {
WaitForWrite waitForWrite = new WaitForWrite(2);
- ResultSetFuture resultSetFuture1 =
dao.updateMetricsIndex(MetricsTable.TWENTY_FOUR_HOUR, scheduleId1,
+ StorageResultSetFuture resultSetFuture1 =
dao.updateMetricsIndex(MetricsTable.TWENTY_FOUR_HOUR, scheduleId1,
hour0().getMillis());
- ResultSetFuture resultSetFuture2 =
dao.updateMetricsIndex(MetricsTable.TWENTY_FOUR_HOUR, scheduleId2,
+ StorageResultSetFuture resultSetFuture2 =
dao.updateMetricsIndex(MetricsTable.TWENTY_FOUR_HOUR, scheduleId2,
hour0().getMillis());
Futures.addCallback(resultSetFuture1, waitForWrite);
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 1ec4771..b4de4e0 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
@@ -43,7 +43,6 @@ 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;
@@ -113,7 +112,7 @@ public class MetricsServerTest extends CassandraIntegrationTest {
dateTimeService.setConfiguration(configuration);
metricsServer.setDateTimeService(dateTimeService);
- dao = new MetricsDAO(session, configuration);
+ dao = new MetricsDAO(storageSession, configuration);
metricsServer.setDAO(dao);
purgeDB();
@@ -230,7 +229,7 @@ public class MetricsServerTest extends CassandraIntegrationTest {
WaitForWrite waitForRawInserts = new WaitForWrite(rawMetrics.size());
for (MeasurementDataNumeric raw : rawMetrics) {
- ResultSetFuture resultSetFuture = dao.insertRawData(raw);
+ StorageResultSetFuture resultSetFuture = dao.insertRawData(raw);
Futures.addCallback(resultSetFuture, waitForRawInserts);
}
waitForRawInserts.await("Failed to insert raw data");
@@ -1008,7 +1007,7 @@ public class MetricsServerTest extends CassandraIntegrationTest {
"SELECT schedule_id, time, value, ttl(value), writetime(value) " +
"FROM " + MetricsTable.RAW + " " +
"WHERE schedule_id = " + scheduleId + " AND time >= "
+ startTime + " AND time < " + endTime;
- return new SimplePagedResult<RawNumericMetric>(cql, new
RawNumericMetricMapper(true), session);
+ return new SimplePagedResult<RawNumericMetric>(cql, new
RawNumericMetricMapper(true), storageSession);
}
private static class WaitForRawInserts implements RawDataInsertedCallback {
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 e2b23ec..dabd110 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
@@ -35,7 +35,6 @@ 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;
@@ -43,7 +42,7 @@ public class TimeoutTest extends CassandraIntegrationTest {
private final Log log = LogFactory.getLog(TimeoutTest.class);
- @Test
+// @Test
public void generateTimeout() throws Exception {
MetricsConfiguration configuration = new MetricsConfiguration();
@@ -54,7 +53,7 @@ public class TimeoutTest extends CassandraIntegrationTest {
dateTimeService.setConfiguration(configuration);
metricsServer.setDateTimeService(dateTimeService);
- MetricsDAO dao = new MetricsDAO(session, configuration);
+ MetricsDAO dao = new MetricsDAO(new StorageSession(session), configuration);
metricsServer.setDAO(dao);
long time = hour0().getMillis();
diff --git a/modules/enterprise/server/server-metrics/src/test/resources/log4j.properties
b/modules/enterprise/server/server-metrics/src/test/resources/log4j.properties
index 67db049..59bd5db 100644
--- a/modules/enterprise/server/server-metrics/src/test/resources/log4j.properties
+++ b/modules/enterprise/server/server-metrics/src/test/resources/log4j.properties
@@ -39,4 +39,4 @@ log4j.appender.CONSOLE.layout=org.apache.log4j.PatternLayout
log4j.appender.CONSOLE.layout.ConversionPattern=%d{ABSOLUTE} %-5p %c %m%n
log4j.logger.org.rhq=DEBUG
-log4j.logger.com.datastax=DEBUG
+log4j.logger.com.datastax=WARN
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 7ff8f6c..40e00bf 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
@@ -60,6 +60,7 @@ 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;
+import org.rhq.server.metrics.StorageSession;
/**
* @author John Sanda
@@ -96,7 +97,9 @@ public class Simulator implements ShutdownManager {
session = createSession(nodes, compression);
}
- MetricsDAO metricsDAO = new MetricsDAO(session,
plan.getMetricsServerConfiguration());
+ StorageSession storageSession = new StorageSession(session);
+
+ MetricsDAO metricsDAO = new MetricsDAO(storageSession,
plan.getMetricsServerConfiguration());
MetricsServer metricsServer = new MetricsServer();
metricsServer.setDAO(metricsDAO);
metricsServer.setConfiguration(plan.getMetricsServerConfiguration());
commit 6c62c7c89a1707660e86fa97320420c2ef817339
Author: John Sanda <jsanda(a)redhat.com>
Date: Thu Jul 25 22:06:31 2013 -0400
upgrade to version 1.0.2 of datastax driver
diff --git a/pom.xml b/pom.xml
index f909033..3f2adbd 100644
--- a/pom.xml
+++ b/pom.xml
@@ -178,7 +178,7 @@
<!-- cassandra dependency versions -->
<cassandra.version>1.2.4</cassandra.version>
<cassandra.thrift.version>0.7.0</cassandra.thrift.version>
- <cassandra.driver.version>1.0.0-rhq-1.2.4</cassandra.driver.version>
+ <cassandra.driver.version>1.0.2-rhq-1.2.4</cassandra.driver.version>
<cassandra.driver.netty.version>3.6.3.Final</cassandra.driver.netty.version>
<cassandra.snappy.version>1.0.4.1-rhq-p1</cassandra.snappy.version>
<cassandra.snakeyaml.version>1.6</cassandra.snakeyaml.version>