modules/core/domain/src/main/java/org/rhq/core/domain/cloud/StorageNode.java
| 3
modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/cloud/StorageNodeManagerBean.java
| 366 +++++++---
modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/cloud/StorageNodeManagerLocal.java
| 9
modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/operation/OperationManagerBean.java
| 7
modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/storage/StorageClusterMonitor.java
| 14
modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/storage/StorageNodeOperationsHandler.java
| 17
modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/storage/StorageNodeOperationsHandlerBean.java
| 90 ++
7 files changed, 418 insertions(+), 88 deletions(-)
New commits:
commit 490af59ea3141e94f865475df51ba061db82a716
Author: John Sanda <jsanda(a)redhat.com>
Date: Tue Jul 30 10:53:24 2013 -0400
adding back support for deploying additional nodes with internode authentication
When the user installs a new storage node the following steps occur,
* The node is committed into inventory (but not part of storage cluster)
* StorageNode entity is created with mode set to INSTALLED
* New node is announced to the cluster
* Server schedules operation for cluster nodes to update internode auth conf
file to include the new node
* Server prepares new node for bootstrap
* Schedules operation on new node to
* Apply cluster settings
* Update internode auth conf settings
* Restart the node so it bootstraps into cluster
* Token ranges will be assigned to node
* New node will stream data from other nodes
* Server is notified that new node is up and part of the cluster
* The operation mode is set to NORMAL
* addNodeMaintenance operation is run on each cluster node
* Updates replication_factor if necessary
* Runs repair if necessary
* Deletes old keys
* Updates seeds property
diff --git a/modules/core/domain/src/main/java/org/rhq/core/domain/cloud/StorageNode.java
b/modules/core/domain/src/main/java/org/rhq/core/domain/cloud/StorageNode.java
index 575edc74..f4d3934 100644
--- a/modules/core/domain/src/main/java/org/rhq/core/domain/cloud/StorageNode.java
+++ b/modules/core/domain/src/main/java/org/rhq/core/domain/cloud/StorageNode.java
@@ -56,6 +56,8 @@ import org.rhq.core.domain.resource.Resource;
+ " FROM StorageNode s " //
+ "LEFT JOIN FETCH s.resource r " //
+ " WHERE s.address = :address"),
+ @NamedQuery(name = StorageNode.QUERY_FIND_ALL_BY_MODE, query =
+ "SELECT s FROM StorageNode s WHERE s.operationMode = :operationMode"),
@NamedQuery(name = StorageNode.QUERY_FIND_ALL_NOT_INSTALLED, query = "SELECT s
FROM StorageNode s WHERE NOT s.operationMode = 'INSTALLED'"),
@NamedQuery(name = StorageNode.QUERY_FIND_ALL_NORMAL, query = "SELECT s FROM
StorageNode s WHERE s.operationMode = 'NORMAL'"),
@NamedQuery(name = StorageNode.QUERY_DELETE_BY_ID, query = "" //
@@ -91,6 +93,7 @@ public class StorageNode implements Serializable {
public static final String QUERY_FIND_ALL = "StorageNode.findAll";
public static final String QUERY_FIND_BY_ADDRESS =
"StorageNode.findByAddress";
+ public static final String QUERY_FIND_ALL_BY_MODE =
"StorageNode.findAllByMode";
public static final String QUERY_FIND_ALL_NOT_INSTALLED =
"StorageNode.findAllCloudMembers";
public static final String QUERY_DELETE_BY_ID = "StorageNode.deleteById";
public static final String QUERY_FIND_ALL_NORMAL =
"StorageNode.findAllNormalCloudMembers";
diff --git
a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/cloud/StorageNodeManagerBean.java
b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/cloud/StorageNodeManagerBean.java
index 049cf42..ee0f406 100644
---
a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/cloud/StorageNodeManagerBean.java
+++
b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/cloud/StorageNodeManagerBean.java
@@ -24,6 +24,8 @@
*/
package org.rhq.enterprise.server.cloud;
+import java.net.InetAddress;
+import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
@@ -47,6 +49,7 @@ import org.quartz.JobDataMap;
import org.quartz.SimpleTrigger;
import org.quartz.Trigger;
+import org.rhq.cassandra.schema.SchemaManager;
import org.rhq.core.domain.alert.Alert;
import org.rhq.core.domain.auth.Subject;
import org.rhq.core.domain.authz.Permission;
@@ -57,6 +60,8 @@ import org.rhq.core.domain.cloud.StorageNodeConfigurationComposite;
import org.rhq.core.domain.cloud.StorageNodeLoadComposite;
import org.rhq.core.domain.common.JobTrigger;
import org.rhq.core.domain.configuration.Configuration;
+import org.rhq.core.domain.configuration.PropertyList;
+import org.rhq.core.domain.configuration.PropertySimple;
import org.rhq.core.domain.criteria.AlertCriteria;
import org.rhq.core.domain.criteria.ResourceGroupCriteria;
import org.rhq.core.domain.criteria.ResourceOperationHistoryCriteria;
@@ -73,6 +78,7 @@ import org.rhq.core.domain.resource.group.ResourceGroup;
import org.rhq.core.domain.util.PageControl;
import org.rhq.core.domain.util.PageList;
import org.rhq.core.domain.util.PageOrdering;
+import org.rhq.core.util.StringUtil;
import org.rhq.enterprise.server.RHQConstants;
import org.rhq.enterprise.server.alert.AlertManagerLocal;
import org.rhq.enterprise.server.auth.SubjectManagerLocal;
@@ -82,6 +88,7 @@ import org.rhq.enterprise.server.cloud.instance.ServerManagerLocal;
import org.rhq.enterprise.server.configuration.ConfigurationManagerLocal;
import org.rhq.enterprise.server.measurement.MeasurementDataManagerLocal;
import org.rhq.enterprise.server.operation.OperationManagerLocal;
+import org.rhq.enterprise.server.resource.ResourceManagerLocal;
import org.rhq.enterprise.server.resource.ResourceTypeManagerLocal;
import org.rhq.enterprise.server.resource.group.ResourceGroupManagerLocal;
import org.rhq.enterprise.server.rest.reporting.MeasurementConverter;
@@ -100,6 +107,14 @@ public class StorageNodeManagerBean implements
StorageNodeManagerLocal, StorageN
private final Log log = LogFactory.getLog(StorageNodeManagerBean.class);
+ private static final String USERNAME_PROPERTY = "rhq.cassandra.username";
+ private static final String PASSWORD_PROPERTY = "rhq.cassandra.password";
+ private final static String MAINTENANCE_OPERATION = "addNodeMaintenance";
+ private final static String MAINTENANCE_OPERATION_NOTE = "Topology change
maintenance.";
+ private final static String RUN_REPAIR_PROPERTY = "runRepair";
+ private final static String UPDATE_SEEDS_LIST = "updateSeedsList";
+ private final static String SEEDS_LIST = "seedsList";
+
private static final String RHQ_STORAGE_CQL_PORT_PROPERTY =
"nativeTransportPort";
private static final String RHQ_STORAGE_JMX_PORT_PROPERTY = "jmxPort";
private static final String RHQ_STORAGE_ADDRESS_PROPERTY = "host";
@@ -139,46 +154,86 @@ public class StorageNodeManagerBean implements
StorageNodeManagerLocal, StorageN
@EJB
private StorageNodeManagerLocal storageNodeManger;
+ @EJB
+ private ResourceManagerLocal resourceManager;
+
@Override
public void linkResource(Resource resource) {
- List<StorageNode> storageNodes = this.getStorageNodes();
-
Configuration resourceConfig = resource.getPluginConfiguration();
- String configAddress =
resourceConfig.getSimpleValue(RHQ_STORAGE_ADDRESS_PROPERTY);
-
- if (configAddress != null) {
- // TODO Do not add the node to the group until we have verified it has joined
the cluster
- // StorageNodeMaintenanceJob currently determines if a new node has
successfully joined the cluster.
- addStorageNodeToGroup(resource);
-
- boolean storageNodeFound = false;
- if (storageNodes != null) {
- for (StorageNode storageNode : storageNodes) {
- if (configAddress.equals(storageNode.getAddress())) {
- storageNode.setResource(resource);
- storageNode.setOperationMode(OperationMode.NORMAL);
- storageNodeFound = true;
- break;
- }
- }
- }
+ String address = resourceConfig.getSimpleValue(RHQ_STORAGE_ADDRESS_PROPERTY);
- if (!storageNodeFound) {
- int cqlPort =
Integer.parseInt(resourceConfig.getSimpleValue(RHQ_STORAGE_CQL_PORT_PROPERTY));
- int jmxPort =
Integer.parseInt(resourceConfig.getSimpleValue(RHQ_STORAGE_JMX_PORT_PROPERTY));
+ if (log.isInfoEnabled()) {
+ log.info("Linking " + resource + " to storage node at " +
address);
+ }
+ try {
+ StorageNode storageNode =
findStorageNodeByAddress(InetAddress.getByName(address));
- StorageNode storageNode = new StorageNode();
- storageNode.setAddress(configAddress);
- storageNode.setCqlPort(cqlPort);
- storageNode.setJmxPort(jmxPort);
+ if (storageNode != null) {
+ if (log.isInfoEnabled()) {
+ log.info(storageNode + " is an existing storage node. No cluster
maintenance is necessary.");
+ }
storageNode.setResource(resource);
storageNode.setOperationMode(OperationMode.NORMAL);
+ addStorageNodeToGroup(resource);
+ } else {
+ storageNode = new StorageNode();
+ storageNode.setAddress(address);
+
storageNode.setCqlPort(Integer.parseInt(resourceConfig.getSimpleValue(RHQ_STORAGE_CQL_PORT_PROPERTY)));
+
storageNode.setJmxPort(Integer.parseInt(resourceConfig.getSimpleValue(RHQ_STORAGE_JMX_PORT_PROPERTY)));
+ storageNode.setResource(resource);
+ storageNode.setOperationMode(OperationMode.INSTALLED);
entityManager.persist(storageNode);
-// scheduleQuartzJob(storageNodes.size());
+ if (log.isInfoEnabled()) {
+ log.info(storageNode + " is a new storage node and not part of
the storage node cluster.");
+ log.info("Scheduling maintenance operations to bring " +
storageNode + " into the cluster...");
+ }
+
+ announceNewNode(storageNode);
}
+ } catch (UnknownHostException e) {
+ throw new RuntimeException("Could not resolve address [" + address
+ "]. The resource " + resource +
+ " cannot be linked to a storage node", e);
+ }
+ }
+
+ private void announceNewNode(StorageNode newStorageNode) {
+ if (log.isInfoEnabled()) {
+ log.info("Announcing " + newStorageNode + " to storage node
cluster.");
+ }
+
+ ResourceGroup storageNodeGroup = getStorageNodeGroup();
+
+ GroupOperationSchedule schedule = new GroupOperationSchedule();
+ schedule.setGroup(storageNodeGroup);
+ schedule.setHaltOnFailure(false);
+ schedule.setExecutionOrder(new
ArrayList<Resource>(storageNodeGroup.getExplicitResources()));
+ schedule.setJobTrigger(JobTrigger.createNowTrigger());
+ schedule.setSubject(subjectManager.getOverlord());
+ schedule.setOperationName("updateKnownNodes");
+
+ Configuration parameters = new Configuration();
+ parameters.put(createPropertyListOfAddresses("ipAddresses",
combine(getStorageNodes(), newStorageNode)));
+ schedule.setParameters(parameters);
+
+ operationManager.scheduleGroupOperation(subjectManager.getOverlord(), schedule);
+ }
+
+ private List<StorageNode> combine(List<StorageNode> storageNodes,
StorageNode storageNode) {
+ List<StorageNode> newList = new
ArrayList<StorageNode>(storageNodes.size() + 1);
+ newList.addAll(storageNodes);
+ newList.add(storageNode);
+
+ return newList;
+ }
+
+ private PropertyList createPropertyListOfAddresses(String propertyName,
List<StorageNode> nodes) {
+ PropertyList list = new PropertyList(propertyName);
+ for (StorageNode storageNode : nodes) {
+ list.add(new PropertySimple("address", storageNode.getAddress()));
}
+ return list;
}
@Override
@@ -214,7 +269,7 @@ public class StorageNodeManagerBean implements
StorageNodeManagerLocal, StorageN
ResourceGroup group = getStorageNodeGroup();
resourceGroupManager.addResourcesToGroup(subjectManager.getOverlord(),
group.getId(),
- new int[] {resource.getId()});
+ new int[]{resource.getId()});
}
@Override
@@ -232,6 +287,13 @@ public class StorageNodeManagerBean implements
StorageNodeManagerLocal, StorageN
}
@Override
+ public void addToStorageNodeGroup(StorageNode storageNode) {
+ storageNode.setOperationMode(OperationMode.NORMAL);
+ entityManager.merge(storageNode);
+ addStorageNodeToGroup(storageNode.getResource());
+ }
+
+ @Override
public ResourceGroup getStorageNodeGroup() {
Subject overlord = subjectManager.getOverlord();
@@ -398,6 +460,19 @@ public class StorageNodeManagerBean implements
StorageNodeManagerLocal, StorageN
return runner.execute();
}
+ public StorageNode findStorageNodeByAddress(InetAddress address) {
+ TypedQuery<StorageNode> query = entityManager.<StorageNode>
createNamedQuery(StorageNode.QUERY_FIND_BY_ADDRESS,
+ StorageNode.class);
+ query.setParameter("address", address.getHostAddress());
+ List<StorageNode> result = query.getResultList();
+
+ if (result != null && result.size() > 0) {
+ return result.get(0);
+ }
+
+ return null;
+ }
+
@Override
@RequiredPermissions({ @RequiredPermission(Permission.MANAGE_SETTINGS),
@RequiredPermission(Permission.MANAGE_INVENTORY) })
@@ -472,19 +547,6 @@ public class StorageNodeManagerBean implements
StorageNodeManagerLocal, StorageN
entityManager.flush();
}
- private StorageNode findStorageNodeByAddress(String address) {
- TypedQuery<StorageNode> query = entityManager.<StorageNode>
createNamedQuery(StorageNode.QUERY_FIND_BY_ADDRESS,
- StorageNode.class);
- query.setParameter("address", address);
- List<StorageNode> result = query.getResultList();
-
- if (result != null && result.size() > 0) {
- return result.get(0);
- }
-
- return null;
- }
-
private StorageNodeLoadComposite.MeasurementAggregateWithUnits
getMeasurementAggregateWithUnits(Subject subject,
int schedId, MeasurementUnits units, long beginTime, long endTime) {
MeasurementAggregate measurementAggregate =
measurementManager.getAggregate(subject, schedId, beginTime,
@@ -641,63 +703,68 @@ public class StorageNodeManagerBean implements
StorageNodeManagerLocal, StorageN
@Override
public boolean updateConfiguration(Subject subject, StorageNodeConfigurationComposite
storageNodeConfiguration) {
- StorageNode storageNode =
findStorageNodeByAddress(storageNodeConfiguration.getStorageNode().getAddress());
+ try {
+ StorageNode storageNode = findStorageNodeByAddress(InetAddress.getByName(
+ storageNodeConfiguration.getStorageNode().getAddress()));
+
+ if (storageNode != null && storageNode.getResource() != null) {
+ Configuration parameters = new Configuration();
+ parameters.setSimpleValue("jmxPort",
storageNodeConfiguration.getJmxPort() + "");
+ if (storageNodeConfiguration.getHeapSize() != null) {
+ parameters.setSimpleValue("heapSize",
storageNodeConfiguration.getHeapSize() + "");
+ }
+ if (storageNodeConfiguration.getHeapNewSize() != null) {
+ parameters.setSimpleValue("heapNewSize",
storageNodeConfiguration.getHeapNewSize() + "");
+ }
+ if (storageNodeConfiguration.getThreadStackSize() != null) {
+ parameters.setSimpleValue("threadStackSize",
storageNodeConfiguration.getThreadStackSize() + "");
+ }
+ parameters.setSimpleValue("restartIfRequired",
"false");
- if (storageNode != null && storageNode.getResource() != null) {
- Configuration parameters = new Configuration();
- parameters.setSimpleValue("jmxPort",
storageNodeConfiguration.getJmxPort() + "");
- if (storageNodeConfiguration.getHeapSize() != null) {
- parameters.setSimpleValue("heapSize",
storageNodeConfiguration.getHeapSize() + "");
- }
- if (storageNodeConfiguration.getHeapNewSize() != null) {
- parameters.setSimpleValue("heapNewSize",
storageNodeConfiguration.getHeapNewSize() + "");
- }
- if (storageNodeConfiguration.getThreadStackSize() != null) {
- parameters.setSimpleValue("threadStackSize",
storageNodeConfiguration.getThreadStackSize() + "");
- }
- parameters.setSimpleValue("restartIfRequired", "false");
+ Resource storageNodeResource = storageNode.getResource();
- Resource storageNodeResource = storageNode.getResource();
+ boolean result = runOperationAndWaitForResult(subject,
storageNodeResource, UPDATE_CONFIGURATION_OPERATION,
+ parameters);
- boolean result = runOperationAndWaitForResult(subject, storageNodeResource,
UPDATE_CONFIGURATION_OPERATION,
- parameters);
+ if (result) {
+ //2. Update the JMX port
+ //this is a fast operation compared to the restart
+ storageNode.setJmxPort(storageNodeConfiguration.getJmxPort());
+ entityManager.merge(storageNode);
- if (result) {
- //2. Update the JMX port
- //this is a fast operation compared to the restart
- storageNode.setJmxPort(storageNodeConfiguration.getJmxPort());
- entityManager.merge(storageNode);
+ //3. Restart the storage node
+ result = runOperationAndWaitForResult(subject, storageNodeResource,
RESTART_OPERATION,
+ new Configuration());
- //3. Restart the storage node
- result = runOperationAndWaitForResult(subject, storageNodeResource,
RESTART_OPERATION,
- new Configuration());
+ //4. Update the plugin configuration to talk with the new server
+ //Up to this point communication with the storage node should not
have been affected by the intermediate
+ //changes
+ Configuration storageNodePluginConfig =
configurationManager.getPluginConfiguration(subject,
+ storageNodeResource.getId());
- //4. Update the plugin configuration to talk with the new server
- //Up to this point communication with the storage node should not have
been affected by the intermediate
- //changes
- Configuration storageNodePluginConfig =
configurationManager.getPluginConfiguration(subject,
- storageNodeResource.getId());
+ String existingJMXPort =
storageNodePluginConfig.getSimpleValue("jmxPort");
+ String newJMXPort = storageNodeConfiguration.getJmxPort() +
"";
- String existingJMXPort =
storageNodePluginConfig.getSimpleValue("jmxPort");
- String newJMXPort = storageNodeConfiguration.getJmxPort() +
"";
+ if (!existingJMXPort.equals(newJMXPort)) {
+ storageNodePluginConfig.setSimpleValue("jmxPort",
newJMXPort);
- if (!existingJMXPort.equals(newJMXPort)) {
- storageNodePluginConfig.setSimpleValue("jmxPort",
newJMXPort);
+ String existingConnectionURL =
storageNodePluginConfig.getSimpleValue("connectorAddress");
+ String newConnectionURL =
existingConnectionURL.replace(":" + existingJMXPort + "/",
":"
+ + storageNodeConfiguration.getJmxPort() + "/");
+
storageNodePluginConfig.setSimpleValue("connectorAddress", newConnectionURL);
- String existingConnectionURL =
storageNodePluginConfig.getSimpleValue("connectorAddress");
- String newConnectionURL = existingConnectionURL.replace(":"
+ existingJMXPort + "/", ":"
- + storageNodeConfiguration.getJmxPort() + "/");
- storageNodePluginConfig.setSimpleValue("connectorAddress",
newConnectionURL);
+ configurationManager.updatePluginConfiguration(subject,
storageNodeResource.getId(),
+ storageNodePluginConfig);
+ }
- configurationManager.updatePluginConfiguration(subject,
storageNodeResource.getId(),
- storageNodePluginConfig);
+ return result;
}
-
- return result;
}
- }
- return false;
+ return false;
+ } catch (UnknownHostException e) {
+ throw new RuntimeException("Failed to resolve address for " +
storageNodeConfiguration, e);
+ }
}
@Override
@@ -754,4 +821,129 @@ public class StorageNodeManagerBean implements
StorageNodeManagerLocal, StorageN
return successResultFound;
}
-}
+
+ @Override
+ public void prepareNewNodesForBootstrap() {
+ List<StorageNode> newStorageNodes =
entityManager.createNamedQuery(StorageNode.QUERY_FIND_ALL_BY_MODE)
+ .setParameter("operationMode",
OperationMode.INSTALLED).getResultList();
+ if (newStorageNodes.isEmpty()) {
+ throw new RuntimeException("Failed to find storage node to bootstrap
into cluster.");
+ }
+ // Right now, without some user input, we can only reliably bootstrap one node at
a
+ // time. To support bootstrapping multiple nodes concurrently, a mechanism will
have
+ // to be put in place for the user to declare in advance the nodes that are
coming
+ // online. Then we can wait until all declared nodes have been committed into
+ // inventory and announced to the cluster
+ StorageNode storageNode = newStorageNodes.get(0);
+
+ if (log.isInfoEnabled()) {
+ log.info("Preparing to bootstrap " + storageNode + " into
cluster...");
+ }
+
+ List<StorageNode> existingStorageNodes = getStorageNodes();
+
+ ResourceOperationSchedule schedule = new ResourceOperationSchedule();
+ schedule.setResource(storageNode.getResource());
+ schedule.setJobTrigger(JobTrigger.createNowTrigger());
+ schedule.setSubject(subjectManager.getOverlord());
+ schedule.setOperationName("prepareForBootstrap");
+
+ Configuration parameters = new Configuration();
+ parameters.put(new PropertySimple("cqlPort",
existingStorageNodes.get(0).getCqlPort()));
+ // TODO need to add support for storage_port in cassandra/storage plugins
+ parameters.put(new PropertySimple("gossipPort", 7100));
+ parameters.put(createPropertyListOfAddresses("storageNodeIPAddresses",
getStorageNodes()));
+
+ schedule.setParameters(parameters);
+
+ operationManager.scheduleResourceOperation(subjectManager.getOverlord(),
schedule);
+ }
+
+ @Override
+ public void runAddNodeMaintenance() {
+ log.info("Preparing to schedule addNodeMaintenance on the storage
cluster...");
+
+ List<StorageNode> storageNodes =
entityManager.createNamedQuery(StorageNode.QUERY_FIND_ALL_BY_MODE,
+ StorageNode.class).setParameter("operationMode",
OperationMode.NORMAL).getResultList();
+
+ int clusterSize = storageNodes.size();
+ boolean isReadRepairNeeded;
+
+ if (clusterSize >= 4) {
+ // At 4 nodes we increase the RF to 3. We are not increasing the RF beyond
+ // that for additional nodes; so, there is no need to run repair if we are
+ // expanding from a 4 node cluster since the RF remains the same.
+ isReadRepairNeeded = false;
+ } else if (clusterSize == 1) {
+ // The RF will increase since we are going from a single to a multi-node
+ // cluster; therefore, we want to run repair.
+ isReadRepairNeeded = true;
+ } else if (clusterSize == 2) {
+ if (storageNodes.size() > 3) {
+ // If we go from 2 to > 3 nodes we will increase the RF to 3;
therefore
+ // we want to run repair.
+ isReadRepairNeeded = true;
+ } else {
+ // If we go from 2 to 3 nodes, we keep the RF at 2 so there is no need
+ // to run repair.
+ isReadRepairNeeded = false;
+ }
+ } else if (clusterSize == 3) {
+ // We are increasing the cluster size > 3 which means the RF will be
+ // updated to 3; therefore, we want to run repair.
+ isReadRepairNeeded = true;
+ } else {
+ // If we cluster size of zero, then something is really screwed up. It
+ // should always be > 0.
+ isReadRepairNeeded = storageNodes.size() > 1;
+ }
+
+ if (isReadRepairNeeded) {
+ updateTopology(storageNodes);
+ }
+
+ ResourceGroup storageNodeGroup = getStorageNodeGroup();
+
+ GroupOperationSchedule schedule = new GroupOperationSchedule();
+ schedule.setGroup(storageNodeGroup);
+ schedule.setHaltOnFailure(false);
+ schedule.setExecutionOrder(new
ArrayList<Resource>(storageNodeGroup.getExplicitResources()));
+ schedule.setJobTrigger(JobTrigger.createNowTrigger());
+ schedule.setSubject(subjectManager.getOverlord());
+ schedule.setOperationName(MAINTENANCE_OPERATION);
+ schedule.setDescription(MAINTENANCE_OPERATION_NOTE);
+
+ Configuration config = new Configuration();
+ config.put(createPropertyListOfAddresses(SEEDS_LIST, storageNodes));
+ config.put(new PropertySimple(RUN_REPAIR_PROPERTY, isReadRepairNeeded));
+ config.put(new PropertySimple(UPDATE_SEEDS_LIST, Boolean.TRUE));
+
+ schedule.setParameters(config);
+
+ operationManager.scheduleGroupOperation(subjectManager.getOverlord(), schedule);
+ }
+
+ private boolean updateTopology(List<StorageNode> storageNodes) {
+ String username = getRequiredStorageProperty(USERNAME_PROPERTY);
+ String password = getRequiredStorageProperty(PASSWORD_PROPERTY);
+ SchemaManager schemaManager = new SchemaManager(username, password,
storageNodes);
+ try{
+ return schemaManager.updateTopology(false);
+ } catch (Exception e) {
+ log.error("An error occurred while applying schema topology
changes", e);
+ }
+
+ return false;
+ }
+
+ private String getRequiredStorageProperty(String property) {
+ String value = System.getProperty(property);
+ if (StringUtil.isEmpty(property)) {
+ throw new IllegalStateException("The system property [" + property
+ "] is not set. The RHQ "
+ + "server will not be able connect to the RHQ storage node(s). This
property should be defined "
+ + "in rhq-server.properties.");
+ }
+ return value;
+ }
+
+}
\ No newline at end of file
diff --git
a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/cloud/StorageNodeManagerLocal.java
b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/cloud/StorageNodeManagerLocal.java
index 15fa85c..00ba9e7 100644
---
a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/cloud/StorageNodeManagerLocal.java
+++
b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/cloud/StorageNodeManagerLocal.java
@@ -18,6 +18,7 @@
*/
package org.rhq.enterprise.server.cloud;
+import java.net.InetAddress;
import java.util.List;
import javax.ejb.Local;
@@ -106,6 +107,8 @@ public interface StorageNodeManagerLocal {
*/
PageList<Alert> findAllStorageNodeAlerts(Subject subject, StorageNode
storageNode);
+ StorageNode findStorageNodeByAddress(InetAddress address);
+
/**
* Find ids for all resources and sub-resources of Storage Nodes that
@@ -170,6 +173,8 @@ public interface StorageNodeManagerLocal {
*/
boolean storageNodeGroupExists();
+ void addToStorageNodeGroup(StorageNode storageNode);
+
/**
* This method assumes the storage node resource group already exists; as such, it
should only be called from places
* in the code that are after the point(s) where the group has been created.
@@ -181,4 +186,8 @@ public interface StorageNodeManagerLocal {
void scheduleOperationInNewTransaction(Subject subject, ResourceOperationSchedule
schedule);
+ void prepareNewNodesForBootstrap();
+
+ void runAddNodeMaintenance();
+
}
diff --git
a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/operation/OperationManagerBean.java
b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/operation/OperationManagerBean.java
index 088f13e..5d3b1ae 100644
---
a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/operation/OperationManagerBean.java
+++
b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/operation/OperationManagerBean.java
@@ -96,6 +96,7 @@ import org.rhq.enterprise.server.resource.ResourceNotFoundException;
import org.rhq.enterprise.server.resource.group.ResourceGroupManagerLocal;
import org.rhq.enterprise.server.resource.group.ResourceGroupNotFoundException;
import org.rhq.enterprise.server.scheduler.SchedulerLocal;
+import org.rhq.enterprise.server.storage.StorageNodeOperationsHandler;
import org.rhq.enterprise.server.util.CriteriaQueryGenerator;
import org.rhq.enterprise.server.util.CriteriaQueryRunner;
@@ -123,6 +124,9 @@ public class OperationManagerBean implements OperationManagerLocal,
OperationMan
@EJB
private SubjectManagerLocal subjectManager;
+ @EJB
+ private StorageNodeOperationsHandler storageNodeOperationsHandler;
+
@SuppressWarnings("unchecked")
public List<IntegerOptionItem> getResourceNameOptionItems(int groupId) {
String queryName = ResourceGroup.QUERY_FIND_RESOURCE_NAMES_BY_GROUP_ID;
@@ -917,7 +921,7 @@ public class OperationManagerBean implements OperationManagerLocal,
OperationMan
if (history.getParameters() != null) {
history.getParameters().getId(); // eagerly reload the parameters
}
-
+ storageNodeOperationsHandler.handleOperationUpdateIfNecessary(history);
notifyAlertConditionCacheManager("updateOperationHistory", history);
return history;
}
@@ -1723,6 +1727,7 @@ public class OperationManagerBean implements OperationManagerLocal,
OperationMan
if (!stillInProgress) {
groupHistory.setErrorMessage((groupErrorMessage == null) ? null :
groupErrorMessage.toString());
groupHistory.setStatus(groupStatus);
+
storageNodeOperationsHandler.handleGroupOperationUpdateIfNecessary(groupHistory);
notifyAlertConditionCacheManager("checkForCompletedGroupOperation",
groupHistory);
}
}
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
index ec28888..0b404bb 100644
---
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
@@ -9,7 +9,9 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.rhq.core.domain.cloud.Server;
+import org.rhq.core.domain.cloud.StorageNode;
import org.rhq.enterprise.server.auth.SubjectManagerLocal;
+import org.rhq.enterprise.server.cloud.StorageNodeManagerLocal;
import org.rhq.enterprise.server.cloud.TopologyManagerLocal;
import org.rhq.enterprise.server.cloud.instance.ServerManagerLocal;
import org.rhq.enterprise.server.util.LookupUtil;
@@ -36,6 +38,18 @@ public class StorageClusterMonitor implements StorageStateListener {
log.info("Taking server out of maintenance mode");
updateServerMode(Server.OperationMode.NORMAL);
}
+
+ StorageNodeManagerLocal storageNodeManager = LookupUtil.getStorageNodeManager();
+ StorageNode newClusterNode =
storageNodeManager.findStorageNodeByAddress(address);
+
+ if (newClusterNode == null) {
+ log.error("Did not find storage node with address [" +
address.getHostAddress() + "]. This should not " +
+ "happen.");
+ } else {
+ log.info("Adding " + newClusterNode + " to storage cluster and
scheduling cluster maintenance...");
+ storageNodeManager.addToStorageNodeGroup(newClusterNode);
+ storageNodeManager.runAddNodeMaintenance();
+ }
}
@Override
diff --git
a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/storage/StorageNodeOperationsHandler.java
b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/storage/StorageNodeOperationsHandler.java
new file mode 100644
index 0000000..96e8de8
--- /dev/null
+++
b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/storage/StorageNodeOperationsHandler.java
@@ -0,0 +1,17 @@
+package org.rhq.enterprise.server.storage;
+
+import javax.ejb.Asynchronous;
+
+import org.rhq.core.domain.operation.GroupOperationHistory;
+import org.rhq.core.domain.operation.OperationHistory;
+
+/**
+ * @author John Sanda
+ */
+public interface StorageNodeOperationsHandler {
+ @Asynchronous
+ void handleOperationUpdateIfNecessary(OperationHistory operationHistory);
+
+ @Asynchronous
+ void handleGroupOperationUpdateIfNecessary(GroupOperationHistory operationHistory);
+}
diff --git
a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/storage/StorageNodeOperationsHandlerBean.java
b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/storage/StorageNodeOperationsHandlerBean.java
new file mode 100644
index 0000000..6da5cca
--- /dev/null
+++
b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/storage/StorageNodeOperationsHandlerBean.java
@@ -0,0 +1,90 @@
+package org.rhq.enterprise.server.storage;
+
+import javax.ejb.Asynchronous;
+import javax.ejb.EJB;
+import javax.ejb.Stateless;
+import javax.persistence.EntityManager;
+import javax.persistence.PersistenceContext;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.rhq.core.domain.cloud.StorageNode;
+import org.rhq.core.domain.operation.GroupOperationHistory;
+import org.rhq.core.domain.operation.OperationDefinition;
+import org.rhq.core.domain.operation.OperationHistory;
+import org.rhq.core.domain.operation.OperationRequestStatus;
+import org.rhq.core.domain.resource.Resource;
+import org.rhq.core.domain.resource.ResourceType;
+import org.rhq.enterprise.server.RHQConstants;
+import org.rhq.enterprise.server.cloud.StorageNodeManagerLocal;
+
+/**
+ * @author John Sanda
+ */
+@Stateless
+public class StorageNodeOperationsHandlerBean implements StorageNodeOperationsHandler {
+
+ private final Log log = LogFactory.getLog(StorageNodeOperationsHandlerBean.class);
+
+ private static final String STORAGE_NODE_TYPE_NAME = "RHQ Storage Node";
+ private static final String STORAGE_NODE_PLUGIN_NAME = "RHQStorage";
+
+ @PersistenceContext(unitName = RHQConstants.PERSISTENCE_UNIT_NAME)
+ private EntityManager entityManager;
+
+ @EJB
+ private StorageNodeManagerLocal storageNodeManager;
+
+ @Override
+ @Asynchronous
+ public void handleOperationUpdateIfNecessary(OperationHistory operationHistory) {
+// if (isStorageNodeOperation(operationHistory.getOperationDefinition())) {
+// if
(operationHistory.getOperationDefinition().getName().equals("prepareForBootstrap"))
{
+// ResourceOperationHistory resourceOperationHistory =
entityManager.find(ResourceOperationHistory.class,
+// operationHistory.getId());
+// if (resourceOperationHistory.getStatus() ==
OperationRequestStatus.SUCCESS) {
+//
+// }
+// StorageNode storageNode =
findStorageNode(resourceOperationHistory.getResource());
+// storageNode.setOperationMode(StorageNode.OperationMode.NORMAL);
+// }
+// }
+ }
+
+ private StorageNode findStorageNode(Resource resource) {
+ for (StorageNode storageNode : storageNodeManager.getStorageNodes()) {
+ if (storageNode.getResource().getId() == resource.getId()) {
+ return storageNode;
+ }
+ }
+ return null;
+ }
+
+ @Override
+ @Asynchronous
+ public void handleGroupOperationUpdateIfNecessary(GroupOperationHistory
groupOperationHistory) {
+ if (isStorageNodeOperation(groupOperationHistory.getOperationDefinition())) {
+ if
(groupOperationHistory.getOperationDefinition().getName().equals("updateKnownNodes"))
{
+ if (groupOperationHistory.getStatus() == OperationRequestStatus.SUCCESS)
{
+ log.info("New storage has been successfully announced to the
storage node cluster.");
+ storageNodeManager.prepareNewNodesForBootstrap();
+ } else if (groupOperationHistory.getStatus() ==
OperationRequestStatus.FAILURE) {
+ log.warn("Failed to announce new storage node to the cluster. It
cannot join the cluster until " +
+ "it has been announced to existing cluster nodes.");
+ } else if (groupOperationHistory.getStatus() ==
OperationRequestStatus.CANCELED) {
+ log.warn("New storage node has not been announced to the
cluster. The group operation " +
+ groupOperationHistory.getOperationDefinition().getName() + "
has been canceled. The new node " +
+ "cannot join the cluster until it has been announced to
existing cluster nodes.");
+ }
+ }
+ }
+ }
+
+ private boolean isStorageNodeOperation(OperationDefinition operationDefinition) {
+ ResourceType resourceType = operationDefinition.getResourceType();
+ return resourceType.getName().equals(STORAGE_NODE_TYPE_NAME) &&
+ resourceType.getPlugin().equals(STORAGE_NODE_PLUGIN_NAME);
+ }
+
+}