diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java index 5a45073ac548..d997dc9e1487 100644 --- a/src/java/org/apache/cassandra/config/Config.java +++ b/src/java/org/apache/cassandra/config/Config.java @@ -1581,6 +1581,7 @@ public static void log(Config config) public volatile DurationSpec.LongMillisecondsBound progress_barrier_timeout = new DurationSpec.LongMillisecondsBound("3600000ms"); public volatile DurationSpec.LongMillisecondsBound progress_barrier_backoff = new DurationSpec.LongMillisecondsBound("1000ms"); public volatile DurationSpec.LongSecondsBound discovery_timeout = new DurationSpec.LongSecondsBound("30s"); + public volatile int discovery_rounds = 5; public boolean unsafe_tcm_mode = false; public boolean legacy_state_listener_sync_local_updates = true; diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index 0885b72d2fd3..f7e21bba6a24 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -6339,6 +6339,11 @@ public static long getDiscoveryTimeout(TimeUnit unit) return conf.discovery_timeout.to(unit); } + public static int getDiscoveryRounds() + { + return conf.discovery_rounds; + } + public static boolean getUnsafeTCMMode() { return conf.unsafe_tcm_mode; diff --git a/src/java/org/apache/cassandra/gms/FailureDetector.java b/src/java/org/apache/cassandra/gms/FailureDetector.java index 13d6266ca039..34f53508093a 100644 --- a/src/java/org/apache/cassandra/gms/FailureDetector.java +++ b/src/java/org/apache/cassandra/gms/FailureDetector.java @@ -328,6 +328,11 @@ public boolean isAlive(InetAddressAndPort ep) // registration via the metadata log, or a full gossip round). This is perfectly harmless, so no need to log // an error in that case. ClusterMetadata metadata = ClusterMetadata.current(); + if (metadata.cmsLookup.isActive() && metadata.fullCMSMembers().contains(ep)) + { + logger.trace("Found endpoint {} in active CMS lookup, assuming it is alive", ep); + return true; + } if (!metadata.directory.allJoinedEndpoints().contains(ep) && !metadata.fullCMSMembers().contains(ep)) logger.error("Unknown endpoint: " + ep, new UnknownEndpointException(ep)); } diff --git a/src/java/org/apache/cassandra/gms/GossipDigestAckVerbHandler.java b/src/java/org/apache/cassandra/gms/GossipDigestAckVerbHandler.java index 604186a7dc8e..9796fc2fd1c8 100644 --- a/src/java/org/apache/cassandra/gms/GossipDigestAckVerbHandler.java +++ b/src/java/org/apache/cassandra/gms/GossipDigestAckVerbHandler.java @@ -56,7 +56,7 @@ public void doVerb(Message message) if (logger.isDebugEnabled()) logger.debug("Received an ack from {}, which may trigger exit from shadow round", from); - NewGossiper.instance.onAck(epStateMap); + NewGossiper.instance.onAck(message.from(), epStateMap); return; } if (epStateMap.size() > 0) diff --git a/src/java/org/apache/cassandra/gms/Gossiper.java b/src/java/org/apache/cassandra/gms/Gossiper.java index 48ef04fa07be..d0890adccfe8 100644 --- a/src/java/org/apache/cassandra/gms/Gossiper.java +++ b/src/java/org/apache/cassandra/gms/Gossiper.java @@ -720,6 +720,17 @@ public void unsafeAnnulEndpoint(InetAddressAndPort endpoint) unreachableEndpoints.remove(endpoint); } + /** + * determine which endpoint started up earlier + */ + public int compareEndpointStartup(InetAddressAndPort addr1, InetAddressAndPort addr2) + { + EndpointState ep1 = getEndpointStateForEndpoint(addr1); + EndpointState ep2 = getEndpointStateForEndpoint(addr2); + assert ep1 != null && ep2 != null; + return ep1.getHeartBeatState().getGeneration() - ep2.getHeartBeatState().getGeneration(); + } + /** * Quarantines the endpoint for QUARANTINE_DELAY * @@ -944,7 +955,7 @@ public boolean isGossipOnlyMember(InetAddressAndPort endpoint) ClusterMetadata metadata = ClusterMetadata.current(); NodeId nodeId = metadata.directory.peerId(endpoint); if (nodeId == null) - return false; + return true; return NodeState.isPreJoin(metadata.directory.states.get(nodeId)); } diff --git a/src/java/org/apache/cassandra/gms/NewGossiper.java b/src/java/org/apache/cassandra/gms/NewGossiper.java index 073943fddc33..fbb061795f5c 100644 --- a/src/java/org/apache/cassandra/gms/NewGossiper.java +++ b/src/java/org/apache/cassandra/gms/NewGossiper.java @@ -20,11 +20,11 @@ import java.util.ArrayList; import java.util.Collection; -import java.util.Collections; import java.util.HashMap; import java.util.HashSet; import java.util.Map; import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; @@ -41,7 +41,6 @@ import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.tcm.ClusterMetadata; import org.apache.cassandra.tcm.compatibility.GossipHelper; -import org.apache.cassandra.utils.concurrent.Accumulator; import org.apache.cassandra.utils.concurrent.AsyncPromise; import org.apache.cassandra.utils.concurrent.Promise; @@ -60,9 +59,14 @@ public class NewGossiper public Map doShadowRound() { Set peers = new HashSet<>(SystemKeyspace.loadHostIds().keySet()); + for (InetAddressAndPort seed : DatabaseDescriptor.getSeeds()) + { + if (!seed.equals(getBroadcastAddressAndPort())) + peers.add(seed); + } + + // implies a single node cluster with only that one node configured as a seed if (peers.isEmpty()) - peers.addAll(DatabaseDescriptor.getSeeds()); - if (peers.equals(Collections.singleton(getBroadcastAddressAndPort()))) return GossipHelper.storedEpstate(); ShadowRoundHandler shadowRoundHandler = new ShadowRoundHandler(peers); @@ -85,6 +89,9 @@ public Map doShadowRound() } } logger.warn("Not able to construct initial cluster metadata from gossip, using system tables instead"); + // Mark done here so that future gossip messages don't get routed to the shadow round handler (see + // GossipDigestSynVerbHandler & GossipDigestAckVerbHandler) + handler.markDone(); return GossipHelper.storedEpstate(); } @@ -94,18 +101,18 @@ public boolean isInShadowRound() return srh != null && !srh.isDone(); } - void onAck( Map epStateMap) + void onAck(InetAddressAndPort from, Map epStateMap) { ShadowRoundHandler srh = handler; if (srh != null && !srh.isDone()) - srh.onAck(epStateMap); + srh.onAck(from, epStateMap); } public static class ShadowRoundHandler { private volatile boolean isDone = false; private final Set peers; - private final Accumulator> responses; + private final Map> responses; private final int requiredResponses; private final MessageDelivery messageDelivery; private final Promise> promise = new AsyncPromise<>(); @@ -117,12 +124,19 @@ public ShadowRoundHandler(Set peers) public ShadowRoundHandler(Set peers, MessageDelivery messageDelivery) { - this.peers = peers; - requiredResponses = Math.max(peers.size() / 10, 1); // todo: is 10% reasonable? - responses = new Accumulator<>(requiredResponses); + this.peers = ConcurrentHashMap.newKeySet(); + this.peers.addAll(peers); + responses = new ConcurrentHashMap<>(); + requiredResponses = this.peers.size() <= 3 ? 1 : Math.max(this.peers.size() / 5, 2); // require response from 20% of the cluster this.messageDelivery = messageDelivery; } + public void markDone() + { + logger.info("Marking NewGossiper shadow round done"); + isDone = true; + } + public boolean isDone() { return isDone; @@ -146,18 +160,21 @@ public Promise> doShadowRound() return promise; } - public void onAck(Map epStateMap) + public void onAck(InetAddressAndPort from, Map epStateMap) { if (!isDone) { if (!epStateMap.isEmpty()) - responses.add(epStateMap); + { + responses.put(from, epStateMap); + peers.addAll(epStateMap.keySet()); // when retrying we should query the endpoints we learned about in the previous round + } logger.debug("Received {} responses. {} required.", responses.size(), requiredResponses); if (responses.size() >= requiredResponses) { isDone = true; - Map merged = merge(responses.snapshot()); + Map merged = merge(responses.values()); if (GossipHelper.isValidForClusterMetadata(merged)) promise.setSuccess(merged); else diff --git a/src/java/org/apache/cassandra/net/MessageDelivery.java b/src/java/org/apache/cassandra/net/MessageDelivery.java index 2e9155f71356..133fdc26de60 100644 --- a/src/java/org/apache/cassandra/net/MessageDelivery.java +++ b/src/java/org/apache/cassandra/net/MessageDelivery.java @@ -53,6 +53,7 @@ static Collection> fanoutAndWait(Messag { return fanoutAndWait(messaging, sendTo, verb, payload, DatabaseDescriptor.getCmsAwaitTimeout().to(TimeUnit.MILLISECONDS), TimeUnit.MILLISECONDS); } + static Collection> fanoutAndWait(MessageDelivery messaging, Set sendTo, Verb verb, REQ payload, long timeout, TimeUnit timeUnit) { Accumulator> responses = new Accumulator<>(sendTo.size()); @@ -62,7 +63,7 @@ static Collection> fanoutAndWait(Messag @Override public void onResponse(Message 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 +71,13 @@ public void onResponse(Message msg) @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); diff --git a/src/java/org/apache/cassandra/net/ResponseVerbHandler.java b/src/java/org/apache/cassandra/net/ResponseVerbHandler.java index 215b575c1002..d44c9510f127 100644 --- a/src/java/org/apache/cassandra/net/ResponseVerbHandler.java +++ b/src/java/org/apache/cassandra/net/ResponseVerbHandler.java @@ -48,6 +48,8 @@ class ResponseVerbHandler implements IVerbHandler Verb.TCM_REPLICATION, Verb.TCM_NOTIFY_RSP, Verb.TCM_DISCOVER_RSP, + Verb.TCM_DISCOVER_PEERS_RSP, + Verb.TCM_DISCOVER_SURVEY_RSP, Verb.TCM_INIT_MIG_RSP); // We skip epoch catchup for PaxosV2 verbs, since we are using PaxosV2 to serially read the log. diff --git a/src/java/org/apache/cassandra/net/Verb.java b/src/java/org/apache/cassandra/net/Verb.java index f0721fe997ff..46c1d843593f 100644 --- a/src/java/org/apache/cassandra/net/Verb.java +++ b/src/java/org/apache/cassandra/net/Verb.java @@ -136,10 +136,13 @@ import org.apache.cassandra.streaming.DataMovement; import org.apache.cassandra.streaming.DataMovementVerbHandler; import org.apache.cassandra.streaming.ReplicationDoneVerbHandler; -import org.apache.cassandra.tcm.Discovery; import org.apache.cassandra.tcm.Epoch; import org.apache.cassandra.tcm.FetchCMSLog; import org.apache.cassandra.tcm.FetchPeerLog; +import org.apache.cassandra.tcm.discovery.Discovery; +import org.apache.cassandra.tcm.discovery.SurveyRequest; +import org.apache.cassandra.tcm.discovery.SurveyRequestHandler; +import org.apache.cassandra.tcm.discovery.SurveyResponse; import org.apache.cassandra.tcm.migration.CMSInitializationRequest; import org.apache.cassandra.tcm.migration.CMSInitializationResponse; import org.apache.cassandra.tcm.migration.Election; @@ -316,6 +319,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, () -> SurveyResponse.serializer, () -> ResponseVerbHandler.instance ), + TCM_DISCOVER_SURVEY_REQ(823, P0, rpcTimeout, INTERNAL_METADATA, () -> SurveyRequest.serializer, () -> SurveyRequestHandler.instance(), 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 ), diff --git a/src/java/org/apache/cassandra/tcm/CMSLookup.java b/src/java/org/apache/cassandra/tcm/CMSLookup.java new file mode 100644 index 000000000000..84c592a09411 --- /dev/null +++ b/src/java/org/apache/cassandra/tcm/CMSLookup.java @@ -0,0 +1,189 @@ +/* + * 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.cassandra.tcm; + +import java.util.HashMap; +import java.util.Map; +import java.util.Objects; +import java.util.function.Predicate; + +import com.google.common.collect.ImmutableMap; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.tcm.listeners.ChangeListener; +import org.apache.cassandra.tcm.membership.EndpointLookup; +import org.apache.cassandra.tcm.membership.NodeId; +import org.apache.cassandra.utils.Pair; + +public class CMSLookup +{ + private static final Logger logger = LoggerFactory.getLogger(CMSLookup.class); + + public enum State { PRE_INIT, ACTIVE, RETIRED }; + + public final static CMSLookup NO_OP = new CMSLookup(State.PRE_INIT, Epoch.EMPTY, ImmutableMap.of()); + public static InitialBuilder builder(ClusterMetadata metadata) + { + return new InitialBuilder(metadata); + } + + private final ImmutableMap> overrides; + private final Epoch lastModified; + private final State state; + + private CMSLookup(State state, Epoch epoch, ImmutableMap> overrides) + { + this.state = state; + this.lastModified = epoch; + this.overrides = overrides; + } + + public boolean isUninitialized() + { + return state == State.PRE_INIT; + } + + public boolean isActive() + { + return state == State.ACTIVE; + } + + public EndpointLookup asNodeLookup(EndpointLookup lookup) + { + return new EndpointLookup() + { + @Override + public InetAddressAndPort endpoint(NodeId id) + { + if (overrides.containsKey(id)) + return overrides.get(id).right; + return lookup.endpoint(id); + } + }; + } + + public CMSLookup rebuild(ClusterMetadata prev, ClusterMetadata next, boolean fromSnapshot) + { + logger.info("Rebuilding CMS lookup {} with metadata from epoch {}", this, next.epoch.getEpoch()); + + // All address changes have been enacted, nothing to do + if (state == State.RETIRED) + return this; + + if (!next.epoch.isEqualOrBefore(Epoch.FIRST) + && !fromSnapshot + && next.directory.lastModified().equals(prev.directory.lastModified())) + return this; + + // Filters from the override list those which are no longer necessary as a transformation has now + // replaced the old address with a new one for that node id + Predicate>> overrideRequired = entry -> { + NodeId nodeId = entry.getKey(); + if (!Objects.equals(prev.directory.getNodeAddresses(nodeId), next.directory.getNodeAddresses(nodeId))) + { + Pair override = overrides.get(nodeId); + if (override != null) + { + // If there was an override for this nodeId && the address being overriden matches the prev + // directory entry, filter out the override from the map which will be used to build the new version + // (i.e. return false from Predicate::test). This indicates the override is no longer required. + return !override.left.equals(prev.directory.endpoint(nodeId)); + } + } + return true; + }; + + logger.info("Current endpoint overrides: {}", overrides); + ImmutableMap.Builder> builder = ImmutableMap.builder(); + for (Map.Entry> override : overrides.entrySet()) + { + if(overrideRequired.test(override)) + builder.put(override.getKey(), override.getValue()); + } + + ImmutableMap> nextOverrides = builder.build(); + if (nextOverrides.equals(overrides)) + { + logger.info("No changes to endpoint overrides detected"); + return this; + } + + logger.info("Proposed endpoint overrides: {}", nextOverrides); + State state = nextOverrides.isEmpty() ? State.RETIRED : State.ACTIVE; + return new CMSLookup(state, next.epoch, nextOverrides); + } + + @Override + public String toString() + { + return "CMSLookup{" + + "state=" + state + + ", epoch=" + lastModified + + ", overrides=" + overrides + + '}'; + } + + public static class InitialBuilder + { + private final Epoch epoch; + private final Map> overrides; + + private InitialBuilder(ClusterMetadata metadata) + { + this.epoch = metadata.epoch; + this.overrides = new HashMap<>(); + } + + public InitialBuilder withOverride(NodeId id, InetAddressAndPort originalAddress, InetAddressAndPort newAddress) + { + overrides.put(id, Pair.create(originalAddress, newAddress)); + return this; + } + + public boolean hasOverrides() + { + return !overrides.isEmpty(); + } + + public CMSLookup build() + { + if (overrides.isEmpty()) + throw new IllegalStateException("No overrides detected"); + return new CMSLookup(State.ACTIVE, epoch, ImmutableMap.copyOf(overrides)); + } + } + + public static class LogListener implements ChangeListener + { + @Override + public void notifyPreCommit(ClusterMetadata prev, ClusterMetadata next, boolean fromSnapshot) + { + logger.info("Reevaluating CMSLookup from {} at epoch {}", prev.epoch, next.epoch); + next.refreshCMSLookup(prev, fromSnapshot); + if (next.cmsLookup.state == State.RETIRED) + { + logger.info("CMSLookup state is RETIRED, removing log listener"); + ClusterMetadataService.instance().log().removeListener(this); + } + } + } +} diff --git a/src/java/org/apache/cassandra/tcm/CMSMembership.java b/src/java/org/apache/cassandra/tcm/CMSMembership.java index 67c30dc86085..9e73ea9e61ac 100644 --- a/src/java/org/apache/cassandra/tcm/CMSMembership.java +++ b/src/java/org/apache/cassandra/tcm/CMSMembership.java @@ -28,6 +28,7 @@ import org.apache.cassandra.locator.MetaStrategy; import org.apache.cassandra.locator.Replica; import org.apache.cassandra.tcm.membership.Directory; +import org.apache.cassandra.tcm.membership.EndpointLookup; import org.apache.cassandra.tcm.membership.NodeId; import org.apache.cassandra.tcm.ownership.DataPlacement; import org.apache.cassandra.tcm.ownership.VersionedEndpoints; @@ -80,18 +81,18 @@ public static CMSMembership reconstruct(DataPlacement placement, Directory direc return new CMSMembership(lm, full, joining); } - public DataPlacement toPlacement(Directory directory) + public DataPlacement toPlacement(EndpointLookup lookup) { DataPlacement.Builder builder = DataPlacement.builder(); for (NodeId id : fullMembers) { - Replica replica = MetaStrategy.replica(directory.endpoint(id)); + Replica replica = MetaStrategy.replica(lookup.endpoint(id)); builder.withReadReplica(lastModified, replica); builder.withWriteReplica(lastModified, replica); } for(NodeId id : joiningMembers) { - builder.withWriteReplica(lastModified, MetaStrategy.replica(directory.endpoint(id))); + builder.withWriteReplica(lastModified, MetaStrategy.replica(lookup.endpoint(id))); } return builder.build(); } diff --git a/src/java/org/apache/cassandra/tcm/CMSOperations.java b/src/java/org/apache/cassandra/tcm/CMSOperations.java index 60f4e67d1a6b..b6df6f8469f2 100644 --- a/src/java/org/apache/cassandra/tcm/CMSOperations.java +++ b/src/java/org/apache/cassandra/tcm/CMSOperations.java @@ -37,6 +37,7 @@ import org.apache.cassandra.db.virtual.ClusterMetadataLogTable; import org.apache.cassandra.schema.ReplicationParams; import org.apache.cassandra.schema.TableId; +import org.apache.cassandra.tcm.membership.EndpointLookup; import org.apache.cassandra.tcm.membership.NodeId; import org.apache.cassandra.tcm.membership.NodeState; import org.apache.cassandra.tcm.membership.NodeVersion; @@ -213,7 +214,12 @@ public Map describeCMS() { Map info = new HashMap<>(); ClusterMetadata metadata = ClusterMetadata.current(); - String members = metadata.fullCMSMembers().stream().sorted().map(Object::toString).collect(Collectors.joining(",")); + EndpointLookup endpoints = metadata.endpointLookup(); + String members = metadata.fullCMSMemberIds() + .stream() + .sorted() + .map(id -> String.format("(nodeid=%s,address=%s)", id.id(), endpoints.endpoint(id))) + .collect(Collectors.joining(",")); info.put(MEMBERS, members); info.put(NEEDS_RECONFIGURATION, Boolean.toString(metadata.epoch.isBefore(Epoch.FIRST) || needsReconfiguration(metadata))); info.put(IS_MEMBER, Boolean.toString(metadata.isCMSMember())); diff --git a/src/java/org/apache/cassandra/tcm/ClusterMetadata.java b/src/java/org/apache/cassandra/tcm/ClusterMetadata.java index 0224ca5bfbb2..5bc0fbe17598 100644 --- a/src/java/org/apache/cassandra/tcm/ClusterMetadata.java +++ b/src/java/org/apache/cassandra/tcm/ClusterMetadata.java @@ -74,6 +74,7 @@ import org.apache.cassandra.tcm.extensions.ExtensionKey; import org.apache.cassandra.tcm.extensions.ExtensionValue; import org.apache.cassandra.tcm.membership.Directory; +import org.apache.cassandra.tcm.membership.EndpointLookup; import org.apache.cassandra.tcm.membership.Location; import org.apache.cassandra.tcm.membership.NodeAddresses; import org.apache.cassandra.tcm.membership.NodeId; @@ -121,6 +122,12 @@ public class ClusterMetadata // This isn't serialized as part of ClusterMetadata it's really just a view over the Directory. public final Locator locator; + // This isn't serialized, it's a helper to apply transient (or not yet committed) changes to CMS member addresses. + // It is initialized with any address changes detected in Startup::initializeCMSLookup when the node boots. It is + // subsequently updated exclusively by CMSLookup.LogListener when enacting a transformation which modifies the + // addresses of CMS member(s), making those changes permanent/durable. + public volatile CMSLookup cmsLookup = CMSLookup.NO_OP; + // These fields are lazy but only for the test purposes, since their computation requires initialization of the log ks private EndpointsForRange fullCMSReplicas; private Set fullCMSEndpoints; @@ -189,7 +196,6 @@ public ClusterMetadata(Epoch epoch, cmsMembership); } - private ClusterMetadata(int metadataIdentifier, Epoch epoch, IPartitioner partitioner, @@ -224,8 +230,8 @@ private ClusterMetadata(int metadataIdentifier, this.locator = Locator.usingDirectory(directory); this.accordStaleReplicas = accordStaleReplicas; this.cmsMembership = cmsMembership; - this.cmsDataPlacement = calculateCMSPlacement(placements, cmsMembership); - + // Build CMS placement using no-op CMS lookup, i.e. using only committed node addresses + this.cmsDataPlacement = calculateCMSPlacement(placements, cmsMembership, CMSLookup.NO_OP); InetAddressAndPort broadcastAddress = FBUtilities.getBroadcastAddressAndPort(); this.localNodeId = directory.allAddresses().contains(broadcastAddress) ? directory.peerId(broadcastAddress) @@ -239,7 +245,25 @@ public Set fullCMSMemberIds() public boolean isCMSMember() { - if (epoch.isEqualOrBefore(Epoch.FIRST)) + // There are two special cases to consider here: + // * Initialization of the CMS for the first time by its first member + // During first initialization, placements are hardcoded to the local address after the PRE_INITIALIZE_CMS is + // committed so that the subsequent INITIALIZE_CMS can be. + // + // * Some changes to the addresses of the CMS members have been changed and whilst this is known (i.e. has been + // discovered) it has not yet been committed. + // When address changes are in flight, at startup the *current* broadcast address for the local node will not + // yet be associated with its NodeId in the directory, causing the localNodeId to be set to + // NodeId.UNREGISTERED. Updating the address requires a Startup transformation to be committed, but if the + // node itself is a CMS member there is a chicken-and-egg problem as NodeId.UNREGISTERED will not be a CMS + // member. + // + // Fortunately, in both cases the placements for the metadata keyspace can be used as the source of truth a + // about CMS membership. As mentioned, during the PRE_INIT/INIT phase, the placement is hardcoded and during + // in-flight address changes a temporary lookup is constructed and the placements derived from that. + // See calculateCMSPlacement for details. + + if (epoch.isEqualOrBefore(Epoch.FIRST) || cmsLookup.isActive()) return isCMSMember(FBUtilities.getBroadcastAddressAndPort()); return fullCMSMemberIds().contains(localNodeId); @@ -260,9 +284,10 @@ public Set fullCMSMembers() if (fullCMSEndpoints == null) { + EndpointLookup lookup = endpointLookup(); fullCMSEndpoints = ImmutableSet.copyOf(cmsMembership.fullMembers() .stream() - .map(directory::endpoint) + .map(lookup::endpoint) .collect(Collectors.toSet())); } return fullCMSEndpoints; @@ -275,15 +300,54 @@ public EndpointsForRange fullCMSMembersAsReplicas() if (fullCMSReplicas == null) { + EndpointLookup lookup = endpointLookup(); EndpointsForRange.Builder builder = EndpointsForRange.builder(MetaStrategy.entireRange); for (NodeId nodeId : fullCMSMemberIds()) - builder.add(MetaStrategy.replica(directory.endpoint(nodeId))); + builder.add(MetaStrategy.replica(lookup.endpoint(nodeId))); fullCMSReplicas = builder.build(); } return fullCMSReplicas; } - private DataPlacement calculateCMSPlacement(DataPlacements placements, CMSMembership cms) + // Synchronization is probably not necessary as this should only be called by a log listener + public synchronized void refreshCMSLookup(ClusterMetadata prev, boolean fromSnapshot) + { + CMSLookup prevLookup = prev.cmsLookup; + CMSLookup proposedLookup = prevLookup.rebuild(prev, this, fromSnapshot); + // rebuild returns `this` if no changes + if (prevLookup != proposedLookup) + logger.info("Replacing CMSLookup, Current: {}, Proposed: {}", prevLookup, proposedLookup); + + // recalculate CMS placement using endpoint mappings from lookup + cmsDataPlacement = calculateCMSPlacement(placements, cmsMembership, proposedLookup); + + // We shouldn't need to null out the other lazily initialized CMS fields when we refresh the CMSLookup as the + // refresh is triggered by a precommit listener, meaning nothing should be accessing the ClusterMetadata + // instance yet. + cmsLookup = proposedLookup; + } + + // Synchronization is probably not necessary as this should only be called once, during startup + public synchronized boolean initCMSLookup(CMSLookup lookup) + { + logger.info("Initializing CMS lookup on ClusterMetadata at epoch {}", epoch.getEpoch()); + boolean isInitial = cmsLookup.isUninitialized(); + logger.info("Current CMS lookup: {}, proposed: {}", cmsLookup, lookup); + if (isInitial) + { + cmsLookup = lookup; + cmsDataPlacement = calculateCMSPlacement(placements, cmsMembership, lookup); + // We need to null out these fields when we init the CMSLookup because post-commit listeners may have + // accessed them during log replay, which happens before this. + fullCMSReplicas = null; + fullCMSEndpoints = null; + } + else + logger.warn("Invalid CMSLookup state for initialization: {}", cmsLookup); + return isInitial; + } + + private DataPlacement calculateCMSPlacement(DataPlacements placements, CMSMembership cms, CMSLookup lookup) { if (epoch.isBefore(Epoch.FIRST) || schema.getKeyspaces().get(SchemaConstants.METADATA_KEYSPACE_NAME).isEmpty()) return DataPlacement.empty(); @@ -326,11 +390,17 @@ private DataPlacement calculateCMSPlacement(DataPlacements placements, CMSMember } else { - // Build a placement based on the CMS membership - return cms.toPlacement(directory); + // Build a placement based on the CMS membership, with any in-flight endpoint changes applied + EndpointLookup endpointLookup = lookup.isActive() ? lookup.asNodeLookup(directory) : directory; + return cms.toPlacement(endpointLookup); } } + public EndpointLookup endpointLookup() + { + return cmsLookup.isActive() ? cmsLookup.asNodeLookup(directory) : directory; + } + public DataPlacement placement(ReplicationParams params) { return params.isMeta() ? cmsDataPlacement : placements.get(params); diff --git a/src/java/org/apache/cassandra/tcm/ClusterMetadataService.java b/src/java/org/apache/cassandra/tcm/ClusterMetadataService.java index 08fadb2aa820..c16c2550bb28 100644 --- a/src/java/org/apache/cassandra/tcm/ClusterMetadataService.java +++ b/src/java/org/apache/cassandra/tcm/ClusterMetadataService.java @@ -61,6 +61,7 @@ import org.apache.cassandra.service.accord.topology.AccordFastPath; import org.apache.cassandra.service.accord.topology.AccordStaleReplicas; import org.apache.cassandra.service.consensus.migration.ConsensusMigrationState; +import org.apache.cassandra.tcm.discovery.Discovery; import org.apache.cassandra.tcm.listeners.SchemaListener; import org.apache.cassandra.tcm.log.Entry; import org.apache.cassandra.tcm.log.LocalLog; @@ -165,7 +166,7 @@ public static State state() 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; @@ -210,7 +211,8 @@ public static State state(ClusterMetadata metadata) Commit.Replicator replicator = CassandraRelevantProperties.TCM_USE_TEST_NO_OP_REPLICATOR.getBoolean() ? Commit.Replicator.NO_OP - : new Commit.DefaultReplicator(() -> log.metadata().directory); + : new Commit.DefaultReplicator(() -> log.metadata().directory, + () -> log.metadata().endpointLookup()); RemoteProcessor remoteProcessor = new RemoteProcessor(log, Discovery.instance::discoveredNodes); GossipProcessor gossipProcessor = new GossipProcessor(); @@ -722,7 +724,7 @@ private static Retry getRetryPolicy(Transformation.Kind kind) Retry retryPolicy; if (kind == Transformation.Kind.STARTUP) { - retryPolicy = Retry.unsafeRetryIndefinitely(); + retryPolicy = Retry.withNoTimeLimit(TCMMetrics.instance.commitRetries, Retry.unsafeRetryIndefinitely()); } else if (kind == Transformation.Kind.SCHEMA_CHANGE) { @@ -825,10 +827,15 @@ public ClusterMetadata fetchLogFromCMS(Epoch awaitAtLeast) return metadata; Epoch ourEpoch = metadata.epoch; - if (ourEpoch.isEqualOrAfter(awaitAtLeast)) return metadata; + if (log.isPaused()) + { + logger.debug("Fetch metadata log from CMS was requested, but log processing is paused"); + return metadata; + } + Retry deadline = Retry.untilElapsed(getCmsAwaitTimeout().to(TimeUnit.NANOSECONDS), TCMMetrics.instance.fetchLogRetries); // responses for ALL withhout knowing we have pending metadata = processor.fetchLogAndWait(awaitAtLeast, deadline); @@ -863,6 +870,12 @@ public Future fetchLogFromPeerAsync(InetAddressAndPort from, Ep awaitAtLeast.isBefore(Epoch.FIRST)) return ImmediateFuture.success(current); + if (log.isPaused()) + { + logger.debug("Fetch metadata log from peer was requested, but log processing is paused"); + return ImmediateFuture.success(current); + } + return peerLogFetcher.asyncFetchLog(from, awaitAtLeast); } @@ -889,6 +902,13 @@ private ClusterMetadata fetchLogFromPeer(ClusterMetadata metadata, InetAddressAn Epoch before = metadata.epoch; if (before.isEqualOrAfter(awaitAtLeast)) return metadata; + + if (log.isPaused()) + { + logger.debug("Fetch metadata log from peer was requested, but log processing is paused"); + return metadata; + } + return peerLogFetcher.fetchLogEntriesAndWait(from, awaitAtLeast); } @@ -944,6 +964,12 @@ public ClusterMetadata fetchLogFromPeerOrCMS(ClusterMetadata metadata, InetAddre if (awaitAtLeast.isBefore(Epoch.FIRST) || FBUtilities.getBroadcastAddressAndPort().equals(from)) return metadata; + if (log.isPaused()) + { + logger.debug("Fetch metadata log from peer or CMS was requested, but log processing is paused"); + return metadata; + } + Epoch before = metadata.epoch; if (before.isEqualOrAfter(awaitAtLeast)) return metadata; diff --git a/src/java/org/apache/cassandra/tcm/Commit.java b/src/java/org/apache/cassandra/tcm/Commit.java index 7eda9b752300..56e74ce91ff4 100644 --- a/src/java/org/apache/cassandra/tcm/Commit.java +++ b/src/java/org/apache/cassandra/tcm/Commit.java @@ -46,6 +46,7 @@ import org.apache.cassandra.tcm.log.Entry; import org.apache.cassandra.tcm.log.LogState; import org.apache.cassandra.tcm.membership.Directory; +import org.apache.cassandra.tcm.membership.EndpointLookup; import org.apache.cassandra.tcm.membership.NodeId; import org.apache.cassandra.tcm.membership.NodeVersion; import org.apache.cassandra.tcm.serialization.Version; @@ -373,8 +374,8 @@ static class Handler implements IVerbHandler public void doVerb(Message message) throws IOException { - checkCMSState(); logger.info("Received commit request {} from {}", message.payload, message.from()); + checkCMSState(); // Reduce our local retry deadline by write_rpc_timeout so we exhaust retries and return an // explicit failure response to the sender before their per-message callback fires to avoid // all the non-CMS nodes being synchronized on their CMS await timeouts expiring at the same time. @@ -436,10 +437,12 @@ public interface Replicator public static class DefaultReplicator implements Replicator { private final Supplier directorySupplier; + private final Supplier lookupSupplier; - public DefaultReplicator(Supplier directorySupplier) + public DefaultReplicator(Supplier directorySupplier, Supplier lookupSupplier) { this.directorySupplier = directorySupplier; + this.lookupSupplier = lookupSupplier; } public void send(Result result, InetAddressAndPort source) @@ -449,21 +452,19 @@ public void send(Result result, InetAddressAndPort source) Result.Success success = result.success(); Directory directory = directorySupplier.get(); + EndpointLookup lookup = lookupSupplier.get(); // Filter the log entries from the commit result for the purposes of replicating to members of the cluster // other than the original submitter. We only need to include the sublist of entries starting at the one // which was newly committed. We exclude entries before that one as the submitter may have been lagging and - // supplied a last known epoch arbitrarily in the past. We include entries after the first newly committed - // one as there may have been a new period automatically triggered and we'd like to push that out to all - // peers too. Of course, there may be other entries interspersed with these but it doesn't harm anything to - // include those too, it may simply be redundant. + // supplied a last known epoch arbitrarily in the past. LogState newlyCommitted = success.logState.retainFrom(success.epoch); assert !newlyCommitted.isEmpty() : String.format("Nothing to replicate after retaining epochs since %s from %s", success.epoch, success.logState); for (NodeId peerId : directory.peerIds()) { - InetAddressAndPort endpoint = directory.endpoint(peerId); + InetAddressAndPort endpoint = lookup.endpoint(peerId); boolean upgraded = directory.version(peerId).isUpgraded(); // Do not replicate to self and to the peer that has requested to commit this message if (endpoint.equals(FBUtilities.getBroadcastAddressAndPort()) || diff --git a/src/java/org/apache/cassandra/tcm/PeerLogFetcher.java b/src/java/org/apache/cassandra/tcm/PeerLogFetcher.java index 4a131954cb06..1f01803d9ff2 100644 --- a/src/java/org/apache/cassandra/tcm/PeerLogFetcher.java +++ b/src/java/org/apache/cassandra/tcm/PeerLogFetcher.java @@ -69,6 +69,7 @@ public ClusterMetadata fetchLogEntriesAndWait(InetAddressAndPort remote, Epoch a catch (ExecutionException | TimeoutException e) { logger.warn("Could not fetch log entries from peer, remote = {}, await = {}", remote, awaitAtleast); + logger.debug("Exception while fetching log entries from peer, remote = {}", remote, e); } return metadata; } diff --git a/src/java/org/apache/cassandra/tcm/RemoteProcessor.java b/src/java/org/apache/cassandra/tcm/RemoteProcessor.java index 7cfc1965ac41..fa718de36565 100644 --- a/src/java/org/apache/cassandra/tcm/RemoteProcessor.java +++ b/src/java/org/apache/cassandra/tcm/RemoteProcessor.java @@ -22,7 +22,9 @@ import java.util.Collection; import java.util.Collections; import java.util.Deque; +import java.util.HashSet; import java.util.List; +import java.util.Set; import java.util.concurrent.ConcurrentLinkedDeque; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; @@ -51,7 +53,7 @@ import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.net.RequestCallbackWithFailure; import org.apache.cassandra.net.Verb; -import org.apache.cassandra.tcm.Discovery.DiscoveredNodes; +import org.apache.cassandra.tcm.discovery.Discovery.DiscoveredNodes; import org.apache.cassandra.tcm.log.Entry; import org.apache.cassandra.tcm.log.LocalLog; import org.apache.cassandra.tcm.log.LogState; @@ -305,8 +307,12 @@ else if (retry.hasExpired()) } InetAddressAndPort candidate = candidates.next(); - long msgExpiresAfterNanos; - if (verb == Verb.TCM_COMMIT_REQ) + long msgExpiresAfterNanos = verb.expiresAfterNanos(); + // When committing a STARTUP the deadline on the Retry will be Long.MAX_VALUE i.e. never stop, retry + // indefinitely. However, we do want to actually expire those specific messages reasonably quickly (i.e. + // in the order of rpc_timeout) so that sending one to a dead or unreachable address doesn't clog things up + // for cms_await_timeout (120s by default) and we can move on to the next candidate from the iterator + if (verb == Verb.TCM_COMMIT_REQ && retry.deadlineNanos != Long.MAX_VALUE) { long cmsAwaitNanos = DatabaseDescriptor.getCmsAwaitTimeout().to(TimeUnit.NANOSECONDS); long remainingNanos = retry.remainingNanos(); @@ -317,10 +323,7 @@ else if (retry.hasExpired()) TimeUnit.NANOSECONDS.toMillis(cmsAwaitNanos), TimeUnit.NANOSECONDS.toMillis(remainingNanos)); } - else - { - msgExpiresAfterNanos = verb.expiresAfterNanos(); - } + long msgExpiresAtNanos = MonotonicClock.Global.preciseTime.now() + msgExpiresAfterNanos; Message msg = Message.outWithFlag(verb, request, MessageFlag.CALL_BACK_ON_FAILURE, msgExpiresAtNanos); MessagingService.instance().sendWithCallback(msg, candidate, new RequestCallbackWithFailure() @@ -418,6 +421,7 @@ public void onFailure(InetAddressAndPort from, RequestFailure failureReason) public static class CandidateIterator extends AbstractIterator { private final Deque candidates; + private final Set elements; private final boolean checkLive; @SuppressWarnings("resource") @@ -430,6 +434,7 @@ public CandidateIterator(Collection initialContacts) public CandidateIterator(Collection initialContacts, boolean checkLive) { this.candidates = new ConcurrentLinkedDeque<>(initialContacts); + this.elements = new HashSet<>(initialContacts); this.checkLive = checkLive; } @@ -441,19 +446,26 @@ public CandidateIterator(Collection initialContacts, boolean public void addCandidates(DiscoveredNodes discoveredNodes) { if (discoveredNodes.kind() == DiscoveredNodes.Kind.CMS_ONLY) - discoveredNodes.nodes().forEach(candidates::addFirst); + discoveredNodes.nodes().forEach(this::maybeAddFirst); else - discoveredNodes.nodes().forEach(candidates::addLast); + discoveredNodes.nodes().forEach(this::maybeAddLast); + } + + private void maybeAddFirst(InetAddressAndPort candidate) + { + if (elements.add(candidate)) + candidates.addFirst(candidate); } - public void notCms(InetAddressAndPort resp) + private void maybeAddLast(InetAddressAndPort candidate) { - candidates.addLast(resp); + if (elements.add(candidate)) + candidates.addLast(candidate); } public void timeout(InetAddressAndPort timedOut) { - candidates.addLast(timedOut); + maybeAddLast(timedOut); } public String toString() @@ -487,14 +499,17 @@ else if (first.equals(ep)) if (checkLive && !FailureDetector.instance.isAlive(ep)) { - if (candidates.isEmpty()) - return ep; - else + // If there are other candidates, just return this one to the back of the deque. It can be added + // directly, not via maybeAddLast as it hasn't been removed from the element set yet + if (!candidates.isEmpty()) { candidates.addLast(ep); continue; } } + // if we have a candidate, it was popped from the deque so make sure it's also removed from the set + if (ep != null) + elements.remove(ep); return ep; } return endOfData(); diff --git a/src/java/org/apache/cassandra/tcm/Retry.java b/src/java/org/apache/cassandra/tcm/Retry.java index 7b46029060bf..2e82cb8640b3 100644 --- a/src/java/org/apache/cassandra/tcm/Retry.java +++ b/src/java/org/apache/cassandra/tcm/Retry.java @@ -24,7 +24,6 @@ import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.config.DurationSpec; -import org.apache.cassandra.metrics.TCMMetrics; import org.apache.cassandra.service.RetryStrategy; import org.apache.cassandra.service.TimeoutStrategy; import org.apache.cassandra.service.TimeoutStrategy.LatencySourceFactory; @@ -37,7 +36,7 @@ public class Retry implements WaitStrategy { private static final WaitStrategy DEFAULT_STRATEGY; - private static final Retry RETRY_INDEFINITELY; + private static final WaitStrategy RETRY_INDEFINITELY; static { DurationSpec.IntMillisecondsBound defaultBackoff = DatabaseDescriptor.getDefaultRetryBackoff(); @@ -51,11 +50,9 @@ public class Retry implements WaitStrategy } DEFAULT_STRATEGY = RetryStrategy.parse(defaultSpec, LatencySourceFactory.none()); - Meter retryMeter = TCMMetrics.instance.commitRetries; String spec = (defaultBackoff == null ? "100ms" : defaultBackoff.toMilliseconds() + "ms") + "*attempts <=" + (defaultMaxBackoff == null ? "10s" : defaultMaxBackoff.toMilliseconds() + "ms"); - WaitStrategy wait = RetryStrategy.parse(spec, TimeoutStrategy.LatencySourceFactory.none()); - RETRY_INDEFINITELY = Retry.withNoTimeLimit(retryMeter, wait); + RETRY_INDEFINITELY = RetryStrategy.parse(spec, TimeoutStrategy.LatencySourceFactory.none()); } public final long deadlineNanos; @@ -74,7 +71,7 @@ public Retry(long deadlineNanos, Meter retryMeter, WaitStrategy delegate) * To be used only when submitting a STARTUP transformation when a node is restarted with a new set of addresses or * running a new release version. */ - static Retry unsafeRetryIndefinitely() + static WaitStrategy unsafeRetryIndefinitely() { return RETRY_INDEFINITELY; } diff --git a/src/java/org/apache/cassandra/tcm/Startup.java b/src/java/org/apache/cassandra/tcm/Startup.java index f9a4313afa71..fb7e02ba69a4 100644 --- a/src/java/org/apache/cassandra/tcm/Startup.java +++ b/src/java/org/apache/cassandra/tcm/Startup.java @@ -19,7 +19,9 @@ 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,9 @@ 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.Verb; import org.apache.cassandra.schema.DistributedSchema; import org.apache.cassandra.schema.KeyspaceMetadata; import org.apache.cassandra.schema.Keyspaces; @@ -60,6 +64,9 @@ import org.apache.cassandra.schema.SchemaKeyspace; import org.apache.cassandra.schema.TableMetadata; import org.apache.cassandra.service.StorageService; +import org.apache.cassandra.tcm.discovery.Discovery; +import org.apache.cassandra.tcm.discovery.SurveyRequest; +import org.apache.cassandra.tcm.discovery.SurveyResponse; import org.apache.cassandra.tcm.log.LocalLog; import org.apache.cassandra.tcm.log.LogStorage; import org.apache.cassandra.tcm.log.SystemKeyspaceStorage; @@ -75,6 +82,7 @@ import org.apache.cassandra.tcm.sequences.InProgressSequences; import org.apache.cassandra.tcm.sequences.ReconfigureCMS; import org.apache.cassandra.tcm.sequences.ReplaceSameAddress; +import org.apache.cassandra.tcm.serialization.Version; import org.apache.cassandra.tcm.transformations.PrepareJoin; import org.apache.cassandra.tcm.transformations.PrepareReplace; import org.apache.cassandra.tcm.transformations.UnsafeJoin; @@ -117,8 +125,52 @@ public static void initialize(Set seeds, break; case NORMAL: logger.info("Initializing as non CMS node"); + // note: if this node was a member of the CMS, log replay will put it back into that state initializeAsNonCmsNode(wrapProcessor); - 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) + { + initMessaging.run(); + } + else + { + NodeId nodeId = NodeId.fromUUID(localHostId); + ClusterMetadata replayed = ClusterMetadata.current(); + InetAddressAndPort oldAddress = replayed.directory.endpoint(nodeId); + InetAddressAndPort newAddress = FBUtilities.getBroadcastAddressAndPort(); + if (!newAddress.equals(oldAddress) && replayed.directory.commonSerializationVersion.isAtLeast(Version.V9)) + { + // Build temporary mappings for addressing CMS nodes who's addresses have been + // changed but not yet committed via the CMS or not yet been enacted locally. + ClusterMetadataService.instance().log().pause(); + initMessaging.run(); + initializeCMSLookup(nodeId, replayed); + ClusterMetadataService.instance().log().unpause(); + + + logger.info("Detected change in local node addresses, committing update to Cluster Metadata Service"); + Transformation transform = new org.apache.cassandra.tcm.transformations.Startup(nodeId, + NodeAddresses.current(), + NodeVersion.CURRENT); + replayed = ClusterMetadataService.instance().commit(transform); + + // Wait for any CMS members which have uncommitted address changes to commit them and + // us to enact them. + while (replayed.cmsLookup.isActive()) + { + logger.info("Waiting for pending CMS address changes to complete {}", replayed.cmsLookup); + TimeUnit.MILLISECONDS.sleep(1000); + replayed = ClusterMetadata.current(); + } + logger.info("In flight CMS address changes have been processed, current epoch is {}", replayed.epoch.getEpoch()); + } + else + { + // nothing more to do, so just initialize messaging + initMessaging.run(); + } + } break; case VOTE: logger.info("Initializing for discovery"); @@ -183,7 +235,6 @@ public static void initializeAsNonCmsNode(Function wrapPro 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 +253,147 @@ public static void initializeAsNonCmsNode(Function wrapPro } } + + /** + * 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; + + logger.info("Initializing CMS lookup to submit STARTUP containing broadcast address change from {} to {}", oldAddress, newAddress); + Map previousCMS = new HashMap<>(); + replayed.fullCMSMemberIds().forEach(id -> previousCMS.put(id, replayed.directory.endpoint(id))); + Map confirmedCMS = new HashMap<>(); + + Set candidates = new HashSet<>(previousCMS.values()); + candidates.add(newAddress); + candidates.addAll(DatabaseDescriptor.getSeeds()); + + // Technically, if this node it not a CMS member it only needs to be able to contact a single peer which is a + // CMS member to submit its STARTUP transformation. However, if this node is a CMS member, it will need to + // confirm a majority of the other members in order to form the consensus group with them to commit its own + // STARTUP. For simplicity we try to confirm a majority of CMS members before proceeding in either case. + int quorum = (previousCMS.size() / 2) + 1; + int rounds = DatabaseDescriptor.getDiscoveryRounds(); + long roundTimeNanos = DatabaseDescriptor.getDiscoveryTimeout(TimeUnit.NANOSECONDS) / rounds; + int currentRound = 0; + logger.info("Running survey and discovery for CMS nodes {} (quorum = {})", previousCMS, quorum); + while (confirmedCMS.size() < quorum && currentRound < rounds) + { + logger.info("In round {} sending survey to {}", currentRound, candidates); + SurveyRequest request = new SurveyRequest(replayed.metadataIdentifier); + Collection> surveyed = MessageDelivery.fanoutAndWait(MessagingService.instance(), + candidates, + Verb.TCM_DISCOVER_SURVEY_REQ, + request, + roundTimeNanos, + TimeUnit.NANOSECONDS); + logger.info("Survey of {} discovered {}", candidates, surveyed); + surveyed.forEach(pair -> { + SurveyResponse response = pair.right; + if (response.metadataId == replayed.metadataIdentifier) + { + if (previousCMS.containsKey(response.nodeId)) + confirmedCMS.put(response.nodeId, pair.left); + } + else + { + logger.info("Mismatching metadata id in survey response from {}, ignoring ({}/{})", pair.left, replayed.metadataIdentifier, response.metadataId); + } + }); + + logger.info("Confirmed CMS members {}", confirmedCMS); + if (confirmedCMS.size() < quorum || (previousCMS.size() == 1 && confirmedCMS.containsKey(nodeId))) + { + // In the single node CMS case we 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(DatabaseDescriptor.getDiscoveryRounds(), 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, prev, next); + builder = builder.withOverride(confirmed, prev, next); + } + } + + if (!builder.hasOverrides()) + { + logger.info("No overrides required for CMS members"); + return replayed; + } + + if (replayed.initCMSLookup(builder.build())) + { + logger.info("Adding CMS lookup log listener"); + ClusterMetadataService.instance().log().addListener(new CMSLookup.LogListener()); + return replayed; + } + else + 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 diff --git a/src/java/org/apache/cassandra/tcm/compatibility/GossipHelper.java b/src/java/org/apache/cassandra/tcm/compatibility/GossipHelper.java index c60373762007..ac0a333fc7d0 100644 --- a/src/java/org/apache/cassandra/tcm/compatibility/GossipHelper.java +++ b/src/java/org/apache/cassandra/tcm/compatibility/GossipHelper.java @@ -22,6 +22,7 @@ import java.io.DataInputStream; import java.io.IOException; import java.net.UnknownHostException; +import java.util.ArrayList; import java.util.Collection; import java.util.Collections; import java.util.EnumSet; @@ -33,7 +34,6 @@ import java.util.UUID; import com.google.common.annotations.VisibleForTesting; -import com.google.common.collect.Lists; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -344,8 +344,41 @@ public static Map storedEpstate() return epstates; } + /** + * During upgrade, in gossip mode we allow nodes to change IP, this means that we can't decide the + * nodeId based on the sorting of the ip address, since that might change during the upgrade + * + * Instead we sort by hostId which can't change - replacements are not allowed etc. + * + * If a node has no hostId, we add the endpoints to the end, sorted by ip (if a node doesn't have a hostid it can't change ip) + * + */ + public static List sortEndpointsByHostId(Map epStates) + { + Map hostIdToEndpoint = new HashMap<>(); + List noHostId = new ArrayList<>(); + for (Map.Entry entry : epStates.entrySet()) + { + VersionedValue vv = entry.getValue().getApplicationState(HOST_ID); + if (vv == null) + noHostId.add(entry.getKey()); + else + hostIdToEndpoint.put(vv.value, entry.getKey()); + } + List sorted = new ArrayList<>(epStates.size()); + List hostIds = new ArrayList<>(hostIdToEndpoint.keySet()); + Collections.sort(hostIds); + for (String hostId : hostIds) + sorted.add(hostIdToEndpoint.get(hostId)); + Collections.sort(noHostId); + sorted.addAll(noHostId); + return sorted; + } + @VisibleForTesting - public static ClusterMetadata fromEndpointStates(Map epStates, IPartitioner partitioner, DistributedSchema schema) + public static ClusterMetadata fromEndpointStates(Map epStates, + IPartitioner partitioner, + DistributedSchema schema) { Directory directory = new Directory().withLastModified(Epoch.UPGRADE_GOSSIP); TokenMap tokenMap = new TokenMap(partitioner).withLastModified(Epoch.UPGRADE_GOSSIP); @@ -355,8 +388,7 @@ public static ClusterMetadata fromEndpointStates(Map sortedEps = Lists.newArrayList(epStates.keySet()); - Collections.sort(sortedEps); + List sortedEps = sortEndpointsByHostId(epStates); Map, ExtensionValue> extensions = new HashMap<>(); for (InetAddressAndPort endpoint : sortedEps) { diff --git a/src/java/org/apache/cassandra/tcm/Discovery.java b/src/java/org/apache/cassandra/tcm/discovery/Discovery.java similarity index 66% rename from src/java/org/apache/cassandra/tcm/Discovery.java rename to src/java/org/apache/cassandra/tcm/discovery/Discovery.java index 99ca8907cc68..9e2e781d696a 100644 --- a/src/java/org/apache/cassandra/tcm/Discovery.java +++ b/src/java/org/apache/cassandra/tcm/discovery/Discovery.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.cassandra.tcm; +package org.apache.cassandra.tcm.discovery; import java.io.IOException; import java.util.ArrayList; @@ -25,7 +25,7 @@ import java.util.Set; import java.util.concurrent.ConcurrentSkipListSet; import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicReference; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Supplier; import com.google.common.annotations.VisibleForTesting; @@ -47,6 +47,8 @@ import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.net.NoPayload; import org.apache.cassandra.net.Verb; +import org.apache.cassandra.tcm.ClusterMetadata; +import org.apache.cassandra.tcm.ClusterMetadataService; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.Pair; @@ -66,7 +68,7 @@ public class Discovery public final IVerbHandler requestHandler; private final Set discovered = new ConcurrentSkipListSet<>(); - private final AtomicReference state = new AtomicReference<>(State.NOT_STARTED); + private final AtomicBoolean inProgress = new AtomicBoolean(false); // These two have to be suppliers because during simulation we can send out discovery requests // rather early, before other node has initialized either MessagingService or DatabaseDescriptor. @@ -94,17 +96,17 @@ public Discovery(Supplier messaging, Supplier candidates = new HashSet<>(); if (initiator != null) @@ -148,12 +152,11 @@ public DiscoveredNodes discoverOnce(InetAddressAndPort initiator, long timeout, else candidates.addAll(discovered); - if (candidates.isEmpty()) - candidates.addAll(seeds.get()); - + candidates.addAll(seeds.get()); candidates.remove(self); - Collection> 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> responses = MessageDelivery.fanoutAndWait(messaging.get(), candidates, verb, NoPayload.noPayload, timeout, timeUnit); for (Pair discoveredNodes : responses) { @@ -167,6 +170,11 @@ public DiscoveredNodes discoverOnce(InetAddressAndPort initiator, long timeout, return new DiscoveredNodes(discovered, DiscoveredNodes.Kind.KNOWN_PEERS); } + public void discovered(InetAddressAndPort peer) + { + discovered.add(peer); + } + private final class DiscoveryRequestHandler implements IVerbHandler { final Supplier messaging; @@ -179,19 +187,48 @@ private final class DiscoveryRequestHandler implements IVerbHandler @Override public void doVerb(Message message) { - Set cms = ClusterMetadata.current().fullCMSMembers(); - logger.debug("Responding to discovery request from {}: {}", message.from(), cms); - + discovered.add(message.from()); + ClusterMetadata metadata = ClusterMetadata.current(); + Set cms = metadata.fullCMSMembers(); 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 (ClusterMetadataService.instance().log().isPaused()) + { + // The fact that the LocalLog is currently paused implies that this node is currently starting + // up and in the process of building a temporary overlay of addresses for the current CMS (see + // Startup::initialize and Startup::initializeCMSLookup). In that case, we don't want to respond + // with a definitive CMS_ONLY response containing the previous set of CMS endpoints. While these + // may still be valid, they may be outdated if the CMS member address have changed. Returning a + // CMS_ONLY response causes the requester to prioritise the returned endpoint. When those are + // outdated, this can mean that the requester must wait for retries to time out and increase the + // time taken to hit a valid CMS endpoint. + logger.info("Responding to discovery request from {}, but this node is in the process of " + + "initializing CMSLookup so not responding with a definitive CMS_ONLY response. " + + "Discovered: {}, CMS: {}", message.from(), discovered, cms); + Set allKnown = new HashSet<>(); + allKnown.addAll(discovered); + allKnown.addAll(cms); + discoveredNodes = new DiscoveredNodes(allKnown, DiscoveredNodes.Kind.KNOWN_PEERS); + } + else 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.info("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; } - - messaging.get().send(message.responseWith(discoveredNodes), message.from()); } } @@ -267,12 +304,4 @@ public long serializedSize(DiscoveredNodes t, int version) return size; } } - - private enum State - { - NOT_STARTED, - IN_PROGRESS, - FINISHED, - FOUND_CMS - } } \ No newline at end of file diff --git a/src/java/org/apache/cassandra/tcm/discovery/SurveyRequest.java b/src/java/org/apache/cassandra/tcm/discovery/SurveyRequest.java new file mode 100644 index 000000000000..b774298b6664 --- /dev/null +++ b/src/java/org/apache/cassandra/tcm/discovery/SurveyRequest.java @@ -0,0 +1,67 @@ +/* + * 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.cassandra.tcm.discovery; + +import java.io.IOException; + +import org.apache.cassandra.db.TypeSizes; +import org.apache.cassandra.io.IVersionedSerializer; +import org.apache.cassandra.io.util.DataInputPlus; +import org.apache.cassandra.io.util.DataOutputPlus; + +public class SurveyRequest +{ + public static final Serializer serializer = new Serializer(); + final int metadataId; + + public SurveyRequest(int metadataId) + { + this.metadataId = metadataId; + } + + @Override + public String toString() + { + return "SurveyRequest{" + + "metadataId=" + metadataId + + '}'; + } + + public static class Serializer implements IVersionedSerializer + { + @Override + public void serialize(SurveyRequest t, DataOutputPlus out, int version) throws IOException + { + out.writeUnsignedVInt32(t.metadataId); + } + + @Override + public SurveyRequest deserialize(DataInputPlus in, int version) throws IOException + { + int metadataId = in.readUnsignedVInt32(); + return new SurveyRequest(metadataId); + } + + @Override + public long serializedSize(SurveyRequest t, int version) + { + return TypeSizes.sizeofUnsignedVInt(t.metadataId); + } + } +} diff --git a/src/java/org/apache/cassandra/tcm/discovery/SurveyRequestHandler.java b/src/java/org/apache/cassandra/tcm/discovery/SurveyRequestHandler.java new file mode 100644 index 000000000000..aaa9ffa2d1eb --- /dev/null +++ b/src/java/org/apache/cassandra/tcm/discovery/SurveyRequestHandler.java @@ -0,0 +1,85 @@ +/* + * 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.cassandra.tcm.discovery; + +import java.io.IOException; +import java.util.function.Supplier; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.cassandra.db.SystemKeyspace; +import org.apache.cassandra.exceptions.InvalidRequestException; +import org.apache.cassandra.net.IVerbHandler; +import org.apache.cassandra.net.Message; +import org.apache.cassandra.net.MessageDelivery; +import org.apache.cassandra.net.MessagingService; +import org.apache.cassandra.tcm.ClusterMetadata; +import org.apache.cassandra.tcm.membership.NodeId; + +public class SurveyRequestHandler implements IVerbHandler +{ + private static final Logger logger = LoggerFactory.getLogger(SurveyRequestHandler.class); + private static volatile SurveyRequestHandler instance; + + final Supplier messaging; + public final int metadataId; + + public static SurveyRequestHandler instance() + { + if (instance == null) + { + synchronized (SurveyRequestHandler.class) + { + if (instance == null) + instance = new SurveyRequestHandler(); + } + } + return instance; + } + + private SurveyRequestHandler() + { + this(ClusterMetadata.current().metadataIdentifier, MessagingService::instance); + } + + public SurveyRequestHandler(int metadataId, Supplier messaging) + { + this.metadataId = metadataId; + this.messaging = messaging; + } + + @Override + public void doVerb(Message message) throws IOException + { + logger.info("Responding to {} request from {}", message.verb(), message.from()); + if (message.payload.metadataId != metadataId) + throw new InvalidRequestException(String.format("Mismatching metadata id in survey request from %s (%d)", + message.from(), + message.payload.metadataId)); + + Discovery.instance.discovered(message.from()); + // Respond with the node id from system.local and not ClusterMetadata.current().myNodeId() because if + // this node is in the process of starting up with a new broadcast address, it will not yet recognise itself + // as being in a REGISTERED state. This results in myNodeId() returning NodeId.UNREGISTERED. + NodeId nodeId = NodeId.fromUUID(SystemKeyspace.getLocalHostId()); + SurveyResponse response = new SurveyResponse(metadataId, nodeId); + messaging.get().respond(response, message); + } +} diff --git a/src/java/org/apache/cassandra/tcm/discovery/SurveyResponse.java b/src/java/org/apache/cassandra/tcm/discovery/SurveyResponse.java new file mode 100644 index 000000000000..83819658b6ac --- /dev/null +++ b/src/java/org/apache/cassandra/tcm/discovery/SurveyResponse.java @@ -0,0 +1,75 @@ +/* + * 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.cassandra.tcm.discovery; + +import java.io.IOException; + +import org.apache.cassandra.db.TypeSizes; +import org.apache.cassandra.io.IVersionedSerializer; +import org.apache.cassandra.io.util.DataInputPlus; +import org.apache.cassandra.io.util.DataOutputPlus; +import org.apache.cassandra.tcm.membership.NodeId; + +public class SurveyResponse +{ + public static final Serializer serializer = new Serializer(); + public final int metadataId; + public final NodeId nodeId; + + public SurveyResponse(int metadataId, NodeId nodeId) + { + this.metadataId = metadataId; + this.nodeId = nodeId; + } + + @Override + public String toString() + { + return "SurveyResponse{" + + "metadataId=" + metadataId + + ", nodeId=" + nodeId + + '}'; + } + + public static class Serializer implements IVersionedSerializer + { + @Override + public void serialize(SurveyResponse t, DataOutputPlus out, int version) throws IOException + { + out.writeUnsignedVInt32(t.metadataId); + out.writeUnsignedVInt32(t.nodeId.id()); + } + + @Override + public SurveyResponse deserialize(DataInputPlus in, int version) throws IOException + { + int metadataId = in.readUnsignedVInt32(); + int nodeId = in.readUnsignedVInt32(); + return new SurveyResponse(metadataId, new NodeId(nodeId)); + } + + @Override + public long serializedSize(SurveyResponse t, int version) + { + return TypeSizes.sizeofUnsignedVInt(t.metadataId) + TypeSizes.sizeofUnsignedVInt(t.nodeId.id()); + } + } + + ; +} diff --git a/src/java/org/apache/cassandra/tcm/log/LocalLog.java b/src/java/org/apache/cassandra/tcm/log/LocalLog.java index 9e283b324491..4622fcad5573 100644 --- a/src/java/org/apache/cassandra/tcm/log/LocalLog.java +++ b/src/java/org/apache/cassandra/tcm/log/LocalLog.java @@ -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,26 @@ public void append(Collection entries) } } + /** + * 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 boolean isPaused() + { + return paused.get(); + } + public void append(Entry entry) { maybeAppend(entry); @@ -469,6 +491,12 @@ private Entry peek() */ void processPendingInternal() { + if (paused.get()) + { + logger.info("Metadata log entry processing is paused, returning without scanning pending buffer"); + return; + } + while (true) { Entry pendingEntry = peek(); diff --git a/src/java/org/apache/cassandra/tcm/membership/Directory.java b/src/java/org/apache/cassandra/tcm/membership/Directory.java index 8d444c644386..1a4f825aa2d7 100644 --- a/src/java/org/apache/cassandra/tcm/membership/Directory.java +++ b/src/java/org/apache/cassandra/tcm/membership/Directory.java @@ -61,7 +61,7 @@ import static org.apache.cassandra.tcm.membership.NodeVersion.CURRENT; import static org.apache.cassandra.tcm.membership.NodeVersion.CURRENT_METADATA_VERSION; -public class Directory implements MetadataValue +public class Directory implements MetadataValue, EndpointLookup { public static final Serializer serializer = new Serializer(); diff --git a/src/java/org/apache/cassandra/tcm/membership/EndpointLookup.java b/src/java/org/apache/cassandra/tcm/membership/EndpointLookup.java new file mode 100644 index 000000000000..03840d948ff4 --- /dev/null +++ b/src/java/org/apache/cassandra/tcm/membership/EndpointLookup.java @@ -0,0 +1,26 @@ +/* + * 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.cassandra.tcm.membership; + +import org.apache.cassandra.locator.InetAddressAndPort; + +public interface EndpointLookup +{ + InetAddressAndPort endpoint(NodeId id); +} diff --git a/src/java/org/apache/cassandra/tcm/migration/GossipCMSListener.java b/src/java/org/apache/cassandra/tcm/migration/GossipCMSListener.java index 8074493baed8..eec1e37dd678 100644 --- a/src/java/org/apache/cassandra/tcm/migration/GossipCMSListener.java +++ b/src/java/org/apache/cassandra/tcm/migration/GossipCMSListener.java @@ -61,8 +61,14 @@ public void onJoin(InetAddressAndPort endpoint, EndpointState epState) { UUID hostId = UUID.fromString(hostIdValue.value); nodeId = metadata.directory.nodeIdFromHostId(hostId); - logger.info("Node {} (hostId = {}) changing IP from {} to {}", nodeId, hostId, metadata.directory.endpoint(nodeId), endpoint); - Gossiper.instance.removeEndpoint(endpoint); + InetAddressAndPort oldEndpoint = metadata.directory.endpoint(nodeId); + if (Gossiper.instance.compareEndpointStartup(oldEndpoint, endpoint) > 0) + { + logger.warn("Host ID collision for {} between {} and {}; ignored {}", hostId, oldEndpoint, endpoint, endpoint); + return; + } + logger.info("Node {} (hostId = {}) changing IP from {} to {}", nodeId, hostId, oldEndpoint, endpoint); + Gossiper.instance.removeEndpoint(oldEndpoint); } else { diff --git a/src/java/org/apache/cassandra/tcm/transformations/Startup.java b/src/java/org/apache/cassandra/tcm/transformations/Startup.java index 00b402352599..caa2e53ce5bf 100644 --- a/src/java/org/apache/cassandra/tcm/transformations/Startup.java +++ b/src/java/org/apache/cassandra/tcm/transformations/Startup.java @@ -22,6 +22,9 @@ import java.util.Map; import java.util.Objects; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + import org.apache.cassandra.io.util.DataInputPlus; import org.apache.cassandra.io.util.DataOutputPlus; import org.apache.cassandra.tcm.ClusterMetadata; @@ -40,6 +43,7 @@ public class Startup implements Transformation { + private static final Logger logger = LoggerFactory.getLogger(Startup.class); public static final Serializer serializer = new Serializer(); private final NodeId nodeId; private final NodeVersion nodeVersion; @@ -123,6 +127,7 @@ public static void maybeExecuteStartupTransformation(NodeId localNodeId) if (!Objects.equals(directory.addresses.get(localNodeId), NodeAddresses.current()) || !Objects.equals(directory.versions.get(localNodeId), NodeVersion.CURRENT)) { + logger.info("Detected change in node addresses or version, committing updates to Cluster Metadata Service"); ClusterMetadataService.instance() .commit(new Startup(localNodeId, NodeAddresses.current(), NodeVersion.CURRENT)); } diff --git a/test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java b/test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java index 006379ae3bfc..1d54bad404d9 100644 --- a/test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java +++ b/test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java @@ -24,7 +24,9 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; +import java.util.Collections; import java.util.HashMap; +import java.util.HashSet; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; @@ -78,6 +80,7 @@ import org.apache.cassandra.gms.VersionedValue; import org.apache.cassandra.io.util.File; import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.locator.SimpleSeedProvider; import org.apache.cassandra.net.Message; import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.net.RequestCallback; @@ -107,6 +110,7 @@ import static org.apache.cassandra.config.CassandraRelevantProperties.BROADCAST_INTERVAL_MS; import static org.apache.cassandra.config.CassandraRelevantProperties.REPLACE_ADDRESS_FIRST_BOOT; import static org.apache.cassandra.config.CassandraRelevantProperties.RING_DELAY; +import static org.apache.cassandra.distributed.impl.TestEndpointCache.fromCassandraInetAddressAndPort; import static org.apache.cassandra.distributed.impl.TestEndpointCache.toCassandraInetAddressAndPort; import static org.assertj.core.api.Assertions.assertThat; @@ -834,6 +838,21 @@ public static Set getCMSMembers(IInvokableInstance inst) .collect(Collectors.toSet())); } + public static Set getCMSMemberIds(IInvokableInstance inst) + { + Set idsAsInts = inst.callOnInstance(() -> ClusterMetadata.current() + .fullCMSMemberIds() + .stream() + .map(NodeId::id) + .collect(Collectors.toSet())); + return idsAsInts.stream().map(NodeId::new).collect(Collectors.toSet()); + } + + public static Set getCMSMemberAddresses(IInvokableInstance inst) + { + return inst.callOnInstance(() -> new HashSet<>(ClusterMetadata.current().fullCMSMembers())); + } + public static boolean decommission(IInvokableInstance leaving) { return leaving.callOnInstance(() -> { @@ -871,6 +890,12 @@ public static int getNodeId(IInvokableInstance target, IInvokableInstance execut }); } + public static InetSocketAddress getEndpoint(IInvokableInstance target, NodeId id) + { + String idString = Integer.toString(id.id()); + return target.callOnInstance(() -> fromCassandraInetAddressAndPort(ClusterMetadata.current().directory.endpoint(NodeId.fromString(idString)))); + } + public static boolean cancelInProgressSequences(IInvokableInstance executor) { return cancelInProgressSequences(getNodeId(executor), executor); @@ -1443,6 +1468,15 @@ public static String getPartitionerName(IInstance instance) return (String) instance.config().get("partitioner"); } + + public static void updateSeed(IInstance instance, String...address) + { + IInstanceConfig conf = instance.config(); + conf.set("seed_provider", + new IInstanceConfig.ParameterizedClass(SimpleSeedProvider.class.getName(), + Collections.singletonMap("seeds", String.join(",", address)))); + } + /** * Changes the instance's address to the new address. This method should only be called while the instance is * down, else has undefined behavior. @@ -1478,12 +1512,14 @@ private static void updateAddress(IInstanceConfig conf, String address) // are a risk if (!conf.broadcastAddress().equals(previous)) { - conf.networkTopology().put(conf.broadcastAddress(), NetworkTopology.dcAndRack(conf.localDatacenter(), conf.localRack())); + NetworkTopology topology = conf.networkTopology(); + NetworkTopology.DcAndRack location = NetworkTopology.dcAndRack(topology.localDC(previous), topology.localRack(previous)); + topology.put(conf.broadcastAddress(), location); try { Field field = NetworkTopology.class.getDeclaredField("map"); field.setAccessible(true); - Map map = (Map) field.get(conf.networkTopology()); + Map map = (Map) field.get(topology); map.remove(previous); } catch (NoSuchFieldException | IllegalAccessException e) diff --git a/test/distributed/org/apache/cassandra/distributed/test/log/AllClusterCMSRediscoveryTest.java b/test/distributed/org/apache/cassandra/distributed/test/log/AllClusterCMSRediscoveryTest.java new file mode 100644 index 000000000000..f508cedf9059 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/log/AllClusterCMSRediscoveryTest.java @@ -0,0 +1,38 @@ +/* + * 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.cassandra.distributed.test.log; + +/** + * Tests a full-cluster restart with every node's broadcast address changing while down. + * The cluster is configured with 3 nodes, all of which are CMS members. + */ +public class AllClusterCMSRediscoveryTest extends CMSRediscoveryTestBase +{ + @Override + int clusterSize() + { + return 3; + } + + @Override + int cmsSize() + { + return 3; + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/test/log/CMSRediscoveryTestBase.java b/test/distributed/org/apache/cassandra/distributed/test/log/CMSRediscoveryTestBase.java new file mode 100644 index 000000000000..364fdcb41f27 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/log/CMSRediscoveryTestBase.java @@ -0,0 +1,224 @@ +/* + * 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.cassandra.distributed.test.log; + +import java.io.IOException; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.UnknownHostException; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +import com.google.common.base.Throwables; + +import org.awaitility.Awaitility; +import org.junit.Before; +import org.junit.Test; + +import org.apache.cassandra.config.CassandraRelevantProperties; +import org.apache.cassandra.cql3.QueryProcessor; +import org.apache.cassandra.cql3.UntypedResultSet; +import org.apache.cassandra.distributed.Cluster; +import org.apache.cassandra.distributed.api.IInvokableInstance; +import org.apache.cassandra.distributed.shared.ClusterUtils; +import org.apache.cassandra.distributed.test.TestBaseImpl; +import org.apache.cassandra.tcm.Epoch; +import org.apache.cassandra.tcm.membership.NodeId; +import org.apache.cassandra.utils.FBUtilities; +import org.apache.cassandra.utils.Pair; + +import static org.apache.cassandra.distributed.api.Feature.GOSSIP; +import static org.apache.cassandra.distributed.api.Feature.NETWORK; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +/** + * Test the rediscovery and recovery of a cluster where every node is stopped and restarted with a new broadcast + * address. Because the routing info (i.e. broadcast addresses) are persisted in ClusterMetadata and changing them + * requires a majority of CMS members to commit a transformation to the metadata log, there is a chicken and egg issue + * when the CMS members cannot locate each other to form a quorum. + */ +public abstract class CMSRediscoveryTestBase extends TestBaseImpl +{ + + @Before + public void disableAccord() + { + CassandraRelevantProperties.DTEST_ACCORD_ENABLED.setBoolean(false); + } + + abstract int clusterSize(); + abstract int cmsSize(); + + @Test + public void testNewCMSDiscovery() throws IOException, ExecutionException, InterruptedException + { + int clusterSize = clusterSize(); + int cmsSize = cmsSize(); + try (Cluster cluster = builder().withNodes(clusterSize) + .withConfig(config -> config.with(NETWORK, GOSSIP) + .set("progress_barrier_backoff", "50ms")) + .createWithoutStarting()) + { + ExecutorService executor = Executors.newFixedThreadPool(clusterSize); + // Some fetchCMSLog operations might temporarily fail and be retried during address changes + String[] expectedErrors = new String[]{ + "Cannot achieve consistency level SERIAL.*", + "Queried for epoch Epoch\\{epoch=\\d+\\}, but could not catch up.*" + }; + cluster.setUncaughtExceptionsFilter((node, t) -> { + Throwable rootCause = Throwables.getRootCause(t); + if (rootCause.getMessage() == null) + return false; + String message = rootCause.getMessage(); + for (String expected : expectedErrors) + if (message.matches(expected)) + return true; + return false; + }); + cluster.startup(); + init(cluster); + IInvokableInstance n1 = cluster.get(1); + if (cmsSize > 1) + n1.nodetoolResult("cms", "reconfigure", "" + cmsSize).asserts().success(); + ClusterUtils.waitForCMSToQuiesce(cluster, n1); + + // Set up the expectations for what address changes are going to happen + Map> addressMapping = new HashMap<>(cluster.size()); + for (IInvokableInstance inst : cluster) + { + InetSocketAddress starting = inst.config().broadcastAddress(); + InetSocketAddress expected = new InetSocketAddress(bumpAddress(starting.getAddress()), starting.getPort()); + NodeId id = ClusterUtils.getNodeId(inst); + addressMapping.put(starting, Pair.create(id, expected)); + } + + // Check the CMS membership at the start of the test & predict what it should be at the end + Set startingCMS = new HashSet<>(cmsSize); + Set expectedCMS = new HashSet<>(cmsSize); + for (InetSocketAddress s : ClusterUtils.getCMSMemberAddresses(n1)) + { + startingCMS.add(s); + expectedCMS.add(addressMapping.get(s).right); + } + Set cmsNodes = ClusterUtils.getCMSMemberIds(n1); + + // Shut down all nodes, modify each one's broadcast address and reconfigure seeds as these + // will be used to rediscover peers. Seed config does not need to be uniform across the cluster + // but there must be enough intersection to enable the CMS members to rediscover each other + for (int i = 1; i <= cluster.size(); i++) + { + IInvokableInstance inst = cluster.get(i); + inst.shutdown().get(); + InetAddress newBroadcastAddress = addressMapping.get(inst.config().broadcastAddress()).right.getAddress(); + byte[] bytes = newBroadcastAddress.getAddress(); + ClusterUtils.updateAddress(inst, addrString(bytes)); + String seed1 = addrString(bytes[0], bytes[1], bytes[2], (byte) i); + String seed2 = addrString(bytes[0], bytes[1], bytes[2], (byte) ((i < cluster.size()) ? i + 1 : 1)); + ClusterUtils.updateSeed(inst, seed1, seed2); + } + + // Start everything up and wait for state to cluster state to quiesce + List> startups = new ArrayList<>(cluster.size()); + for (IInvokableInstance inst : cluster) + { + Future f = executor.submit(() -> { + inst.startup(); + return true; + }); + startups.add(f); + } + + FBUtilities.waitOnFutures(startups, 60, TimeUnit.SECONDS); + ClusterUtils.waitForCMSToQuiesce(cluster, n1); + + // wait until each node's STARTUP transformation has been enacted by all nodes + Awaitility.waitAtMost(30, TimeUnit.SECONDS).until(() -> allAddressChangesEnacted(cluster)); + Epoch afterAllAddressChanges = getEpochAfterAllAddressChanges(cluster); + ClusterUtils.waitForCMSToQuiesce(cluster, afterAllAddressChanges, true); + + // Assert that: + // * The membership of the CMS (i.e. which node ids) remains the same + // * The set of CMS addresses matches the prediction made at the start + // * Every node has successfully changed its address. The previous check + // is a logical consequence of this, but it doesn't hurt to verify both + for (IInvokableInstance inst : cluster) + { + assertEquals(cmsNodes, ClusterUtils.getCMSMemberIds(inst)); + Set finalCMS = ClusterUtils.getCMSMemberAddresses(inst); + assertEquals(startingCMS.size(), finalCMS.size()); + assertEquals(expectedCMS.size(), finalCMS.size()); + assertTrue(expectedCMS.containsAll(finalCMS)); + for (Pair peer : addressMapping.values()) + { + InetSocketAddress fromInst = ClusterUtils.getEndpoint(inst, peer.left); + assertEquals("Check failed on instance " + inst.config().num(), peer.right, fromInst); + } + } + + cluster.schemaChange("Create keyspace ks1 with replication = {'class': 'SimpleStrategy', 'replication_factor': 1}"); + cluster.schemaChange("Create keyspace ks2 with replication = {'class': 'SimpleStrategy', 'replication_factor': 1}"); + cluster.schemaChange("Create keyspace ks3 with replication = {'class': 'SimpleStrategy', 'replication_factor': 1}"); + } + } + + private boolean allAddressChangesEnacted(Cluster cluster) + { + return getEpochAfterAllAddressChanges(cluster).isAfter(Epoch.FIRST); + } + + private Epoch getEpochAfterAllAddressChanges(Cluster cluster) + { + int nodes = cluster.size(); + long epochAfterAllStartups = cluster.get(1).callOnInstance(() -> { + UntypedResultSet rs = QueryProcessor.executeInternal("SELECT epoch, kind from system_views.cluster_metadata_log where kind = 'STARTUP'"); + if (rs == null || rs.isEmpty() || rs.size() < nodes) + return -1; + + long epoch = rs.stream().mapToLong(r -> r.getLong("epoch")).max().orElse(-1); + return epoch; + }).longValue(); + return Epoch.create(epochAfterAllStartups); + } + + private InetAddress bumpAddress(InetAddress address) throws UnknownHostException + { + // ipv4 addresses for this test + assert address.getAddress().length == 4; + byte[] bytes = address.getAddress(); + bytes[2]++; + return InetAddress.getByAddress(bytes); + } + + private String addrString(byte...b) throws UnknownHostException + { + assert b.length == 4; + return InetAddress.getByAddress(new byte[] { b[0], b[1], b[2], b[3] }).getHostAddress(); + } + +} diff --git a/test/distributed/org/apache/cassandra/distributed/test/log/CoordinatorPathTestBase.java b/test/distributed/org/apache/cassandra/distributed/test/log/CoordinatorPathTestBase.java index 28125692acf2..17f247b9b33d 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/log/CoordinatorPathTestBase.java +++ b/test/distributed/org/apache/cassandra/distributed/test/log/CoordinatorPathTestBase.java @@ -79,13 +79,13 @@ import org.apache.cassandra.tcm.ClusterMetadata; import org.apache.cassandra.tcm.ClusterMetadataService; import org.apache.cassandra.tcm.Commit; -import org.apache.cassandra.tcm.Discovery; import org.apache.cassandra.tcm.Epoch; import org.apache.cassandra.tcm.FetchCMSLog; import org.apache.cassandra.tcm.MetadataSnapshots; import org.apache.cassandra.tcm.Processor; import org.apache.cassandra.tcm.Retry; import org.apache.cassandra.tcm.Transformation; +import org.apache.cassandra.tcm.discovery.Discovery; import org.apache.cassandra.tcm.log.Entry; import org.apache.cassandra.tcm.log.LocalLog; import org.apache.cassandra.tcm.log.LogState; diff --git a/test/distributed/org/apache/cassandra/distributed/test/log/MultiNodeCMSRediscoveryTest.java b/test/distributed/org/apache/cassandra/distributed/test/log/MultiNodeCMSRediscoveryTest.java new file mode 100644 index 000000000000..70e34b340204 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/log/MultiNodeCMSRediscoveryTest.java @@ -0,0 +1,38 @@ +/* + * 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.cassandra.distributed.test.log; + +/** + * Tests a full-cluster restart with every node's broadcast address changing while down. + * The cluster is configured with 6 nodes, 3 of which are CMS members. + */ +public class MultiNodeCMSRediscoveryTest extends CMSRediscoveryTestBase +{ + @Override + int clusterSize() + { + return 6; + } + + @Override + int cmsSize() + { + return 3; + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/test/log/SingleNodeCMSRediscoveryTest.java b/test/distributed/org/apache/cassandra/distributed/test/log/SingleNodeCMSRediscoveryTest.java new file mode 100644 index 000000000000..9de7ffaae916 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/log/SingleNodeCMSRediscoveryTest.java @@ -0,0 +1,38 @@ +/* + * 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.cassandra.distributed.test.log; + +/** + * Tests a full-cluster restart with every node's broadcast address changing while down. + * The cluster is configured with 3 nodes, only 1 of which is a CMS member. + */ +public class SingleNodeCMSRediscoveryTest extends CMSRediscoveryTestBase +{ + @Override + int clusterSize() + { + return 3; + } + + @Override + int cmsSize() + { + return 1; + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradeChangeAllIPTest.java b/test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradeChangeAllIPTest.java new file mode 100644 index 000000000000..82fcb3f0dafc --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradeChangeAllIPTest.java @@ -0,0 +1,34 @@ +/* + * 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.cassandra.distributed.upgrade; + +import java.util.stream.IntStream; + +import org.junit.Test; + + +public class ClusterMetadataUpgradeChangeAllIPTest extends ClusterMetadataUpgradeChangeIPTestBase +{ + @Test + public void gossipModeAllIPChangeTest() throws Throwable + { + // all nodes upgraded, bouncing all nodes to new ips while in gossip mode + ipChangeTestHelper(IntStream.rangeClosed(1, NODE_COUNT).toArray()); + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradeChangeGossipIPTest.java b/test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradeChangeGossipIPTest.java new file mode 100644 index 000000000000..1dc734cfed80 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradeChangeGossipIPTest.java @@ -0,0 +1,34 @@ +/* + * 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.cassandra.distributed.upgrade; + +import java.util.stream.IntStream; + +import org.junit.Test; + + +public class ClusterMetadataUpgradeChangeGossipIPTest extends ClusterMetadataUpgradeChangeIPTestBase +{ + @Test + public void gossipModeIPChangeTest() throws Throwable + { + // half the nodes upgraded, bouncing all nodes to new ips (and upgrade the remaining ones) + ipChangeTestHelper(IntStream.rangeClosed(1, NODE_COUNT / 2).toArray()); + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradeChangeIPTest.java b/test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradeChangeIPTest.java index c363ba5cf18a..440f71df0b15 100644 --- a/test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradeChangeIPTest.java +++ b/test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradeChangeIPTest.java @@ -20,54 +20,13 @@ import org.junit.Test; -import org.apache.cassandra.distributed.Constants; -import org.apache.cassandra.distributed.api.Feature; -import org.apache.cassandra.distributed.api.IInstanceConfig; -import org.apache.cassandra.distributed.api.IUpgradeableInstance; -import org.apache.cassandra.distributed.api.TokenSupplier; -import org.apache.cassandra.distributed.impl.AbstractCluster; -import org.apache.cassandra.distributed.shared.NetworkTopology; -public class ClusterMetadataUpgradeChangeIPTest extends UpgradeTestBase +public class ClusterMetadataUpgradeChangeIPTest extends ClusterMetadataUpgradeChangeIPTestBase { - @Test - public void gossipModeIPChangeTest() throws Throwable - { - // all nodes upgraded, bouncing node3 to new ip while in gossip mode - ipChangeTestHelper(1, 2, 3); - } - @Test public void upgradeChangeIPTest() throws Throwable { - // changing IP while upgrading node 3 - ipChangeTestHelper(1, 2); - } - - private void ipChangeTestHelper(int ... toUpgrade) throws Throwable - { - TokenSupplier ts = TokenSupplier.evenlyDistributedTokens(3); - new TestCase() - .nodesToUpgrade(toUpgrade) - .withConfig((cfg) -> cfg.with(Feature.NETWORK, Feature.GOSSIP) - .set(Constants.KEY_DTEST_FULL_STARTUP, true)) - .withBuilder(builder -> builder.withNodeIdTopology(NetworkTopology.singleDcNetworkTopology(4, "dc0", "rack0")) - .withTokenSupplier((TokenSupplier) i -> i == 4 ? ts.tokens(3) : ts.tokens(i))) - .nodes(3) - .upgradesToCurrentFrom(v50) - .setup((cluster) -> {}) - .runAfterClusterUpgrade((cluster) -> { - cluster.get(3).shutdown().get(); - IInstanceConfig nodeConfig = cluster.newInstanceConfig(); - nodeConfig.set("data_file_directories", cluster.get(3).config().get("data_file_directories")); - IUpgradeableInstance newInstance = cluster.bootstrap(nodeConfig, AbstractCluster.CURRENT_VERSION); - newInstance.startup(); - cluster.get(1).nodetoolResult("cms", "initialize").asserts().success(); - - cluster.get(2).shutdown().get(); - cluster.get(2).startup(); - - cluster.get(2).nodetoolResult("cms", "reconfigure", "3").asserts().success(); - }).run(); + // changing all ips while upgrading + ipChangeTestHelper(); } } diff --git a/test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradeChangeIPTestBase.java b/test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradeChangeIPTestBase.java new file mode 100644 index 000000000000..eb57a3db7c97 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradeChangeIPTestBase.java @@ -0,0 +1,91 @@ +/* + * 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.cassandra.distributed.upgrade; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Random; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.cassandra.distributed.Constants; +import org.apache.cassandra.distributed.api.Feature; +import org.apache.cassandra.distributed.api.IInstanceConfig; +import org.apache.cassandra.distributed.api.IUpgradeableInstance; +import org.apache.cassandra.distributed.api.TokenSupplier; +import org.apache.cassandra.distributed.impl.AbstractCluster; +import org.apache.cassandra.distributed.shared.NetworkTopology; +import org.apache.cassandra.locator.SimpleSeedProvider; + +public class ClusterMetadataUpgradeChangeIPTestBase extends UpgradeTestBase +{ + private static final Logger logger = LoggerFactory.getLogger(ClusterMetadataUpgradeChangeIPTestBase.class); + static final int NODE_COUNT = 4; + + void ipChangeTestHelper(int ... toUpgrade) throws Throwable + { + long seed = System.currentTimeMillis(); + Random r = new Random(seed); + logger.info("SEED={}", seed); + + TokenSupplier ts = TokenSupplier.evenlyDistributedTokens(NODE_COUNT); + + new UpgradeTestBase.TestCase() + .nodesToUpgrade(toUpgrade) + .withConfig((cfg) -> cfg.with(Feature.NETWORK, Feature.GOSSIP) + .set(Constants.KEY_DTEST_FULL_STARTUP, true)) + .withBuilder(builder -> builder.withNodeIdTopology(NetworkTopology.singleDcNetworkTopology(NODE_COUNT * 2, "dc0", "rack0")) + .withTokenSupplier((TokenSupplier) i -> i > NODE_COUNT ? ts.tokens(i - NODE_COUNT) : ts.tokens(i))) + .nodes(NODE_COUNT) + .upgradesToCurrentFrom(v50) + .setup((cluster) -> {}) + .runAfterClusterUpgrade((cluster) -> { + IInstanceConfig.ParameterizedClass seedConf = + new IInstanceConfig.ParameterizedClass(SimpleSeedProvider.class.getName(), + Collections.singletonMap("seeds", "127.0.0.1,127.0.0.2,127.0.0.5,127.0.0.6,127.0.0.7")); + List ipChangeOrder = new ArrayList<>(NODE_COUNT); + for (int i = 1; i <= NODE_COUNT; i++) + { + cluster.get(i).config().set("seed_provider", seedConf); + ipChangeOrder.add(i); + } + Collections.shuffle(ipChangeOrder, r); + + for (int i : ipChangeOrder) + { + cluster.get(i).shutdown().get(); + IInstanceConfig nodeConfig = cluster.newInstanceConfig(); + nodeConfig.set("seed_provider", seedConf); + nodeConfig.set("data_file_directories", cluster.get(i).config().get("data_file_directories")); + IUpgradeableInstance newInstance = cluster.bootstrap(nodeConfig, AbstractCluster.CURRENT_VERSION); + newInstance.startup(); + } + + cluster.get(randomNode(r)).nodetoolResult("cms", "initialize").asserts().success(); + cluster.get(randomNode(r)).nodetoolResult("cms", "reconfigure", String.valueOf(NODE_COUNT)).asserts().success(); + }).run(); + } + + static int randomNode(Random r) + { + return NODE_COUNT + r.nextInt(NODE_COUNT) + 1; + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/upgrade/UpgradeTestBase.java b/test/distributed/org/apache/cassandra/distributed/upgrade/UpgradeTestBase.java index 7fb41b843f43..2c60941dcece 100644 --- a/test/distributed/org/apache/cassandra/distributed/upgrade/UpgradeTestBase.java +++ b/test/distributed/org/apache/cassandra/distributed/upgrade/UpgradeTestBase.java @@ -163,6 +163,7 @@ public static class TestCase implements ThrowingRunnable private RunOnCluster runBeforeClusterUpgrade; private RunOnCluster runAfterClusterUpgrade; private final Set nodesToUpgrade = new LinkedHashSet<>(); + private boolean noNodesToUpgrade = false; private Consumer configConsumer; private Consumer builderConsumer; private TokenSupplier tokenSupplier; @@ -370,7 +371,7 @@ public void run() throws Throwable runAfterClusterUpgrade = (c) -> {}; if (runAfterNodeUpgrade == null) runAfterNodeUpgrade = (c, n) -> {}; - if (nodesToUpgrade.isEmpty()) + if (nodesToUpgrade.isEmpty() && !noNodesToUpgrade) for (int n = 1; n <= nodeCount; n++) nodesToUpgrade.add(n); @@ -416,6 +417,11 @@ public void run() throws Throwable public TestCase nodesToUpgrade(int ... nodes) { + if (nodes.length == 0) + { + noNodesToUpgrade = true; + return this; + } Set set = new HashSet<>(nodes.length); for (int n : nodes) { diff --git a/test/unit/org/apache/cassandra/gms/NewGossiperTest.java b/test/unit/org/apache/cassandra/gms/NewGossiperTest.java index a11b3d0208f2..4f15cf555ad9 100644 --- a/test/unit/org/apache/cassandra/gms/NewGossiperTest.java +++ b/test/unit/org/apache/cassandra/gms/NewGossiperTest.java @@ -70,9 +70,11 @@ public void mergeTest() throws InterruptedException, ExecutionException NewGossiper.ShadowRoundHandler srh = new NewGossiper.ShadowRoundHandler(fakePeers, new NoOpMessageDelivery()); Future> states = srh.doShadowRound(); Map firstResp = buildEpstates(fakePeers, r); - srh.onAck(firstResp); + srh.onAck(InetAddressAndPort.getByNameUnchecked("127.0.0.1"), firstResp); Map secondResp = buildEpstates(fakePeers, r); - srh.onAck(secondResp); + srh.onAck(InetAddressAndPort.getByNameUnchecked("127.0.0.2"), secondResp); + srh.onAck(InetAddressAndPort.getByNameUnchecked("127.0.0.3"), secondResp); + srh.onAck(InetAddressAndPort.getByNameUnchecked("127.0.0.4"), secondResp); Map result = states.get(); verifyResult(result, firstResp, secondResp); } diff --git a/test/unit/org/apache/cassandra/tcm/RemoteProcessorTest.java b/test/unit/org/apache/cassandra/tcm/RemoteProcessorTest.java index 85dcfac8a1bd..b06d69c94ac1 100644 --- a/test/unit/org/apache/cassandra/tcm/RemoteProcessorTest.java +++ b/test/unit/org/apache/cassandra/tcm/RemoteProcessorTest.java @@ -31,6 +31,7 @@ import org.apache.cassandra.config.Config; import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.locator.Locator; +import org.apache.cassandra.tcm.discovery.Discovery; import org.apache.cassandra.tcm.membership.Location; import static org.junit.Assert.assertEquals; @@ -79,30 +80,6 @@ public void timeoutTest() } } - @Test - public void notCMSTest() - { - // make sure that a node marked as notCMS will not be returned until we've cycled through all other candidates - // when using the iterator in a RemoteProcessor::sendWithCallback call, the Backoff will trigger the breaking - // out of the cycle. - int endpointCount = 10; - List allEndpoints = eps(endpointCount); - Set discovery = new HashSet<>(allEndpoints.subList(0, 4)); - RemoteProcessor.CandidateIterator iter = new RemoteProcessor.CandidateIterator(discovery, false); - InetAddressAndPort notcms = iter.peek(); - for (int i = 1; i < 10; i++) - { - assertTrue(iter.hasNext()); - InetAddressAndPort returned = iter.next(); - assertTrue(discovery.contains(returned)); - if (returned.equals(notcms)) - { - iter.notCms(returned); - assertEquals(notcms, iter.peekLast()); - } - } - } - private List eps(int endpointCount) { List allEndpoints = new ArrayList<>(endpointCount); diff --git a/test/unit/org/apache/cassandra/tcm/RetryTest.java b/test/unit/org/apache/cassandra/tcm/RetryTest.java index 5fe2c7ebbe28..2e401761c5b9 100644 --- a/test/unit/org/apache/cassandra/tcm/RetryTest.java +++ b/test/unit/org/apache/cassandra/tcm/RetryTest.java @@ -91,16 +91,14 @@ public long computeWait(int attempts, TimeUnit units) @Test public void testProcessorIndefiniteRetryBehaviour() { - Retry retryPolicy = Retry.unsafeRetryIndefinitely(); + WaitStrategy retryPolicy = Retry.unsafeRetryIndefinitely(); // Assert the properties of the Retry provided by the private static Processor::unsafeRetryIndefinitely for (int i = 1; i < 1000; i++) { // backoff increases in 100ms steps, up to a max of 10000ms long waitTime = retryPolicy.computeWait(i, MILLISECONDS); - assertEquals(Math.min((i + 1) * 100, 10000), waitTime); + assertEquals(Math.min((i) * 100, 10000), waitTime); } - // Retry indefinitely means no explicit deadline is set - assertEquals(Long.MAX_VALUE, retryPolicy.deadlineNanos); } @Test diff --git a/test/unit/org/apache/cassandra/tcm/DiscoverySimulationTest.java b/test/unit/org/apache/cassandra/tcm/discovery/DiscoverySimulationTest.java similarity index 96% rename from test/unit/org/apache/cassandra/tcm/DiscoverySimulationTest.java rename to test/unit/org/apache/cassandra/tcm/discovery/DiscoverySimulationTest.java index 85fa61ed9b31..17b579b3c4fa 100644 --- a/test/unit/org/apache/cassandra/tcm/DiscoverySimulationTest.java +++ b/test/unit/org/apache/cassandra/tcm/discovery/DiscoverySimulationTest.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.cassandra.tcm; +package org.apache.cassandra.tcm.discovery; import java.io.IOException; import java.util.Collections; @@ -44,6 +44,11 @@ import org.apache.cassandra.net.MessageDelivery; import org.apache.cassandra.net.RequestCallback; import org.apache.cassandra.net.Verb; +import org.apache.cassandra.tcm.AtomicLongBackedProcessor; +import org.apache.cassandra.tcm.ClusterMetadata; +import org.apache.cassandra.tcm.ClusterMetadataService; +import org.apache.cassandra.tcm.Commit; +import org.apache.cassandra.tcm.MetadataSnapshots; import org.apache.cassandra.tcm.log.LocalLog; import org.apache.cassandra.tcm.ownership.UniformRangePlacement; import org.apache.cassandra.utils.concurrent.Future; @@ -101,7 +106,7 @@ public void discoveryTest() throws Throwable Map> 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 discovered = new HashMap<>(); diff --git a/test/unit/org/apache/cassandra/tcm/discovery/SurveyRequestHandlerTest.java b/test/unit/org/apache/cassandra/tcm/discovery/SurveyRequestHandlerTest.java new file mode 100644 index 000000000000..50a2da973abe --- /dev/null +++ b/test/unit/org/apache/cassandra/tcm/discovery/SurveyRequestHandlerTest.java @@ -0,0 +1,106 @@ +/* + * 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.cassandra.tcm.discovery; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; + +import org.junit.BeforeClass; +import org.junit.Test; + +import org.apache.cassandra.ServerTestUtils; +import org.apache.cassandra.db.SystemKeyspace; +import org.apache.cassandra.exceptions.InvalidRequestException; +import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.net.ConnectionType; +import org.apache.cassandra.net.Message; +import org.apache.cassandra.net.MessageDelivery; +import org.apache.cassandra.net.RequestCallback; +import org.apache.cassandra.net.Verb; +import org.apache.cassandra.tcm.ClusterMetadata; +import org.apache.cassandra.tcm.membership.NodeId; +import org.apache.cassandra.utils.concurrent.Future; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.fail; + +public class SurveyRequestHandlerTest +{ + @BeforeClass + public static void initClusterMetadata() + { + ServerTestUtils.prepareServerNoRegister(); + } + + @Test + public void testRequestWithMismatchingMetadataIdIsRejected() throws IOException + { + StubMessageDelivery messaging = new StubMessageDelivery(); + SurveyRequestHandler handler = new SurveyRequestHandler(999, () -> messaging); + try + { + handler.doVerb(Message.out(Verb.TCM_DISCOVER_SURVEY_REQ, new SurveyRequest(0))); + fail("Expected InvalidRequestException"); + } + catch (InvalidRequestException e) + { + assertEquals("Mismatching metadata id in survey request from /127.0.0.1:7012 (0)",e.getMessage()); + } + } + + @Test + public void testRespondWithNodeIdFromSystemTable() throws IOException + { + ClusterMetadata metadata = ClusterMetadata.current(); + StubMessageDelivery messaging = new StubMessageDelivery(); + SurveyRequestHandler handler = new SurveyRequestHandler(metadata.metadataIdentifier, () -> messaging); + assertEquals(NodeId.UNREGISTERED, ClusterMetadata.current().myNodeId()); + NodeId id = new NodeId(555); + SystemKeyspace.setLocalHostId(id.toUUID()); + handler.doVerb(Message.out(Verb.TCM_DISCOVER_SURVEY_REQ, new SurveyRequest(0))); + assertEquals(1, messaging.responses.size()); + SurveyResponse response = (SurveyResponse) messaging.responses.get(0); + assertEquals(id, response.nodeId); + assertEquals(metadata.metadataIdentifier, response.metadataId); + } + + private static class StubMessageDelivery implements MessageDelivery + { + + List responses = new ArrayList<>(); + @Override + public void respond(V response, Message message) + { + responses.add(response); + } + + @Override + public void send(Message message, InetAddressAndPort to) {} + + @Override + public void sendWithCallback(Message message, InetAddressAndPort to, RequestCallback cb) {} + + @Override + public void sendWithCallback(Message message, InetAddressAndPort to, RequestCallback cb, ConnectionType specifyConnection) {} + + @Override + public Future> sendWithResult(Message message, InetAddressAndPort to) {return null;} + } +}