From 6c998c379e9526ac3931792c91bd7c6647e9df3a Mon Sep 17 00:00:00 2001 From: Manish Khandelwal Date: Fri, 2 May 2025 15:41:42 +0530 Subject: [PATCH] CASSANDRA-20291 Filter unreachable private IPs from batch log writing Add filtering logic to remove reachable private IPs from list of nodes to be selected for batch log writes. Patch By Manish Khandelwal, Reviewed by TBD for CASSANDRA-20291 --- .../cassandra/locator/ReplicaPlans.java | 113 +++++++++++++++++- 1 file changed, 109 insertions(+), 4 deletions(-) diff --git a/src/java/org/apache/cassandra/locator/ReplicaPlans.java b/src/java/org/apache/cassandra/locator/ReplicaPlans.java index b90c65c785..cc65fd5c5f 100644 --- a/src/java/org/apache/cassandra/locator/ReplicaPlans.java +++ b/src/java/org/apache/cassandra/locator/ReplicaPlans.java @@ -18,6 +18,10 @@ package org.apache.cassandra.locator; +import java.io.IOException; +import java.net.InetSocketAddress; +import java.net.Socket; +import java.net.UnknownHostException; import java.util.ArrayList; import java.util.Collection; import java.util.Collections; @@ -25,7 +29,9 @@ import java.util.HashSet; import java.util.LinkedHashSet; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ThreadLocalRandom; import java.util.function.Consumer; import java.util.function.Function; @@ -55,7 +61,11 @@ import org.apache.cassandra.db.PartitionPosition; import org.apache.cassandra.dht.AbstractBounds; import org.apache.cassandra.dht.Token; import org.apache.cassandra.exceptions.UnavailableException; +import org.apache.cassandra.gms.ApplicationState; +import org.apache.cassandra.gms.EndpointState; import org.apache.cassandra.gms.FailureDetector; +import org.apache.cassandra.gms.Gossiper; +import org.apache.cassandra.gms.VersionedValue; import org.apache.cassandra.schema.SchemaConstants; import org.apache.cassandra.service.StorageService; import org.apache.cassandra.service.reads.AlwaysSpeculativeRetryPolicy; @@ -80,6 +90,9 @@ public class ReplicaPlans private static final int REQUIRED_BATCHLOG_REPLICA_COUNT = Math.max(1, Math.min(2, CassandraRelevantProperties.REQUIRED_BATCHLOG_REPLICA_COUNT.getInt())); + private static final Map reachabilityCache = new ConcurrentHashMap<>(); + private static final int CACHE_TTL_MS = 30000; // 30 seconds + static { int batchlogReplicaCount = CassandraRelevantProperties.REQUIRED_BATCHLOG_REPLICA_COUNT.getInt(); @@ -295,8 +308,27 @@ public class ReplicaPlans for (Map.Entry entry : endpoints.entries()) { InetAddressAndPort addr = entry.getValue(); - if (!addr.equals(FBUtilities.getBroadcastAddressAndPort()) && isAlive.test(addr)) - validated.put(entry.getKey(), entry.getValue()); + + // Skip local address and nodes marked down by FailureDetector + if (addr.equals(FBUtilities.getBroadcastAddressAndPort())) + continue; + + if (!isAlive.test(addr)) + continue; + + // Check for intra dc connectivity on private network. Cache the result for 30 seconds. + // Valid for topology where two newtwork interfaces are used. + Optional maybeInternal = getInternalAddressAndPort(addr); + if (maybeInternal.isPresent()) { + InetAddressAndPort internal = maybeInternal.get(); + if (isReachableWithCache(internal, CACHE_TTL_MS)) { + validated.put(entry.getKey(), addr); + } + } else { + // No internal address means either it's a single-interface node or gossip isn't set up; + // trust isAlive (failure detector) in this case. + validated.put(entry.getKey(), addr); + } } // return early if no more than 2 nodes: @@ -413,14 +445,87 @@ public class ReplicaPlans continue; if (result.contains(endpoint)) continue; - result.add(endpoint); } } - return result; } + private static Optional getInternalAddressAndPort(InetAddressAndPort endpoint) + { + EndpointState state = Gossiper.instance.getEndpointStateForEndpoint(endpoint); + if (state == null) + { + logger.debug("No EndpointState found for endpoint: {}", endpoint); + return Optional.empty(); + } + + VersionedValue internal = state.getApplicationState(ApplicationState.INTERNAL_ADDRESS_AND_PORT); + if (internal == null || internal.value == null || internal.value.trim().isEmpty()) + { + logger.debug("No INTERNAL_ADDRESS_AND_PORT state for endpoint: {}", endpoint); + return Optional.empty(); + } + + try + { + InetAddressAndPort internalIp = InetAddressAndPort.getByName(internal.value.trim()); + return Optional.of(InetAddressAndPort.getByAddressOverrideDefaults(internalIp.getAddress(), internalIp.getPort())); + } + catch (UnknownHostException e) + { + logger.warn("Failed to parse INTERNAL_ADDRESS_AND_PORT [{}] for endpoint: {} due to {}", internal.value, endpoint, e.getMessage()); + return Optional.empty(); + } + } + + @VisibleForTesting + public static class CacheEntry + { + boolean reachable; + long timestampMillis; + + public CacheEntry(boolean reachable, long timestampMillis) + { + this.reachable = reachable; + this.timestampMillis = timestampMillis; + } + } + + // Check reachability with cache + private static boolean isReachableWithCache(InetAddressAndPort address, int timeoutMs) + { + if (address == null) + { + logger.debug("Null address provided to isReachableWithCache, treating as unreachable."); + return false; + } + + long now = FBUtilities.now().toEpochMilli(); + CacheEntry entry = reachabilityCache.get(address); + + if (entry != null && (now - entry.timestampMillis) < CACHE_TTL_MS) { + logger.trace("Using cached reachability for {}: {}", address, entry.reachable); + return entry.reachable; + } + + boolean reachable = isReachableOnce(address, timeoutMs); + reachabilityCache.put(address, new CacheEntry(reachable, now)); + logger.debug("Refreshed reachability for {}: {}", address, reachable); + return reachable; + } + + private static boolean isReachableOnce(InetAddressAndPort address, int timeoutMs) + { + try (Socket socket = new Socket()) { + socket.connect(new InetSocketAddress(address.getAddress(), address.getPort()), timeoutMs); + return true; + } catch (IOException e) { + logger.trace("Unreachable: {} due to {}", address, e.toString()); + return false; + } + } + @VisibleForTesting public static List sortByProximity(Collection endpoints) {