natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

[CASSANDRA-20476] Perform rediscovery of CMS at startup if addresses change

This commit is contained in:
Sam Tunnicliffe 2025-06-02 11:50:04 +01:00
parent e67e83f92d
commit ccd89e4926
7 changed files with 223 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
src/java/org/apache/cassandra/net/MessageDelivery.java
View File

@ -62,7 +62,7 @@ public interface MessageDelivery
@Override
public void onResponse(Message<RSP> msg)
{
logger.info("Received a {} response from {}: {}", msg.verb(), msg.from(), msg.payload);
logger.debug("Received a {} response from {}: {}", msg.verb(), msg.from(), msg.payload);
responses.add(Pair.create(msg.from(), msg.payload));
cdl.decrement();
}
@ -70,13 +70,13 @@ public interface MessageDelivery
@Override
public void onFailure(InetAddressAndPort from, RequestFailure reason)
{
logger.info("Received failure in response to {} from {}: {}", verb, from, reason);
logger.debug("Received failure in response to {} from {}: {}", verb, from, reason);
cdl.decrement();
}
};
sendTo.forEach((ep) -> {
logger.info("Election for metadata migration sending {} ({}) to {}", verb, payload.toString(), ep);
logger.debug("Sending {} ({}) to {}", verb, payload.toString(), ep);
messaging.sendWithCallback(Message.out(verb, payload), ep, callback);
});
cdl.awaitUninterruptibly(timeout, timeUnit);

4
src/java/org/apache/cassandra/net/Verb.java
View File

@ -316,6 +316,10 @@ public enum Verb
TCM_DISCOVER_REQ (813, P0, rpcTimeout, INTERNAL_METADATA, () -> NoPayload.serializer, () -> Discovery.instance.requestHandler, TCM_DISCOVER_RSP ),
TCM_FETCH_PEER_LOG_RSP (818, P0, shortTimeout, FETCH_METADATA, MessageSerializers::logStateSerializer, RESPONSE_HANDLER ),
TCM_FETCH_PEER_LOG_REQ (819, P0, rpcTimeout, FETCH_METADATA, () -> FetchPeerLog.serializer, () -> FetchPeerLog.Handler.instance, TCM_FETCH_PEER_LOG_RSP ),
TCM_DISCOVER_PEERS_RSP (820, P0, rpcTimeout, INTERNAL_METADATA, () -> Discovery.serializer, () -> ResponseVerbHandler.instance ),
TCM_DISCOVER_PEERS_REQ (821, P0, rpcTimeout, INTERNAL_METADATA, () -> NoPayload.serializer, () -> Discovery.instance.requestHandler, TCM_DISCOVER_PEERS_RSP),
TCM_DISCOVER_SURVEY_RSP(822, P0, rpcTimeout, INTERNAL_METADATA, () -> Discovery.nodeIdSerializer, () -> ResponseVerbHandler.instance ),
TCM_DISCOVER_SURVEY_REQ(823, P0, rpcTimeout, INTERNAL_METADATA, () -> NoPayload.serializer, () -> Discovery.instance.requestHandler, TCM_DISCOVER_SURVEY_RSP),
INITIATE_DATA_MOVEMENTS_RSP (814, P1, rpcTimeout, MISC, () -> NoPayload.serializer, RESPONSE_HANDLER ),
INITIATE_DATA_MOVEMENTS_REQ (815, P1, rpcTimeout, MISC, () -> DataMovement.serializer, () -> DataMovementVerbHandler.instance, INITIATE_DATA_MOVEMENTS_RSP ),

2
src/java/org/apache/cassandra/tcm/ClusterMetadataService.java
View File

@ -165,7 +165,7 @@ public class ClusterMetadataService
return state(ClusterMetadata.current());
}
public static State state(ClusterMetadata metadata)
public static ClusterMetadataService.State state(ClusterMetadata metadata)
{
if (CassandraRelevantProperties.TCM_UNSAFE_BOOT_WITH_CLUSTERMETADATA.isPresent())
return RESET;

74
src/java/org/apache/cassandra/tcm/Discovery.java
View File

@ -36,6 +36,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.SystemKeyspace;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.util.DataInputPlus;
@ -47,6 +48,7 @@ import org.apache.cassandra.net.MessageDelivery;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.net.NoPayload;
import org.apache.cassandra.net.Verb;
import org.apache.cassandra.tcm.membership.NodeId;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Pair;
@ -63,6 +65,28 @@ public class Discovery
public static final Discovery instance = new Discovery();
public static final Serializer serializer = new Serializer();
// TODO add this to MessageSerializers properly or define a real response format
public static final IVersionedSerializer<NodeId> nodeIdSerializer = new IVersionedSerializer<>()
{
@Override
public void serialize(NodeId t, DataOutputPlus out, int version) throws IOException
{
out.writeUnsignedVInt32(t.id());
}
@Override
public NodeId deserialize(DataInputPlus in, int version) throws IOException
{
int id = in.readUnsignedVInt32();
return new NodeId(id);
}
@Override
public long serializedSize(NodeId t, int version)
{
return TypeSizes.sizeofUnsignedVInt(t.id());
}
};
public final IVerbHandler<NoPayload> requestHandler;
private final Set<InetAddressAndPort> discovered = new ConcurrentSkipListSet<>();
@ -94,12 +118,14 @@ public class Discovery
public DiscoveredNodes discover()
{
return discover(5);
return discover(5, false);
}
public DiscoveredNodes discover(int rounds)
public DiscoveredNodes discover(int rounds, boolean allPeers)
{
boolean res = state.compareAndSet(State.NOT_STARTED, State.IN_PROGRESS);
if (!res)
res = state.compareAndSet(State.FINISHED, State.IN_PROGRESS);
assert res : String.format("Can not start discovery as it is in state %s", state.get());
long deadline = nanoTime() + DatabaseDescriptor.getDiscoveryTimeout(TimeUnit.NANOSECONDS);
@ -114,7 +140,7 @@ public class Discovery
while (nanoTime() <= deadline || unchangedFor < rounds)
{
long startTimeNanos = nanoTime();
last = discoverOnce(null, roundTimeNanos, TimeUnit.NANOSECONDS);
last = discoverOnce(allPeers, null, roundTimeNanos, TimeUnit.NANOSECONDS);
if (last.kind == DiscoveredNodes.Kind.CMS_ONLY)
break;
@ -136,11 +162,13 @@ public class Discovery
assert res : String.format("Can not finish discovery as it is in state %s", state.get());
return last;
}
public DiscoveredNodes discoverOnce(InetAddressAndPort initiator)
{
return discoverOnce(initiator, 1, TimeUnit.SECONDS);
return discoverOnce(false, initiator, 1, TimeUnit.SECONDS);
}
public DiscoveredNodes discoverOnce(InetAddressAndPort initiator, long timeout, TimeUnit timeUnit)
public DiscoveredNodes discoverOnce(boolean allPeers, InetAddressAndPort initiator, long timeout, TimeUnit timeUnit)
{
Set<InetAddressAndPort> candidates = new HashSet<>();
if (initiator != null)
@ -148,12 +176,11 @@ public class Discovery
else
candidates.addAll(discovered);
if (candidates.isEmpty())
candidates.addAll(seeds.get());
candidates.addAll(seeds.get());
candidates.remove(self);
Collection<Pair<InetAddressAndPort, DiscoveredNodes>> responses = MessageDelivery.fanoutAndWait(messaging.get(), candidates, Verb.TCM_DISCOVER_REQ, NoPayload.noPayload, timeout, timeUnit);
Verb verb = allPeers ? Verb.TCM_DISCOVER_PEERS_REQ : Verb.TCM_DISCOVER_REQ;
Collection<Pair<InetAddressAndPort, DiscoveredNodes>> responses = MessageDelivery.fanoutAndWait(messaging.get(), candidates, verb, NoPayload.noPayload, timeout, timeUnit);
for (Pair<InetAddressAndPort, DiscoveredNodes> discoveredNodes : responses)
{
@ -179,19 +206,30 @@ public class Discovery
@Override
public void doVerb(Message<NoPayload> message)
{
discovered.add(message.from());
Set<InetAddressAndPort> cms = ClusterMetadata.current().fullCMSMembers();
logger.debug("Responding to discovery request from {}: {}", message.from(), cms);
DiscoveredNodes discoveredNodes;
if (!cms.isEmpty())
discoveredNodes = new DiscoveredNodes(cms, DiscoveredNodes.Kind.CMS_ONLY);
else
switch (message.verb())
{
discovered.add(message.from());
discoveredNodes = new DiscoveredNodes(new HashSet<>(discovered), DiscoveredNodes.Kind.KNOWN_PEERS);
case TCM_DISCOVER_REQ:
logger.trace("Responding to discovery request from {}: {}", message.from(), cms);
if (!cms.isEmpty())
discoveredNodes = new DiscoveredNodes(cms, DiscoveredNodes.Kind.CMS_ONLY);
else
discoveredNodes = new DiscoveredNodes(new HashSet<>(discovered), DiscoveredNodes.Kind.KNOWN_PEERS);
messaging.get().send(message.responseWith(discoveredNodes), message.from());
break;
case TCM_DISCOVER_PEERS_REQ:
logger.trace("Responding to {} request from {}", message.verb(), message.from());
discoveredNodes = new DiscoveredNodes(new HashSet<>(discovered), DiscoveredNodes.Kind.KNOWN_PEERS);
messaging.get().send(message.responseWith(discoveredNodes), message.from());
break;
case TCM_DISCOVER_SURVEY_REQ:
logger.trace("Responding to {} request from {}", message.verb(), message.from());
NodeId id = NodeId.fromUUID(SystemKeyspace.getLocalHostId());
messaging.get().send(message.responseWith(id), message.from());
break;
}
messaging.get().send(message.responseWith(discoveredNodes), message.from());
}
}

136
src/java/org/apache/cassandra/tcm/Startup.java
View File

@ -19,7 +19,9 @@ package org.apache.cassandra.tcm;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
@ -52,7 +54,10 @@ import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.gms.NewGossiper;
import org.apache.cassandra.gms.VersionedValue;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.net.MessageDelivery;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.net.NoPayload;
import org.apache.cassandra.net.Verb;
import org.apache.cassandra.schema.DistributedSchema;
import org.apache.cassandra.schema.KeyspaceMetadata;
import org.apache.cassandra.schema.Keyspaces;
@ -118,7 +123,16 @@ import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort;
case NORMAL:
logger.info("Initializing as non CMS node");
initializeAsNonCmsNode(wrapProcessor);
ClusterMetadataService.instance().log().pause();
initMessaging.run();
UUID localHostId = SystemKeyspace.getLocalHostId();
// If null, this node has not yet registered so there will be more nothing to do here
if (localHostId != null)
{
NodeId nodeId = NodeId.fromUUID(localHostId);
initializeCMSLookup(nodeId, ClusterMetadata.current());
}
ClusterMetadataService.instance().log().unpause();
break;
case VOTE:
logger.info("Initializing for discovery");
@ -183,7 +197,6 @@ import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort;
ClusterMetadataService::state,
logSpec));
ClusterMetadataService.instance().log().ready();
NodeId nodeId = ClusterMetadata.current().myNodeId();
UUID currentHostId = SystemKeyspace.getLocalHostId();
if (nodeId != NodeId.UNREGISTERED && !Objects.equals(nodeId.toUUID(), currentHostId))
@ -202,6 +215,127 @@ import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort;
}
}
/**
* If the broadcast address of this node has changed, we must verify the endpoints it knows for
* the members of the CMS are still reachable and valid. This is necessary for the node to submit
* a STARTUP transformation which updates its broadcast address in ClusterMetadata.
*
* If the node is itself a CMS member, it is also a requirement to be able to contact a
* majority of the other CMS members in order to perform the serial reads and writes which
* constitute committing to and fetching from the distributed metadata log.
*
* To do this, we use a simple protocol:
* 1. For each CMS member in our replayed ClusterMetadata, ping the associated broadcast address
* to query for id of the node at that address. This determines whether the endpoint still
* belongs to that same node (which is/was a CMS member).
* 2. While we don't have confirmed current addresses for a majority of CMS nodes:
* 2a. Run discovery to locate as many peer addresses as possible.
* 2b. Query every discovered endpoint and ask for its node id.
* If we still don't have confirmed addresses for a majority of CMS members, go to 2a and
* repeat as peers may themselves still be starting up and so may have become discoverable.
*
* This process builds up a mapping of id -> current address for CMS members which can then be
* used to construct a set of temporary redirects between addresses according to ClusterMetadata
* and the newly discovered ones.
*
* As each CMS node with a changed address goes through the startup process, it will commit its
* STARTUP transformation and the new broadcast address will be found in ClusterMetadata. A log
* listener is used to react to these transformations by removing redundant address overrides
* as they are enacted.
*
* @param nodeId derived from the persisted id of this node from the system.peers table
* @param replayed current ClusterMetadata after replaying the metadata log for startup
*/
private static ClusterMetadata initializeCMSLookup(NodeId nodeId, ClusterMetadata replayed)
{
InetAddressAndPort oldAddress = replayed.directory.endpoint(nodeId);
InetAddressAndPort newAddress = FBUtilities.getBroadcastAddressAndPort();
if (newAddress.equals(oldAddress))
return replayed;
Map<NodeId, InetAddressAndPort> previousCMS = new HashMap<>();
replayed.fullCMSMemberIds().forEach(id -> previousCMS.put(id, replayed.directory.endpoint(id)));
Map<NodeId, InetAddressAndPort> confirmedCMS = new HashMap<>();
Set<InetAddressAndPort> candidates = new HashSet<>(previousCMS.values());
candidates.add(newAddress);
int maxRounds = 5;
int currentRound = 0;
long roundTimeNanos = Math.min(TimeUnit.SECONDS.toNanos(4),
DatabaseDescriptor.getDiscoveryTimeout(TimeUnit.NANOSECONDS) / maxRounds);
// TODO a non-CMS node only needs to be able to contact a single CMS member to commit its STARTUP
int quorum = (previousCMS.size() / 2) + 1;
logger.info("Running survey and discovery for CMS nodes {} (quorum = {})", previousCMS, quorum);
while (confirmedCMS.size() < quorum && currentRound < maxRounds)
{
logger.info("In round {} sending survey to {}", currentRound, candidates);
Collection<Pair<InetAddressAndPort, NodeId>> surveyed =
MessageDelivery.fanoutAndWait(MessagingService.instance(),
candidates,
Verb.TCM_DISCOVER_SURVEY_REQ,
NoPayload.noPayload,
roundTimeNanos,
TimeUnit.NANOSECONDS);
logger.info("Survey of {} discovered {}", candidates, surveyed);
surveyed.forEach(pair -> {
if (previousCMS.containsKey(pair.right))
confirmedCMS.put(pair.right, pair.left);
});
logger.info("Confirmed CMS members {}", confirmedCMS);
if (confirmedCMS.size() < quorum || (previousCMS.size() == 1 && confirmedCMS.containsKey(nodeId)))
{
// In the single node CMS case, run discovery simply to propagate this node's new address to the rest of
// the cluster via the seeds & discovery meshing. Otherwise, if every node has a new address the non-CMS
// members have no way to discover the CMS and it has no way to know the new places to push updates to.
logger.info("Running discovery round; either CMS quorum was not confirmed or this is the only CMS member");
Discovery.DiscoveredNodes nodes = Discovery.instance.discover(5, true);
candidates.addAll(nodes.nodes());
logger.info("Rediscovery completed, discovered nodes: {}", nodes);
}
currentRound++;
}
if (confirmedCMS.size() >= quorum)
{
logger.info("Identified a quorum of CMS members (found {}, required {}).", confirmedCMS.size(), quorum);
if (confirmedCMS.values().containsAll(previousCMS.values()))
{
logger.info("No endpoint changes found for CMS members");
return replayed;
}
else
{
logger.info("Applying temporary address overrides for uncommitted CMS endpoint changes");
CMSLookup.InitialBuilder builder = CMSLookup.builder(replayed);
for (NodeId confirmed : confirmedCMS.keySet())
{
InetAddressAndPort prev = previousCMS.get(confirmed);
InetAddressAndPort next = confirmedCMS.get(confirmed);
if (!next.equals(prev))
{
logger.info("Added override for {}, ({} -> {})", confirmed, previousCMS.get(confirmed), confirmedCMS.get(confirmed));
builder = builder.withOverride(confirmed, previousCMS.get(confirmed), confirmedCMS.get(confirmed));
}
}
if (replayed.initCMSLookup(builder.build()))
{
ClusterMetadataService.instance().log().addListener(new CMSLookup.LogListener());
return replayed;
}
throw new RuntimeException("Could not initialize CMS lookup");
}
}
else
{
throw new RuntimeException(String.format("Unable to identify a quorum of CMS members (found %s, required %s)",
confirmedCMS.size(), quorum));
}
}
public static void scrubDataDirectories(ClusterMetadata metadata) throws StartupException
{
// clean up debris in the rest of the keyspaces

23
src/java/org/apache/cassandra/tcm/log/LocalLog.java
View File

@ -103,6 +103,8 @@ public abstract class LocalLog implements Closeable
// notification to them all.
private final AtomicBoolean replayComplete = new AtomicBoolean();
private final AtomicBoolean paused = new AtomicBoolean();
public static LogSpec logSpec()
{
return new LogSpec();
@ -378,6 +380,21 @@ public abstract class LocalLog implements Closeable
}
}
/**
* Temporarily halts processing of the pending buffer. While paused, entries may be appended to the buffer but
* they will not be enacted. Currently only used sparingly during startup if any address changes for CMS members are
* detected. In this case, we pause just while a mapping from old to new addresses is being built.
*/
public void pause()
{
paused.set(true);
}
public void unpause()
{
paused.set(false);
}
public void append(Entry entry)
{
maybeAppend(entry);
@ -469,6 +486,12 @@ public abstract class LocalLog implements Closeable
*/
void processPendingInternal()
{
if (paused.get())
{
logger.trace("Entry processing is paused, returning without scanning pending buffer");
return;
}
while (true)
{
Entry pendingEntry = peek();

2
test/unit/org/apache/cassandra/tcm/DiscoverySimulationTest.java
View File

@ -101,7 +101,7 @@ public class DiscoverySimulationTest
Map<InetAddressAndPort, CompletableFuture<Discovery.DiscoveredNodes>> futures = new HashMap<>();
nodes.forEach((addr, discovery) -> {
futures.put(addr, CompletableFuture.supplyAsync(() -> discovery.discover(5), executor));
futures.put(addr, CompletableFuture.supplyAsync(() -> discovery.discover(5, false), executor));
});
Map<InetAddressAndPort, Discovery.DiscoveredNodes> discovered = new HashMap<>();