You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ozone.apache.org by ha...@apache.org on 2021/12/17 15:54:26 UTC
[ozone] branch master updated: HDDS-5490. Remove (Decommission) an OM node from HA ring (#2886)
This is an automated email from the ASF dual-hosted git repository.
hanishakoneru pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/master by this push:
new fe7f5b4 HDDS-5490. Remove (Decommission) an OM node from HA ring (#2886)
fe7f5b4 is described below
commit fe7f5b4ecad81dfff7d7703b69e2eec1907a4a95
Author: Hanisha Koneru <ha...@apache.org>
AuthorDate: Fri Dec 17 07:54:14 2021 -0800
HDDS-5490. Remove (Decommission) an OM node from HA ring (#2886)
---
.../common/src/main/resources/ozone-default.xml | 15 +-
.../java/org/apache/hadoop/hdds/ExitManager.java | 4 +
.../org/apache/hadoop/hdds/cli/OzoneAdmin.java | 11 ++
.../main/java/org/apache/hadoop/ozone/OmUtils.java | 94 +++++----
.../org/apache/hadoop/ozone/om/OMConfigKeys.java | 2 +
.../ozone/om/ha/OMFailoverProxyProvider.java | 3 +-
.../hadoop/ozone/om/helpers/OMNodeDetails.java | 19 +-
.../hadoop/ozone/om/protocol/OMAdminProtocol.java | 6 +
.../hadoop/ozone/om/protocol/OMConfiguration.java | 30 ++-
.../protocolPB/OMAdminProtocolClientSideImpl.java | 112 +++++++++--
.../hadoop/ozone/TestOzoneConfigurationFields.java | 3 +
...otstrap.java => TestAddRemoveOzoneManager.java} | 107 ++++++++++-
.../src/main/proto/OMAdminProtocol.proto | 29 ++-
.../org/apache/hadoop/ozone/om/OzoneManager.java | 132 +++++++++----
.../apache/hadoop/ozone/om/ha/OMHANodeDetails.java | 3 +-
.../ozone/om/ratis/OzoneManagerRatisServer.java | 74 +++++--
.../om/snapshot/OzoneManagerSnapshotProvider.java | 7 +
.../protocolPB/OMAdminProtocolServerSideImpl.java | 50 +++++
.../hadoop/ozone/s3/OzoneServiceProvider.java | 3 +-
.../ozone/admin/om/DecommissionOMSubcommand.java | 212 +++++++++++++++++++++
.../org/apache/hadoop/ozone/admin/om/OMAdmin.java | 7 +-
21 files changed, 792 insertions(+), 131 deletions(-)
diff --git a/hadoop-hdds/common/src/main/resources/ozone-default.xml b/hadoop-hdds/common/src/main/resources/ozone-default.xml
index e26ad1a..057d80c 100644
--- a/hadoop-hdds/common/src/main/resources/ozone-default.xml
+++ b/hadoop-hdds/common/src/main/resources/ozone-default.xml
@@ -458,6 +458,10 @@
EXAMPLEOMSERVICEID). The OM service ID should be the value (one of the
values if there are multiple) set for the parameter ozone.om.service.ids.
+ Decommissioned nodes (represented by node Ids in
+ ozone.om.decommissioned.nodes config list) will be ignored and not
+ included in the OM HA setup even if added to this list.
+
Unique identifiers for each OM Node, delimited by commas. This will be
used by OzoneManagers in HA setup to determine all the OzoneManagers
belonging to the same OMservice in the cluster. For example, if you
@@ -468,6 +472,15 @@
</description>
</property>
<property>
+ <name>ozone.om.decommissioned.nodes.EXAMPLEOMSERVICEID</name>
+ <value/>
+ <tag>OM, HA</tag>
+ <description>
+ Comma-separated list of OM node Ids which have been decommissioned. OMs
+ present in this list will not be included in the OM HA ring.
+ </description>
+ </property>
+ <property>
<name>ozone.om.node.id</name>
<value/>
<tag>OM, HA</tag>
@@ -2303,7 +2316,7 @@
<tag>OM, MANAGEMENT</tag>
<description>
Expert only. The maximum number of retries for Ozone Manager Admin
- protocol.
+ protocol on each OM.
</description>
</property>
<property>
diff --git a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/ExitManager.java b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/ExitManager.java
index 425a1f8..3bcf641 100644
--- a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/ExitManager.java
+++ b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/ExitManager.java
@@ -40,4 +40,8 @@ public class ExitManager {
public void forceExit(int status, Exception ex, Logger log) {
ExitUtils.terminate(status, ex.getLocalizedMessage(), ex, log);
}
+
+ public void forceExit(int status, String exMsg, Logger log) {
+ ExitUtils.terminate(status, exMsg, log);
+ }
}
diff --git a/hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/cli/OzoneAdmin.java b/hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/cli/OzoneAdmin.java
index e2cc307..8ba56c1 100644
--- a/hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/cli/OzoneAdmin.java
+++ b/hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/cli/OzoneAdmin.java
@@ -18,8 +18,10 @@
package org.apache.hadoop.hdds.cli;
import com.google.common.annotations.VisibleForTesting;
+import java.io.IOException;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.tracing.TracingUtil;
+import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.util.NativeCodeLoader;
import org.apache.log4j.ConsoleAppender;
@@ -43,6 +45,8 @@ public class OzoneAdmin extends GenericCli {
private OzoneConfiguration ozoneConf;
+ private UserGroupInformation user;
+
public OzoneAdmin() {
super(OzoneAdmin.class);
}
@@ -60,6 +64,13 @@ public class OzoneAdmin extends GenericCli {
return ozoneConf;
}
+ public UserGroupInformation getUser() throws IOException {
+ if (user == null) {
+ user = UserGroupInformation.getCurrentUser();
+ }
+ return user;
+ }
+
/**
* Main for the Ozone Admin shell Command handling.
*
diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/OmUtils.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/OmUtils.java
index 433fc11..96c5658 100644
--- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/OmUtils.java
+++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/OmUtils.java
@@ -39,7 +39,6 @@ import java.util.Set;
import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.hdds.HddsUtils;
import org.apache.hadoop.hdds.conf.ConfigurationSource;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.scm.client.HddsClientUtils;
@@ -60,6 +59,7 @@ import static org.apache.hadoop.hdds.HddsUtils.getPortNumberFromConfigKeys;
import static org.apache.hadoop.ozone.OzoneConsts.OM_KEY_PREFIX;
import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_ADDRESS_KEY;
import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_BIND_HOST_DEFAULT;
+import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_DECOMMISSIONED_NODES_KEY;
import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_HTTPS_ADDRESS_KEY;
import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_HTTPS_BIND_HOST_KEY;
import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_HTTPS_BIND_PORT_DEFAULT;
@@ -69,8 +69,6 @@ import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_HTTP_BIND_PORT_DE
import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_INTERNAL_SERVICE_ID;
import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_NODES_KEY;
import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_PORT_DEFAULT;
-import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_RATIS_PORT_DEFAULT;
-import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_RATIS_PORT_KEY;
import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_SERVICE_IDS_KEY;
import org.slf4j.Logger;
@@ -123,7 +121,7 @@ public final class OmUtils {
if (!result.containsKey(serviceId)) {
result.put(serviceId, new ArrayList<>());
}
- for (String nodeId : getOMNodeIds(conf, serviceId)) {
+ for (String nodeId : getActiveOMNodeIds(conf, serviceId)) {
String rpcAddr = getOmRpcAddress(conf,
ConfUtils.addKeySuffixes(OZONE_OM_ADDRESS_KEY, serviceId, nodeId));
if (rpcAddr != null) {
@@ -329,12 +327,47 @@ public final class OmUtils {
}
/**
- * Get a collection of all omNodeIds for the given omServiceId.
+ * Get a collection of all active omNodeIds (excluding decommissioned nodes)
+ * for the given omServiceId.
*/
- public static Collection<String> getOMNodeIds(ConfigurationSource conf,
+ public static Collection<String> getActiveOMNodeIds(ConfigurationSource conf,
String omServiceId) {
- String key = ConfUtils.addSuffix(OZONE_OM_NODES_KEY, omServiceId);
- return conf.getTrimmedStringCollection(key);
+ String nodeIdsKey = ConfUtils.addSuffix(OZONE_OM_NODES_KEY, omServiceId);
+ Collection<String> nodeIds = conf.getTrimmedStringCollection(nodeIdsKey);
+ String decommNodesKey = ConfUtils.addKeySuffixes(
+ OZONE_OM_DECOMMISSIONED_NODES_KEY, omServiceId);
+ Collection<String> decommNodeIds = conf.getTrimmedStringCollection(
+ decommNodesKey);
+ nodeIds.removeAll(decommNodeIds);
+
+ return nodeIds;
+ }
+
+ /**
+ * Get a collection of all omNodeIds (active and decommissioned) for a
+ * gived omServiceId.
+ */
+ public static Collection<String> getAllOMNodeIds(ConfigurationSource conf,
+ String omServiceId) {
+ Set<String> nodeIds = new HashSet<>();
+ String nodeIdsKey = ConfUtils.addSuffix(OZONE_OM_NODES_KEY, omServiceId);
+ String decommNodesKey = ConfUtils.addKeySuffixes(
+ OZONE_OM_DECOMMISSIONED_NODES_KEY, omServiceId);
+
+ nodeIds.addAll(conf.getTrimmedStringCollection(nodeIdsKey));
+ nodeIds.addAll(conf.getTrimmedStringCollection(decommNodesKey));
+
+ return nodeIds;
+ }
+
+ /**
+ * Get a collection of nodeIds of all decommissioned OMs for a given
+ * omServideId.
+ */
+ public static Collection<String> getDecommissionedNodes(
+ ConfigurationSource conf, String omServiceId) {
+ return conf.getTrimmedStringCollection(ConfUtils.addKeySuffixes(
+ OZONE_OM_DECOMMISSIONED_NODES_KEY, omServiceId));
}
/**
@@ -657,14 +690,14 @@ public final class OmUtils {
/**
- * For a given service ID, return th of configured OM hosts.
+ * For a given service ID, return list of configured OM hosts.
* @param conf configuration
* @param omServiceId service id
* @return Set of hosts.
*/
public static Set<String> getOmHostsFromConfig(OzoneConfiguration conf,
String omServiceId) {
- Collection<String> omNodeIds = OmUtils.getOMNodeIds(conf,
+ Collection<String> omNodeIds = OmUtils.getActiveOMNodeIds(conf,
omServiceId);
Set<String> omHosts = new HashSet<>();
for (String nodeId : OmUtils.emptyAsSingletonNull(omNodeIds)) {
@@ -680,44 +713,33 @@ public final class OmUtils {
/**
* Get a list of all OM details (address and ports) from the specified config.
*/
- public static List<OMNodeDetails> getAllOMAddresses(OzoneConfiguration conf,
- String omServiceId, String currentOMNodeId) {
+ public static List<OMNodeDetails> getAllOMHAAddresses(OzoneConfiguration conf,
+ String omServiceId, boolean includeDecommissionedNodes) {
List<OMNodeDetails> omNodesList = new ArrayList<>();
- Collection<String> omNodeIds = OmUtils.getOMNodeIds(conf, omServiceId);
+ Collection<String> omNodeIds;
+ if (includeDecommissionedNodes) {
+ omNodeIds = OmUtils.getAllOMNodeIds(conf, omServiceId);
+ } else {
+ omNodeIds = OmUtils.getActiveOMNodeIds(conf, omServiceId);
+ }
+ Collection<String> decommNodeIds = OmUtils.getDecommissionedNodes(conf,
+ omServiceId);
String rpcAddrStr, hostAddr, httpAddr, httpsAddr;
int rpcPort, ratisPort;
if (omNodeIds.size() == 0) {
- //Check if it is non-HA cluster
- rpcAddrStr = OmUtils.getOmRpcAddress(conf, OZONE_OM_ADDRESS_KEY);
- if (rpcAddrStr == null || rpcAddrStr.isEmpty()) {
- return omNodesList;
- }
- hostAddr = HddsUtils.getHostName(rpcAddrStr).orElse(null);
- rpcPort = HddsUtils.getHostPort(rpcAddrStr).orElse(0);
- ratisPort = conf.getInt(OZONE_OM_RATIS_PORT_KEY,
- OZONE_OM_RATIS_PORT_DEFAULT);
- httpAddr = OmUtils.getHttpAddressForOMPeerNode(conf,
- null, null, hostAddr);
- httpsAddr = OmUtils.getHttpsAddressForOMPeerNode(conf,
- null, null, hostAddr);
-
- omNodesList.add(new OMNodeDetails.Builder()
- .setOMNodeId(currentOMNodeId)
- .setHostAddress(hostAddr)
- .setRpcPort(rpcPort)
- .setRatisPort(ratisPort)
- .setHttpAddress(httpAddr)
- .setHttpsAddress(httpsAddr)
- .build());
- return omNodesList;
+ // If there are no nodeIds present, return empty list
+ return Collections.EMPTY_LIST;
}
for (String nodeId : omNodeIds) {
try {
OMNodeDetails omNodeDetails = OMNodeDetails.getOMNodeDetailsFromConf(
conf, omServiceId, nodeId);
+ if (decommNodeIds.contains(omNodeDetails.getNodeId())) {
+ omNodeDetails.setDecommissioningState();
+ }
omNodesList.add(omNodeDetails);
} catch (IOException e) {
String omRpcAddressStr = OMNodeDetails.getOMNodeAddressFromConf(conf,
diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/OMConfigKeys.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/OMConfigKeys.java
index 08bb64c..e3e7f41 100644
--- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/OMConfigKeys.java
+++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/OMConfigKeys.java
@@ -49,6 +49,8 @@ public final class OMConfigKeys {
"ozone.om.nodes";
public static final String OZONE_OM_NODE_ID_KEY =
"ozone.om.node.id";
+ public static final String OZONE_OM_DECOMMISSIONED_NODES_KEY =
+ "ozone.om.decommissioned.nodes";
public static final String OZONE_OM_ADDRESS_KEY =
"ozone.om.address";
diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/OMFailoverProxyProvider.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/OMFailoverProxyProvider.java
index 7c45593..5432468 100644
--- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/OMFailoverProxyProvider.java
+++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/OMFailoverProxyProvider.java
@@ -132,7 +132,8 @@ public class OMFailoverProxyProvider<T> implements
Collection<String> omServiceIds = Collections.singletonList(omSvcId);
for (String serviceId : OmUtils.emptyAsSingletonNull(omServiceIds)) {
- Collection<String> omNodeIds = OmUtils.getOMNodeIds(config, serviceId);
+ Collection<String> omNodeIds = OmUtils.getActiveOMNodeIds(config,
+ serviceId);
for (String nodeId : OmUtils.emptyAsSingletonNull(omNodeIds)) {
diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OMNodeDetails.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OMNodeDetails.java
index 71c51d8..3de2f12 100644
--- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OMNodeDetails.java
+++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OMNodeDetails.java
@@ -24,6 +24,7 @@ import org.apache.hadoop.net.NetUtils;
import org.apache.hadoop.ozone.OmUtils;
import org.apache.hadoop.ozone.ha.ConfUtils;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerAdminProtocolProtos.OMNodeInfo;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerAdminProtocolProtos.NodeState;
import org.apache.hadoop.hdds.NodeDetails;
import java.io.IOException;
@@ -40,6 +41,7 @@ import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_RATIS_PORT_KEY;
*/
public final class OMNodeDetails extends NodeDetails {
private int rpcPort;
+ private boolean isDecommissioned = false;
/**
* Constructs OMNodeDetails object.
@@ -61,6 +63,14 @@ public final class OMNodeDetails extends NodeDetails {
this.rpcPort = rpcPort;
}
+ public void setDecommissioningState() {
+ isDecommissioned = true;
+ }
+
+ public boolean isDecommissioned() {
+ return isDecommissioned;
+ }
+
@Override
public String toString() {
return "OMNodeDetails["
@@ -218,15 +228,22 @@ public final class OMNodeDetails extends NodeDetails {
.setHostAddress(getHostAddress())
.setRpcPort(getRpcPort())
.setRatisPort(getRatisPort())
+ .setNodeState(isDecommissioned ?
+ NodeState.DECOMMISSIONED : NodeState.ACTIVE)
.build();
}
public static OMNodeDetails getFromProtobuf(OMNodeInfo omNodeInfo) {
- return new Builder()
+ OMNodeDetails nodeDetails = new Builder()
.setOMNodeId(omNodeInfo.getNodeID())
.setHostAddress(omNodeInfo.getHostAddress())
.setRpcPort(omNodeInfo.getRpcPort())
.setRatisPort(omNodeInfo.getRatisPort())
.build();
+ if (omNodeInfo.hasNodeState() &&
+ omNodeInfo.getNodeState().equals(NodeState.DECOMMISSIONED)) {
+ nodeDetails.setDecommissioningState();
+ }
+ return nodeDetails;
}
}
diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocol/OMAdminProtocol.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocol/OMAdminProtocol.java
index 1a90ee7..3569408 100644
--- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocol/OMAdminProtocol.java
+++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocol/OMAdminProtocol.java
@@ -20,6 +20,7 @@ package org.apache.hadoop.ozone.om.protocol;
import java.io.Closeable;
import java.io.IOException;
import org.apache.hadoop.ozone.om.OMConfigKeys;
+import org.apache.hadoop.ozone.om.helpers.OMNodeDetails;
import org.apache.hadoop.security.KerberosInfo;
/**
@@ -33,4 +34,9 @@ public interface OMAdminProtocol extends Closeable {
* Get the OM configuration.
*/
OMConfiguration getOMConfiguration() throws IOException;
+
+ /**
+ * Remove OM from HA ring.
+ */
+ void decommission(OMNodeDetails removeOMNode) throws IOException;
}
diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocol/OMConfiguration.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocol/OMConfiguration.java
index 8bae3ea..f91b119 100644
--- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocol/OMConfiguration.java
+++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocol/OMConfiguration.java
@@ -32,9 +32,9 @@ import org.apache.hadoop.ozone.om.helpers.OMNodeDetails;
*/
public final class OMConfiguration {
- // OM nodes present in OM's memory
+ // OM nodes present in OM's memory (does not include Decommissioned nodes)
private List<OMNodeDetails> omNodesInMemory = new ArrayList<>();
- // OM nodes reloaded from new config on disk
+ // OM nodes reloaded from new config on disk (includes Decommissioned nodes)
private List<OMNodeDetails> omNodesInNewConf = new ArrayList<>();
private OMConfiguration(List<OMNodeDetails> inMemoryNodeList,
@@ -80,13 +80,25 @@ public final class OMConfiguration {
}
/**
- * Reload configuration from disk and return all the OM nodes present in
- * the new conf under current serviceId.
+ * Reload configuration from disk and return all active OM nodes (excludes
+ * decommissioned nodes) present in the new conf under current serviceId.
*/
- public Map<String, OMNodeDetails> getOmNodesInNewConf() {
- return omNodesInNewConf.stream().collect(Collectors.toMap(
- NodeDetails::getNodeId,
- omNodeDetails -> omNodeDetails,
- (nodeId, omNodeDetails) -> omNodeDetails));
+ public Map<String, OMNodeDetails> getActiveOmNodesInNewConf() {
+ return omNodesInNewConf.stream()
+ .filter(omNodeDetails -> !omNodeDetails.isDecommissioned())
+ .collect(Collectors.toMap(NodeDetails::getNodeId,
+ omNodeDetails -> omNodeDetails,
+ (nodeId, omNodeDetails) -> omNodeDetails));
+ }
+
+ /**
+ * Return all decommissioned nodes in the reloaded config.
+ */
+ public Map<String, OMNodeDetails> getDecommissionedNodesInNewConf() {
+ return omNodesInNewConf.stream()
+ .filter(omNodeDetails -> omNodeDetails.isDecommissioned())
+ .collect(Collectors.toMap(NodeDetails::getNodeId,
+ omNodeDetails -> omNodeDetails,
+ (nodeId, omNodeDetails) -> omNodeDetails));
}
}
diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OMAdminProtocolClientSideImpl.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OMAdminProtocolClientSideImpl.java
index 35ba130..05e73d4 100644
--- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OMAdminProtocolClientSideImpl.java
+++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OMAdminProtocolClientSideImpl.java
@@ -20,14 +20,15 @@ package org.apache.hadoop.ozone.om.protocolPB;
import com.google.protobuf.RpcController;
import com.google.protobuf.ServiceException;
import java.io.IOException;
+import java.util.List;
import java.util.concurrent.TimeUnit;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hdds.conf.ConfigurationSource;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.utils.LegacyHadoopConfigurationSource;
import org.apache.hadoop.io.retry.RetryPolicies;
import org.apache.hadoop.io.retry.RetryPolicy;
import org.apache.hadoop.io.retry.RetryProxy;
+import org.apache.hadoop.ipc.ProtobufHelper;
import org.apache.hadoop.ipc.ProtobufRpcEngine;
import org.apache.hadoop.ipc.RPC;
import org.apache.hadoop.net.NetUtils;
@@ -36,6 +37,11 @@ import org.apache.hadoop.ozone.om.OMConfigKeys;
import org.apache.hadoop.ozone.om.helpers.OMNodeDetails;
import org.apache.hadoop.ozone.om.protocol.OMConfiguration;
import org.apache.hadoop.ozone.om.protocol.OMAdminProtocol;
+import org.apache.hadoop.ozone.om.exceptions.OMLeaderNotReadyException;
+import org.apache.hadoop.ozone.om.exceptions.OMNotLeaderException;
+import org.apache.hadoop.ozone.om.ha.OMFailoverProxyProvider;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerAdminProtocolProtos.DecommissionOMRequest;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerAdminProtocolProtos.DecommissionOMResponse;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerAdminProtocolProtos.OMConfigurationRequest;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerAdminProtocolProtos.OMConfigurationResponse;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerAdminProtocolProtos.OMNodeInfo;
@@ -46,7 +52,7 @@ import org.slf4j.LoggerFactory;
/**
* Protocol implementation for OM admin operations.
*/
-public class OMAdminProtocolClientSideImpl implements OMAdminProtocol {
+public final class OMAdminProtocolClientSideImpl implements OMAdminProtocol {
/**
* RpcController is not used and hence is set to null.
@@ -56,18 +62,27 @@ public class OMAdminProtocolClientSideImpl implements OMAdminProtocol {
private static final Logger LOG =
LoggerFactory.getLogger(OMAdminProtocolClientSideImpl.class);
- private final OMNodeDetails remoteOmNodeDetails;
private final OMAdminProtocolPB rpcProxy;
+ private final String omPrintInfo; // For targeted OM proxy
- public OMAdminProtocolClientSideImpl(ConfigurationSource conf,
- UserGroupInformation ugi, OMNodeDetails omNodeDetails)
- throws IOException {
+ private OMAdminProtocolClientSideImpl(OMAdminProtocolPB proxy,
+ String printInfo) {
+ this.rpcProxy = proxy;
+ this.omPrintInfo = printInfo;
+ }
+
+ /**
+ * Create OM Admin Protocol Client for contacting the given OM (does not
+ * failover to different OM). Use for admin commands such as
+ * getOMConfiguration which are targeted to a specific OM.
+ */
+ public static OMAdminProtocolClientSideImpl createProxyForSingleOM(
+ OzoneConfiguration conf, UserGroupInformation ugi,
+ OMNodeDetails omNodeDetails) throws IOException {
RPC.setProtocolEngine(OzoneConfiguration.of(conf),
OMAdminProtocolPB.class, ProtobufRpcEngine.class);
- this.remoteOmNodeDetails = omNodeDetails;
-
int maxRetries = conf.getInt(
OMConfigKeys.OZONE_OM_ADMIN_PROTOCOL_MAX_RETRIES_KEY,
OMConfigKeys.OZONE_OM_ADMIN_PROTOCOL_MAX_RETRIES_DEFAULT);
@@ -86,7 +101,7 @@ public class OMAdminProtocolClientSideImpl implements OMAdminProtocol {
OMAdminProtocolPB proxy = RPC.getProtocolProxy(
OMAdminProtocolPB.class,
RPC.getProtocolVersion(OMAdminProtocolPB.class),
- remoteOmNodeDetails.getRpcAddress(), ugi, hadoopConf,
+ omNodeDetails.getRpcAddress(), ugi, hadoopConf,
NetUtils.getDefaultSocketFactory(hadoopConf),
(int) OmUtils.getOMClientRpcTimeOut(conf), connectionRetryPolicy)
.getProxy();
@@ -94,8 +109,45 @@ public class OMAdminProtocolClientSideImpl implements OMAdminProtocol {
RetryPolicy retryPolicy = RetryPolicies.retryUpToMaximumCountWithFixedSleep(
10, 1000, TimeUnit.MILLISECONDS);
- this.rpcProxy = (OMAdminProtocolPB) RetryProxy.create(
+ OMAdminProtocolPB rpcProxy = (OMAdminProtocolPB) RetryProxy.create(
OMAdminProtocolPB.class, proxy, retryPolicy);
+
+ return new OMAdminProtocolClientSideImpl(rpcProxy,
+ omNodeDetails.getOMPrintInfo());
+ }
+
+ /**
+ * Create OM Admin Protocol Client for contacting the OM ring (failover
+ * till the current OM leader is reached). Use for admin commands such as
+ * decommissionOM which should reach the OM leader.
+ */
+ public static OMAdminProtocolClientSideImpl createProxyForOMHA(
+ OzoneConfiguration conf, UserGroupInformation ugi, String omServiceId)
+ throws IOException {
+
+ RPC.setProtocolEngine(OzoneConfiguration.of(conf),
+ OMAdminProtocolPB.class, ProtobufRpcEngine.class);
+
+ OMFailoverProxyProvider omFailoverProxyProvider =
+ new OMFailoverProxyProvider(conf, ugi, omServiceId,
+ OMAdminProtocolPB.class);
+
+ // Multiple the max number of retries with number of OMs to calculate the
+ // max number of failovers.
+ int maxFailovers = conf.getInt(
+ OMConfigKeys.OZONE_OM_ADMIN_PROTOCOL_MAX_RETRIES_KEY,
+ OMConfigKeys.OZONE_OM_ADMIN_PROTOCOL_MAX_RETRIES_DEFAULT) *
+ omFailoverProxyProvider.getOMProxies().size();
+
+ OMAdminProtocolPB retryProxy = (OMAdminProtocolPB) RetryProxy.create(
+ OMAdminProtocolPB.class, omFailoverProxyProvider,
+ omFailoverProxyProvider.getRetryPolicy(maxFailovers));
+
+ List<OMNodeDetails> allOMNodeDetails = OmUtils.getAllOMHAAddresses(conf,
+ omServiceId, false);
+
+ return new OMAdminProtocolClientSideImpl(retryProxy,
+ OmUtils.getOMAddressListPrintString(allOMNodeDetails));
}
@Override
@@ -123,13 +175,49 @@ public class OMAdminProtocolClientSideImpl implements OMAdminProtocol {
}
return omMedatataBuilder.build();
} catch (ServiceException e) {
- LOG.error("Failed to retrieve configuration of OM {}",
- remoteOmNodeDetails.getOMPrintInfo(), e);
+ LOG.error("Failed to retrieve configuration of OM {}", omPrintInfo, e);
}
return null;
}
@Override
+ public void decommission(OMNodeDetails removeOMNode) throws IOException {
+ DecommissionOMRequest decommOMRequest = DecommissionOMRequest.newBuilder()
+ .setNodeId(removeOMNode.getNodeId())
+ .setNodeAddress(removeOMNode.getHostAddress())
+ .build();
+
+ DecommissionOMResponse response;
+ try {
+ response = rpcProxy.decommission(NULL_RPC_CONTROLLER, decommOMRequest);
+ } catch (ServiceException e) {
+ OMNotLeaderException notLeaderException =
+ OMFailoverProxyProvider.getNotLeaderException(e);
+ if (notLeaderException != null) {
+ throwException(notLeaderException.getMessage());
+ }
+
+ OMLeaderNotReadyException leaderNotReadyException =
+ OMFailoverProxyProvider.getLeaderNotReadyException(e);
+ if (leaderNotReadyException != null) {
+ throwException(leaderNotReadyException.getMessage());
+ }
+ throw ProtobufHelper.getRemoteException(e);
+ }
+
+ if (!response.getSuccess()) {
+ throwException("Request to decommission" + removeOMNode.getOMPrintInfo() +
+ ", sent to " + omPrintInfo + " failed with error: " +
+ response.getErrorMsg());
+ }
+ }
+
+ private void throwException(String errorMsg)
+ throws IOException {
+ throw new IOException("Failed to Decommission OM. Error: " + errorMsg);
+ }
+
+ @Override
public void close() throws IOException {
RPC.stopProxy(rpcProxy);
}
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestOzoneConfigurationFields.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestOzoneConfigurationFields.java
index 8462caa..2be761e 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestOzoneConfigurationFields.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestOzoneConfigurationFields.java
@@ -59,6 +59,8 @@ public class TestOzoneConfigurationFields extends TestConfigurationFieldsBase {
errorIfMissingXmlProps = true;
xmlPropsToSkipCompare.add("hadoop.tags.custom");
xmlPropsToSkipCompare.add("ozone.om.nodes.EXAMPLEOMSERVICEID");
+ xmlPropsToSkipCompare.add("ozone.om.decommissioned.nodes" +
+ ".EXAMPLEOMSERVICEID");
xmlPropsToSkipCompare.add("ozone.scm.nodes.EXAMPLESCMSERVICEID");
xmlPrefixToSkipCompare.add("ipc.client.rpc-timeout.ms");
xmlPropsToSkipCompare.add("ozone.om.leader.election.minimum.timeout" +
@@ -82,6 +84,7 @@ public class TestOzoneConfigurationFields extends TestConfigurationFieldsBase {
HddsConfigKeys.HDDS_SECURITY_PROVIDER,
HddsConfigKeys.HDDS_X509_CRL_NAME, // HDDS-2873
OMConfigKeys.OZONE_OM_NODES_KEY,
+ OMConfigKeys.OZONE_OM_DECOMMISSIONED_NODES_KEY,
ScmConfigKeys.OZONE_SCM_NODES_KEY,
ScmConfigKeys.OZONE_SCM_ADDRESS_KEY,
OMConfigKeys.OZONE_FS_TRASH_INTERVAL_KEY,
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerBootstrap.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestAddRemoveOzoneManager.java
similarity index 78%
rename from hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerBootstrap.java
rename to hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestAddRemoveOzoneManager.java
index 115c9b3..bb808cf 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerBootstrap.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestAddRemoveOzoneManager.java
@@ -23,6 +23,7 @@ import java.io.File;
import java.io.FileFilter;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
@@ -37,7 +38,11 @@ import org.apache.hadoop.ozone.client.ObjectStore;
import org.apache.hadoop.ozone.client.OzoneBucket;
import org.apache.hadoop.ozone.client.OzoneClientFactory;
import org.apache.hadoop.ozone.client.OzoneVolume;
+import org.apache.hadoop.ozone.om.helpers.OMNodeDetails;
+import org.apache.hadoop.ozone.om.protocolPB.OMAdminProtocolClientSideImpl;
import org.apache.hadoop.ozone.om.ratis.OzoneManagerRatisServer;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.util.StringUtils;
import org.apache.ozone.test.GenericTestUtils;
import org.apache.ratis.grpc.server.GrpcLogAppender;
import org.apache.ratis.server.leader.FollowerInfo;
@@ -56,25 +61,27 @@ import static org.apache.hadoop.ozone.om.TestOzoneManagerHA.createKey;
/**
* Test for OM bootstrap process.
*/
-public class TestOzoneManagerBootstrap {
+public class TestAddRemoveOzoneManager {
@Rule
public ExpectedException exception = ExpectedException.none();
@Rule
- public Timeout timeout = new Timeout(500_000);
+ public Timeout timeout = new Timeout(5000_000);
private MiniOzoneHAClusterImpl cluster = null;
private ObjectStore objectStore;
private OzoneConfiguration conf;
private final String clusterId = UUID.randomUUID().toString();
private final String scmId = UUID.randomUUID().toString();
+ private long lastTransactionIndex;
+ private UserGroupInformation user;
- private static final String OM_SERVICE_ID = "om-bootstrap";
+ private static final String OM_SERVICE_ID = "om-add-remove";
private static final String VOLUME_NAME;
private static final String BUCKET_NAME;
-
- private long lastTransactionIndex;
+ private static final String DECOMM_NODES_CONFIG_KEY =
+ "ozone.om.decommissioned.nodes." + OM_SERVICE_ID;
static {
VOLUME_NAME = "volume" + RandomStringUtils.randomNumeric(5);
@@ -118,12 +125,12 @@ public class TestOzoneManagerBootstrap {
// server's peer list
for (OzoneManager om : cluster.getOzoneManagersList()) {
Assert.assertTrue("New OM node " + nodeId + " not present in Peer list " +
- "of OM " + om.getOMNodeId(), om.doesPeerExist(nodeId));
+ "of OM " + om.getOMNodeId(), om.doesPeerExist(nodeId));
Assert.assertTrue("New OM node " + nodeId + " not present in Peer list " +
- "of OM " + om.getOMNodeId() + " RatisServer",
+ "of OM " + om.getOMNodeId() + " RatisServer",
om.getOmRatisServer().doesPeerExist(nodeId));
Assert.assertTrue("New OM node " + nodeId + " not present in " +
- "OM " + om.getOMNodeId() + "RatisServer's RaftConf",
+ "OM " + om.getOMNodeId() + "RatisServer's RaftConf",
om.getOmRatisServer().getCurrentPeersFromRaftConf().contains(nodeId));
}
@@ -164,8 +171,8 @@ public class TestOzoneManagerBootstrap {
/**
* 1. Add 2 new OMs to an existing 1 node OM cluster.
- * 2. Verify that one of the new OMs must becomes the leader by stopping
- * the old OM.
+ * 2. Verify that one of the new OMs becomes the leader by stopping the old
+ * OM.
*/
@Test
public void testBootstrap() throws Exception {
@@ -333,4 +340,84 @@ public class TestOzoneManagerBootstrap {
// Verify that the newly bootstrapped OM is running
Assert.assertTrue(newOM.isRunning());
}
+
+ /**
+ * Decommissioning Tests:
+ * 1. Stop an OM and decommission it from a 3 node cluster
+ * 2. Decommission another OM without stopping it.
+ * 3.
+ */
+ @Test
+ public void testDecommission() throws Exception {
+ setupCluster(3);
+ user = UserGroupInformation.getCurrentUser();
+
+ // Stop the 3rd OM and decommission it
+ String omNodeId3 = cluster.getOzoneManager(2).getOMNodeId();
+ cluster.stopOzoneManager(omNodeId3);
+ decommissionOM(omNodeId3);
+
+ // Decommission the non leader OM and then stop it. Stopping OM before will
+ // lead to no quorum and there will not be a elected leader OM to process
+ // the decommission request.
+ String omNodeId2;
+ if (cluster.getOMLeader().getOMNodeId().equals(
+ cluster.getOzoneManager(1).getOMNodeId())) {
+ omNodeId2 = cluster.getOzoneManager(0).getOMNodeId();
+ } else {
+ omNodeId2 = cluster.getOzoneManager(1).getOMNodeId();
+ }
+ decommissionOM(omNodeId2);
+ cluster.stopOzoneManager(omNodeId2);
+
+ // Verify that we can read/ write to the cluster with only 1 OM.
+ OzoneVolume volume = objectStore.getVolume(VOLUME_NAME);
+ OzoneBucket bucket = volume.getBucket(BUCKET_NAME);
+ String key = createKey(bucket);
+
+ Assert.assertNotNull(bucket.getKey(key));
+
+ }
+
+ /**
+ * Decommission given OM and verify that the other OM's peer nodes are
+ * updated after decommissioning.
+ */
+ private void decommissionOM(String decommNodeId) throws Exception {
+ Collection<String> decommNodes = conf.getTrimmedStringCollection(
+ DECOMM_NODES_CONFIG_KEY);
+ decommNodes.add(decommNodeId);
+ conf.set(DECOMM_NODES_CONFIG_KEY, StringUtils.join(",", decommNodes));
+ List<OzoneManager> activeOMs = new ArrayList<>();
+ for (OzoneManager om : cluster.getOzoneManagersList()) {
+ String omNodeId = om.getOMNodeId();
+ if (cluster.isOMActive(omNodeId)) {
+ om.setConfiguration(conf);
+ activeOMs.add(om);
+ }
+ }
+
+ // Create OMAdmin protocol client to send decommission request
+ OMAdminProtocolClientSideImpl omAdminProtocolClient =
+ OMAdminProtocolClientSideImpl.createProxyForOMHA(conf, user,
+ OM_SERVICE_ID);
+ OMNodeDetails decommNodeDetails = new OMNodeDetails.Builder()
+ .setOMNodeId(decommNodeId)
+ .setHostAddress("localhost")
+ .build();
+ omAdminProtocolClient.decommission(decommNodeDetails);
+
+ // Verify decomm node is removed from the HA ring
+ GenericTestUtils.waitFor(() -> {
+ for (OzoneManager om : activeOMs) {
+ if (om.getPeerNodes().contains(decommNodeId)) {
+ return false;
+ }
+ }
+ return true;
+ }, 100, 100000);
+
+ // Wait for new leader election if required
+ GenericTestUtils.waitFor(() -> cluster.getOMLeader() != null, 500, 30000);
+ }
}
diff --git a/hadoop-ozone/interface-client/src/main/proto/OMAdminProtocol.proto b/hadoop-ozone/interface-client/src/main/proto/OMAdminProtocol.proto
index bfba946..60e8ea1 100644
--- a/hadoop-ozone/interface-client/src/main/proto/OMAdminProtocol.proto
+++ b/hadoop-ozone/interface-client/src/main/proto/OMAdminProtocol.proto
@@ -30,8 +30,8 @@ option java_generate_equals_and_hash = true;
package hadoop.ozone;
/**
-This file contains the admin protocol for Ozone Manager(s). This involves
-getting the meta information about an individual OM.
+This file contains the admin protocol for Ozone Manager(s). These
+communications should be instantiated only via the Admin Cli or through an OM.
*/
message OMConfigurationRequest {
@@ -40,11 +40,10 @@ message OMConfigurationRequest {
message OMConfigurationResponse {
required bool success = 1;
optional string errorMsg = 2;
- // OM nodes present in OM's memory
+ // OM nodes present in OM's memory (does not include Decommissioned nodes)
repeated OMNodeInfo nodesInMemory = 3;
- // OM nodes reloaded from new config on disk
+ // OM nodes reloaded from new config on disk (includes Decommissioned nodes)
repeated OMNodeInfo nodesInNewConf = 4;
-
}
message OMNodeInfo {
@@ -52,6 +51,22 @@ message OMNodeInfo {
required string hostAddress = 2;
required uint32 rpcPort = 3;
required uint32 ratisPort = 4;
+ optional NodeState nodeState = 5 [default=ACTIVE];
+}
+
+enum NodeState {
+ ACTIVE = 1;
+ DECOMMISSIONED = 2;
+}
+
+message DecommissionOMRequest {
+ required string nodeId = 1;
+ required string nodeAddress = 2;
+}
+
+message DecommissionOMResponse {
+ required bool success = 1;
+ optional string errorMsg = 3;
}
/**
@@ -62,4 +77,8 @@ service OzoneManagerAdminService {
// and the anticipated nodes list from the config files (upon reloading).
rpc getOMConfiguration(OMConfigurationRequest)
returns(OMConfigurationResponse);
+
+ // RPC request from admin to remove an OM from the cluster
+ rpc decommission(DecommissionOMRequest)
+ returns(DecommissionOMResponse);
}
\ No newline at end of file
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManager.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManager.java
index 8077d07..af30947 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManager.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManager.java
@@ -1055,7 +1055,7 @@ public final class OzoneManager extends ServiceRuntimeInfoImpl
private RPC.Server startRpcServer(OzoneConfiguration conf,
InetSocketAddress addr, BlockingService clientProtocolService,
BlockingService interOMProtocolService,
- BlockingService omMetadataProtocolService,
+ BlockingService omAdminProtocolService,
int handlerCount)
throws IOException {
RPC.Server rpcServer = new RPC.Builder(conf)
@@ -1071,7 +1071,7 @@ public final class OzoneManager extends ServiceRuntimeInfoImpl
HddsServerUtil.addPBProtocol(conf, OMInterServiceProtocolPB.class,
interOMProtocolService, rpcServer);
HddsServerUtil.addPBProtocol(conf, OMAdminProtocolPB.class,
- omMetadataProtocolService, rpcServer);
+ omAdminProtocolService, rpcServer);
if (conf.getBoolean(CommonConfigurationKeys.HADOOP_SECURITY_AUTHORIZATION,
false)) {
@@ -1511,12 +1511,12 @@ public final class OzoneManager extends ServiceRuntimeInfoImpl
for (Map.Entry<String, OMNodeDetails> entry : peerNodesMap.entrySet()) {
String remoteNodeId = entry.getKey();
OMNodeDetails remoteNodeDetails = entry.getValue();
- try (OMAdminProtocolClientSideImpl omMetadataProtocolClient =
- new OMAdminProtocolClientSideImpl(configuration,
- getRemoteUser(), entry.getValue())) {
+ try (OMAdminProtocolClientSideImpl omAdminProtocolClient =
+ OMAdminProtocolClientSideImpl.createProxyForSingleOM(
+ configuration, getRemoteUser(), entry.getValue())) {
OMConfiguration remoteOMConfiguration =
- omMetadataProtocolClient.getOMConfiguration();
+ omAdminProtocolClient.getOMConfiguration();
checkRemoteOMConfig(remoteNodeId, remoteOMConfiguration);
} catch (IOException ioe) {
LOG.error("Remote OM config check failed on OM {}", remoteNodeId, ioe);
@@ -1549,7 +1549,7 @@ public final class OzoneManager extends ServiceRuntimeInfoImpl
}
OMNodeDetails omNodeDetailsInRemoteConfig = remoteOMConfig
- .getOmNodesInNewConf().get(getOMNodeId());
+ .getActiveOmNodesInNewConf().get(getOMNodeId());
if (omNodeDetailsInRemoteConfig == null) {
throw new IOException("Remote OM " + remoteNodeId + " does not have the" +
" bootstrapping OM(" + getOMNodeId() + ") information on reloading " +
@@ -1563,6 +1563,12 @@ public final class OzoneManager extends ServiceRuntimeInfoImpl
omNodeDetailsInRemoteConfig.getRpcAddress() + " where the " +
"bootstrapping OM address is " + omNodeDetails.getRpcAddress());
}
+
+ if (omNodeDetailsInRemoteConfig.isDecommissioned()) {
+ throw new IOException("Remote OM " + remoteNodeId + " configuration has" +
+ " bootstrapping OM(" + getOMNodeId() + ") in decommissioned " +
+ "nodes list - " + OMConfigKeys.OZONE_OM_DECOMMISSIONED_NODES_KEY);
+ }
}
@Override
@@ -1590,37 +1596,70 @@ public final class OzoneManager extends ServiceRuntimeInfoImpl
/**
* When OMStateMachine receives a configuration change update, it calls
- * this function to update the peers list, if required.
- */
- public void updatePeerList(List<String> omNodeIds) {
- List<String> ratisServerPeerIdsList = omRatisServer.getPeerIds();
- for (String omNodeId : omNodeIds) {
- // Check if the OM NodeID is already present in the peer list or its
- // the local NodeID.
- if (!peerNodesMap.containsKey(omNodeId) && !isCurrentNode(omNodeId)) {
+ * this function to update the peers list, if required. The configuration
+ * change could be to add or to remove an OM from the ring.
+ */
+ public void updatePeerList(List<String> newPeers) {
+ List<String> currentPeers = omRatisServer.getPeerIds();
+
+ // NodeIds present in new node list and not in current peer list are the
+ // bootstapped OMs and should be added to the peer list
+ List<String> bootstrappedOMs = new ArrayList<>();
+ bootstrappedOMs.addAll(newPeers);
+ bootstrappedOMs.removeAll(currentPeers);
+
+ // NodeIds present in current peer list but not in new node list are the
+ // decommissioned OMs and should be removed from the peer list
+ List<String> decommissionedOMs = new ArrayList<>();
+ decommissionedOMs.addAll(currentPeers);
+ decommissionedOMs.removeAll(newPeers);
+
+ // Add bootstrapped OMs to peer list
+ for (String omNodeId : bootstrappedOMs) {
+ // Check if its the local nodeId (bootstrapping OM)
+ if (isCurrentNode(omNodeId)) {
+ // For a Bootstrapping OM, none of the peers are added to it's
+ // RatisServer's peer list and it needs to be updated here after
+ // receiving the conf change notification from Ratis.
+ for (String peerNodeId : newPeers) {
+ if (peerNodeId.equals(omNodeId)) {
+ omRatisServer.addRaftPeer(omNodeDetails);
+ } else {
+ omRatisServer.addRaftPeer(peerNodesMap.get(peerNodeId));
+ }
+ }
+ } else {
+ // For other nodes, add bootstrapping OM to OM peer list (which
+ // internally adds to Ratis peer list too)
try {
addOMNodeToPeers(omNodeId);
} catch (IOException e) {
- LOG.error("Fatal Error: Shutting down the system as otherwise it " +
+ LOG.error("Fatal Error while adding bootstrapped node to " +
+ "peer list. Shutting down the system as otherwise it " +
"could lead to OM state divergence.", e);
exitManager.forceExit(1, e, LOG);
}
+ }
+ }
+
+ // Remove decommissioned OMs from peer list
+ for (String omNodeId : decommissionedOMs) {
+ if (isCurrentNode(omNodeId)) {
+ // Decommissioning Node should not receive the configuration change
+ // request. Shut it down.
+ String errorMsg = "Shutting down as OM has been decommissioned.";
+ LOG.error("Fatal Error: {}", errorMsg);
+ exitManager.forceExit(1, errorMsg, LOG);
} else {
- // Check if the OMNodeID is present in the RatisServer's peer list
- if (!ratisServerPeerIdsList.contains(omNodeId)) {
- // This can happen on a bootstrapping OM. The peer information
- // would be present in OzoneManager but OMRatisServer peer list
- // would not have the peers list. OMRatisServer peer list of
- // bootstrapping node should be updated after it gets the RaftConf
- // through Ratis.
- if (isCurrentNode(omNodeId)) {
- // OM Ratis server has the current node also in the peer list as
- // this is the Raft Group peers list. Hence, add the current node
- // also to Ratis peers list if not present.
- omRatisServer.addRaftPeer(omNodeDetails);
- } else {
- omRatisServer.addRaftPeer(peerNodesMap.get(omNodeId));
- }
+ // Remove decommissioned node from peer list (which internally
+ // removed from Ratis peer list too)
+ try {
+ removeOMNodeFromPeers(omNodeId);
+ } catch (IOException e) {
+ LOG.error("Fatal Error while removing decommissioned node from " +
+ "peer list. Shutting down the system as otherwise it " +
+ "could lead to OM state divergence.", e);
+ exitManager.forceExit(1, e, LOG);
}
}
}
@@ -1674,6 +1713,24 @@ public final class OzoneManager extends ServiceRuntimeInfoImpl
}
/**
+ * Remove an OM Node from the peers list. This call comes from OMStateMachine
+ * after a SetConfiguration request has been successfully executed by the
+ * Ratis server.
+ */
+ private void removeOMNodeFromPeers(String decommNodeId) throws IOException {
+ OMNodeDetails decommOMNodeDetails = peerNodesMap.get(decommNodeId);
+ if (decommOMNodeDetails == null) {
+ throw new IOException("Decommissioned Node " + decommNodeId + " not " +
+ "present in peer list");
+ }
+
+ omSnapshotProvider.removeDecommissionedPeerNode(decommNodeId);
+ omRatisServer.removeRaftPeer(decommOMNodeDetails);
+ peerNodesMap.remove(decommNodeId);
+ LOG.info("Removed OM {} from OM Peer Nodes.", decommNodeId);
+ }
+
+ /**
* Check if the input nodeId exists in the peers list.
* @return true if the nodeId is self or it exists in peer node list,
* false otherwise.
@@ -1706,7 +1763,14 @@ public final class OzoneManager extends ServiceRuntimeInfoImpl
*/
public List<OMNodeDetails> getAllOMNodesInNewConf() {
OzoneConfiguration newConf = reloadConfiguration();
- return OmUtils.getAllOMAddresses(newConf, getOMServiceId(), getOMNodeId());
+ List<OMNodeDetails> allOMNodeDetails = OmUtils.getAllOMHAAddresses(
+ newConf, getOMServiceId(), true);
+ if (allOMNodeDetails.isEmpty()) {
+ // There are no addresses configured for HA. Return only current OM
+ // details.
+ return Collections.singletonList(omNodeDetails);
+ }
+ return allOMNodeDetails;
}
/**
@@ -3372,6 +3436,10 @@ public final class OzoneManager extends ServiceRuntimeInfoImpl
return new ArrayList<>(peerNodesMap.values());
}
+ public OMNodeDetails getPeerNode(String nodeID) {
+ return peerNodesMap.get(nodeID);
+ }
+
@VisibleForTesting
public CertificateClient getCertificateClient() {
return certClient;
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ha/OMHANodeDetails.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ha/OMHANodeDetails.java
index 41f33f1..d04266a 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ha/OMHANodeDetails.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ha/OMHANodeDetails.java
@@ -124,7 +124,8 @@ public class OMHANodeDetails {
boolean isOMAddressSet = false;
for (String serviceId : omServiceIds) {
- Collection<String> omNodeIds = OmUtils.getOMNodeIds(conf, serviceId);
+ Collection<String> omNodeIds = OmUtils.getActiveOMNodeIds(conf,
+ serviceId);
if (omNodeIds.size() == 0) {
throwConfException("Configuration does not have any value set for %s " +
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerRatisServer.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerRatisServer.java
index 918d99b..ba7bd47 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerRatisServer.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerRatisServer.java
@@ -20,7 +20,7 @@ package org.apache.hadoop.ozone.om.ratis;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
-import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
import com.google.protobuf.InvalidProtocolBufferException;
import com.google.protobuf.ServiceException;
@@ -106,7 +106,7 @@ public final class OzoneManagerRatisServer {
private final RaftGroupId raftGroupId;
private final RaftGroup raftGroup;
private final RaftPeerId raftPeerId;
- private final List<RaftPeer> raftPeers;
+ private final Map<String, RaftPeer> raftPeerMap;
private final OzoneManager ozoneManager;
private final OzoneManagerStateMachine omStateMachine;
@@ -144,8 +144,8 @@ public final class OzoneManagerRatisServer {
this.raftPeerId = localRaftPeerId;
this.raftGroupId = RaftGroupId.valueOf(
getRaftGroupIdFromOmServiceId(raftGroupIdStr));
- this.raftPeers = Lists.newArrayList();
- this.raftPeers.addAll(peers);
+ this.raftPeerMap = Maps.newHashMap();
+ peers.forEach(e -> raftPeerMap.put(e.getId().toString(), e));
this.raftGroup = RaftGroup.valueOf(raftGroupId, peers);
if (isBootstrapping) {
@@ -311,7 +311,7 @@ public final class OzoneManagerRatisServer {
newRaftPeer, raftGroup);
List<RaftPeer> newPeersList = new ArrayList<>();
- newPeersList.addAll(raftPeers);
+ newPeersList.addAll(raftPeerMap.values());
newPeersList.add(newRaftPeer);
checkLeaderStatus();
@@ -331,13 +331,44 @@ public final class OzoneManagerRatisServer {
}
/**
+ * Remove decommissioned OM node from Ratis ring.
+ */
+ public void removeOMFromRatisRing(OMNodeDetails removeOMNode)
+ throws IOException {
+ Preconditions.checkNotNull(removeOMNode);
+
+ String removeNodeId = removeOMNode.getNodeId();
+ LOG.info("{}: Submitting SetConfiguration request to Ratis server to " +
+ "remove OM peer {} from Ratis group {}", ozoneManager.getOMNodeId(),
+ removeNodeId, raftGroup);
+
+ List<RaftPeer> newPeersList = new ArrayList<>();
+ newPeersList.addAll(raftPeerMap.values());
+ newPeersList.remove(raftPeerMap.get(removeNodeId));
+
+ checkLeaderStatus();
+ SetConfigurationRequest request = new SetConfigurationRequest(clientId,
+ server.getId(), raftGroupId, nextCallId(), newPeersList);
+
+ RaftClientReply raftClientReply = server.setConfiguration(request);
+ if (raftClientReply.isSuccess()) {
+ LOG.info("Removed OM {} from Ratis group {}.", removeNodeId,
+ raftGroupId);
+ } else {
+ LOG.error("Failed to remove OM {} from Ratis group {}. Ratis " +
+ "SetConfiguration reply: {}", removeNodeId, raftGroupId,
+ raftClientReply);
+ throw new IOException("Failed to remove OM " + removeNodeId + " from " +
+ "Ratis ring.");
+ }
+ }
+
+ /**
* Return a list of peer NodeIds.
*/
public List<String> getPeerIds() {
List<String> peerIds = new ArrayList<>();
- for (RaftPeer raftPeer : raftPeers) {
- peerIds.add(raftPeer.getId().toString());
- }
+ peerIds.addAll(raftPeerMap.keySet());
return peerIds;
}
@@ -348,12 +379,7 @@ public final class OzoneManagerRatisServer {
*/
@VisibleForTesting
public boolean doesPeerExist(String peerId) {
- for (RaftPeer raftPeer : raftPeers) {
- if (raftPeer.getId().toString().equals(peerId)) {
- return true;
- }
- }
- return false;
+ return raftPeerMap.containsKey(peerId);
}
/**
@@ -363,12 +389,24 @@ public final class OzoneManagerRatisServer {
InetSocketAddress newOMRatisAddr = new InetSocketAddress(
omNodeDetails.getHostAddress(), omNodeDetails.getRatisPort());
- raftPeers.add(RaftPeer.newBuilder()
- .setId(RaftPeerId.valueOf(omNodeDetails.getNodeId()))
+ String newNodeId = omNodeDetails.getNodeId();
+ RaftPeerId newPeerId = RaftPeerId.valueOf(newNodeId);
+ RaftPeer raftPeer = RaftPeer.newBuilder()
+ .setId(newPeerId)
.setAddress(newOMRatisAddr)
- .build());
+ .build();
+ raftPeerMap.put(newNodeId, raftPeer);
+
+ LOG.info("Added OM {} to Ratis Peers list.", newNodeId);
+ }
- LOG.info("Added OM {} to Ratis Peers list.", omNodeDetails.getNodeId());
+ /**
+ * Remove given node from list of RaftPeers.
+ */
+ public void removeRaftPeer(OMNodeDetails omNodeDetails) {
+ String removeNodeID = omNodeDetails.getNodeId();
+ raftPeerMap.remove(removeNodeID);
+ LOG.info("{}: Removed OM {} from Ratis Peers list.", this, removeNodeID);
}
/**
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/OzoneManagerSnapshotProvider.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/OzoneManagerSnapshotProvider.java
index b24ca4b..01fade0 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/OzoneManagerSnapshotProvider.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/OzoneManagerSnapshotProvider.java
@@ -163,4 +163,11 @@ public class OzoneManagerSnapshotProvider {
public void addNewPeerNode(OMNodeDetails newOMNode) {
peerNodesMap.put(newOMNode.getNodeId(), newOMNode);
}
+
+ /**
+ * When an OM is decommissioned, remove it from the peerNode map.
+ */
+ public void removeDecommissionedPeerNode(String decommNodeId) {
+ peerNodesMap.remove(decommNodeId);
+ }
}
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/protocolPB/OMAdminProtocolServerSideImpl.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/protocolPB/OMAdminProtocolServerSideImpl.java
index efc9f0b..f419301 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/protocolPB/OMAdminProtocolServerSideImpl.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/protocolPB/OMAdminProtocolServerSideImpl.java
@@ -18,11 +18,16 @@ package org.apache.hadoop.ozone.protocolPB;
import com.google.protobuf.RpcController;
import com.google.protobuf.ServiceException;
+import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import org.apache.hadoop.ozone.om.OzoneManager;
import org.apache.hadoop.ozone.om.helpers.OMNodeDetails;
import org.apache.hadoop.ozone.om.protocolPB.OMAdminProtocolPB;
+import org.apache.hadoop.ozone.om.ratis.OzoneManagerRatisServer;
+import org.apache.hadoop.ozone.om.ratis.utils.OzoneManagerRatisUtils;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerAdminProtocolProtos.DecommissionOMRequest;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerAdminProtocolProtos.DecommissionOMResponse;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerAdminProtocolProtos.OMConfigurationRequest;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerAdminProtocolProtos.OMConfigurationResponse;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerAdminProtocolProtos.OMNodeInfo;
@@ -63,4 +68,49 @@ public class OMAdminProtocolServerSideImpl implements OMAdminProtocolPB {
.addAllNodesInNewConf(omNodesInNewConf)
.build();
}
+
+ @Override
+ public DecommissionOMResponse decommission(RpcController controller,
+ DecommissionOMRequest request) throws ServiceException {
+ if (request == null) {
+ return null;
+ }
+ if (!ozoneManager.isRatisEnabled()) {
+ return DecommissionOMResponse.newBuilder()
+ .setSuccess(false)
+ .setErrorMsg("OM node cannot be decommissioned as Ratis is " +
+ "not enabled.")
+ .build();
+ }
+
+ OzoneManagerRatisServer omRatisServer = ozoneManager.getOmRatisServer();
+ checkLeaderStatus(omRatisServer);
+
+ OMNodeDetails decommNode = ozoneManager.getPeerNode(request.getNodeId());
+ if (decommNode == null) {
+ return DecommissionOMResponse.newBuilder()
+ .setSuccess(false)
+ .setErrorMsg("OM node not present in the OM peer list.")
+ .build();
+ }
+
+ try {
+ omRatisServer.removeOMFromRatisRing(decommNode);
+ } catch (IOException ex) {
+ return DecommissionOMResponse.newBuilder()
+ .setSuccess(false)
+ .setErrorMsg(ex.getMessage())
+ .build();
+ }
+
+ return DecommissionOMResponse.newBuilder()
+ .setSuccess(true)
+ .build();
+ }
+
+ private void checkLeaderStatus(OzoneManagerRatisServer omRatisServer)
+ throws ServiceException {
+ OzoneManagerRatisUtils.checkLeaderStatus(omRatisServer.checkLeaderStatus(),
+ omRatisServer.getRaftPeerId());
+ }
}
diff --git a/hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/OzoneServiceProvider.java b/hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/OzoneServiceProvider.java
index 0530fa6..351163c 100644
--- a/hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/OzoneServiceProvider.java
+++ b/hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/OzoneServiceProvider.java
@@ -59,7 +59,8 @@ public class OzoneServiceProvider {
"configured. " + Arrays.toString(serviceIdList.toArray()));
} else {
String serviceId = serviceIdList.iterator().next();
- Collection<String> omNodeIds = OmUtils.getOMNodeIds(conf, serviceId);
+ Collection<String> omNodeIds = OmUtils.getActiveOMNodeIds(conf,
+ serviceId);
if (omNodeIds.size() == 0) {
throw new IllegalArgumentException(OZONE_OM_NODES_KEY
+ "." + serviceId + " is not defined");
diff --git a/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/admin/om/DecommissionOMSubcommand.java b/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/admin/om/DecommissionOMSubcommand.java
new file mode 100644
index 0000000..455b3ca
--- /dev/null
+++ b/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/admin/om/DecommissionOMSubcommand.java
@@ -0,0 +1,212 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ozone.admin.om;
+
+import java.io.IOException;
+import java.net.InetAddress;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import org.apache.hadoop.hdds.cli.HddsVersionProvider;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.ozone.OmUtils;
+import org.apache.hadoop.ozone.ha.ConfUtils;
+import org.apache.hadoop.ozone.om.helpers.OMNodeDetails;
+import org.apache.hadoop.ozone.om.protocol.OMConfiguration;
+import org.apache.hadoop.ozone.om.protocolPB.OMAdminProtocolClientSideImpl;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.logging.log4j.util.Strings;
+import picocli.CommandLine;
+
+import java.util.concurrent.Callable;
+
+import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_ADDRESS_KEY;
+import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_DECOMMISSIONED_NODES_KEY;
+
+/**
+ * Handler of om roles command.
+ */
+@CommandLine.Command(
+ name = "decommission",
+ customSynopsis = "ozone admin om decommission -id=<om-service-id> " +
+ "-nodeid=<decommission-om-node-id> " +
+ "-hostname=<decommission-om-node-address> [options]",
+ description = "Decommission an OzoneManager. Ensure that the node being " +
+ "decommissioned is shutdown first." +
+ "\nNote - Add the node to be decommissioned to " +
+ OZONE_OM_DECOMMISSIONED_NODES_KEY + "config in ozone-site.xml of all " +
+ "OzoneManagers before proceeding with decommission." +
+ "\nNote - DECOMMISSIONING AN OM MIGHT RENDER THE CLUSTER TO LOSE " +
+ "HIGH AVAILABILITY." +
+ "\nNote - When there are only two OzoneManagers, do not stop the " +
+ "OzoneManager before decommissioning as both OzoneManagers are " +
+ "required to reach quorum." + "\n" +
+ "It is recommended to add another OzoneManager(s) before " +
+ "decommissioning one to maintain HA.",
+ mixinStandardHelpOptions = true,
+ versionProvider = HddsVersionProvider.class)
+public class DecommissionOMSubcommand implements Callable<Void> {
+
+ @CommandLine.ParentCommand
+ private OMAdmin parent;
+
+ @CommandLine.Option(names = {"-id", "--service-id"},
+ description = "OM Service ID",
+ required = true)
+ private String omServiceId;
+
+ @CommandLine.Option(names = {"-nodeid"},
+ description = "NodeID of the OM to be decommissioned.",
+ required = true)
+ private String decommNodeId;
+
+ @CommandLine.Option(names = {"-hostname", "--node-host-address"},
+ description = "Host name/address of the OM to be decommissioned.",
+ required = true)
+ private String hostname;
+
+ private InetAddress hostInetAddress;
+
+ @CommandLine.Option(
+ names = {"--force"},
+ description = "This option will skip checking whether OM configs " +
+ "have been updated with the decommissioned node added to " +
+ "ozone.om.decommissioned.nodes config in ozone-site.xml."
+ )
+ private boolean force;
+
+ private OzoneConfiguration ozoneConf;
+ private UserGroupInformation user;
+
+ @Override
+ public Void call() throws IOException {
+ ozoneConf = parent.getParent().getOzoneConf();
+ user = parent.getParent().getUser();
+
+ verifyNodeIdAndHostAddress();
+ if (!force) {
+ verifyConfigUpdatedOnAllOMs();
+ }
+
+ // Proceed with decommissioning the OM by contacting the current OM
+ // leader.
+ try (OMAdminProtocolClientSideImpl omAdminProtocolClient =
+ OMAdminProtocolClientSideImpl.createProxyForOMHA(ozoneConf, user,
+ omServiceId)) {
+ OMNodeDetails decommNodeDetails = new OMNodeDetails.Builder()
+ .setOMNodeId(decommNodeId)
+ .setHostAddress(hostInetAddress.getHostAddress())
+ .build();
+ omAdminProtocolClient.decommission(decommNodeDetails);
+
+ System.out.println("Successfully decommissioned OM " + decommNodeId);
+ } catch (IOException e) {
+ System.out.println("Failed to decommission OM " + decommNodeId);
+ throw e;
+ }
+ return null;
+ }
+
+ /**
+ * Verify that the provided nodeId and host address correspond to the same
+ * OM in the configs.
+ */
+ private void verifyNodeIdAndHostAddress() throws IOException {
+ String rpcAddrKey = ConfUtils.addKeySuffixes(OZONE_OM_ADDRESS_KEY,
+ omServiceId, decommNodeId);
+ String rpcAddrStr = OmUtils.getOmRpcAddress(ozoneConf, rpcAddrKey);
+ if (rpcAddrStr == null || rpcAddrStr.isEmpty()) {
+ throw new IOException("There is no OM corresponding to " + decommNodeId
+ + "in the configuration.");
+ }
+
+ hostInetAddress = InetAddress.getByName(hostname);
+ InetAddress rpcAddressFromConfig = InetAddress.getByName(
+ rpcAddrStr.split(":")[0]);
+
+ if (!hostInetAddress.equals(rpcAddressFromConfig)) {
+ throw new IOException("OM " + decommNodeId + "'s host address in " +
+ "config - " + rpcAddressFromConfig.getHostAddress() + " does not " +
+ "match the provided host address " + hostInetAddress);
+ }
+ }
+
+ /**
+ * Verify that the to be decommissioned node is added to the
+ * OZONE_OM_DECOMMISSIONED_NODES_KEY.<SERVICE_ID> config in ozone-site.xml
+ * of all OMs.
+ */
+ private void verifyConfigUpdatedOnAllOMs() throws IOException {
+ String decommNodesKey = ConfUtils.addKeySuffixes(
+ OZONE_OM_DECOMMISSIONED_NODES_KEY, omServiceId);
+ Collection<String> decommNodes = ozoneConf.getTrimmedStringCollection(
+ decommNodesKey);
+ if (!decommNodes.contains(decommNodeId)) {
+ throw new IOException("Please add the to be decommissioned OM "
+ + decommNodeId + " to the " + decommNodesKey + " config in " +
+ "ozone-site.xml of all nodes.");
+ }
+
+ // For each OM, we need to get the reloaded config and check that the
+ // decommissioned node is either removed from ozone.om.nodes config or
+ // added to ozone.om.decommissioned.nodes
+ List<OMNodeDetails> activeOMNodeDetails = OmUtils.getAllOMHAAddresses(
+ ozoneConf, omServiceId, false);
+ if (activeOMNodeDetails.isEmpty()) {
+ throw new IOException("Cannot decommission OM " + decommNodeId + " as " +
+ "it is the only node in the ring.");
+ }
+
+ List<String> staleOMConfigs = new ArrayList<>();
+ for (OMNodeDetails nodeDetails : activeOMNodeDetails) {
+ if (!checkOMConfig(nodeDetails)) {
+ staleOMConfigs.add(nodeDetails.getNodeId());
+ }
+ }
+ if (!staleOMConfigs.isEmpty()) {
+ throw new IOException("OM(s) " + Strings.join(staleOMConfigs, ',') +
+ " have not been updated with decommissioned nodes list or their" +
+ " address for the decommissioning node does not match");
+ }
+ }
+
+ /**
+ * Check whether the to be decommissioned node is added to the
+ * OZONE_OM_DECOMMISSIONED_NODES_KEY.<SERVICE_ID> config in ozone-site.xml
+ * of given OM.
+ */
+ private boolean checkOMConfig(OMNodeDetails omNodeDetails)
+ throws IOException {
+ try (OMAdminProtocolClientSideImpl omAdminProtocolClient =
+ OMAdminProtocolClientSideImpl.createProxyForSingleOM(ozoneConf,
+ user, omNodeDetails)) {
+ OMConfiguration omConfig = omAdminProtocolClient.getOMConfiguration();
+ OMNodeDetails decommNodeDetails = omConfig
+ .getDecommissionedNodesInNewConf().get(decommNodeId);
+ if (decommNodeDetails == null) {
+ return false;
+ }
+ if (!decommNodeDetails.getRpcAddress().getAddress().equals(
+ hostInetAddress)) {
+ return false;
+ }
+ }
+ return true;
+ }
+}
diff --git a/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/admin/om/OMAdmin.java b/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/admin/om/OMAdmin.java
index a11ce25..ac03b46 100644
--- a/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/admin/om/OMAdmin.java
+++ b/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/admin/om/OMAdmin.java
@@ -32,7 +32,6 @@ import org.apache.hadoop.ozone.om.protocolPB.Hadoop3OmTransportFactory;
import org.apache.hadoop.ozone.om.protocolPB.OmTransport;
import org.apache.hadoop.ozone.om.protocolPB.OzoneManagerProtocolClientSideTranslatorPB;
import org.apache.hadoop.ozone.om.protocolPB.OzoneManagerProtocolPB;
-import org.apache.hadoop.security.UserGroupInformation;
import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_ADDRESS_KEY;
import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_SERVICE_IDS_KEY;
@@ -57,7 +56,8 @@ import java.util.Collection;
GetServiceRolesSubcommand.class,
PrepareSubCommand.class,
CancelPrepareSubCommand.class,
- FinalizationStatusSubCommand.class
+ FinalizationStatusSubCommand.class,
+ DecommissionOMSubcommand.class
})
@MetaInfServices(SubcommandWithParent.class)
public class OMAdmin extends GenericCli implements SubcommandWithParent {
@@ -110,13 +110,12 @@ public class OMAdmin extends GenericCli implements SubcommandWithParent {
} else if (omServiceID == null || omServiceID.isEmpty()) {
omServiceID = getTheOnlyConfiguredOmServiceIdOrThrow();
}
- UserGroupInformation ugi = UserGroupInformation.getCurrentUser();
RPC.setProtocolEngine(conf, OzoneManagerProtocolPB.class,
ProtobufRpcEngine.class);
String clientId = ClientId.randomId().toString();
if (!forceHA || (forceHA && OmUtils.isOmHAServiceId(conf, omServiceID))) {
OmTransport omTransport = new Hadoop3OmTransportFactory()
- .createOmTransport(conf, ugi, omServiceID);
+ .createOmTransport(conf, parent.getUser(), omServiceID);
return new OzoneManagerProtocolClientSideTranslatorPB(omTransport,
clientId);
} else {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@ozone.apache.org
For additional commands, e-mail: commits-help@ozone.apache.org