HDDS-15533. DNS refresh on heartbeat failure for DN to SCM

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

@@ -39,6 +39,13 @@ public final class HddsConfigKeys {
       "hdds.heartbeat.recon.initial-interval";
   public static final String HDDS_RECON_INITIAL_HEARTBEAT_INTERVAL_DEFAULT =
       "2s";
+  /**
+   * Number of consecutive heartbeat failures the DN tolerates against the same SCM endpoint before
+   * re-resolving its hostname. Only consulted when ozone.client.failover.resolve-needed is true.
+   */
+  public static final String HDDS_HEARTBEAT_ADDRESS_REFRESH_THRESHOLD =
+      "hdds.heartbeat.address.refresh.threshold";
+  public static final int HDDS_HEARTBEAT_ADDRESS_REFRESH_THRESHOLD_DEFAULT = 3;
   public static final String HDDS_NODE_REPORT_INTERVAL =
       "hdds.node.report.interval";
   public static final String HDDS_NODE_REPORT_INTERVAL_DEFAULT =

hadoop-hdds/common/src/main/resources/ozone-default.xml

@@ -3953,6 +3953,19 @@
     </description>
   </property>
 
+  <property>
+    <name>hdds.heartbeat.address.refresh.threshold</name>
+    <value>3</value>
+    <tag>OZONE, DATANODE, HA</tag>
+    <description>Number of consecutive heartbeat failures the
+      DataNode tolerates against the same SCM endpoint before
+      attempting a DNS re-resolution. Only consulted when
+      ozone.client.failover.resolve-needed is true. Conservative
+      default avoids re-resolution on transient blips while still
+      recovering from a peer pod IP change within seconds.
+    </description>
+  </property>
+
   <property>
     <name>ozone.directory.deleting.service.interval</name>
     <value>1m</value>

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

@@ -799,7 +799,9 @@ private String reconfigScmNodes(String value) {
         continue;
       }
       try {
-        connectionManager.addSCMServer(scmAddress, context.getThreadNamePrefix());
+        String hostAndPort = scmAddress.getHostString() + ":" + scmAddress.getPort();
+        connectionManager.addSCMServer(scmAddress, hostAndPort,
+            context.getThreadNamePrefix());
         context.addEndpoint(scmAddress);
         effectiveScmNodeIds.add(scmNodeId);
         LOG.info("Reconfiguration successfully add SCM address {} for SCM service {}", scmAddress, scmServiceId);

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

@@ -22,6 +22,7 @@
 
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import java.io.Closeable;
+import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.time.ZonedDateTime;
 import java.util.concurrent.ExecutorService;
@@ -31,6 +32,7 @@
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 import org.apache.hadoop.hdds.conf.ConfigurationSource;
+import org.apache.hadoop.net.NetUtils;
 import org.apache.hadoop.ozone.protocol.VersionResponse;
 import org.apache.hadoop.ozone.protocolPB.ReconDatanodeProtocolPB;
 import org.apache.hadoop.ozone.protocolPB.StorageContainerDatanodeProtocolClientSideTranslatorPB;
@@ -47,6 +49,19 @@ public class EndpointStateMachine
   private final StorageContainerDatanodeProtocolClientSideTranslatorPB endPoint;
   private final AtomicLong missedCount;
   private final InetSocketAddress address;
+  /**
+   * The original "host:port" string used to resolve {@link #address}.
+   * Preserved so that we can re-resolve DNS when the cached IP becomes
+   * stale (e.g. after a Kubernetes pod restart of the SCM peer). Since
+   * {@link InetSocketAddress} performs a one-shot DNS lookup at
+   * construction and freezes the IP, we cannot recover from a peer IP
+   * change without rebuilding the address from this hostname string.
+   * Null only in the legacy code path where the caller did not preserve
+   * the original config string (in which case re-resolution is disabled
+   * for that endpoint and the operator must restart the DN to pick up
+   * the new IP).
+   */
+  private final String hostAndPort;
   private final Lock lock;
   private final ConfigurationSource conf;
   private EndPointStates state = EndPointStates.FIRST;
@@ -67,9 +82,23 @@ public class EndpointStateMachine
   public EndpointStateMachine(InetSocketAddress address,
       StorageContainerDatanodeProtocolClientSideTranslatorPB endPoint,
       ConfigurationSource conf, String threadNamePrefix) {
+    this(address, null, endPoint, conf, threadNamePrefix);
+  }
+
+  /**
+   * Constructs RPC Endpoints, preserving the original host:port string
+   * so DNS can be re-resolved on heartbeat failure.
+   * @param address     resolved address used to build the RPC proxy
+   * @param hostAndPort the original "host:port" string, or null if the
+   *                    caller did not preserve it (re-resolution disabled)
+   */
+  public EndpointStateMachine(InetSocketAddress address, String hostAndPort,
+      StorageContainerDatanodeProtocolClientSideTranslatorPB endPoint,
+      ConfigurationSource conf, String threadNamePrefix) {
     this.endPoint = endPoint;
     this.missedCount = new AtomicLong(0);
     this.address = address;
+    this.hostAndPort = hostAndPort;
     lock = new ReentrantLock();
     this.conf = conf;
     executorService = Executors.newSingleThreadExecutor(
@@ -79,6 +108,48 @@ public EndpointStateMachine(InetSocketAddress address,
             .build());
   }
 
+  /**
+   * The original "host:port" string used to construct {@link #getAddress()}.
+   * @return the host:port string, or null if not preserved at construction
+   */
+  public String getHostAndPort() {
+    return hostAndPort;
+  }
+
+  /**
+   * Re-resolves the configured hostname and reports whether the resolved IP differs from the cached
+   * {@link #address}. Does not mutate any state on this endpoint -- the caller is responsible for
+   * swapping this endpoint out (via {@link SCMConnectionManager#refreshSCMServer}) if a refresh is
+   * desired.
+   * <p>
+   * Returns null when {@link #hostAndPort} was not preserved at construction, when the resolved IP
+   * is unchanged, or when DNS resolution produced no address. Returns the freshly-resolved
+   * {@link InetSocketAddress} when the IP has changed under the same hostname, and throws
+   * {@link IllegalStateException} if {@link #hostAndPort} is malformed.
+   */
+  public InetSocketAddress resolveLatestAddress() {
+    if (hostAndPort == null) {
+      return null;
+    }
+    final InetSocketAddress refreshed;
+    try {
+      refreshed = NetUtils.createSocketAddr(hostAndPort);
+    } catch (IllegalArgumentException ex) {
+      throw new IllegalStateException("Malformed host address: " + hostAndPort, ex);
+    }
+    final InetAddress refreshedIp = refreshed.getAddress();
+    if (refreshedIp == null) {
+      LOG.warn("Failed to resolve {}; reusing previous address {}.", hostAndPort, address);
+      return null;
+    }
+    // A previously-unresolved cached address (address.getAddress() == null) compares unequal here, so
+    // a now-successful resolution correctly counts as a change.
+    if (refreshedIp.equals(address.getAddress())) {
+      return null;
+    }
+    return refreshed;
+  }
+
   /**
    * Takes a lock on this EndPoint so that other threads don't use this while we
    * are trying to communicate via this endpoint.
@@ -150,13 +221,23 @@ public ExecutorService getExecutorService() {
 
   /**
    * Closes the connection.
+   * <p>
+   * The underlying {@code endPoint.close()} delegates to
+   * {@code RPC.stopProxy} which can raise a RuntimeException; without
+   * the try/finally below, the per-endpoint executor would leak its
+   * thread on close failure. Callers (notably
+   * {@code SCMConnectionManager.refreshSCMServer}) catch RuntimeException
+   * from close, so the leak would otherwise be silent.
    */
   @Override
   public void close() {
-    if (endPoint != null) {
-      endPoint.close();
+    try {
+      if (endPoint != null) {
+        endPoint.close();
+      }
+    } finally {
+      executorService.shutdown();
     }
-    executorService.shutdown();
   }
 
   /**