HDFS: Allow overriding datanode registration addresses

CHANGELOG.md

@@ -7,6 +7,7 @@ All notable changes to this project will be documented in this file.
 ### Added
 
 - omid: init at 1.1.0 ([#493]).
+- hadoop: Allow datanodes to override their registration addresses ([#506]).
 
 ### Changed
 
@@ -17,6 +18,7 @@ All notable changes to this project will be documented in this file.
 - hadoop: Build from source ([#526]).
 
 [#493]: https://github.com/stackabletech/docker-images/pull/493
+[#506]: https://github.com/stackabletech/docker-images/pull/506
 [#514]: https://github.com/stackabletech/docker-images/pull/514
 [#517]: https://github.com/stackabletech/docker-images/pull/517
 [#519]: https://github.com/stackabletech/docker-images/pull/519

hadoop/stackable/patches/3.3.4/002-datanode-registration-override-3.3.4.patch

@@ -0,0 +1,247 @@
+diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java
+index 7196def4221..2c00fb4fb1a 100755
+--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java
++++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java
+@@ -139,6 +139,13 @@ public class DFSConfigKeys extends CommonConfigurationKeys {
+   public static final boolean DFS_DATANODE_DROP_CACHE_BEHIND_READS_DEFAULT = false;
+   public static final String  DFS_DATANODE_USE_DN_HOSTNAME = "dfs.datanode.use.datanode.hostname";
+   public static final boolean DFS_DATANODE_USE_DN_HOSTNAME_DEFAULT = false;
++
++  public static final String  DFS_DATANODE_REGISTERED_HOSTNAME = "dfs.datanode.registered.hostname";
++  public static final String  DFS_DATANODE_REGISTERED_DATA_PORT = "dfs.datanode.registered.port";
++  public static final String  DFS_DATANODE_REGISTERED_HTTP_PORT = "dfs.datanode.registered.http.port";
++  public static final String  DFS_DATANODE_REGISTERED_HTTPS_PORT = "dfs.datanode.registered.https.port";
++  public static final String  DFS_DATANODE_REGISTERED_IPC_PORT = "dfs.datanode.registered.ipc.port";
++
+   public static final String  DFS_DATANODE_MAX_LOCKED_MEMORY_KEY = "dfs.datanode.max.locked.memory";
+   public static final long    DFS_DATANODE_MAX_LOCKED_MEMORY_DEFAULT = 0;
+   public static final String  DFS_DATANODE_FSDATASETCACHE_MAX_THREADS_PER_VOLUME_KEY = "dfs.datanode.fsdatasetcache.max.threads.per.volume";
+@@ -446,6 +453,8 @@ public class DFSConfigKeys extends CommonConfigurationKeys {
+   public static final long DFS_DATANODE_PROCESS_COMMANDS_THRESHOLD_DEFAULT =
+       TimeUnit.SECONDS.toMillis(2);
+
++  public static final String DFS_NAMENODE_DATANODE_REGISTRATION_UNSAFE_ALLOW_ADDRESS_OVERRIDE_KEY = "dfs.namenode.datanode.registration.unsafe.allow-address-override";
++  public static final boolean DFS_NAMENODE_DATANODE_REGISTRATION_UNSAFE_ALLOW_ADDRESS_OVERRIDE_DEFAULT = false;
+   public static final String DFS_NAMENODE_DATANODE_REGISTRATION_IP_HOSTNAME_CHECK_KEY = "dfs.namenode.datanode.registration.ip-hostname-check";
+   public static final boolean DFS_NAMENODE_DATANODE_REGISTRATION_IP_HOSTNAME_CHECK_DEFAULT = true;
+
+diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeManager.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeManager.java
+index 44dffcbed11..54f6d63fa78 100644
+--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeManager.java
++++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeManager.java
+@@ -179,6 +179,8 @@ public class DatanodeManager {
+   private boolean hasClusterEverBeenMultiRack = false;
+
+   private final boolean checkIpHostnameInRegistration;
++  private final boolean allowRegistrationAddressOverride;
++
+   /**
+    * Whether we should tell datanodes what to cache in replies to
+    * heartbeat messages.
+@@ -326,6 +328,11 @@ public class DatanodeManager {
+         + ": configured=" + configuredBlockInvalidateLimit
+         + ", counted=" + countedBlockInvalidateLimit
+         + ", effected=" + blockInvalidateLimit);
++    this.allowRegistrationAddressOverride = conf.getBoolean(
++        DFSConfigKeys.DFS_NAMENODE_DATANODE_REGISTRATION_UNSAFE_ALLOW_ADDRESS_OVERRIDE_KEY,
++        DFSConfigKeys.DFS_NAMENODE_DATANODE_REGISTRATION_UNSAFE_ALLOW_ADDRESS_OVERRIDE_DEFAULT);
++    LOG.info(DFSConfigKeys.DFS_NAMENODE_DATANODE_REGISTRATION_UNSAFE_ALLOW_ADDRESS_OVERRIDE_KEY
++        + "=" + allowRegistrationAddressOverride);
+     this.checkIpHostnameInRegistration = conf.getBoolean(
+         DFSConfigKeys.DFS_NAMENODE_DATANODE_REGISTRATION_IP_HOSTNAME_CHECK_KEY,
+         DFSConfigKeys.DFS_NAMENODE_DATANODE_REGISTRATION_IP_HOSTNAME_CHECK_DEFAULT);
+@@ -1133,27 +1140,29 @@ void startAdminOperationIfNecessary(DatanodeDescriptor nodeReg) {
+    */
+   public void registerDatanode(DatanodeRegistration nodeReg)
+       throws DisallowedDatanodeException, UnresolvedTopologyException {
+-    InetAddress dnAddress = Server.getRemoteIp();
+-    if (dnAddress != null) {
+-      // Mostly called inside an RPC, update ip and peer hostname
+-      String hostname = dnAddress.getHostName();
+-      String ip = dnAddress.getHostAddress();
+-      if (checkIpHostnameInRegistration && !isNameResolved(dnAddress)) {
+-        // Reject registration of unresolved datanode to prevent performance
+-        // impact of repetitive DNS lookups later.
+-        final String message = "hostname cannot be resolved (ip="
+-            + ip + ", hostname=" + hostname + ")";
+-        LOG.warn("Unresolved datanode registration: " + message);
+-        throw new DisallowedDatanodeException(nodeReg, message);
++    if (!allowRegistrationAddressOverride) {
++      InetAddress dnAddress = Server.getRemoteIp();
++      if (dnAddress != null) {
++        // Mostly called inside an RPC, update ip and peer hostname
++        String hostname = dnAddress.getHostName();
++        String ip = dnAddress.getHostAddress();
++        if (checkIpHostnameInRegistration && !isNameResolved(dnAddress)) {
++          // Reject registration of unresolved datanode to prevent performance
++          // impact of repetitive DNS lookups later.
++          final String message = "hostname cannot be resolved (ip="
++              + ip + ", hostname=" + hostname + ")";
++          LOG.warn("Unresolved datanode registration: " + message);
++          throw new DisallowedDatanodeException(nodeReg, message);
++        }
++        // update node registration with the ip and hostname from rpc request
++        nodeReg.setIpAddr(ip);
++        nodeReg.setPeerHostName(hostname);
+       }
+-      // update node registration with the ip and hostname from rpc request
+-      nodeReg.setIpAddr(ip);
+-      nodeReg.setPeerHostName(hostname);
+     }
+-    
++
+     try {
+       nodeReg.setExportedKeys(blockManager.getBlockKeys());
+-  
++
+       // Checks if the node is not on the hosts list.  If it is not, then
+       // it will be disallowed from registering. 
+       if (!hostConfigManager.isIncluded(nodeReg)) {
+diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DNConf.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DNConf.java
+index d61a17e83fe..eaf4a6d7c1d 100644
+--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DNConf.java
++++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DNConf.java
+@@ -99,6 +99,11 @@ public class DNConf {
+   final boolean syncOnClose;
+   final boolean encryptDataTransfer;
+   final boolean connectToDnViaHostname;
++  private final String registeredHostname;
++  private final int registeredDataPort;
++  private final int registeredHttpPort;
++  private final int registeredHttpsPort;
++  private final int registeredIpcPort;
+   final boolean overwriteDownstreamDerivedQOP;
+   private final boolean pmemCacheRecoveryEnabled;
+
+@@ -187,6 +192,11 @@ public DNConf(final Configurable dn) {
+     connectToDnViaHostname = getConf().getBoolean(
+         DFSConfigKeys.DFS_DATANODE_USE_DN_HOSTNAME,
+         DFSConfigKeys.DFS_DATANODE_USE_DN_HOSTNAME_DEFAULT);
++    registeredHostname = getConf().get(DFSConfigKeys.DFS_DATANODE_REGISTERED_HOSTNAME);
++    registeredDataPort = getConf().getInt(DFSConfigKeys.DFS_DATANODE_REGISTERED_DATA_PORT, -1);
++    registeredHttpPort = getConf().getInt(DFSConfigKeys.DFS_DATANODE_REGISTERED_HTTP_PORT, -1);
++    registeredHttpsPort = getConf().getInt(DFSConfigKeys.DFS_DATANODE_REGISTERED_HTTPS_PORT, -1);
++    registeredIpcPort = getConf().getInt(DFSConfigKeys.DFS_DATANODE_REGISTERED_IPC_PORT, -1);
+     this.blockReportInterval = getConf().getLong(
+         DFS_BLOCKREPORT_INTERVAL_MSEC_KEY,
+         DFS_BLOCKREPORT_INTERVAL_MSEC_DEFAULT);
+@@ -360,6 +370,66 @@ public boolean getConnectToDnViaHostname() {
+     return connectToDnViaHostname;
+   }
+
++  /**
++   * Returns a hostname to register with the cluster instead of the system
++   * hostname.
++   * This is an expert setting and can be used in multihoming scenarios to
++   * override the detected hostname.
++   *
++   * @return null if the system hostname should be used, otherwise a hostname
++   */
++  public String getRegisteredHostname() {
++      return registeredHostname;
++  }
++
++  /**
++   * Returns a port number to register with the cluster instead of the
++   * data port that the node is listening on.
++   * This is an expert setting and can be used in multihoming scenarios to
++   * override the detected port.
++   *
++   * @return -1 if the actual port should be used, otherwise a port number
++   */
++  public int getRegisteredDataPort() {
++    return registeredDataPort;
++  }
++
++  /**
++   * Returns a port number to register with the cluster instead of the
++   * HTTP port that the node is listening on.
++   * This is an expert setting and can be used in multihoming scenarios to
++   * override the detected port.
++   *
++   * @return -1 if the actual port should be used, otherwise a port number
++   */
++  public int getRegisteredHttpPort() {
++    return registeredHttpPort;
++  }
++
++  /**
++   * Returns a port number to register with the cluster instead of the
++   * HTTPS port that the node is listening on.
++   * This is an expert setting and can be used in multihoming scenarios to
++   * override the detected port.
++   *
++   * @return -1 if the actual port should be used, otherwise a port number
++   */
++  public int getRegisteredHttpsPort() {
++    return registeredHttpsPort;
++  }
++
++  /**
++   * Returns a port number to register with the cluster instead of the
++   * IPC port that the node is listening on.
++   * This is an expert setting and can be used in multihoming scenarios to
++   * override the detected port.
++   *
++   * @return -1 if the actual port should be used, otherwise a port number
++   */
++  public int getRegisteredIpcPort() {
++    return registeredIpcPort;
++  }
++
+   /**
+    * Returns socket timeout
+    * 
+diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
+index c1507a45120..d253779e70d 100644
+--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
++++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
+@@ -82,6 +82,7 @@
+ import java.util.Iterator;
+ import java.util.List;
+ import java.util.Map;
++import java.util.Optional;
+ import java.util.Map.Entry;
+ import java.util.Set;
+ import java.util.UUID;
+@@ -1556,11 +1557,35 @@ DatanodeRegistration createBPRegistration(NamespaceInfo nsInfo) {
+           NodeType.DATA_NODE);
+     }
+
+-    DatanodeID dnId = new DatanodeID(
+-        streamingAddr.getAddress().getHostAddress(), hostName, 
+-        storage.getDatanodeUuid(), getXferPort(), getInfoPort(),
+-            infoSecurePort, getIpcPort());
+-    return new DatanodeRegistration(dnId, storageInfo, 
++    String registeredHostname = Optional
++      .ofNullable(dnConf.getRegisteredHostname())
++      .orElseGet(() -> streamingAddr.getAddress().getHostAddress());
++    int registeredDataPort = dnConf.getRegisteredDataPort();
++    if (registeredDataPort == -1) {
++      registeredDataPort = getXferPort();
++    }
++    int registeredHttpPort = dnConf.getRegisteredHttpPort();
++    if (registeredHttpPort == -1) {
++      registeredHttpPort = getInfoPort();
++    }
++    int registeredHttpsPort = dnConf.getRegisteredHttpsPort();
++    if (registeredHttpsPort == -1) {
++      registeredHttpsPort = getInfoSecurePort();
++    }
++    int registeredIpcPort = dnConf.getRegisteredIpcPort();
++    if (registeredIpcPort == -1) {
++      registeredIpcPort = getIpcPort();
++    }
++
++    DatanodeID dnId = new DatanodeID(registeredHostname,
++                                     hostName,
++                                     storage.getDatanodeUuid(),
++                                     registeredDataPort,
++                                     registeredHttpPort,
++                                     registeredHttpsPort,
++                                     registeredIpcPort);
++
++    return new DatanodeRegistration(dnId, storageInfo,
+         new ExportedBlockKeys(), VersionInfo.getVersion());
+   }
+