HDDS-13890. Datanode supports dynamic configuration of SCM (#9385)

Author: ivandika3

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/HddsUtils.java

@@ -31,6 +31,7 @@
 import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_CLIENT_ADDRESS_KEY;
 import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_CLIENT_PORT_DEFAULT;
 import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_CLIENT_PORT_KEY;
+import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_DATANODE_ADDRESS_KEY;
 import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_DATANODE_PORT_DEFAULT;
 import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_DATANODE_PORT_KEY;
 import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_DEFAULT_SERVICE_ID;
@@ -57,10 +58,12 @@
 import java.util.Objects;
 import java.util.Optional;
 import java.util.OptionalInt;
+import java.util.Set;
 import java.util.TreeMap;
 import java.util.UUID;
 import javax.management.ObjectName;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.tuple.Pair;
 import org.apache.hadoop.conf.ConfigRedactor;
 import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
 import org.apache.hadoop.hdds.annotation.InterfaceAudience;
@@ -331,6 +334,38 @@ public static Collection<InetSocketAddress> getSCMAddressForDatanodes(
     }
   }
 
+  /**
+   * Returns the SCM address for datanodes based on the service ID and the SCM addresses.
+   * @param conf Configuration
+   * @param scmServiceId SCM service ID
+   * @param scmNodeIds Requested SCM node IDs
+   * @return A collection with addresses of the request SCM node IDs.
+   * Null if there is any wrongly configured SCM address. Note that the returned collection
+   * might not be ordered the same way as the requested SCM node IDs
+   */
+  public static Collection<Pair<String, InetSocketAddress>> getSCMAddressForDatanodes(
+      ConfigurationSource conf, String scmServiceId, Set<String> scmNodeIds) {
+    Collection<Pair<String, InetSocketAddress>> scmNodeAddress = new HashSet<>(scmNodeIds.size());
+    for (String scmNodeId : scmNodeIds) {
+      String addressKey = ConfUtils.addKeySuffixes(
+          OZONE_SCM_ADDRESS_KEY, scmServiceId, scmNodeId);
+      String scmAddress = conf.get(addressKey);
+      if (scmAddress == null) {
+        LOG.warn("The SCM address configuration {} is not defined, return nothing", addressKey);
+        return null;
+      }
+
+      int scmDatanodePort = SCMNodeInfo.getPort(conf, scmServiceId, scmNodeId,
+          OZONE_SCM_DATANODE_ADDRESS_KEY, OZONE_SCM_DATANODE_PORT_KEY,
+          OZONE_SCM_DATANODE_PORT_DEFAULT);
+
+      String scmDatanodeAddressStr = SCMNodeInfo.buildAddress(scmAddress, scmDatanodePort);
+      InetSocketAddress scmDatanodeAddress = NetUtils.createSocketAddr(scmDatanodeAddressStr);
+      scmNodeAddress.add(Pair.of(scmNodeId, scmDatanodeAddress));
+    }
+    return scmNodeAddress;
+  }
+
   /**
    * Retrieve the socket addresses of recon.
    *

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMNodeInfo.java

@@ -181,12 +181,11 @@ public static List<SCMNodeInfo> buildNodeInfo(ConfigurationSource conf) {
 
   }
 
-  private static String buildAddress(String address, int port) {
-    return new StringBuilder().append(address).append(':')
-        .append(port).toString();
+  public static String buildAddress(String address, int port) {
+    return address + ':' + port;
   }
 
-  private static int getPort(ConfigurationSource conf,
+  public static int getPort(ConfigurationSource conf,
       String scmServiceId, String scmNodeId, String configKey,
       String portKey, int defaultPort) {
     String suffixKey = ConfUtils.addKeySuffixes(configKey, scmServiceId,

hadoop-hdds/config/src/main/java/org/apache/hadoop/hdds/conf/ConfigurationSource.java

@@ -98,6 +98,10 @@ default String getTrimmed(String key, String defaultValue) {
 
   default String[] getTrimmedStrings(String name) {
     String valueString = get(name);
+    return getTrimmedStringsFromValue(valueString);
+  }
+
+  static String[] getTrimmedStringsFromValue(String valueString) {
     if (null == valueString) {
       return EMPTY_STRING_ARRAY;
     }

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/HddsDatanodeService.java

@@ -19,6 +19,7 @@
 
 import static org.apache.hadoop.hdds.protocol.DatanodeDetails.Port.Name.HTTP;
 import static org.apache.hadoop.hdds.protocol.DatanodeDetails.Port.Name.HTTPS;
+import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_NODES_KEY;
 import static org.apache.hadoop.hdds.utils.HddsServerUtil.getRemoteUser;
 import static org.apache.hadoop.hdds.utils.HddsServerUtil.getScmSecurityClientWithMaxRetry;
 import static org.apache.hadoop.ozone.OzoneConfigKeys.HDDS_DATANODE_PLUGINS_KEY;
@@ -34,23 +35,33 @@
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
+import com.google.common.collect.Sets;
 import java.io.File;
 import java.io.IOException;
+import java.net.InetSocketAddress;
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Set;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 import javax.management.ObjectName;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.tuple.Pair;
 import org.apache.hadoop.conf.Configurable;
 import org.apache.hadoop.hdds.DatanodeVersion;
 import org.apache.hadoop.hdds.HddsConfigKeys;
 import org.apache.hadoop.hdds.HddsUtils;
 import org.apache.hadoop.hdds.cli.GenericCli;
 import org.apache.hadoop.hdds.cli.HddsVersionProvider;
+import org.apache.hadoop.hdds.conf.ConfigurationSource;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
 import org.apache.hadoop.hdds.conf.ReconfigurationHandler;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
@@ -72,6 +83,9 @@
 import org.apache.hadoop.ozone.container.common.DatanodeLayoutStorage;
 import org.apache.hadoop.ozone.container.common.helpers.ContainerUtils;
 import org.apache.hadoop.ozone.container.common.statemachine.DatanodeStateMachine;
+import org.apache.hadoop.ozone.container.common.statemachine.DatanodeStateMachine.DatanodeStates;
+import org.apache.hadoop.ozone.container.common.statemachine.SCMConnectionManager;
+import org.apache.hadoop.ozone.container.common.statemachine.StateContext;
 import org.apache.hadoop.ozone.container.common.statemachine.commandhandler.DeleteBlocksCommandHandler;
 import org.apache.hadoop.ozone.container.common.utils.StorageVolumeUtil;
 import org.apache.hadoop.ozone.container.common.volume.HddsVolume;
@@ -124,6 +138,7 @@ public class HddsDatanodeService extends GenericCli implements Callable<Void>, S
   private HddsDatanodeClientProtocolServer clientProtocolServer;
   private OzoneAdmins admins;
   private ReconfigurationHandler reconfigurationHandler;
+  private String scmServiceId;
 
   //Constructor for DataNode PluginService
   public HddsDatanodeService() { }
@@ -207,6 +222,7 @@ public void start(OzoneConfiguration configuration) {
     start();
   }
 
+  @SuppressWarnings("methodlength")
   public void start() {
     serviceRuntimeInfo = new DNMXBeanImpl(HddsVersionInfo.HDDS_VERSION_INFO) {
       @Override
@@ -294,6 +310,12 @@ public String getNamespace() {
               .register(REPLICATION_STREAMS_LIMIT_KEY,
                   this::reconfigReplicationStreamsLimit);
 
+      scmServiceId = HddsUtils.getScmServiceId(conf);
+      if (scmServiceId != null) {
+        reconfigurationHandler.register(OZONE_SCM_NODES_KEY + "." + scmServiceId,
+            this::reconfigScmNodes);
+      }
+
       reconfigurationHandler.setReconfigurationCompleteCallback(reconfigurationHandler.defaultLoggingCallback());
 
       datanodeStateMachine = new DatanodeStateMachine(this, datanodeDetails, conf,
@@ -680,6 +702,112 @@ private String reconfigBlockDeletingServiceTimeout(String value) {
     return value;
   }
 
+  /**
+   * Reconfigure the SCM nodes configuration which will trigger the creation and removal of
+   * SCM connections based on the difference between the old and the new SCM nodes configuration.
+   * <p>
+   * The assumption is that the SCM node address configurations exists for all the involved node IDs
+   * This is because reconfiguration can only support one configuration field at a time
+   * @param value The new configuration value for "ozone.scm.nodes.SERVICEID"
+   * @return new configuration for "ozone.scm.nodes.SERVICEID" which reflects the SCMs that the datanode has
+   *         is not connected to.
+   */
+  private String reconfigScmNodes(String value) {
+    if (StringUtils.isBlank(value)) {
+      throw new IllegalArgumentException("Reconfiguration failed since setting the empty SCM nodes " +
+          "configuration is not allowed");
+    }
+    Set<String> previousNodeIds = new HashSet<>(HddsUtils.getSCMNodeIds(getConf(), scmServiceId));
+    Set<String> newScmNodeIds = Stream.of(ConfigurationSource.getTrimmedStringsFromValue(value))
+        .collect(Collectors.toSet());
+
+    if (newScmNodeIds.isEmpty()) {
+      throw new IllegalArgumentException("Reconfiguration failed since setting the empty SCM nodes " +
+          "configuration is not allowed");
+    }
+
+    Set<String> scmNodesIdsToAdd = Sets.difference(newScmNodeIds, previousNodeIds);
+    Set<String> scmNodesIdsToRemove = Sets.difference(previousNodeIds, newScmNodeIds);
+
+    // We should only update configuration with the SCMs that are actually added / removed
+    // If there is partial reconfiguration (e.g. one successful add and one failed add),
+    // we want to be able to retry on the failed node reconfiguration.
+    // If we don't handle this, the subsequent reconfiguration will not work since the node
+    // configuration is already exists / removed.
+    Set<String> effectiveScmNodeIds = new HashSet<>(previousNodeIds);
+
+    LOG.info("Reconfiguring SCM nodes for service ID {} with new SCM nodes {} and remove SCM nodes {}",
+        scmServiceId, scmNodesIdsToAdd, scmNodesIdsToRemove);
+
+    Collection<Pair<String, InetSocketAddress>> scmToAdd = HddsUtils.getSCMAddressForDatanodes(
+        getConf(), scmServiceId, scmNodesIdsToAdd);
+    if (scmToAdd == null) {
+      throw new IllegalStateException("Reconfiguration failed to get SCM address to add due to wrong configuration");
+    }
+    Collection<Pair<String, InetSocketAddress>> scmToRemove = HddsUtils.getSCMAddressForDatanodes(
+        getConf(), scmServiceId, scmNodesIdsToRemove);
+    if (scmToRemove == null) {
+      throw new IllegalArgumentException(
+          "Reconfiguration failed to get SCM address to remove due to wrong configuration");
+    }
+
+    StateContext context = datanodeStateMachine.getContext();
+    SCMConnectionManager connectionManager = datanodeStateMachine.getConnectionManager();
+
+    // Assert that the datanode is in RUNNING state since
+    // 1. If the datanode state is INIT, there might be concurrent connection manager operations
+    //    that might cause unpredictable behaviors
+    // 2. If the datanode state is SHUTDOWN, it means that datanode is shutting down and there is no need
+    //    to reconfigure the connections.
+    if (!DatanodeStates.RUNNING.equals(context.getState())) {
+      throw new IllegalStateException("Reconfiguration failed since the datanode the current state" +
+          context.getState().toString() + " is not in RUNNING state");
+    }
+
+    // Add the new SCM servers
+    for (Pair<String, InetSocketAddress> pair : scmToAdd) {
+      String scmNodeId = pair.getLeft();
+      InetSocketAddress scmAddress = pair.getRight();
+      if (scmAddress.isUnresolved()) {
+        LOG.warn("Reconfiguration failed to add SCM address {} for SCM service {} since it can't " +
+            "be resolved, skipping", scmAddress, scmServiceId);
+        continue;
+      }
+      try {
+        connectionManager.addSCMServer(scmAddress, context.getThreadNamePrefix());
+        context.addEndpoint(scmAddress);
+        effectiveScmNodeIds.add(scmNodeId);
+        LOG.info("Reconfiguration successfully add SCM address {} for SCM service {}", scmAddress, scmServiceId);
+      } catch (IOException e) {
+        LOG.error("Reconfiguration failed to add SCM address {} for SCM service {}", scmAddress, scmServiceId, e);
+      }
+    }
+
+    // Remove the old SCM server
+    for (Pair<String, InetSocketAddress> pair : scmToRemove) {
+      String scmNodeId = pair.getLeft();
+      InetSocketAddress scmAddress = pair.getRight();
+      try {
+        connectionManager.removeSCMServer(scmAddress);
+        context.removeEndpoint(scmAddress);
+        effectiveScmNodeIds.remove(scmNodeId);
+        LOG.info("Reconfiguration successfully remove SCM address {} for SCM service {}",
+            scmAddress, scmServiceId);
+      } catch (IOException e) {
+        LOG.error("Reconfiguration failed to remove SCM address {} for SCM service {}", scmAddress, scmServiceId, e);
+      }
+    }
+
+    // Resize the executor pool size to (number of SCMs + 1 Recon)
+    // Refer to DatanodeStateMachine#getEndPointTaskThreadPoolSize
+    datanodeStateMachine.resizeExecutor(connectionManager.getNumOfConnections());
+
+    // TODO: In the future, we might also do some assertions on the SCM
+    //  - The SCM cannot be a leader since this causes the datanode to disappear
+    //  - The SCM should be decommissioned
+    return String.join(",", effectiveScmNodeIds);
+  }
+
   /**
    * Returns the initial version of the datanode.
    */

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeQueueMetrics.java

@@ -19,6 +19,7 @@
 
 import static org.apache.hadoop.metrics2.lib.Interns.info;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.CaseFormat;
 import java.net.InetSocketAddress;
 import java.util.HashMap;
@@ -171,6 +172,27 @@ public void addEndpoint(InetSocketAddress endpoint) {
                 .to(CaseFormat.UPPER_CAMEL, k.getHostName())));
   }
 
+  public void removeEndpoint(InetSocketAddress endpoint) {
+    incrementalReportsQueueMap.remove(endpoint);
+    containerActionQueueMap.remove(endpoint);
+    pipelineActionQueueMap.remove(endpoint);
+  }
+
+  @VisibleForTesting
+  public int getIncrementalReportsQueueMapSize() {
+    return incrementalReportsQueueMap.size();
+  }
+
+  @VisibleForTesting
+  public int getContainerActionQueueMapSize() {
+    return containerActionQueueMap.size();
+  }
+
+  @VisibleForTesting
+  public int getPipelineActionQueueMapSize() {
+    return pipelineActionQueueMap.size();
+  }
+
   private MetricsInfo getMetricsInfo(String prefix, String metricName) {
     String metric = prefix + WordUtils.capitalize(metricName) + "Size";
     String description = "Queue size of " + metricName + " from " + prefix;

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeStateMachine.java

@@ -45,6 +45,7 @@
 import org.apache.hadoop.hdds.security.symmetric.SecretKeyClient;
 import org.apache.hadoop.hdds.security.x509.certificate.client.CertificateClient;
 import org.apache.hadoop.hdds.upgrade.HDDSLayoutVersionManager;
+import org.apache.hadoop.hdds.utils.HddsServerUtil;
 import org.apache.hadoop.hdds.utils.IOUtils;
 import org.apache.hadoop.hdds.utils.NettyMetrics;
 import org.apache.hadoop.hdfs.util.EnumCounters;
@@ -795,4 +796,19 @@ public VolumeChoosingPolicy getVolumeChoosingPolicy() {
   public void setNextHB(long time) {
     nextHB.set(time);
   }
+
+  @VisibleForTesting
+  public ExecutorService getExecutorService() {
+    return executorService;
+  }
+
+  /**
+   * Resize the executor based on the number of active endpoint tasks.
+   */
+  public void resizeExecutor(int size) {
+    if (executorService instanceof ThreadPoolExecutor) {
+      ThreadPoolExecutor tpe = (ThreadPoolExecutor) executorService;
+      HddsServerUtil.setPoolSize(tpe, size, LOG);
+    }
+  }
 }

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/SCMConnectionManager.java

@@ -44,6 +44,7 @@
 import org.apache.hadoop.ipc_.RPC;
 import org.apache.hadoop.metrics2.util.MBeans;
 import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.ozone.container.common.statemachine.EndpointStateMachine.EndPointStates;
 import org.apache.hadoop.ozone.protocolPB.ReconDatanodeProtocolPB;
 import org.apache.hadoop.ozone.protocolPB.StorageContainerDatanodeProtocolClientSideTranslatorPB;
 import org.apache.hadoop.ozone.protocolPB.StorageContainerDatanodeProtocolPB;
@@ -227,15 +228,14 @@ public void addReconServer(InetSocketAddress address,
   public void removeSCMServer(InetSocketAddress address) throws IOException {
     writeLock();
     try {
-      if (!scmMachines.containsKey(address)) {
+      EndpointStateMachine endPoint = scmMachines.remove(address);
+      if (endPoint == null) {
         LOG.warn("Trying to remove a non-existent SCM machine. " +
             "Ignoring the request.");
         return;
       }
-
-      EndpointStateMachine endPoint = scmMachines.get(address);
+      endPoint.setState(EndPointStates.SHUTDOWN);
       endPoint.close();
-      scmMachines.remove(address);
     } finally {
       writeUnlock();
     }
@@ -274,4 +274,16 @@ public List<EndpointStateMachineMBean> getSCMServers() {
       readUnlock();
     }
   }
+
+  /**
+   * @return the number of connections (both SCM and Recon)
+   */
+  public int getNumOfConnections() {
+    readLock();
+    try {
+      return scmMachines.size();
+    } finally {
+      readUnlock();
+    }
+  }
 }