HDDS-16300. Allow the Ozone Manager to dynamically reconfigure its SCM node list (ozone.scm.nodes / ozone.scm.address) without a restart

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<SCMNodeInfo> scmNodeInfoList = SCMNodeInfo.buildNodeInfo(conf);
-    scmNodeIds = new ArrayList<>();
-
+    List<String> newScmNodeIds = new ArrayList<>();
+    Map<String, SCMProxyInfo> 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.<serviceId>} and
+   * {@code ozone.scm.address.<serviceId>.<nodeId>} 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<String, SCMProxyInfo> 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<String, SCMProxyInfo> entry : oldProxyInfoMap.entrySet()) {
+      String nodeId = entry.getKey();
+      SCMProxyInfo newInfo = scmProxyInfoMap.get(nodeId);
+      if (newInfo == null
+          || !newInfo.getAddress().equals(entry.getValue().getAddress())) {
+        ProxyInfo<T> 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

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,18 +132,39 @@ 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);
   }
 
   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(

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<String> 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<String> 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<String> nodeIds = provider.getSCMNodeIds();
+    assertEquals(2, nodeIds.size());
+    assertTrue(nodeIds.contains("scm1"));
+    assertTrue(nodeIds.contains("scm2"));
+  }
+}