diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/HddsConfigKeys.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/HddsConfigKeys.java index 8e9d4e9e098b..973d2c654c0b 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/HddsConfigKeys.java +++ b/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 = diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMNodeInfo.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMNodeInfo.java index 0006b2762d40..1faa834f638b 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMNodeInfo.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMNodeInfo.java @@ -229,15 +229,15 @@ public String getNodeId() { } public String getBlockClientAddress() { - return blockClientAddress.getHostAndPortString(); + return blockClientAddress == null ? null : blockClientAddress.getHostAndPortString(); } public String getScmClientAddress() { - return scmClientAddress.getHostAndPortString(); + return scmClientAddress == null ? null : scmClientAddress.getHostAndPortString(); } public String getScmSecurityAddress() { - return scmSecurityAddress.getHostAndPortString(); + return scmSecurityAddress == null ? null : scmSecurityAddress.getHostAndPortString(); } public HostAndPort getScmDatanodeHostPortAddress() { @@ -245,6 +245,6 @@ public HostAndPort getScmDatanodeHostPortAddress() { } public String getScmDatanodeAddress() { - return scmDatanodeAddress.getHostAndPortString(); + return scmDatanodeAddress == null ? null : scmDatanodeAddress.getHostAndPortString(); } } diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/net/HostAndPort.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/net/HostAndPort.java index c56776d74b55..55500bae553a 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/net/HostAndPort.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/net/HostAndPort.java @@ -17,6 +17,7 @@ package org.apache.hadoop.hdds.scm.net; +import java.net.InetAddress; import java.net.InetSocketAddress; import org.apache.hadoop.net.NetUtils; @@ -44,6 +45,12 @@ public HostAndPort(String host, int port) { this(host, port, null); } + /** + * Derives the host from {@code address.getHostName()}, which can differ from the configured + * spelling used by {@link #HostAndPort(String, int)} (IP literal vs hostname). Since equals/hashCode + * key on the host string, callers must use one consistent spelling per node or instances will not + * match in the endpoint maps. + */ public HostAndPort(InetSocketAddress address) { this(address.getHostName(), address.getPort(), address); } @@ -64,6 +71,26 @@ public synchronized InetSocketAddress getAddress() { return address; } + /** + * Re-resolves the host:port and swaps the cached address if the IP changed. The DNS lookup runs + * outside the monitor; only the swap is synchronized, mirroring {@link #getAddress()}. + * @return true if the cached address changed + */ + public boolean refresh() { + final InetSocketAddress latest = NetUtils.createSocketAddr(hostAndPortString); + final InetAddress latestIp = latest.getAddress(); + if (latestIp == null) { + return false; + } + synchronized (this) { + if (latestIp.equals(address.getAddress())) { + return false; + } + address = latest; + return true; + } + } + @Override public int hashCode() { return hash; diff --git a/hadoop-hdds/common/src/main/resources/ozone-default.xml b/hadoop-hdds/common/src/main/resources/ozone-default.xml index dd33a02b02a7..8ad5b0f8796b 100644 --- a/hadoop-hdds/common/src/main/resources/ozone-default.xml +++ b/hadoop-hdds/common/src/main/resources/ozone-default.xml @@ -3964,6 +3964,19 @@ + + hdds.heartbeat.address.refresh.threshold + 3 + OZONE, DATANODE, HA + Number of consecutive heartbeat failures the DataNode + tolerates against the same SCM endpoint before re-resolving its + hostname. 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. + + + ozone.directory.deleting.service.interval 1m diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/SCMConnectionManager.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/SCMConnectionManager.java index 25be82cd0414..643053df07cc 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/SCMConnectionManager.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/SCMConnectionManager.java @@ -139,33 +139,7 @@ public void addSCMServer(HostAndPort address, "Ignoring the request."); return; } - - Configuration hadoopConfig = - LegacyHadoopConfigurationSource.asHadoopConfiguration(this.conf); - RPC.setProtocolEngine( - hadoopConfig, - StorageContainerDatanodeProtocolPB.class, - ProtobufRpcEngine.class); - long version = - RPC.getProtocolVersion(StorageContainerDatanodeProtocolPB.class); - - RetryPolicy retryPolicy = - RetryPolicies.retryUpToMaximumCountWithFixedSleep( - getScmRpcRetryCount(conf), getScmRpcRetryInterval(conf), - TimeUnit.MILLISECONDS); - - StorageContainerDatanodeProtocolPB rpcProxy = RPC.getProtocolProxy( - StorageContainerDatanodeProtocolPB.class, version, - address.getAddress(), UserGroupInformation.getCurrentUser(), hadoopConfig, - NetUtils.getDefaultSocketFactory(hadoopConfig), getRpcTimeout(), - retryPolicy).getProxy(); - - StorageContainerDatanodeProtocolClientSideTranslatorPB rpcClient = - new StorageContainerDatanodeProtocolClientSideTranslatorPB( - rpcProxy); - - EndpointStateMachine endPoint = new EndpointStateMachine(address, - rpcClient, this.conf, threadNamePrefix); + EndpointStateMachine endPoint = buildScmEndpoint(address, threadNamePrefix); endPoint.setPassive(false); scmMachines.put(address, endPoint); } finally { @@ -173,6 +147,73 @@ public void addSCMServer(HostAndPort address, } } + /** + * Builds (but does not register) an active-SCM endpoint dialing {@code address.getAddress()}. + * Factored out so {@link #refreshSCMServer} can rebuild an endpoint on a freshly-resolved IP. + */ + private EndpointStateMachine buildScmEndpoint(HostAndPort address, String threadNamePrefix) + throws IOException { + Configuration hadoopConfig = + LegacyHadoopConfigurationSource.asHadoopConfiguration(this.conf); + RPC.setProtocolEngine(hadoopConfig, StorageContainerDatanodeProtocolPB.class, + ProtobufRpcEngine.class); + long version = RPC.getProtocolVersion(StorageContainerDatanodeProtocolPB.class); + RetryPolicy retryPolicy = RetryPolicies.retryUpToMaximumCountWithFixedSleep( + getScmRpcRetryCount(conf), getScmRpcRetryInterval(conf), TimeUnit.MILLISECONDS); + StorageContainerDatanodeProtocolPB rpcProxy = RPC.getProtocolProxy( + StorageContainerDatanodeProtocolPB.class, version, + address.getAddress(), UserGroupInformation.getCurrentUser(), hadoopConfig, + NetUtils.getDefaultSocketFactory(hadoopConfig), getRpcTimeout(), + retryPolicy).getProxy(); + StorageContainerDatanodeProtocolClientSideTranslatorPB rpcClient = + new StorageContainerDatanodeProtocolClientSideTranslatorPB(rpcProxy); + return new EndpointStateMachine(address, rpcClient, this.conf, threadNamePrefix); + } + + /** + * Re-resolves the hostname of the active SCM endpoint at {@code address} and, if the IP changed, + * rebuilds the endpoint in place under the same key. Because the key is the stable host:port, no + * re-keying or StateContext queue migration is needed -- only the {@link EndpointStateMachine} + * value (which holds the frozen RPC proxy) is replaced. + * + * @return true if the endpoint was rebuilt on a new address + */ + public boolean refreshSCMServer(HostAndPort address, String threadNamePrefix) + throws IOException { + final EndpointStateMachine current; + readLock(); + try { + current = scmMachines.get(address); + if (current == null || current.isPassive()) { + return false; + } + } finally { + readUnlock(); + } + // DNS lookup runs outside the map lock; HostAndPort swaps its cached address only on a change. + if (!address.refresh()) { + return false; + } + final EndpointStateMachine stale; + writeLock(); + try { + if (scmMachines.get(address) != current) { + // Lost the race: another refresh or removeSCMServer replaced this entry. Abandon. + return false; + } + EndpointStateMachine rebuilt = buildScmEndpoint(address, threadNamePrefix); + rebuilt.setPassive(false); + scmMachines.put(address, rebuilt); + stale = current; + } finally { + writeUnlock(); + } + // Close the stale proxy outside the lock; teardown can block on socket close. + stale.close(); + LOG.info("Re-resolved SCM endpoint {} after connection failure.", address); + return true; + } + /** * Adds a new Recon server to the set of endpoints. * diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java index 7cb24558c7cb..497d0042ea0f 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java @@ -19,8 +19,12 @@ import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_CONTAINER_ACTION_MAX_LIMIT; import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_CONTAINER_ACTION_MAX_LIMIT_DEFAULT; +import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_ADDRESS_REFRESH_THRESHOLD; +import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_ADDRESS_REFRESH_THRESHOLD_DEFAULT; import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_PIPELINE_ACTION_MAX_LIMIT; import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_PIPELINE_ACTION_MAX_LIMIT_DEFAULT; +import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_CLIENT_FAILOVER_RESOLVE_NEEDED_DEFAULT; +import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_CLIENT_FAILOVER_RESOLVE_NEEDED_KEY; import static org.apache.hadoop.ozone.container.upgrade.UpgradeUtils.toLayoutVersionProto; import com.google.common.base.Preconditions; @@ -44,6 +48,7 @@ import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMHeartbeatRequestProto; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMHeartbeatResponseProto; import org.apache.hadoop.hdds.upgrade.HDDSLayoutVersionManager; +import org.apache.hadoop.hdds.utils.ConnectionFailureUtils; import org.apache.hadoop.hdfs.util.EnumCounters; import org.apache.hadoop.ozone.container.common.helpers.DeletedContainerBlocksSummary; import org.apache.hadoop.ozone.container.common.statemachine.EndpointStateMachine; @@ -77,6 +82,8 @@ public class HeartbeatEndpointTask private int maxContainerActionsPerHB; private int maxPipelineActionsPerHB; private HDDSLayoutVersionManager layoutVersionManager; + private final boolean resolveOnFailureEnabled; + private final int refreshThreshold; /** * Constructs a SCM heart beat. @@ -100,6 +107,10 @@ public HeartbeatEndpointTask(EndpointStateMachine rpcEndpoint, } else { this.layoutVersionManager = context.getParent().getLayoutVersionManager(); } + this.resolveOnFailureEnabled = conf.getBoolean(OZONE_CLIENT_FAILOVER_RESOLVE_NEEDED_KEY, + OZONE_CLIENT_FAILOVER_RESOLVE_NEEDED_DEFAULT); + this.refreshThreshold = Math.max(1, conf.getInt(HDDS_HEARTBEAT_ADDRESS_REFRESH_THRESHOLD, + HDDS_HEARTBEAT_ADDRESS_REFRESH_THRESHOLD_DEFAULT)); } /** @@ -157,12 +168,34 @@ public EndpointStateMachine.EndPointStates call() throws Exception { // put back the reports which failed to be sent putBackIncrementalReports(requestBuilder); rpcEndpoint.logIfNeeded(ex); + maybeRefreshScmAddress(ex); } finally { rpcEndpoint.unlock(); } return rpcEndpoint.getState(); } + /** + * On a connection-class heartbeat failure, once the missed-count threshold is reached and DNS + * refresh is enabled, ask the connection manager to re-resolve this SCM peer and rebuild the + * endpoint if its IP changed. Off by default (ozone.client.failover.resolve-needed). + */ + private void maybeRefreshScmAddress(IOException heartbeatFailure) { + if (!resolveOnFailureEnabled + || rpcEndpoint.isPassive() + || rpcEndpoint.getMissedCount() < refreshThreshold + || !ConnectionFailureUtils.isConnectionFailure(heartbeatFailure)) { + return; + } + try { + context.getParent().getConnectionManager() + .refreshSCMServer(rpcEndpoint.getAddress(), context.getThreadNamePrefix()); + } catch (IOException ex) { + LOG.warn("Failed to refresh SCM address {} after {} missed heartbeats", + rpcEndpoint.getAddress(), rpcEndpoint.getMissedCount(), ex); + } + } + // TODO: Make it generic. private void putBackIncrementalReports( SCMHeartbeatRequestProto.Builder requestBuilder) { diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/states/endpoint/TestHeartbeatEndpointTaskDnsRefresh.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/states/endpoint/TestHeartbeatEndpointTaskDnsRefresh.java new file mode 100644 index 000000000000..485a33fca5f9 --- /dev/null +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/states/endpoint/TestHeartbeatEndpointTaskDnsRefresh.java @@ -0,0 +1,136 @@ +/* + * 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.ozone.container.common.states.endpoint; + +import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_ADDRESS_REFRESH_THRESHOLD; +import static org.apache.hadoop.hdds.upgrade.HDDSLayoutVersionManager.maxLayoutVersion; +import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_CLIENT_FAILOVER_RESOLVE_NEEDED_KEY; +import static org.mockito.Mockito.any; +import static org.mockito.Mockito.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.io.IOException; +import java.net.ConnectException; +import java.util.UUID; +import org.apache.hadoop.hdds.conf.OzoneConfiguration; +import org.apache.hadoop.hdds.protocol.DatanodeDetails; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMCommandProto; +import org.apache.hadoop.hdds.scm.net.HostAndPort; +import org.apache.hadoop.hdds.upgrade.HDDSLayoutVersionManager; +import org.apache.hadoop.hdfs.util.EnumCounters; +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.EndpointStateMachine; +import org.apache.hadoop.ozone.container.common.statemachine.SCMConnectionManager; +import org.apache.hadoop.ozone.container.common.statemachine.StateContext; +import org.apache.hadoop.ozone.protocolPB.StorageContainerDatanodeProtocolClientSideTranslatorPB; +import org.junit.jupiter.api.Test; + +/** + * Verifies that a connection-class heartbeat failure triggers a DNS re-resolution of the SCM peer + * (HDDS-15533), gated by the resolve-needed flag, the passive (Recon) check, the missed-heartbeat + * threshold, and the connection-class classifier. Because #10612 keys SCM identity on the stable + * host:port, the trigger only asks the connection manager to rebuild the endpoint -- there is no + * re-keying or StateContext queue migration to assert here. + */ +public class TestHeartbeatEndpointTaskDnsRefresh { + + private static final HostAndPort SCM = new HostAndPort("test-scm-1", 9861); + + @Test + public void connectionFailureAtThresholdTriggersRefresh() throws Exception { + OzoneConfiguration conf = new OzoneConfiguration(); + conf.setBoolean(OZONE_CLIENT_FAILOVER_RESOLVE_NEEDED_KEY, true); + conf.setInt(HDDS_HEARTBEAT_ADDRESS_REFRESH_THRESHOLD, 2); + SCMConnectionManager cm = runHeartbeat(conf, 3, new ConnectException("refused")); + verify(cm, times(1)).refreshSCMServer(eq(SCM), any()); + } + + @Test + public void flagOffSuppressesRefresh() throws Exception { + OzoneConfiguration conf = new OzoneConfiguration(); + conf.setBoolean(OZONE_CLIENT_FAILOVER_RESOLVE_NEEDED_KEY, false); + SCMConnectionManager cm = runHeartbeat(conf, 5, new ConnectException("refused")); + verify(cm, never()).refreshSCMServer(any(), any()); + } + + @Test + public void applicationErrorDoesNotTriggerRefresh() throws Exception { + OzoneConfiguration conf = new OzoneConfiguration(); + conf.setBoolean(OZONE_CLIENT_FAILOVER_RESOLVE_NEEDED_KEY, true); + SCMConnectionManager cm = runHeartbeat(conf, 5, new IOException("application-level")); + verify(cm, never()).refreshSCMServer(any(), any()); + } + + @Test + public void belowThresholdDoesNotTriggerRefresh() throws Exception { + OzoneConfiguration conf = new OzoneConfiguration(); + conf.setBoolean(OZONE_CLIENT_FAILOVER_RESOLVE_NEEDED_KEY, true); + conf.setInt(HDDS_HEARTBEAT_ADDRESS_REFRESH_THRESHOLD, 5); + SCMConnectionManager cm = runHeartbeat(conf, 1, new ConnectException("refused")); + verify(cm, never()).refreshSCMServer(any(), any()); + } + + /** + * Drives a single heartbeat that fails with {@code failure}, with the endpoint reporting + * {@code missedCount} missed heartbeats, and returns the mocked connection manager to verify. + */ + private SCMConnectionManager runHeartbeat(OzoneConfiguration conf, long missedCount, + IOException failure) throws Exception { + StorageContainerDatanodeProtocolClientSideTranslatorPB proxy = + mock(StorageContainerDatanodeProtocolClientSideTranslatorPB.class); + when(proxy.sendHeartbeat(any())).thenThrow(failure); + + EndpointStateMachine endpoint = mock(EndpointStateMachine.class); + when(endpoint.getEndPoint()).thenReturn(proxy); + when(endpoint.getAddress()).thenReturn(SCM); + when(endpoint.getMissedCount()).thenReturn(missedCount); + when(endpoint.isPassive()).thenReturn(false); + + SCMConnectionManager connectionManager = mock(SCMConnectionManager.class); + DatanodeStateMachine dsm = mock(DatanodeStateMachine.class); + when(dsm.getConnectionManager()).thenReturn(connectionManager); + when(dsm.getQueuedCommandCount()) + .thenReturn(new EnumCounters<>(SCMCommandProto.Type.class)); + StateContext context = new StateContext(conf, DatanodeStates.RUNNING, dsm, ""); + + HDDSLayoutVersionManager lvm = mock(HDDSLayoutVersionManager.class); + when(lvm.getSoftwareLayoutVersion()).thenReturn(maxLayoutVersion()); + when(lvm.getMetadataLayoutVersion()).thenReturn(maxLayoutVersion()); + + DatanodeDetails dn = DatanodeDetails.newBuilder() + .setUuid(UUID.randomUUID()) + .setHostName("localhost") + .setIpAddress("127.0.0.1") + .build(); + + HeartbeatEndpointTask.newBuilder() + .setConfig(conf) + .setDatanodeDetails(dn) + .setContext(context) + .setLayoutVersionManager(lvm) + .setEndpointStateMachine(endpoint) + .build() + .call(); + return connectionManager; + } +}