HDFS: Allow overriding datanode registration addresses

conf.py

@@ -61,7 +61,7 @@
             {"product": "3.2.2", "java-base": "11", "jmx_exporter": "0.20.0"},
             {"product": "3.2.4", "java-base": "11", "jmx_exporter": "0.20.0"},
             {"product": "3.3.4", "java-base": "11", "jmx_exporter": "0.20.0"},
-            {"product": "3.3.6", "java-base": "11", "jmx_exporter": "0.20.0"},
+            {"product": "3.3.6", "java-base": "11", "jmx_exporter": "0.20.0", "node": "18.16.0"},
         ],
     },
     {

hadoop/Dockerfile

@@ -2,40 +2,28 @@
 FROM stackable/image/java-base AS builder
 
 ARG PRODUCT
+ARG NODE
 ARG JMX_EXPORTER
 
 # https://github.com/hadolint/hadolint/wiki/DL4006
 SHELL ["/bin/bash", "-o", "pipefail", "-c"]
 
-# unzip & zip are required for log4shell.sh
 # All others are required for the FUSE build
 RUN microdnf update && \
     microdnf install \
-    cmake \
-    cyrus-sasl-devel \
-    fuse-devel \
-    gcc \
-    gcc-c++ \
-    java-11-openjdk-devel \
-    maven \
-    openssl-devel \
-    tar \
-    unzip \
-    zip && \
+    # Required for Hadoop build
+    cmake cyrus-sasl-devel fuse-devel gcc gcc-c++ java-11-openjdk-devel maven openssl-devel tar xz git \
+    # Required for log4shell.sh
+    unzip zip && \
     microdnf clean all
 
 WORKDIR /stackable
 
 # This is needed here because it creates the JMX directory, we could create it any other way but this works
 COPY hadoop/stackable /stackable
 
-# The source is needed to build FUSE. The rest of the src package will not make it into the final image.
-# Both the src and binary variants extract into different root folders
-RUN curl --fail -L "https://repo.stackable.tech/repository/packages/hadoop/hadoop-${PRODUCT}-src.tar.gz" | tar -xzC . && \
-    curl --fail -L "https://repo.stackable.tech/repository/packages/hadoop/hadoop-${PRODUCT}.tar.gz" | tar -xzC . && \
-    ln -s "/stackable/hadoop-${PRODUCT}" /stackable/hadoop && \
-    rm -rf /stackable/hadoop/lib/native/examples && \
-    rm -rf /stackable/hadoop/share/doc
+# Build from source to enable FUSE module, and to apply custom patches.
+RUN curl --fail -L "https://repo.stackable.tech/repository/packages/hadoop/hadoop-${PRODUCT}-src.tar.gz" | tar -xzC .
 
 # The symlink from JMX Exporter 0.16.1 to the versionless link exists because old HDFS Operators (up until and including 23.7) used to hardcode
 # the version of JMX Exporter like this: "-javaagent:/stackable/jmx/jmx_prometheus_javaagent-0.16.1.jar"
@@ -48,6 +36,40 @@ RUN curl --fail "https://repo.stackable.tech/repository/packages/jmx-exporter/jm
     ln -s "/stackable/jmx/jmx_prometheus_javaagent-${JMX_EXPORTER}.jar" /stackable/jmx/jmx_prometheus_javaagent.jar && \
     ln -s /stackable/jmx/jmx_prometheus_javaagent.jar /stackable/jmx/jmx_prometheus_javaagent-0.16.1.jar
 
+# This Protobuf version is the exact version as used in the Hadoop Dockerfile
+# See https://github.com/apache/hadoop/blob/trunk/dev-support/docker/pkg-resolver/install-protobuf.sh
+# (this was hardcoded in the Dockerfile in earlier versions of Hadoop, make sure to look at the exact version in Github)
+# For now all versions of Hadoop we support use Protobuf 3.7.1 so we can hardcode it here.
+# Should it ever differ between versions we'll need to make this a variable as well.
+RUN mkdir -p /opt/protobuf-src && \
+    curl --fail -L -s -S https://repo.stackable.tech/repository/packages/protobuf/protobuf-java-3.7.1.tar.gz -o /opt/protobuf.tar.gz && \
+    tar xzf /opt/protobuf.tar.gz --strip-components 1 -C /opt/protobuf-src --no-same-owner && \
+    cd /opt/protobuf-src && \
+    ./configure --prefix=/opt/protobuf && \
+    make "-j$(nproc)" && \
+    make install && \
+    cd /root && \
+    rm -rf /opt/protobuf-src
+
+ENV PROTOBUF_HOME /opt/protobuf
+ENV PATH "${PATH}:/opt/protobuf/bin"
+
+RUN curl --fail -L https://repo.stackable.tech/repository/packages/node/node-v${NODE}-linux-x64.tar.xz | tar -xJC . && \
+    ln -s /stackable/node-v${NODE}-linux-x64 /stackable/node
+ENV PATH "${PATH}:/stackable/node/bin"
+RUN corepack enable yarn && \
+    yarn config set ignore-engines true
+
+RUN cd /stackable/hadoop-${PRODUCT}-src && \
+    # Hadoop Pipes requires libtirpc to build, whose headers are not packaged in RedHat UBI
+    git apply < /stackable/patches/0001-disable-pipes.patch && \
+    # Datanode registration override is required for Listener Operator integration
+    # Developed at https://github.com/stackabletech/hadoop/tree/spike/override-datanode-id
+    git apply < /stackable/patches/0002-datanode-registration-override.patch && \
+    mvn clean package -Pdist,native -Drequire.fuse=true -DskipTests -Dmaven.javadoc.skip=true && \
+    cp -r hadoop-dist/target/hadoop-${PRODUCT} /stackable/hadoop-${PRODUCT} && \
+    # HDFS fuse-dfs is not part of the regular dist output, so we need to copy it in ourselves
+    cp hadoop-hdfs-project/hadoop-hdfs-native-client/target/main/native/fuse-dfs/fuse_dfs /stackable/hadoop-${PRODUCT}/bin
 
 # ===
 # Mitigation for CVE-2021-44228 (Log4Shell)
@@ -72,33 +94,6 @@ COPY shared/log4shell_scanner /bin/log4shell_scanner
 RUN /bin/log4shell_scanner s "/stackable/hadoop-${PRODUCT}"
 # ===
 
-
-# This Protobuf version is the exact version as used in the Hadoop Dockerfile
-# See https://github.com/apache/hadoop/blob/trunk/dev-support/docker/pkg-resolver/install-protobuf.sh
-# (this was hardcoded in the Dockerfile in earlier versions of Hadoop, make sure to look at the exact version in Github)
-# For now all versions of Hadoop we support use Protobuf 3.7.1 so we can hardcode it here.
-# Should it ever differ between versions we'll need to make this a variable as well.
-RUN mkdir -p /opt/protobuf-src && \
-    curl --fail -L -s -S https://repo.stackable.tech/repository/packages/protobuf/protobuf-java-3.7.1.tar.gz -o /opt/protobuf.tar.gz && \
-    tar xzf /opt/protobuf.tar.gz --strip-components 1 -C /opt/protobuf-src --no-same-owner && \
-    cd /opt/protobuf-src && \
-    ./configure --prefix=/opt/protobuf && \
-    make "-j$(nproc)" && \
-    make install && \
-    cd /root && \
-    rm -rf /opt/protobuf-src
-
-ENV PROTOBUF_HOME /opt/protobuf
-ENV PATH "${PATH}:/opt/protobuf/bin"
-
-WORKDIR /stackable/hadoop-${PRODUCT}-src/hadoop-hdfs-project/hadoop-hdfs-native-client
-
-# This command comes from hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/fuse-dfs/doc/README
-RUN mvn clean package -Pnative -Drequire.fuse=true -DskipTests -Dmaven.javadoc.skip=true && \
-    cp target/main/native/fuse-dfs/fuse_dfs /stackable/hadoop/bin && \
-    rm -rf /stackable/hadoop-${PRODUCT}-src
-
-
 # Final Image
 FROM stackable/image/java-base
 

hadoop/stackable/patches/0001-disable-pipes.patch

@@ -0,0 +1,30 @@
+diff --git a/hadoop-tools/hadoop-tools-dist/pom.xml b/hadoop-tools/hadoop-tools-dist/pom.xml
+index 8a3e93c1037..8604a3325d9 100644
+--- a/hadoop-tools/hadoop-tools-dist/pom.xml
++++ b/hadoop-tools/hadoop-tools-dist/pom.xml
+@@ -85,13 +85,6 @@
+       <artifactId>hadoop-gridmix</artifactId>
+       <scope>compile</scope>
+     </dependency>
+-    <dependency>
+-      <groupId>org.apache.hadoop</groupId>
+-      <artifactId>hadoop-pipes</artifactId>
+-      <scope>compile</scope>
+-      <type>pom</type>
+-      <version>${project.version}</version>
+-    </dependency>
+     <dependency>
+       <groupId>org.apache.hadoop</groupId>
+       <artifactId>hadoop-aws</artifactId>
+diff --git a/hadoop-tools/pom.xml b/hadoop-tools/pom.xml
+index 4e934cd101f..2654dea5dd6 100644
+--- a/hadoop-tools/pom.xml
++++ b/hadoop-tools/pom.xml
+@@ -41,7 +41,6 @@
+     <module>hadoop-datajoin</module>
+     <module>hadoop-tools-dist</module>
+     <module>hadoop-extras</module>
+-    <module>hadoop-pipes</module>
+     <module>hadoop-openstack</module>
+     <module>hadoop-sls</module>
+     <module>hadoop-resourceestimator</module>

hadoop/stackable/patches/0002-datanode-registration-override.patch

@@ -0,0 +1,213 @@
+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 88a18d9cf07..b07fcb0b17a 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
+@@ -152,6 +152,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_ADVERTISED_HOSTNAME = "dfs.datanode.advertised.hostname";
++  public static final String  DFS_DATANODE_ADVERTISED_DATA_PORT = "dfs.datanode.advertised.port";
++  public static final String  DFS_DATANODE_ADVERTISED_HTTP_PORT = "dfs.datanode.advertised.http.port";
++  public static final String  DFS_DATANODE_ADVERTISED_HTTPS_PORT = "dfs.datanode.advertised.https.port";
++  public static final String  DFS_DATANODE_ADVERTISED_IPC_PORT = "dfs.datanode.advertised.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";
+@@ -484,6 +491,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 bdd20d7e276..c10db0611c9 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
+@@ -181,6 +181,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.
+@@ -314,6 +316,11 @@ public class DatanodeManager {
+     // Block invalidate limit also has some dependency on heartbeat interval.
+     // Check setBlockInvalidateLimit().
+     setBlockInvalidateLimit(configuredBlockInvalidateLimit);
++    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);
+@@ -1146,27 +1153,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 9b5343321d3..8ce6a61204b 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
+@@ -100,6 +100,11 @@ public class DNConf {
+   final boolean syncOnClose;
+   final boolean encryptDataTransfer;
+   final boolean connectToDnViaHostname;
++  private final String advertisedHostname;
++  private final int advertisedDataPort;
++  private final int advertisedHttpPort;
++  private final int advertisedHttpsPort;
++  private final int advertisedIpcPort;
+   final boolean overwriteDownstreamDerivedQOP;
+   private final boolean pmemCacheRecoveryEnabled;
+
+@@ -188,6 +193,11 @@ public DNConf(final Configurable dn) {
+     connectToDnViaHostname = getConf().getBoolean(
+         DFSConfigKeys.DFS_DATANODE_USE_DN_HOSTNAME,
+         DFSConfigKeys.DFS_DATANODE_USE_DN_HOSTNAME_DEFAULT);
++    advertisedHostname = getConf().get(DFSConfigKeys.DFS_DATANODE_ADVERTISED_HOSTNAME);
++    advertisedDataPort = getConf().getInt(DFSConfigKeys.DFS_DATANODE_ADVERTISED_DATA_PORT, -1);
++    advertisedHttpPort = getConf().getInt(DFSConfigKeys.DFS_DATANODE_ADVERTISED_HTTP_PORT, -1);
++    advertisedHttpsPort = getConf().getInt(DFSConfigKeys.DFS_DATANODE_ADVERTISED_HTTPS_PORT, -1);
++    advertisedIpcPort = getConf().getInt(DFSConfigKeys.DFS_DATANODE_ADVERTISED_IPC_PORT, -1);
+     this.blockReportInterval = getConf().getLong(
+         DFS_BLOCKREPORT_INTERVAL_MSEC_KEY,
+         DFS_BLOCKREPORT_INTERVAL_MSEC_DEFAULT);
+@@ -362,6 +372,32 @@ public boolean getConnectToDnViaHostname() {
+     return connectToDnViaHostname;
+   }
+
++  /**
++   * Returns a hostname to advertise 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 getAdvertisedHostname() {
++    return advertisedHostname;
++  }
++
++  public int getAdvertisedDataPort() {
++    return advertisedDataPort;
++  }
++
++  public int getAdvertisedHttpPort() {
++    return advertisedHttpPort;
++  }
++
++  public int getAdvertisedHttpsPort() {
++    return advertisedHttpsPort;
++  }
++
++  public int getAdvertisedIpcPort() {
++    return advertisedIpcPort;
++  }
++
+   /**
+    * 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 8fb009dab85..228bcce62b3 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
+@@ -133,6 +133,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;
+@@ -2053,11 +2054,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 advertisedHostname = Optional
++      .ofNullable(dnConf.getAdvertisedHostname())
++      .orElseGet(() -> streamingAddr.getAddress().getHostAddress());
++    int advertisedDataPort = dnConf.getAdvertisedDataPort();
++    if (advertisedDataPort == -1) {
++      advertisedDataPort = getXferPort();
++    }
++    int advertisedHttpPort = dnConf.getAdvertisedHttpPort();
++    if (advertisedHttpPort == -1) {
++      advertisedHttpPort = getInfoPort();
++    }
++    int advertisedHttpsPort = dnConf.getAdvertisedHttpPort();
++    if (advertisedHttpsPort == -1) {
++      advertisedHttpPort = getInfoSecurePort();
++    }
++    int advertisedIpcPort = dnConf.getAdvertisedIpcPort();
++    if (advertisedIpcPort == -1) {
++      advertisedIpcPort = getIpcPort();
++    }
++
++    DatanodeID dnId = new DatanodeID(advertisedHostname,
++                                     hostName,
++                                     storage.getDatanodeUuid(),
++                                     advertisedDataPort,
++                                     advertisedHttpPort,
++                                     advertisedHttpsPort,
++                                     advertisedIpcPort);
++
++    return new DatanodeRegistration(dnId, storageInfo,
+         new ExportedBlockKeys(), VersionInfo.getVersion());
+   }
+