modules/common/cassandra-ccm/cassandra-ccm-arquillian/src/main/java/org/rhq/cassandra/ccm/arquillian/CCMSuiteDeploymentExtension.java
| 121 +++++-----
modules/common/cassandra-ccm/cassandra-ccm-testng/src/main/java/org/rhq/cassandra/CCMTestNGListener.java
| 12
modules/common/cassandra-schema/src/main/java/org/rhq/cassandra/schema/AbstractManager.java
| 59 +---
modules/common/cassandra-schema/src/main/java/org/rhq/cassandra/schema/SchemaManager.java
| 25 +-
modules/common/cassandra-schema/src/main/java/org/rhq/cassandra/schema/SessionManager.java
| 60 ++++
modules/common/cassandra-schema/src/main/java/org/rhq/cassandra/schema/TopologyManager.java
| 25 --
modules/common/cassandra-schema/src/main/java/org/rhq/cassandra/schema/VersionManager.java
| 113 ++++-----
modules/enterprise/server/installer/src/main/java/org/rhq/enterprise/server/installer/InstallerServiceImpl.java
| 41 ++-
modules/enterprise/server/installer/src/main/java/org/rhq/enterprise/server/installer/ServerInstallUtil.java
| 3
modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/storage/StorageClientManagerBean.java
| 2
modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/CassandraIntegrationTest.java
| 2
modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/Simulator.java
| 13 -
modules/plugins/rhq-storage/src/test/java/org/rhq/plugins/storage/StorageNodeComponentITest.java
| 36 --
13 files changed, 279 insertions(+), 233 deletions(-)
New commits:
commit ea444e2b561ea03c4e513eb5a06ad6e8685bda40
Author: Stefan Negrea <snegrea(a)redhat.com>
Date: Tue Oct 22 16:52:22 2013 -0500
[BZ 1004050] Minor updates for the way create() works since the session is now
externally managed. Also simplified the abstract class code.
Conflicts:
modules/common/cassandra-schema/src/main/java/org/rhq/cassandra/schema/AbstractManager.java
diff --git
a/modules/common/cassandra-schema/src/main/java/org/rhq/cassandra/schema/AbstractManager.java
b/modules/common/cassandra-schema/src/main/java/org/rhq/cassandra/schema/AbstractManager.java
index 07a1de4..dc78601 100644
---
a/modules/common/cassandra-schema/src/main/java/org/rhq/cassandra/schema/AbstractManager.java
+++
b/modules/common/cassandra-schema/src/main/java/org/rhq/cassandra/schema/AbstractManager.java
@@ -29,12 +29,9 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
-import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Row;
-import com.datastax.driver.core.Session;
import com.datastax.driver.core.exceptions.AuthenticationException;
-import com.datastax.driver.core.exceptions.NoHostAvailableException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -61,8 +58,6 @@ abstract class AbstractManager {
}
}
- private Cluster cluster;
- private Session session;
private SessionManager sessionManager;
private final String username;
private final String password;
@@ -72,15 +67,12 @@ abstract class AbstractManager {
protected AbstractManager(String username, String password, String[] nodes, int
cqlPort,
SessionManager sessionManager) {
- try {
- this.username = username;
- this.password = password;
- this.cqlPort = cqlPort;
- this.nodes = nodes;
- this.sessionManager = sessionManager;
- } catch (NoHostAvailableException e) {
- throw new RuntimeException("Unable create storage node session.",
e);
- }
+
+ this.username = username;
+ this.password = password;
+ this.cqlPort = cqlPort;
+ this.nodes = nodes;
+ this.sessionManager = sessionManager;
try {
UpdateFolder managementFolder = new UpdateFolder(MANAGEMENT_BASE_FOLDER);
diff --git
a/modules/common/cassandra-schema/src/main/java/org/rhq/cassandra/schema/VersionManager.java
b/modules/common/cassandra-schema/src/main/java/org/rhq/cassandra/schema/VersionManager.java
index 2f89df5..ac406d5 100644
---
a/modules/common/cassandra-schema/src/main/java/org/rhq/cassandra/schema/VersionManager.java
+++
b/modules/common/cassandra-schema/src/main/java/org/rhq/cassandra/schema/VersionManager.java
@@ -78,19 +78,11 @@ class VersionManager extends AbstractManager {
public void install() throws Exception {
log.info("Preparing to install storage schema");
- boolean clusterSessionInitialized = false;
try {
initClusterSession();
- clusterSessionInitialized = true;
} catch (AuthenticationException e) {
log.debug("Authentication exception. Will now attempt to create the
storage schema.");
log.debug(e);
- shutdownClusterConnection();
- } finally {
- //shutdownClusterConnection();
- }
-
- if (!clusterSessionInitialized) {
create();
}
@@ -114,6 +106,10 @@ class VersionManager extends AbstractManager {
*/
//1. Execute the creation of RHQ schema, version table, admin user.
try {
+ //shutdown existing connection
+ shutdownClusterConnection();
+
+ //re-initialize the cluster connection with default cassandra password
initClusterSession(DEFAULT_CASSANDRA_USER, DEFAULT_CASSANDRA_PASSWORD);
updateFolder = new UpdateFolder(Task.Create.getFolder());
commit d31a3167baf295177ec9e10e59bde8b28d7df1b4
Author: John Sanda <jsanda(a)redhat.com>
Date: Mon Oct 21 21:53:00 2013 -0400
[BZ 1004050] This commit includes the bulk of the changes for this bug
Conflicts:
modules/enterprise/server/installer/src/main/java/org/rhq/enterprise/server/installer/InstallerServiceImpl.java
modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/Simulator.java
diff --git
a/modules/common/cassandra-ccm/cassandra-ccm-arquillian/src/main/java/org/rhq/cassandra/ccm/arquillian/CCMSuiteDeploymentExtension.java
b/modules/common/cassandra-ccm/cassandra-ccm-arquillian/src/main/java/org/rhq/cassandra/ccm/arquillian/CCMSuiteDeploymentExtension.java
index 50aaab1..9164a81 100644
---
a/modules/common/cassandra-ccm/cassandra-ccm-arquillian/src/main/java/org/rhq/cassandra/ccm/arquillian/CCMSuiteDeploymentExtension.java
+++
b/modules/common/cassandra-ccm/cassandra-ccm-arquillian/src/main/java/org/rhq/cassandra/ccm/arquillian/CCMSuiteDeploymentExtension.java
@@ -103,77 +103,80 @@ public class CCMSuiteDeploymentExtension implements
LoadableExtension {
executeInClassScope(new Callable<Void>() {
public Void call() throws Exception {
+ SchemaManager schemaManager = null;
+ try {
+ ClusterInitService clusterInitService = new
ClusterInitService();
+
+ String[] nodes = null;
+ int[] jmxPorts = null;
+ int cqlPort = -1;
+
+ if
(!Boolean.valueOf(System.getProperty("itest.use-external-storage-node",
"false"))) {
+
+ DeploymentOptionsFactory factory = new
DeploymentOptionsFactory();
+ DeploymentOptions options = factory.newDeploymentOptions();
+ File basedir = new File("target");
+ File clusterDir = new File(basedir, "cassandra");
+
+ options.setUsername("rhqadmin");
+
options.setPassword("1eeb2f255e832171df8592078de921bc");
+ options.setClusterDir(clusterDir.getAbsolutePath());
+ options.setHeapSize("256M");
+ options.setHeapNewSize("64M");
+ options.setStartRpc(true);
+
+ ccm = new CassandraClusterManager(options);
+ ccm.createCluster();
+
+ nodes = ccm.getNodes();
+ jmxPorts = ccm.getJmxPorts();
+ cqlPort = ccm.getCqlPort();
+
+ ccm.startCluster(false);
+
+ try {
+ clusterInitService.waitForClusterToStart(nodes, jmxPorts,
nodes.length, 2000, 20, 10);
+ schemaManager = new SchemaManager("rhqadmin",
"1eeb2f255e832171df8592078de921bc", nodes,
+ cqlPort);
+ } catch (Exception e) {
+ if (null != ccm) {
+ ccm.shutdownCluster();
+ }
+ throw new RuntimeException("Cassandra cluster
initialization failed", e);
+ }
+ } else {
+ try {
+ String nodesString =
System.getProperty("rhq.storage.nodes", "127.0.0.1");
+ nodes = nodesString.split(",");
- SchemaManager schemaManager;
- ClusterInitService clusterInitService = new ClusterInitService();
-
- String[] nodes = null;
- int[] jmxPorts = null;
- int cqlPort = -1;
-
- if
(!Boolean.valueOf(System.getProperty("itest.use-external-storage-node",
"false"))) {
-
- DeploymentOptionsFactory factory = new
DeploymentOptionsFactory();
- DeploymentOptions options = factory.newDeploymentOptions();
- File basedir = new File("target");
- File clusterDir = new File(basedir, "cassandra");
-
- options.setUsername("rhqadmin");
-
options.setPassword("1eeb2f255e832171df8592078de921bc");
- options.setClusterDir(clusterDir.getAbsolutePath());
- options.setHeapSize("256M");
- options.setHeapNewSize("64M");
- options.setStartRpc(true);
-
- ccm = new CassandraClusterManager(options);
- ccm.createCluster();
+ String cqlPortString =
System.getProperty("rhq.storage.cql-port", "9042");
+ cqlPort = Integer.parseInt(cqlPortString);
- nodes = ccm.getNodes();
- jmxPorts = ccm.getJmxPorts();
- cqlPort = ccm.getCqlPort();
+ String jmxPortString =
System.getProperty("rhq.storage.jmx-port", "7299");
+ jmxPorts = new int[] { Integer.parseInt(jmxPortString)
};
- ccm.startCluster(false);
+ schemaManager = new SchemaManager("rhqadmin",
"1eeb2f255e832171df8592078de921bc", nodes, cqlPort);
+ } catch (Exception e) {
+ throw new RuntimeException("External Cassandra
initialization failed", e);
+ }
+ }
try {
- clusterInitService.waitForClusterToStart(nodes, jmxPorts,
nodes.length, 2000, 20, 10);
- schemaManager = new SchemaManager("rhqadmin",
"1eeb2f255e832171df8592078de921bc", nodes, cqlPort);
-
+ schemaManager.install();
+ clusterInitService.waitForSchemaAgreement(nodes, jmxPorts);
+ schemaManager.updateTopology();
} catch (Exception e) {
if (null != ccm) {
ccm.shutdownCluster();
}
- throw new RuntimeException("Cassandra cluster
initialization failed", e);
- }
- } else {
- try {
- String nodesString =
System.getProperty("rhq.storage.nodes", "127.0.0.1");
- nodes = nodesString.split(",");
-
- String cqlPortString =
System.getProperty("rhq.storage.cql-port", "9042");
- cqlPort = Integer.parseInt(cqlPortString);
-
- String jmxPortString =
System.getProperty("rhq.storage.jmx-port", "7299");
- jmxPorts = new int[] { Integer.parseInt(jmxPortString) };
-
- schemaManager = new SchemaManager("rhqadmin",
"1eeb2f255e832171df8592078de921bc", nodes, cqlPort);
-
- } catch (Exception e) {
- throw new RuntimeException("External Cassandra
initialization failed", e);
+ throw new RuntimeException("Cassandra schema
initialization failed", e);
}
- }
-
- try {
- schemaManager.install();
- clusterInitService.waitForSchemaAgreement(nodes, jmxPorts);
- schemaManager.updateTopology();
- } catch (Exception e) {
- if (null != ccm) {
- ccm.shutdownCluster();
+ return null;
+ } finally {
+ if (schemaManager != null) {
+ schemaManager.shutdown();
}
- throw new RuntimeException("Cassandra schema initialization
failed", e);
}
-
- return null;
}
});
}
diff --git
a/modules/common/cassandra-ccm/cassandra-ccm-testng/src/main/java/org/rhq/cassandra/CCMTestNGListener.java
b/modules/common/cassandra-ccm/cassandra-ccm-testng/src/main/java/org/rhq/cassandra/CCMTestNGListener.java
index 286a5ff..ae9a980 100644
---
a/modules/common/cassandra-ccm/cassandra-ccm-testng/src/main/java/org/rhq/cassandra/CCMTestNGListener.java
+++
b/modules/common/cassandra-ccm/cassandra-ccm-testng/src/main/java/org/rhq/cassandra/CCMTestNGListener.java
@@ -119,11 +119,15 @@ public class CCMTestNGListener implements IInvokedMethodListener {
SchemaManager schemaManager = new SchemaManager(annotation.username(),
annotation.password(), nodes,
ccm.getCqlPort());
- schemaManager.install();
- if (annotation.waitForSchemaAgreement()) {
- clusterInitService.waitForSchemaAgreement(nodes, jmxPorts);
+ try {
+ schemaManager.install();
+ if (annotation.waitForSchemaAgreement()) {
+ clusterInitService.waitForSchemaAgreement(nodes, jmxPorts);
+ }
+ schemaManager.updateTopology();
+ } finally {
+ schemaManager.shutdown();
}
- schemaManager.updateTopology();
}
private void shutdownCluster() throws Exception {
diff --git
a/modules/common/cassandra-schema/src/main/java/org/rhq/cassandra/schema/AbstractManager.java
b/modules/common/cassandra-schema/src/main/java/org/rhq/cassandra/schema/AbstractManager.java
index 1c3d51b..07a1de4 100644
---
a/modules/common/cassandra-schema/src/main/java/org/rhq/cassandra/schema/AbstractManager.java
+++
b/modules/common/cassandra-schema/src/main/java/org/rhq/cassandra/schema/AbstractManager.java
@@ -30,7 +30,6 @@ import java.util.List;
import java.util.Properties;
import com.datastax.driver.core.Cluster;
-import com.datastax.driver.core.ProtocolOptions.Compression;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Row;
import com.datastax.driver.core.Session;
@@ -40,7 +39,6 @@ import com.datastax.driver.core.exceptions.NoHostAvailableException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
-import org.rhq.cassandra.util.ClusterBuilder;
import org.rhq.core.util.obfuscation.PicketBoxObfuscator;
/**
@@ -65,18 +63,21 @@ abstract class AbstractManager {
private Cluster cluster;
private Session session;
+ private SessionManager sessionManager;
private final String username;
private final String password;
private final int cqlPort;
private final String[] nodes;
private final UpdateFile managementTasks;
- protected AbstractManager(String username, String password, String[] nodes, int
cqlPort) {
+ protected AbstractManager(String username, String password, String[] nodes, int
cqlPort,
+ SessionManager sessionManager) {
try {
this.username = username;
this.password = password;
this.cqlPort = cqlPort;
this.nodes = nodes;
+ this.sessionManager = sessionManager;
} catch (NoHostAvailableException e) {
throw new RuntimeException("Unable create storage node session.",
e);
}
@@ -104,17 +105,7 @@ abstract class AbstractManager {
* @param password
*/
protected void initClusterSession(String username, String password) {
- shutdownClusterConnection();
-
- log.info("Initializing storage node session.");
-
- cluster = new
ClusterBuilder().addContactPoints(nodes).withCredentialsObfuscated(username, password)
- .withPort(this.getCqlPort()).withCompression(Compression.NONE).build();
-
- log.info("Cluster connection configured.");
-
- session = cluster.connect("system");
- log.info("Cluster connected.");
+ sessionManager.initSession(username, password, cqlPort, nodes);
}
/**
@@ -122,18 +113,14 @@ abstract class AbstractManager {
*/
protected void shutdownClusterConnection() {
log.info("Shutting down existing cluster connections");
- if (cluster != null) {
- cluster.shutdown();
- }
+ sessionManager.shutdownCluster();
}
/**
- * Get storage cluster size.
- *
- * @return cluster size
+ * @return The actual size of the cluster which includes both specified and
discovered nodes
*/
- protected int getClusterSize() {
- return nodes.length;
+ protected int getActualClusterSize() {
+ return
sessionManager.getSession().getCluster().getMetadata().getAllHosts().size();
}
/**
@@ -222,9 +209,10 @@ abstract class AbstractManager {
*/
protected int calculateNewReplicationFactor() {
int replicationFactor;
- if (getClusterSize() < 3) {
- replicationFactor = getClusterSize();
- } else if (getClusterSize() < 4) {
+ int actualClusterSize = getActualClusterSize();
+ if (actualClusterSize < 3) {
+ replicationFactor = actualClusterSize;
+ } else if (actualClusterSize < 4) {
replicationFactor = 2;
} else {
replicationFactor = 3;
@@ -362,7 +350,8 @@ abstract class AbstractManager {
* @return result for the query
*/
protected ResultSet execute(String query) {
- return session.execute(query);
+// return session.execute(query);
+ return sessionManager.getSession().execute(query);
}
}
diff --git
a/modules/common/cassandra-schema/src/main/java/org/rhq/cassandra/schema/SchemaManager.java
b/modules/common/cassandra-schema/src/main/java/org/rhq/cassandra/schema/SchemaManager.java
index 1e67b1f..31b3137 100644
---
a/modules/common/cassandra-schema/src/main/java/org/rhq/cassandra/schema/SchemaManager.java
+++
b/modules/common/cassandra-schema/src/main/java/org/rhq/cassandra/schema/SchemaManager.java
@@ -27,6 +27,7 @@ package org.rhq.cassandra.schema;
import java.util.Arrays;
import java.util.List;
+import java.util.Set;
import org.apache.log4j.ConsoleAppender;
import org.apache.log4j.Level;
@@ -60,6 +61,8 @@ public class SchemaManager {
*/
private final int cqlPort;
+ private SessionManager sessionManager;
+
/**
*
* @param username The username RHQ will use to connect to the storage cluster
@@ -73,6 +76,7 @@ public class SchemaManager {
this.password = password;
this.cqlPort = cqlPort;
this.nodes = nodes;
+ sessionManager = new SessionManager();
}
/**
@@ -84,10 +88,7 @@ public class SchemaManager {
* @param cqlPort The native CQL port for the storage cluster
*/
public SchemaManager(String username, String password, List<String> nodes, int
cqlPort) {
- this.username = username;
- this.password = password;
- this.cqlPort = cqlPort;
- this.nodes = nodes.toArray(new String[nodes.size()]);
+ this(username, password, nodes.toArray(new String[nodes.size()]), cqlPort);
}
/**
@@ -96,7 +97,7 @@ public class SchemaManager {
* @throws Exception
*/
public void install() throws Exception {
- VersionManager version = new VersionManager(username, password, nodes, cqlPort);
+ VersionManager version = new VersionManager(username, password, nodes, cqlPort,
sessionManager);
version.install();
}
@@ -107,7 +108,7 @@ public class SchemaManager {
* @throws Exception
*/
public void checkCompatibility() throws Exception {
- VersionManager version = new VersionManager(username, password, nodes, cqlPort);
+ VersionManager version = new VersionManager(username, password, nodes, cqlPort,
sessionManager);
version.checkCompatibility();
}
@@ -117,7 +118,7 @@ public class SchemaManager {
* @throws Exception
*/
public void drop() throws Exception {
- VersionManager version = new VersionManager(username, password, nodes, cqlPort);
+ VersionManager version = new VersionManager(username, password, nodes, cqlPort,
sessionManager);
version.drop();
}
@@ -129,10 +130,14 @@ public class SchemaManager {
* @throws Exception
*/
public void updateTopology() throws Exception {
- TopologyManager topology = new TopologyManager(username, password, nodes,
cqlPort);
+ TopologyManager topology = new TopologyManager(username, password, nodes,
cqlPort, sessionManager);
topology.updateTopology();
}
+ public void shutdown() {
+ sessionManager.shutdownCluster();
+ }
+
/**
* Returns the list of storage nodes.
*
@@ -142,6 +147,10 @@ public class SchemaManager {
return nodes;
}
+ public Set<String> getStorageNodeAddresses() {
+ return sessionManager.getNodeAdresses();
+ }
+
/**
* A main runner used for direct usage of the schema manager.
*
diff --git
a/modules/common/cassandra-schema/src/main/java/org/rhq/cassandra/schema/TopologyManager.java
b/modules/common/cassandra-schema/src/main/java/org/rhq/cassandra/schema/TopologyManager.java
index 481c006..f56e3a0 100644
---
a/modules/common/cassandra-schema/src/main/java/org/rhq/cassandra/schema/TopologyManager.java
+++
b/modules/common/cassandra-schema/src/main/java/org/rhq/cassandra/schema/TopologyManager.java
@@ -52,8 +52,9 @@ class TopologyManager extends AbstractManager {
}
}
- public TopologyManager(String username, String password, String[] nodes, int cqlPort)
{
- super(username, password, nodes, cqlPort);
+ public TopologyManager(String username, String password, String[] nodes, int
cqlPort,
+ SessionManager sessionManager) {
+ super(username, password, nodes, cqlPort, sessionManager);
}
/**
@@ -64,17 +65,13 @@ class TopologyManager extends AbstractManager {
* @return true if update successful, false otherwise.
*/
public void updateTopology() {
- try {
- initClusterSession();
- if (schemaExists()) {
- log.info("Applying topology updates...");
- updateReplicationFactor();
- updateGCGrace();
- } else {
- log.info("Topology updates cannot be applied because the schema is
not installed.");
- }
- } finally {
- shutdownClusterConnection();
+ initClusterSession();
+ if (schemaExists()) {
+ log.info("Applying topology updates...");
+ updateReplicationFactor();
+ updateGCGrace();
+ } else {
+ log.info("Topology updates cannot be applied because the schema is not
installed.");
}
}
@@ -106,7 +103,7 @@ class TopologyManager extends AbstractManager {
log.info("Starting to execute " + Task.UpdateGCGrace + "
task.");
int gcGraceSeconds = 864000;
- if (getClusterSize() == 1) {
+ if (getActualClusterSize() == 1) {
gcGraceSeconds = 0;
} else {
gcGraceSeconds = 691200; // 8 days
diff --git
a/modules/common/cassandra-schema/src/main/java/org/rhq/cassandra/schema/VersionManager.java
b/modules/common/cassandra-schema/src/main/java/org/rhq/cassandra/schema/VersionManager.java
index 4c77b08..2f89df5 100644
---
a/modules/common/cassandra-schema/src/main/java/org/rhq/cassandra/schema/VersionManager.java
+++
b/modules/common/cassandra-schema/src/main/java/org/rhq/cassandra/schema/VersionManager.java
@@ -63,8 +63,9 @@ class VersionManager extends AbstractManager {
}
}
- public VersionManager(String username, String password, String[] nodes, int cqlPort)
throws Exception {
- super(username, password, nodes, cqlPort);
+ public VersionManager(String username, String password, String[] nodes, int cqlPort,
SessionManager sessionManager)
+ throws Exception {
+ super(username, password, nodes, cqlPort, sessionManager);
}
/**
@@ -84,8 +85,9 @@ class VersionManager extends AbstractManager {
} catch (AuthenticationException e) {
log.debug("Authentication exception. Will now attempt to create the
storage schema.");
log.debug(e);
- } finally {
shutdownClusterConnection();
+ } finally {
+ //shutdownClusterConnection();
}
if (!clusterSessionInitialized) {
@@ -101,13 +103,8 @@ class VersionManager extends AbstractManager {
* @throws Exception
*/
private void create() throws Exception {
- UpdateFolder updateFolder = new UpdateFolder(Task.Create.getFolder());
-
- Properties properties = new Properties(System.getProperties());
- properties.put("replication_factor", calculateNewReplicationFactor() +
"");
- properties.put("cassandra_user_password", UUID.randomUUID() +
"");
- properties.put("rhq_admin_username", getUsername());
- properties.put("rhq_admin_password", getPassword());
+ UpdateFolder updateFolder = null;
+ Properties properties = null;
/**
* NOTE: Before applying any schema, we need to create the rhqadmin user. If we
have more
@@ -118,6 +115,15 @@ class VersionManager extends AbstractManager {
//1. Execute the creation of RHQ schema, version table, admin user.
try {
initClusterSession(DEFAULT_CASSANDRA_USER, DEFAULT_CASSANDRA_PASSWORD);
+
+ updateFolder = new UpdateFolder(Task.Create.getFolder());
+
+ properties = new Properties(System.getProperties());
+ properties.put("replication_factor",
calculateNewReplicationFactor() + "");
+ properties.put("cassandra_user_password", UUID.randomUUID() +
"");
+ properties.put("rhq_admin_username", getUsername());
+ properties.put("rhq_admin_password", getPassword());
+
if (!schemaExists()) {
execute(updateFolder.getUpdateFiles().get(0), properties);
} else {
@@ -131,12 +137,8 @@ class VersionManager extends AbstractManager {
}
//2. Change Cassandra default user privileges and password.
- try {
- initClusterSession();
- execute(updateFolder.getUpdateFiles().get(1), properties);
- } finally {
- shutdownClusterConnection();
- }
+ initClusterSession();
+ execute(updateFolder.getUpdateFiles().get(1), properties);
}
/**
@@ -145,52 +147,49 @@ class VersionManager extends AbstractManager {
* @throws Exception
*/
private void update() throws Exception {
- try {
- initClusterSession();
+ initClusterSession();
- if (!schemaExists()) {
- log.error("Storage schema not installed.");
- throw new RuntimeException("Storage schema not installed propertly,
cannot apply schema updates.");
- }
+ if (!schemaExists()) {
+ log.error("Storage schema not installed.");
+ throw new RuntimeException("Storage schema not installed propertly,
cannot apply schema updates.");
+ }
- UpdateFolder updateFolder = new UpdateFolder(Task.Update.getFolder());
+ UpdateFolder updateFolder = new UpdateFolder(Task.Update.getFolder());
- int installedSchemaVersion = getInstalledSchemaVersion();
- log.info("Installed storage schema version is " +
installedSchemaVersion);
+ int installedSchemaVersion = getInstalledSchemaVersion();
+ log.info("Installed storage schema version is " +
installedSchemaVersion);
- int requiredSchemaVersion = updateFolder.getLatestVersion();
- log.info("Required storage schema version is " +
requiredSchemaVersion);
+ int requiredSchemaVersion = updateFolder.getLatestVersion();
+ log.info("Required storage schema version is " +
requiredSchemaVersion);
- if (requiredSchemaVersion == installedSchemaVersion) {
- log.info("Storage schema version is current ( " +
installedSchemaVersion + " ). No updates applied.");
- } else if (requiredSchemaVersion < installedSchemaVersion) {
- log.error("Installed storage cluster schema version: " +
installedSchemaVersion +
- ". Required schema version: " + requiredSchemaVersion
- + ". Storage cluster schema has been updated beyond the
capability of the existing server installation.");
- throw new InstalledSchemaTooAdvancedException();
- } else {
- log.info("Storage schema requires udpates. Updating from version
" + installedSchemaVersion
- + " to version " + requiredSchemaVersion + ".");
+ if (requiredSchemaVersion == installedSchemaVersion) {
+ log.info("Storage schema version is current ( " +
installedSchemaVersion + " ). No updates applied.");
+ } else if (requiredSchemaVersion < installedSchemaVersion) {
+ log.error("Installed storage cluster schema version: " +
installedSchemaVersion +
+ ". Required schema version: " + requiredSchemaVersion
+ +
+ ". Storage cluster schema has been updated beyond the capability of
the existing server installation.");
+ throw new InstalledSchemaTooAdvancedException();
+ } else {
+ log.info("Storage schema requires udpates. Updating from version "
+ installedSchemaVersion
+ + " to version " + requiredSchemaVersion + ".");
- updateFolder.removeAppliedUpdates(installedSchemaVersion);
+ updateFolder.removeAppliedUpdates(installedSchemaVersion);
- if (updateFolder.getUpdateFiles().size() == 0) {
- log.info("Storage schema is current! No updates
applied.");
- } else {
- for (UpdateFile updateFile : updateFolder.getUpdateFiles()) {
- execute(updateFile);
+ if (updateFolder.getUpdateFiles().size() == 0) {
+ log.info("Storage schema is current! No updates applied.");
+ } else {
+ for (UpdateFile updateFile : updateFolder.getUpdateFiles()) {
+ execute(updateFile);
- Properties versionProperties = new Properties();
- versionProperties.put("version",
updateFile.extractVersion() + "");
- versionProperties.put("time",
System.currentTimeMillis() + "");
- executeManagementQuery(Query.INSERT_SCHEMA_VERSION,
versionProperties);
+ Properties versionProperties = new Properties();
+ versionProperties.put("version",
updateFile.extractVersion() + "");
+ versionProperties.put("time", System.currentTimeMillis() +
"");
+ executeManagementQuery(Query.INSERT_SCHEMA_VERSION,
versionProperties);
- log.info("Storage schema update " + updateFile + "
applied.");
- }
+ log.info("Storage schema update " + updateFile + "
applied.");
}
}
- } finally {
- shutdownClusterConnection();
}
}
@@ -246,8 +245,6 @@ class VersionManager extends AbstractManager {
}
} catch (Exception e) {
throw new RuntimeException(e);
- } finally {
- shutdownClusterConnection();
}
}
@@ -292,8 +289,6 @@ class VersionManager extends AbstractManager {
} catch (Exception e3) {
throw new RuntimeException(e3);
} finally {
- shutdownClusterConnection();
-
log.info("Completed storage schema compatibility check.");
}
}
diff --git
a/modules/enterprise/server/installer/src/main/java/org/rhq/enterprise/server/installer/InstallerServiceImpl.java
b/modules/enterprise/server/installer/src/main/java/org/rhq/enterprise/server/installer/InstallerServiceImpl.java
index 6f426b2..8758246 100644
---
a/modules/enterprise/server/installer/src/main/java/org/rhq/enterprise/server/installer/InstallerServiceImpl.java
+++
b/modules/enterprise/server/installer/src/main/java/org/rhq/enterprise/server/installer/InstallerServiceImpl.java
@@ -22,6 +22,8 @@ import java.io.File;
import java.io.IOException;
import java.net.InetAddress;
import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -29,10 +31,12 @@ import java.util.Map;
import java.util.Properties;
import java.util.Random;
import java.util.Set;
+import java.util.TreeSet;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import com.datastax.driver.core.exceptions.AuthenticationException;
+import com.google.common.collect.ImmutableSet;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -536,6 +540,7 @@ public class InstallerServiceImpl implements InstallerService {
}
SchemaManager storageNodeSchemaManager = null;
+ Set<String> storageNodeAddresses = Collections.emptySet();
try {
storageNodeSchemaManager = createStorageNodeSchemaManager(serverProperties);
@@ -552,13 +557,18 @@ public class InstallerServiceImpl implements InstallerService {
log("Install RHQ schema along with updates to storage
nodes.");
storageNodeSchemaManager.install();
storageNodeSchemaManager.updateTopology();
+ storageNodeAddresses =
storageNodeSchemaManager.getStorageNodeAddresses();
} catch (SchemaNotInstalledException e2) {
log("Install RHQ schema along with updates to storage
nodes.");
storageNodeSchemaManager.install();
storageNodeSchemaManager.updateTopology();
+ storageNodeAddresses =
storageNodeSchemaManager.getStorageNodeAddresses();
} catch (InstalledSchemaTooOldException e3) {
log("Install RHQ schema updates to storage cluster.");
storageNodeSchemaManager.install();
+ storageNodeAddresses =
storageNodeSchemaManager.getStorageNodeAddresses();
+ } finally {
+ storageNodeSchemaManager.shutdown();
}
} else {
log("Ignoring storage cluster schema - installer will assume it
exists and is already up-to-date.");
@@ -572,7 +582,7 @@ public class InstallerServiceImpl implements InstallerService {
// ensure the server info is up to date and stored in the DB
ServerInstallUtil.storeServerDetails(serverProperties, clearTextDbPassword,
serverDetails);
ServerInstallUtil.persistStorageNodesIfNecessary(serverProperties,
clearTextDbPassword,
- parseNodeInformation(serverProperties));
+ parseNodeInformation(serverProperties, storageNodeAddresses));
ServerInstallUtil.persistStorageClusterSettingsIfNecessary(serverProperties,
clearTextDbPassword);
// For sanity, make sure the server props file is in sync with the db settings.
@@ -1197,26 +1207,35 @@ public class InstallerServiceImpl implements InstallerService {
}
}
- private List<StorageNode> parseNodeInformation(Map<String, String>
serverProps) {
- String[] nodes =
serverProps.get(ServerProperties.PROP_STORAGE_NODES).split(",");
- String cqlPort = serverProps.get(ServerProperties.PROP_STORAGE_CQL_PORT);
+ private Set<StorageNode> parseNodeInformation(Map<String, String>
serverProps, Set<String> storageNodeAddresses) {
+ int cqlPort =
Integer.parseInt(serverProps.get(ServerProperties.PROP_STORAGE_CQL_PORT));
- List<StorageNode> parsedNodes = new ArrayList<StorageNode>();
- for (String node : nodes) {
- StorageNode storageNode = new StorageNode();
- storageNode.setAddress(node);
- storageNode.setCqlPort(Integer.parseInt(cqlPort));
- parsedNodes.add(storageNode);
+ Set<StorageNode> parsedNodes = new TreeSet<StorageNode>(new
Comparator<StorageNode>() {
+ @Override
+ public int compare(StorageNode left, StorageNode right) {
+ return left.getAddress().compareTo(right.getAddress());
+ }
+ });
+ for (String address : storageNodeAddresses) {
+ StorageNode node = new StorageNode();
+ node.setAddress(address);
+ node.setCqlPort(cqlPort);
+ parsedNodes.add(node);
}
return parsedNodes;
}
+ private Set<StorageNode> parseNodeInformation(Map<String, String>
serverProps) {
+ return parseNodeInformation(serverProps, ImmutableSet.copyOf(serverProps.get(
+ ServerProperties.PROP_STORAGE_NODES).split(",")));
+ }
+
private SchemaManager createStorageNodeSchemaManager(Map<String, String>
serverProps) {
String username = serverProps.get(ServerProperties.PROP_STORAGE_USERNAME);
String password = serverProps.get(ServerProperties.PROP_STORAGE_PASSWORD);
- List<StorageNode> storageNodes = this.parseNodeInformation(serverProps);
+ List<StorageNode> storageNodes = new
ArrayList<StorageNode>(parseNodeInformation(serverProps));
String[] nodes = new String[storageNodes.size()];
for (int index = 0; index < storageNodes.size(); index++) {
nodes[index] = storageNodes.get(index).getAddress();
diff --git
a/modules/enterprise/server/installer/src/main/java/org/rhq/enterprise/server/installer/ServerInstallUtil.java
b/modules/enterprise/server/installer/src/main/java/org/rhq/enterprise/server/installer/ServerInstallUtil.java
index b511369..f2e1ec6 100644
---
a/modules/enterprise/server/installer/src/main/java/org/rhq/enterprise/server/installer/ServerInstallUtil.java
+++
b/modules/enterprise/server/installer/src/main/java/org/rhq/enterprise/server/installer/ServerInstallUtil.java
@@ -36,6 +36,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
+import java.util.Set;
import javax.security.auth.login.AppConfigurationEntry;
@@ -912,7 +913,7 @@ public class ServerInstallUtil {
* @throws Exception
*/
public static void persistStorageNodesIfNecessary(HashMap<String, String>
serverProperties, String password,
- List<StorageNode> storageNodes) throws Exception {
+ Set<StorageNode> storageNodes) throws Exception {
DatabaseType db = null;
Connection connection = null;
Statement queryStatement = null;
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 98d1405..c57d7cd 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
@@ -176,6 +176,8 @@ public class StorageClientManagerBean {
throw e;
} catch (Exception e) {
throw new RuntimeException(e.getMessage(), e);
+ } finally {
+ schemaManager.shutdown();
}
}
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 07bcedb..f7125e5 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
@@ -75,7 +75,7 @@ public class CassandraIntegrationTest {
dateTimeService = new DateTimeService();
Cluster cluster = new ClusterBuilder()
- .addContactPoints("127.0.0.1", "127.0.0.2")
+ .addContactPoints("127.0.0.1")
.withCredentialsObfuscated(RHQADMIN, RHQADMIN_PASSWORD)
.build();
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 32321e9..7c4d697 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
@@ -62,6 +62,9 @@ public class Simulator implements ShutdownManager {
final ScheduledExecutorService readers =
Executors.newScheduledThreadPool(plan.getNumReaders(),
new SimulatorThreadFactory());
+ Metrics metrics = new Metrics();
+ final ConsoleReporter consoleReporter = createConsoleReporter(metrics,
plan.getMetricsReportInterval());
+
Runtime.getRuntime().addShutdownHook(new Thread() {
@Override
public void run() {
@@ -69,6 +72,7 @@ public class Simulator implements ShutdownManager {
shutdown(readers, "readers", 5);
shutdown(aggregators, "aggregators", 1);
shutdown(aggregationQueue, "aggregationQueue",
Integer.MAX_VALUE);
+ consoleReporter.stop();
}
});
@@ -84,13 +88,9 @@ public class Simulator implements ShutdownManager {
metricsServer.setDateTimeService(plan.getDateTimeService());
- Metrics metrics = new Metrics();
-
MeasurementAggregator measurementAggregator = new
MeasurementAggregator(metricsServer, this, metrics,
aggregationQueue);
- ConsoleReporter consoleReporter = createConsoleReporter(metrics,
plan.getMetricsReportInterval());
-
for (int i = 0; i < plan.getNumMeasurementCollectors(); ++i) {
collectors.scheduleAtFixedRate(new MeasurementCollector(plan.getBatchSize(),
plan.getBatchSize() * i, metrics, metricsServer,
plan.getDateTimeService()), 0,
@@ -153,12 +153,15 @@ public class Simulator implements ShutdownManager {
}
private void createSchema(String[] nodes, int cqlPort) {
+ SchemaManager schemaManager = new SchemaManager("rhqadmin",
"1eeb2f255e832171df8592078de921bc", nodes,
+ cqlPort);
try {
log.info("Creating schema");
- SchemaManager schemaManager = new SchemaManager("rhqadmin",
"1eeb2f255e832171df8592078de921bc", nodes, 9142);
schemaManager.install();
} catch (Exception e) {
throw new RuntimeException("Failed to start simulator. An error occurred
during schema creation.", e);
+ } finally {
+ schemaManager.shutdown();
}
}
diff --git
a/modules/plugins/rhq-storage/src/test/java/org/rhq/plugins/storage/StorageNodeComponentITest.java
b/modules/plugins/rhq-storage/src/test/java/org/rhq/plugins/storage/StorageNodeComponentITest.java
index bae3c1d..67fed0c 100644
---
a/modules/plugins/rhq-storage/src/test/java/org/rhq/plugins/storage/StorageNodeComponentITest.java
+++
b/modules/plugins/rhq-storage/src/test/java/org/rhq/plugins/storage/StorageNodeComponentITest.java
@@ -120,34 +120,6 @@ public class StorageNodeComponentITest {
deploymentOptions.setHeapNewSize("64M");
deploymentOptions.load();
-
-// Deployer deployer = new Deployer();
-// deployer.setDeploymentOptions(deploymentOptions);
-//
-// deployer.unzipDistro();
-// deployer.applyConfigChanges();
-// deployer.updateFilePerms();
-// deployer.updateStorageAuthConf(Sets.newHashSet(node1Address, node2Address));
-//
-// File confDir = new File(basedir, "conf");
-// File cassandraJvmPropsFile = new File(confDir,
"cassandra-jvm.properties");
-// PropertiesFileUpdate propertiesUpdater = new
PropertiesFileUpdate(cassandraJvmPropsFile.getAbsolutePath());
-// Properties properties = propertiesUpdater.loadExistingProperties();
-//
-// String jvmOpts = properties.getProperty("JVM_OPTS");
-// jvmOpts = jvmOpts.substring(0, jvmOpts.lastIndexOf("\""));
-// jvmOpts = jvmOpts + " -Dcassandra.ring_delay_ms=100\"";
-// properties.setProperty("JVM_OPTS", jvmOpts);
-//
-// propertiesUpdater.update(properties);
-//
-// File binDir = new File(basedir, "bin");
-// SystemInfo systemInfo = SystemInfoFactory.createSystemInfo();
-//
-// ProcessExecution processExecution = getProcessExecution(binDir);
-// ProcessExecutionResults results = systemInfo.executeProcess(processExecution);
-//
-// assertEquals(results.getExitCode(), (Integer) 0, "Cassandra failed to
start: " + results.getCapturedOutput());
doDeployment(deploymentOptions);
String[] addresses = new String[] {"127.0.0.1"};
@@ -158,8 +130,12 @@ public class StorageNodeComponentITest {
SchemaManager schemaManager = new SchemaManager("rhqadmin",
"1eeb2f255e832171df8592078de921bc",
addresses, 9142);
- schemaManager.install();
- schemaManager.updateTopology();
+ try {
+ schemaManager.install();
+ schemaManager.updateTopology();
+ } finally {
+ schemaManager.shutdown();
+ }
}
private void doDeployment(DeploymentOptions deploymentOptions) throws Exception {
commit 94d9faba6fd4217954c4e454511ae0357aa59e0c
Author: John Sanda <jsanda(a)redhat.com>
Date: Mon Oct 21 21:26:58 2013 -0400
[BZ 1004050] do not require user to specify all storage nodes during install
If you install multiple storage nodes before installing the server and fail to
specifiy any of them in the rhq.storage.nodes property in rhq-server.properties,
then you will run into problems at runtime as the nodes not specified will go
through the deployment process. Any requests sent to those nodes during that
deployment process will get rejected. This commit makes the install process a
bit more robust by removing the requirement to specify all nodes. The server
installer now queries the driver for all cluster nodes, and storage node
entities will be created for all nodes reported by the driver.
This involved a refactoring of the classes in the rhq-cassandra-schema maven
module. A session is no longer created/destroyed for each operation, i.e.,
install/update schema, version check, etc. Callers of SchemaManager need to
invoke SchemaManager.shutdown to ensure that the cluster and session objects
are properly shutdown.
diff --git
a/modules/common/cassandra-schema/src/main/java/org/rhq/cassandra/schema/SessionManager.java
b/modules/common/cassandra-schema/src/main/java/org/rhq/cassandra/schema/SessionManager.java
new file mode 100644
index 0000000..897e98d
--- /dev/null
+++
b/modules/common/cassandra-schema/src/main/java/org/rhq/cassandra/schema/SessionManager.java
@@ -0,0 +1,60 @@
+package org.rhq.cassandra.schema;
+
+import java.util.Arrays;
+import java.util.Set;
+import java.util.TreeSet;
+
+import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.Host;
+import com.datastax.driver.core.ProtocolOptions;
+import com.datastax.driver.core.Session;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.rhq.cassandra.util.ClusterBuilder;
+
+/**
+ * @author John Sanda
+ */
+public class SessionManager {
+
+ private final Log log = LogFactory.getLog(SessionManager.class);
+
+
+ private Cluster cluster;
+
+ private Session session;
+
+ public void initSession(String username, String password, int cqlPort, String...
nodes) {
+ if (session == null) {
+ log.info("Initializing session for [username: " + username +
", cqlPort: " + cqlPort + ", nodes: " +
+ Arrays.toString(nodes) + "]");
+ cluster = new
ClusterBuilder().addContactPoints(nodes).withCredentialsObfuscated(username, password)
+
.withPort(cqlPort).withCompression(ProtocolOptions.Compression.NONE).build();
+ session = cluster.connect("system");
+ }
+ }
+
+ public void shutdownCluster() {
+ log.info("Shutting down cluster");
+ if (cluster != null) {
+ cluster.shutdown();
+ cluster = null;
+ session = null;
+ }
+ }
+
+ public Session getSession() {
+ return session;
+ }
+
+ public Set<String> getNodeAdresses() {
+ Set<String> nodes = new TreeSet<String>();
+ for (Host host : cluster.getMetadata().getAllHosts()) {
+ nodes.add(host.getAddress().getHostAddress());
+ }
+ return nodes;
+ }
+
+}