diff --git a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/proxy/SCMFailoverProxyProviderBase.java b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/proxy/SCMFailoverProxyProviderBase.java index 05bcbf6bffa2..f1530c047a2d 100644 --- a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/proxy/SCMFailoverProxyProviderBase.java +++ b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/proxy/SCMFailoverProxyProviderBase.java @@ -175,8 +175,8 @@ synchronized void replaceProxyInfoForTest(String nodeId, SCMProxyInfo info) { @VisibleForTesting protected synchronized void loadConfigs() { List scmNodeInfoList = SCMNodeInfo.buildNodeInfo(conf); - scmNodeIds = new ArrayList<>(); - + List newScmNodeIds = new ArrayList<>(); + Map newScmProxyInfoMap = new HashMap<>(); for (SCMNodeInfo scmNodeInfo : scmNodeInfoList) { String protocolAddress = getProtocolAddress(scmNodeInfo); @@ -188,16 +188,82 @@ protected synchronized void loadConfigs() { String scmServiceId = scmNodeInfo.getServiceId(); String scmNodeId = scmNodeInfo.getNodeId(); - scmNodeIds.add(scmNodeId); + newScmNodeIds.add(scmNodeId); // Preserve the original config string so DNS can be re-resolved // on connection failure when the SCM peer is rescheduled to a // new IP (Kubernetes pod-IP-change recovery). See // refreshProxyAddressIfChanged(String). SCMProxyInfo scmProxyInfo = new SCMProxyInfo(scmServiceId, scmNodeId, protocolAddr, protocolAddress); - scmProxyInfoMap.put(scmNodeId, scmProxyInfo); + newScmProxyInfoMap.put(scmNodeId, scmProxyInfo); + } + } + + // Commit only after the whole configuration parsed successfully. A dynamic + // reconfiguration that adds an SCM needs two properties updated (the node + // list and the new node's address) and they can be applied in either order; + // if the node list is updated first, buildNodeInfo above throws and the + // previous state is left intact so the operator can retry. + scmNodeIds = newScmNodeIds; + scmProxyInfoMap.clear(); + scmProxyInfoMap.putAll(newScmProxyInfoMap); + } + + /** + * Reload the SCM node list and their addresses from the (already updated) + * configuration. Used for dynamic reconfiguration of + * {@code ozone.scm.nodes.} and + * {@code ozone.scm.address..} so that a newly added SCM + * can be reached without restarting the service. Cached proxies for removed + * nodes, or nodes whose address changed, are stopped so that the next call + * dials the fresh address. If the new configuration is incomplete this throws + * and leaves the current state intact. + */ + public synchronized void changeConfig() { + Map oldProxyInfoMap = new HashMap<>(scmProxyInfoMap); + loadConfigs(); + + // Keep the current proxy pointer valid before touching any proxy: if the + // node it referenced was removed (or the list shrank), fall back to the + // first node. Otherwise keep pointing to the same node but re-sync the index + // to the rebuilt list. Doing this before stopping stale proxies means a + // stopProxy failure cannot leave the pointer naming a node absent from the + // rebuilt map. + if (!scmNodeIds.contains(currentProxySCMNodeId)) { + currentProxyIndex = 0; + currentProxySCMNodeId = scmNodeIds.get(currentProxyIndex); + } else { + currentProxyIndex = scmNodeIds.indexOf(currentProxySCMNodeId); + } + + // A pending failover target (set on a retriable-no-failover error) may name + // a node that this reload removed; clear it so performFailover does not + // point at a node absent from the rebuilt proxy map, which would NPE in + // createSCMProxy on the next failover. + if (updatedLeaderNodeID != null + && !scmProxyInfoMap.containsKey(updatedLeaderNodeID)) { + updatedLeaderNodeID = null; + } + + for (Map.Entry entry : oldProxyInfoMap.entrySet()) { + String nodeId = entry.getKey(); + SCMProxyInfo newInfo = scmProxyInfoMap.get(nodeId); + if (newInfo == null + || !newInfo.getAddress().equals(entry.getValue().getAddress())) { + ProxyInfo staleProxy = scmProxies.remove(nodeId); + if (staleProxy != null && staleProxy.proxy != null) { + try { + RPC.stopProxy(staleProxy.proxy); + } catch (RuntimeException stopEx) { + getLogger().warn("Failed to stop stale proxy for SCM node {}", + nodeId, stopEx); + } + } } } + + getLogger().info("Reloaded SCM proxy configuration for protocol {} with {} nodes: {}", + protocolClass.getSimpleName(), scmNodeIds.size(), scmProxyInfoMap.values()); } @VisibleForTesting diff --git a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/HAUtils.java b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/HAUtils.java index 3ae1b451e1fc..11d8077fca0f 100644 --- a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/HAUtils.java +++ b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/HAUtils.java @@ -57,6 +57,7 @@ import org.apache.hadoop.hdds.scm.proxy.SCMBlockLocationFailoverProxyProvider; import org.apache.hadoop.hdds.scm.proxy.SCMClientConfig; import org.apache.hadoop.hdds.scm.proxy.SCMContainerLocationFailoverProxyProvider; +import org.apache.hadoop.hdds.scm.proxy.SCMFailoverProxyProviderBase; import org.apache.hadoop.hdds.security.exception.SCMSecurityException; import org.apache.hadoop.hdds.tracing.TracingUtil; import org.apache.hadoop.hdds.utils.db.DBDefinition; @@ -131,9 +132,20 @@ public static boolean addSCM(OzoneConfiguration conf, AddSCMRequest request, */ public static ScmBlockLocationProtocol getScmBlockClient( OzoneConfiguration conf) { + return getScmBlockClient(conf, + new SCMBlockLocationFailoverProxyProvider(conf)); + } + + /** + * Create a scm block client backed by the given proxy provider. The caller + * keeps the provider reference so it can dynamically reload the SCM node list + * (see {@link SCMFailoverProxyProviderBase#changeConfig()}). + */ + public static ScmBlockLocationProtocol getScmBlockClient( + OzoneConfiguration conf, + SCMBlockLocationFailoverProxyProvider proxyProvider) { ScmBlockLocationProtocolClientSideTranslatorPB scmBlockLocationClient = - new ScmBlockLocationProtocolClientSideTranslatorPB( - new SCMBlockLocationFailoverProxyProvider(conf), conf); + new ScmBlockLocationProtocolClientSideTranslatorPB(proxyProvider, conf); return TracingUtil .createProxy(scmBlockLocationClient, ScmBlockLocationProtocol.class, conf); @@ -141,8 +153,18 @@ public static ScmBlockLocationProtocol getScmBlockClient( public static StorageContainerLocationProtocol getScmContainerClient( ConfigurationSource conf) { - SCMContainerLocationFailoverProxyProvider proxyProvider = - new SCMContainerLocationFailoverProxyProvider(conf, null); + return getScmContainerClient(conf, + new SCMContainerLocationFailoverProxyProvider(conf, null)); + } + + /** + * Create a scm container client backed by the given proxy provider. The + * caller keeps the provider reference so it can dynamically reload the SCM + * node list (see {@link SCMFailoverProxyProviderBase#changeConfig()}). + */ + public static StorageContainerLocationProtocol getScmContainerClient( + ConfigurationSource conf, + SCMContainerLocationFailoverProxyProvider proxyProvider) { StorageContainerLocationProtocol scmContainerClient = TracingUtil.createProxy( new StorageContainerLocationProtocolClientSideTranslatorPB( diff --git a/hadoop-hdds/framework/src/test/java/org/apache/hadoop/hdds/scm/proxy/TestSCMFailoverProxyProviderChangeConfig.java b/hadoop-hdds/framework/src/test/java/org/apache/hadoop/hdds/scm/proxy/TestSCMFailoverProxyProviderChangeConfig.java new file mode 100644 index 000000000000..c53ccd1d37f3 --- /dev/null +++ b/hadoop-hdds/framework/src/test/java/org/apache/hadoop/hdds/scm/proxy/TestSCMFailoverProxyProviderChangeConfig.java @@ -0,0 +1,122 @@ +/* + * 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.hdds.scm.proxy; + +import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_ADDRESS_KEY; +import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_NODES_KEY; +import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_SERVICE_IDS_KEY; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.List; +import org.apache.hadoop.hdds.conf.ConfigurationException; +import org.apache.hadoop.hdds.conf.OzoneConfiguration; +import org.apache.hadoop.ozone.ha.ConfUtils; +import org.junit.jupiter.api.Test; + +/** + * Verifies that {@link SCMFailoverProxyProviderBase#changeConfig()} reloads the + * SCM node list from an updated configuration (the dynamic SCM reconfiguration + * scenario), adding and removing nodes and keeping the current proxy pointer + * valid, while leaving the previous state intact if the new configuration is + * incomplete. + */ +public class TestSCMFailoverProxyProviderChangeConfig { + + private static final String SERVICE_ID = "scmservice"; + + private static OzoneConfiguration haConf(String nodes) { + OzoneConfiguration conf = new OzoneConfiguration(); + conf.set(OZONE_SCM_SERVICE_IDS_KEY, SERVICE_ID); + conf.set(ConfUtils.addSuffix(OZONE_SCM_NODES_KEY, SERVICE_ID), nodes); + return conf; + } + + private static void setAddress(OzoneConfiguration conf, String nodeId, + String host) { + conf.set(ConfUtils.addKeySuffixes(OZONE_SCM_ADDRESS_KEY, SERVICE_ID, nodeId), + host); + } + + @Test + public void testChangeConfigAddsNode() { + OzoneConfiguration conf = haConf("scm1,scm2"); + setAddress(conf, "scm1", "host1"); + setAddress(conf, "scm2", "host2"); + + SCMBlockLocationFailoverProxyProvider provider = + new SCMBlockLocationFailoverProxyProvider(conf); + assertEquals(2, provider.getSCMNodeIds().size()); + + // Operator adds a third SCM: its address key first, then the node list. + setAddress(conf, "scm3", "host3"); + conf.set(ConfUtils.addSuffix(OZONE_SCM_NODES_KEY, SERVICE_ID), + "scm1,scm2,scm3"); + provider.changeConfig(); + + List nodeIds = provider.getSCMNodeIds(); + assertEquals(3, nodeIds.size()); + assertTrue(nodeIds.contains("scm3")); + } + + @Test + public void testChangeConfigRemovesNode() { + OzoneConfiguration conf = haConf("scm1,scm2,scm3"); + setAddress(conf, "scm1", "host1"); + setAddress(conf, "scm2", "host2"); + setAddress(conf, "scm3", "host3"); + + SCMBlockLocationFailoverProxyProvider provider = + new SCMBlockLocationFailoverProxyProvider(conf); + // Point the current proxy at the node that is about to be removed. + provider.changeCurrentProxy("scm3"); + + conf.set(ConfUtils.addSuffix(OZONE_SCM_NODES_KEY, SERVICE_ID), "scm1,scm2"); + provider.changeConfig(); + + List nodeIds = provider.getSCMNodeIds(); + assertEquals(2, nodeIds.size()); + assertTrue(nodeIds.contains("scm1")); + assertTrue(nodeIds.contains("scm2")); + // The current proxy pointer must fall back to a still-configured node. + assertTrue(nodeIds.contains(provider.getCurrentProxySCMNodeId())); + } + + @Test + public void testChangeConfigFailsWhenAddressMissing() { + OzoneConfiguration conf = haConf("scm1,scm2"); + setAddress(conf, "scm1", "host1"); + setAddress(conf, "scm2", "host2"); + + SCMBlockLocationFailoverProxyProvider provider = + new SCMBlockLocationFailoverProxyProvider(conf); + + // Node list references scm3 but its address has not been set yet. This + // mimics reconfiguring the node list before the new node's address; the + // reload must fail and leave the previous node set intact for a retry. + conf.set(ConfUtils.addSuffix(OZONE_SCM_NODES_KEY, SERVICE_ID), + "scm1,scm2,scm3"); + assertThrows(ConfigurationException.class, provider::changeConfig); + + List nodeIds = provider.getSCMNodeIds(); + assertEquals(2, nodeIds.size()); + assertTrue(nodeIds.contains("scm1")); + assertTrue(nodeIds.contains("scm2")); + } +} diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestOmSCMNodesReconfiguration.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestOmSCMNodesReconfiguration.java new file mode 100644 index 000000000000..787b81764c48 --- /dev/null +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestOmSCMNodesReconfiguration.java @@ -0,0 +1,152 @@ +/* + * 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.hdds.scm; + +import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_ADDRESS_KEY; +import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_NODES_KEY; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import org.apache.hadoop.conf.ReconfigurationException; +import org.apache.hadoop.hdds.conf.OzoneConfiguration; +import org.apache.hadoop.hdds.conf.ReconfigurationHandler; +import org.apache.hadoop.hdds.scm.server.StorageContainerManager; +import org.apache.hadoop.ozone.MiniOzoneCluster; +import org.apache.hadoop.ozone.MiniOzoneHAClusterImpl; +import org.apache.hadoop.ozone.ha.ConfUtils; +import org.apache.hadoop.ozone.om.OzoneManager; +import org.apache.hadoop.ozone.om.ScmClient; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +/** + * Test the OM's SCM nodes reconfiguration wiring: the SCM node list and the + * per-node SCM addresses must be reconfigurable on a running OM so that the OM + * can reload its SCM failover proxies without a restart. The proxy-level + * add/remove behavior is covered by + * {@link org.apache.hadoop.hdds.scm.proxy.SCMFailoverProxyProviderBase}'s unit + * tests; this verifies the OM-side registration and callback end to end. + */ +@Timeout(300) +public class TestOmSCMNodesReconfiguration { + + private MiniOzoneHAClusterImpl cluster = null; + private String scmServiceId; + + @BeforeEach + public void init() throws Exception { + OzoneConfiguration conf = new OzoneConfiguration(); + scmServiceId = "scm-service-test1"; + cluster = MiniOzoneCluster.newHABuilder(conf) + .setOMServiceId("om-service-test1") + .setSCMServiceId(scmServiceId) + .setNumOfStorageContainerManagers(3) + .setNumOfOzoneManagers(1) + .build(); + cluster.waitForClusterToBeReady(); + } + + @AfterEach + public void shutdown() { + if (cluster != null) { + cluster.shutdown(); + } + } + + /** + * The SCM node list and each SCM's address (registered as a prefix) must be + * reconfigurable on the OM. + */ + @Test + void testScmNodesAndAddressReconfigurableOnOm() throws Exception { + ReconfigurationHandler handler = + cluster.getOzoneManager().getReconfigurationHandler(); + String scmNodesKey = + ConfUtils.addKeySuffixes(OZONE_SCM_NODES_KEY, scmServiceId); + + assertTrue(handler.isPropertyReconfigurable(scmNodesKey)); + assertTrue(handler.listReconfigureProperties().contains(scmNodesKey)); + + // The per-node SCM address keys are registered as a prefix, so any node's + // address key is reconfigurable even though it was not registered by name. + for (StorageContainerManager scm : cluster.getStorageContainerManagers()) { + String scmAddrKey = ConfUtils.addKeySuffixes( + OZONE_SCM_ADDRESS_KEY, scmServiceId, scm.getSCMNodeId()); + assertTrue(handler.isPropertyReconfigurable(scmAddrKey)); + } + } + + /** + * Setting an empty SCM node list must be rejected, leaving the OM's SCM + * proxies untouched. + */ + @Test + void testReconfigureScmNodesToBlankThrows() { + ReconfigurationHandler handler = + cluster.getOzoneManager().getReconfigurationHandler(); + String scmNodesKey = + ConfUtils.addKeySuffixes(OZONE_SCM_NODES_KEY, scmServiceId); + + assertThrows(ReconfigurationException.class, + () -> handler.reconfigureProperty(scmNodesKey, "")); + } + + /** + * Reconfiguring the SCM node list on a running OM must reload the SCM failover + * proxies to the new membership. Dropping one SCM from the list has to shrink + * the proxy node set for both the block and container providers; the reload + * reads the list from the (freshly written) configuration, so reconfiguring to + * a genuinely different value is what exercises the wiring. + */ + @Test + void testReconfigureScmNodesReloadsProxies() throws Exception { + OzoneManager om = cluster.getOzoneManager(); + ReconfigurationHandler handler = om.getReconfigurationHandler(); + ScmClient scmClient = om.getScmClient(); + String scmNodesKey = + ConfUtils.addKeySuffixes(OZONE_SCM_NODES_KEY, scmServiceId); + + List before = + new ArrayList<>(scmClient.getContainerProxyProvider().getSCMNodeIds()); + assertEquals(3, before.size()); + + // Drop one SCM from the OM's view. Its address stays in the configuration, + // so the reload of the remaining nodes succeeds. + String dropped = before.get(before.size() - 1); + List remaining = new ArrayList<>(before.subList(0, before.size() - 1)); + Set expected = new HashSet<>(remaining); + + handler.reconfigureProperty(scmNodesKey, String.join(",", remaining)); + + Set afterContainer = + new HashSet<>(scmClient.getContainerProxyProvider().getSCMNodeIds()); + Set afterBlock = + new HashSet<>(scmClient.getBlockProxyProvider().getSCMNodeIds()); + assertEquals(expected, afterContainer); + assertEquals(expected, afterBlock); + assertFalse(afterContainer.contains(dropped)); + } +} 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 a53c0217efdc..f280cfbf394e 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 @@ -24,6 +24,8 @@ import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_BLOCK_TOKEN_ENABLED; import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_BLOCK_TOKEN_ENABLED_DEFAULT; import static org.apache.hadoop.hdds.HddsUtils.getScmAddressForClients; +import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_ADDRESS_KEY; +import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_NODES_KEY; import static org.apache.hadoop.hdds.server.ServerUtils.updateRPCListenAddress; import static org.apache.hadoop.hdds.utils.HAUtils.getScmInfo; import static org.apache.hadoop.hdds.utils.HddsServerUtil.getRemoteUser; @@ -201,6 +203,8 @@ import org.apache.hadoop.hdds.scm.net.NetworkTopology; import org.apache.hadoop.hdds.scm.protocol.ScmBlockLocationProtocol; import org.apache.hadoop.hdds.scm.protocol.StorageContainerLocationProtocol; +import org.apache.hadoop.hdds.scm.proxy.SCMBlockLocationFailoverProxyProvider; +import org.apache.hadoop.hdds.scm.proxy.SCMContainerLocationFailoverProxyProvider; import org.apache.hadoop.hdds.security.SecurityConfig; import org.apache.hadoop.hdds.security.exception.OzoneSecurityException; import org.apache.hadoop.hdds.security.symmetric.DefaultSecretKeyClient; @@ -250,6 +254,7 @@ import org.apache.hadoop.ozone.audit.OMAction; import org.apache.hadoop.ozone.audit.OMSystemAction; import org.apache.hadoop.ozone.common.Storage.StorageState; +import org.apache.hadoop.ozone.ha.ConfUtils; import org.apache.hadoop.ozone.om.exceptions.OMException; import org.apache.hadoop.ozone.om.exceptions.OMException.ResultCodes; import org.apache.hadoop.ozone.om.exceptions.OMLeaderNotReadyException; @@ -572,6 +577,17 @@ private OzoneManager(OzoneConfiguration conf, StartupOption startupOption) .register(OZONE_READ_BLACKLIST_USERS, this::reconfOzoneReadBlacklistUsers) .register(OZONE_READ_BLACKLIST_GROUPS, this::reconfOzoneReadBlacklistGroups); + // Allow the SCM node list and the per-node SCM addresses to be reconfigured + // so that a newly added SCM can be reached without restarting the OM. The + // address keys are registered as a prefix since the new node's key does not + // exist at startup and cannot be registered by name in advance. + String scmServiceId = HddsUtils.getScmServiceId(conf); + if (scmServiceId != null) { + reconfigurationHandler + .registerPrefix(ConfUtils.addKeySuffixes(OZONE_SCM_ADDRESS_KEY, scmServiceId)) + .register(OZONE_SCM_NODES_KEY + "." + scmServiceId, this::reconfScmNodes); + } + reconfigurationHandler.setReconfigurationCompleteCallback(reconfigurationHandler.defaultLoggingCallback()); reconfigurationHandler.registerCompleteCallback(tracingReconfigurationCallback); @@ -650,12 +666,21 @@ private OzoneManager(OzoneConfiguration conf, StartupOption startupOption) // Honor property 'hadoop.security.token.service.use_ip' omRpcAddressTxt = new Text(SecurityUtil.buildTokenService(omNodeRpcAddr)); - final StorageContainerLocationProtocol scmContainerClient = getScmContainerClient(configuration); + // Keep references to the SCM proxy providers so the OM can dynamically + // reload the SCM node list / addresses on reconfiguration (see + // reconfScmNodes) without a restart. + final SCMContainerLocationFailoverProxyProvider scmContainerProxyProvider = + new SCMContainerLocationFailoverProxyProvider(configuration, null); + final StorageContainerLocationProtocol scmContainerClient = + HAUtils.getScmContainerClient(configuration, scmContainerProxyProvider); // verifies that the SCM info in the OM Version file is correct. - final ScmBlockLocationProtocol scmBlockClient = getScmBlockClient(configuration); + final SCMBlockLocationFailoverProxyProvider scmBlockProxyProvider = + new SCMBlockLocationFailoverProxyProvider(configuration); + final ScmBlockLocationProtocol scmBlockClient = + HAUtils.getScmBlockClient(configuration, scmBlockProxyProvider); scmTopologyClient = new ScmTopologyClient(scmBlockClient); this.scmClient = new ScmClient(scmBlockClient, scmContainerClient, - configuration); + scmBlockProxyProvider, scmContainerProxyProvider, configuration); this.ozoneLockProvider = new OzoneLockProvider(getKeyPathLockEnabled(), getEnableFileSystemPaths()); @@ -1409,26 +1434,6 @@ private static void loginOMUser(OzoneConfiguration conf) LOG.info("Ozone Manager login successful."); } - /** - * Create a scm block client, used by putKey() and getKey(). - * - * @return {@link ScmBlockLocationProtocol} - */ - private static ScmBlockLocationProtocol getScmBlockClient( - OzoneConfiguration conf) { - return HAUtils.getScmBlockClient(conf); - } - - /** - * Returns a scm container client. - * - * @return {@link StorageContainerLocationProtocol} - */ - private static StorageContainerLocationProtocol getScmContainerClient( - OzoneConfiguration conf) { - return HAUtils.getScmContainerClient(conf); - } - /** * Creates a new instance of rpc server. If an earlier instance is already * running then returns the same. @@ -5764,6 +5769,48 @@ public ListSnapshotDiffJobResponse listSnapshotDiffJobs( } } + /** + * Reload the block and container SCM failover proxies after the SCM node list + * ({@code ozone.scm.nodes.}) is reconfigured, so the OM can reach a + * newly added SCM without a restart. The per-node address keys + * ({@code ozone.scm.address..}) must already be present for + * the involved nodes. + * + * Scope: only the block and container proxies are reloaded here. Changing an + * address key alone does not trigger a reload; touch the node list to apply + * it. The secure-mode SCM security and secret-key proxy providers are not + * reloaded and continue to use the node list captured at startup. + */ + private String reconfScmNodes(String value) { + if (StringUtils.isBlank(value)) { + throw new IllegalArgumentException("Reconfiguration failed since setting an empty SCM nodes " + + "configuration is not allowed"); + } + // ReconfigurableBase stores the new value into the configuration only after + // this callback returns, but reloadScmNodes() rebuilds the SCM proxies from + // that same live configuration. Publish the new node list first so the + // reload sees the intended membership, and roll it back if the reload fails + // (e.g. a newly added SCM's address is not set yet) so the property is not + // left naming a node set the proxies never adopted; that also lets the + // reconfiguration be retried once both keys are updated. + String scmNodesKey = ConfUtils.addKeySuffixes(OZONE_SCM_NODES_KEY, + HddsUtils.getScmServiceId(configuration)); + String previousValue = configuration.get(scmNodesKey); + configuration.set(scmNodesKey, value); + try { + scmClient.reloadScmNodes(); + } catch (RuntimeException e) { + if (previousValue == null) { + configuration.unset(scmNodesKey); + } else { + configuration.set(scmNodesKey, previousValue); + } + throw e; + } + LOG.info("Reloaded SCM proxy configuration for {} : {}", OZONE_SCM_NODES_KEY, value); + return value; + } + private String reconfOzoneAdmins(String newVal) { Collection admins = OzoneAdmins.getOzoneAdminsFromConfigValue(newVal, omStarterUser); diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ScmClient.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ScmClient.java index 868a248e909e..f174934b2729 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ScmClient.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ScmClient.java @@ -24,6 +24,7 @@ import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_CONTAINER_LOCATION_DATANODE_CACHE_SIZE; import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_CONTAINER_LOCATION_DATANODE_CACHE_SIZE_DEFAULT; +import com.google.common.annotations.VisibleForTesting; import com.google.common.cache.Cache; import com.google.common.cache.CacheBuilder; import com.google.common.cache.CacheLoader; @@ -46,6 +47,7 @@ import org.apache.hadoop.hdds.scm.pipeline.Pipeline; import org.apache.hadoop.hdds.scm.protocol.ScmBlockLocationProtocol; import org.apache.hadoop.hdds.scm.protocol.StorageContainerLocationProtocol; +import org.apache.hadoop.hdds.scm.proxy.SCMFailoverProxyProviderBase; import org.apache.hadoop.ozone.util.CacheMetrics; /** @@ -55,6 +57,8 @@ public class ScmClient { private final ScmBlockLocationProtocol blockClient; private final StorageContainerLocationProtocol containerClient; + private final SCMFailoverProxyProviderBase blockProxyProvider; + private final SCMFailoverProxyProviderBase containerProxyProvider; private final LoadingCache containerLocationCache; private final CacheMetrics containerCacheMetrics; private final CacheMetrics datanodeDetailsCacheMetrics; @@ -62,8 +66,18 @@ public class ScmClient { ScmClient(ScmBlockLocationProtocol blockClient, StorageContainerLocationProtocol containerClient, OzoneConfiguration configuration) { + this(blockClient, containerClient, null, null, configuration); + } + + ScmClient(ScmBlockLocationProtocol blockClient, + StorageContainerLocationProtocol containerClient, + SCMFailoverProxyProviderBase blockProxyProvider, + SCMFailoverProxyProviderBase containerProxyProvider, + OzoneConfiguration configuration) { this.containerClient = containerClient; this.blockClient = blockClient; + this.blockProxyProvider = blockProxyProvider; + this.containerProxyProvider = containerProxyProvider; Cache datanodeDetailsCache = createDatanodeDetailsCache(configuration); this.containerLocationCache = @@ -144,6 +158,31 @@ static Pipeline newPipelineWithDNCache(Pipeline pipeline, return builder.build(); } + /** + * Reload the SCM node list and addresses for both the block and container + * SCM clients after a dynamic reconfiguration, so the OM can reach a newly + * added SCM without a restart. No-op when the providers are not available + * (e.g. clients created directly with mocks in tests). + */ + public void reloadScmNodes() { + if (blockProxyProvider != null) { + blockProxyProvider.changeConfig(); + } + if (containerProxyProvider != null) { + containerProxyProvider.changeConfig(); + } + } + + @VisibleForTesting + public SCMFailoverProxyProviderBase getBlockProxyProvider() { + return blockProxyProvider; + } + + @VisibleForTesting + public SCMFailoverProxyProviderBase getContainerProxyProvider() { + return containerProxyProvider; + } + public ScmBlockLocationProtocol getBlockClient() { return this.blockClient; } diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/TestScmClient.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/TestScmClient.java index 89d8c2162c38..b96ed9f96012 100644 --- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/TestScmClient.java +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/TestScmClient.java @@ -20,6 +20,7 @@ import static com.google.common.collect.Sets.newHashSet; import static java.util.Arrays.asList; import static org.apache.hadoop.hdds.client.ReplicationConfig.fromTypeAndFactor; +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertThrows; @@ -51,6 +52,7 @@ import org.apache.hadoop.hdds.scm.pipeline.PipelineID; import org.apache.hadoop.hdds.scm.protocol.ScmBlockLocationProtocol; import org.apache.hadoop.hdds.scm.protocol.StorageContainerLocationProtocol; +import org.apache.hadoop.hdds.scm.proxy.SCMFailoverProxyProviderBase; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; @@ -210,6 +212,29 @@ public void testDatanodeDetailsCacheUpdatesIpAddressChange() { assertSame(updated, datanodeDetailsCache.getIfPresent(original.getID())); } + @Test + public void testReloadScmNodesDelegatesToBothProviders() { + SCMFailoverProxyProviderBase blockProvider = + mock(SCMFailoverProxyProviderBase.class); + SCMFailoverProxyProviderBase containerProvider = + mock(SCMFailoverProxyProviderBase.class); + ScmClient client = new ScmClient(mock(ScmBlockLocationProtocol.class), + mock(StorageContainerLocationProtocol.class), blockProvider, + containerProvider, new OzoneConfiguration()); + + client.reloadScmNodes(); + + verify(blockProvider, times(1)).changeConfig(); + verify(containerProvider, times(1)).changeConfig(); + } + + @Test + public void testReloadScmNodesIsNoOpWithoutProviders() { + // The 3-arg constructor leaves both providers null (e.g. clients created + // directly with mocks); reloading must not fail. + assertDoesNotThrow(scmClient::reloadScmNodes); + } + ContainerWithPipeline createPipeline(long containerId, List dnList) { ContainerInfo containerInfo = new ContainerInfo.Builder()