From 2b01b5fa79369179cb264ddcb4b3013a878efa04 Mon Sep 17 00:00:00 2001 From: Caleb Rackliffe Date: Tue, 16 Jul 2024 17:40:49 -0500 Subject: [PATCH] Command to Exclude Replicas from Durability Status Coordination patch by Caleb Rackliffe; reviewed by David Capwell, Sam Tunnicliffe, and Benedict Elliott Smith for CASSANDRA-19321 --- modules/accord | 2 +- .../cassandra/service/CassandraDaemon.java | 2 + .../service/accord/AccordOperations.java | 76 +++++++++ .../service/accord/AccordOperationsMBean.java | 31 ++++ .../service/accord/AccordStaleReplicas.java | 137 ++++++++++++++++ .../service/accord/AccordTopology.java | 25 ++- .../serializers/TopologySerializers.java | 5 +- .../apache/cassandra/tcm/ClusterMetadata.java | 80 ++++++++-- .../apache/cassandra/tcm/MetadataKeys.java | 2 + .../tcm/StubClusterMetadataService.java | 4 +- .../apache/cassandra/tcm/Transformation.java | 4 + .../tcm/compatibility/GossipHelper.java | 10 +- .../cassandra/tcm/serialization/Version.java | 1 + .../transformations/AccordMarkRejoining.java | 123 ++++++++++++++ .../tcm/transformations/AccordMarkStale.java | 151 ++++++++++++++++++ .../org/apache/cassandra/tools/NodeProbe.java | 10 ++ .../org/apache/cassandra/tools/NodeTool.java | 7 + .../cassandra/tools/nodetool/AccordAdmin.java | 69 ++++++++ .../mock/nodetool/InternalNodeProbe.java | 2 + .../test/accord/AccordNodetoolTest.java | 99 ++++++++++++ .../test/log/ClusterMetadataTestHelper.java | 10 +- .../cassandra/locator/MetaStrategyTest.java | 10 +- .../accord/AccordFastPathCoordinatorTest.java | 17 +- .../accord/AccordStaleReplicasTest.java | 54 +++++++ .../ClusterMetadataTransformationTest.java | 2 + .../AsymmetricMetadataSerializers.java | 42 +++++ .../AccordMarkRejoiningTest.java | 56 +++++++ .../transformations/AccordMarkStaleTest.java | 56 +++++++ 28 files changed, 1047 insertions(+), 40 deletions(-) create mode 100644 src/java/org/apache/cassandra/service/accord/AccordOperations.java create mode 100644 src/java/org/apache/cassandra/service/accord/AccordOperationsMBean.java create mode 100644 src/java/org/apache/cassandra/service/accord/AccordStaleReplicas.java create mode 100644 src/java/org/apache/cassandra/tcm/transformations/AccordMarkRejoining.java create mode 100644 src/java/org/apache/cassandra/tcm/transformations/AccordMarkStale.java create mode 100644 src/java/org/apache/cassandra/tools/nodetool/AccordAdmin.java create mode 100644 test/distributed/org/apache/cassandra/distributed/test/accord/AccordNodetoolTest.java create mode 100644 test/unit/org/apache/cassandra/service/accord/AccordStaleReplicasTest.java create mode 100644 test/unit/org/apache/cassandra/tcm/serialization/AsymmetricMetadataSerializers.java create mode 100644 test/unit/org/apache/cassandra/tcm/transformations/AccordMarkRejoiningTest.java create mode 100644 test/unit/org/apache/cassandra/tcm/transformations/AccordMarkStaleTest.java diff --git a/modules/accord b/modules/accord index 449b2b4d0b..81c02769f9 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit 449b2b4d0bf4bb44d55a3c57f712a4d5a15e7220 +Subproject commit 81c02769f9ad73ef3aba0675c2217fc74b8a4a4c diff --git a/src/java/org/apache/cassandra/service/CassandraDaemon.java b/src/java/org/apache/cassandra/service/CassandraDaemon.java index 57f38e93e9..45740f1916 100644 --- a/src/java/org/apache/cassandra/service/CassandraDaemon.java +++ b/src/java/org/apache/cassandra/service/CassandraDaemon.java @@ -78,6 +78,7 @@ import org.apache.cassandra.net.StartupClusterConnectivityChecker; import org.apache.cassandra.schema.Schema; import org.apache.cassandra.schema.SchemaConstants; import org.apache.cassandra.security.ThreadAwareSecurityManager; +import org.apache.cassandra.service.accord.AccordOperations; import org.apache.cassandra.service.paxos.PaxosState; import org.apache.cassandra.service.snapshot.SnapshotManager; import org.apache.cassandra.streaming.StreamManager; @@ -269,6 +270,7 @@ public class CassandraDaemon Startup.initialize(DatabaseDescriptor.getSeeds()); disableAutoCompaction(Schema.instance.distributedKeyspaces().names()); CMSOperations.initJmx(); + AccordOperations.initJmx(); if (ClusterMetadata.current().myNodeId() != null) RegistrationStatus.instance.onRegistration(); } diff --git a/src/java/org/apache/cassandra/service/accord/AccordOperations.java b/src/java/org/apache/cassandra/service/accord/AccordOperations.java new file mode 100644 index 0000000000..e7820919e1 --- /dev/null +++ b/src/java/org/apache/cassandra/service/accord/AccordOperations.java @@ -0,0 +1,76 @@ +/* + * 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.service.accord; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.stream.Collectors; + +import org.apache.cassandra.tcm.ClusterMetadata; +import org.apache.cassandra.tcm.ClusterMetadataService; +import org.apache.cassandra.tcm.membership.NodeId; +import org.apache.cassandra.tcm.transformations.AccordMarkStale; +import org.apache.cassandra.tcm.transformations.AccordMarkRejoining; +import org.apache.cassandra.utils.MBeanWrapper; + +public class AccordOperations implements AccordOperationsMBean +{ + public static final String MBEAN_OBJECT_NAME = "org.apache.cassandra.service.accord:type=AccordOperations"; + public static final AccordOperations instance = new AccordOperations(ClusterMetadataService.instance()); + + private final ClusterMetadataService cms; + + public static void initJmx() + { + MBeanWrapper.instance.registerMBean(instance, MBEAN_OBJECT_NAME); + } + + private AccordOperations(ClusterMetadataService cms) + { + this.cms = cms; + } + + @Override + public Map describe() + { + Map info = new HashMap<>(); + ClusterMetadata metadata = ClusterMetadata.current(); + + info.put("EPOCH", Long.toString(metadata.epoch.getEpoch())); + String staleReplicas = metadata.accordStaleReplicas.ids().stream().sorted().map(Object::toString).collect(Collectors.joining(",")); + info.put("STALE_REPLICAS", staleReplicas); + return info; + } + + @Override + public void accordMarkStale(List nodeIdStrings) + { + Set nodeIds = nodeIdStrings.stream().map(NodeId::fromString).collect(Collectors.toSet()); + cms.commit(new AccordMarkStale(nodeIds)); + } + + @Override + public void accordMarkRejoining(List nodeIdStrings) + { + Set nodeIds = nodeIdStrings.stream().map(NodeId::fromString).collect(Collectors.toSet()); + cms.commit(new AccordMarkRejoining(nodeIds)); + } +} diff --git a/src/java/org/apache/cassandra/service/accord/AccordOperationsMBean.java b/src/java/org/apache/cassandra/service/accord/AccordOperationsMBean.java new file mode 100644 index 0000000000..e0b0884733 --- /dev/null +++ b/src/java/org/apache/cassandra/service/accord/AccordOperationsMBean.java @@ -0,0 +1,31 @@ +/* + * 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.service.accord; + +import java.util.List; +import java.util.Map; + +public interface AccordOperationsMBean +{ + Map describe(); + + void accordMarkStale(List nodeIds); + + void accordMarkRejoining(List nodeIds); +} diff --git a/src/java/org/apache/cassandra/service/accord/AccordStaleReplicas.java b/src/java/org/apache/cassandra/service/accord/AccordStaleReplicas.java new file mode 100644 index 0000000000..2502fa1a63 --- /dev/null +++ b/src/java/org/apache/cassandra/service/accord/AccordStaleReplicas.java @@ -0,0 +1,137 @@ +/* + * 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.service.accord; + +import java.io.IOException; +import java.util.Objects; +import java.util.Set; + +import javax.annotation.concurrent.Immutable; + +import com.google.common.collect.ImmutableSet; + +import accord.local.Node; +import org.apache.cassandra.io.util.DataInputPlus; +import org.apache.cassandra.io.util.DataOutputPlus; +import org.apache.cassandra.service.accord.serializers.TopologySerializers; +import org.apache.cassandra.tcm.Epoch; +import org.apache.cassandra.tcm.MetadataValue; +import org.apache.cassandra.tcm.serialization.MetadataSerializer; +import org.apache.cassandra.tcm.serialization.Version; +import org.apache.cassandra.utils.CollectionSerializers; + +@Immutable +public class AccordStaleReplicas implements MetadataValue +{ + public static final AccordStaleReplicas EMPTY = new AccordStaleReplicas(ImmutableSet.of(), Epoch.EMPTY); + + private final Set staleIds; + private final Epoch lastModified; + + AccordStaleReplicas(Set staleIds, Epoch lastModified) + { + this.staleIds = staleIds; + this.lastModified = lastModified; + } + + @Override + public AccordStaleReplicas withLastModified(Epoch epoch) + { + return new AccordStaleReplicas(staleIds, epoch); + } + + @Override + public Epoch lastModified() + { + return lastModified; + } + + public AccordStaleReplicas withNodeIds(Set ids) + { + ImmutableSet.Builder builder = new ImmutableSet.Builder<>(); + Set newIds = builder.addAll(staleIds).addAll(ids).build(); + return new AccordStaleReplicas(newIds, lastModified); + } + + public AccordStaleReplicas without(Set ids) + { + ImmutableSet.Builder builder = new ImmutableSet.Builder<>(); + + for (Node.Id staleId : staleIds) + if (!ids.contains(staleId)) + builder.add(staleId); + + return new AccordStaleReplicas(builder.build(), lastModified); + } + + public boolean contains(Node.Id nodeId) + { + return staleIds.contains(nodeId); + } + + public Set ids() + { + return staleIds; + } + + @Override + public String toString() + { + return "AccordStaleReplicas{staleIds=" + staleIds + ", lastModified=" + lastModified + '}'; + } + + @Override + public boolean equals(Object o) + { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + AccordStaleReplicas that = (AccordStaleReplicas) o; + return Objects.equals(staleIds, that.staleIds) && Objects.equals(lastModified, that.lastModified); + } + + @Override + public int hashCode() + { + return Objects.hash(staleIds, lastModified); + } + + public static final MetadataSerializer serializer = new MetadataSerializer<>() + { + @Override + public void serialize(AccordStaleReplicas replicas, DataOutputPlus out, Version version) throws IOException + { + CollectionSerializers.serializeCollection(replicas.staleIds, out, version, TopologySerializers.nodeId); + Epoch.serializer.serialize(replicas.lastModified, out, version); + } + + @Override + public AccordStaleReplicas deserialize(DataInputPlus in, Version version) throws IOException + { + return new AccordStaleReplicas(CollectionSerializers.deserializeSet(in, version, TopologySerializers.nodeId), + Epoch.serializer.deserialize(in, version)); + } + + @Override + public long serializedSize(AccordStaleReplicas replicas, Version version) + { + return CollectionSerializers.serializedCollectionSize(replicas.staleIds, version, TopologySerializers.nodeId) + + Epoch.serializer.serializedSize(replicas.lastModified, version); + } + }; +} diff --git a/src/java/org/apache/cassandra/service/accord/AccordTopology.java b/src/java/org/apache/cassandra/service/accord/AccordTopology.java index 7bbfc0c250..45f3b3fd76 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordTopology.java +++ b/src/java/org/apache/cassandra/service/accord/AccordTopology.java @@ -78,7 +78,7 @@ public class AccordTopology } } - static class KeyspaceShard + public static class KeyspaceShard { private final KeyspaceMetadata keyspace; private final Range range; @@ -139,7 +139,7 @@ public class AccordTopology return new KeyspaceShard(keyspace, range, nodes, pending); } - public static List forKeyspace(KeyspaceMetadata keyspace, DataPlacements placements, Directory directory, ShardLookup lookup) + public static List forKeyspace(KeyspaceMetadata keyspace, DataPlacements placements, Directory directory) { ReplicationParams replication = keyspace.params.replication; DataPlacement placement = placements.get(replication); @@ -154,6 +154,16 @@ public class AccordTopology } return shards; } + + public List nodes() + { + return nodes; + } + + public Range range() + { + return range; + } } static TokenRange minRange(TableId table, Token token) @@ -219,7 +229,9 @@ public class AccordTopology return builder.build(); } - public static Topology createAccordTopology(Epoch epoch, DistributedSchema schema, DataPlacements placements, Directory directory, AccordFastPath accordFastPath, ShardLookup lookup) + public static Topology createAccordTopology(Epoch epoch, DistributedSchema schema, DataPlacements placements, + Directory directory, AccordFastPath accordFastPath, ShardLookup lookup, + AccordStaleReplicas staleReplicas) { List shards = new ArrayList<>(); Set unavailable = accordFastPath.unavailableIds(); @@ -230,17 +242,18 @@ public class AccordTopology List tables = keyspace.tables.stream().filter(TableMetadata::requiresAccordSupport).collect(Collectors.toList()); if (tables.isEmpty()) continue; - List ksShards = KeyspaceShard.forKeyspace(keyspace, placements, directory, lookup); + List ksShards = KeyspaceShard.forKeyspace(keyspace, placements, directory); tables.forEach(table -> ksShards.forEach(shard -> shards.add(shard.createForTable(table, unavailable, dcMap, lookup)))); } shards.sort((a, b) -> a.range.compare(b.range)); - return new Topology(epoch.getEpoch(), shards.toArray(new Shard[0])); + + return new Topology(epoch.getEpoch(), staleReplicas.ids(), shards.toArray(new Shard[0])); } public static Topology createAccordTopology(ClusterMetadata metadata, ShardLookup lookup) { - return createAccordTopology(metadata.epoch, metadata.schema, metadata.placements, metadata.directory, metadata.accordFastPath, lookup); + return createAccordTopology(metadata.epoch, metadata.schema, metadata.placements, metadata.directory, metadata.accordFastPath, lookup, metadata.accordStaleReplicas); } public static Topology createAccordTopology(ClusterMetadata metadata, Topology current) diff --git a/src/java/org/apache/cassandra/service/accord/serializers/TopologySerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/TopologySerializers.java index 782ecbf5ed..73708c125f 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/TopologySerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/TopologySerializers.java @@ -159,6 +159,7 @@ public class TopologySerializers { out.writeLong(topology.epoch()); ArraySerializers.serializeArray(topology.unsafeGetShards(), out, version, shard); + CollectionSerializers.serializeCollection(topology.staleIds(), out, version, TopologySerializers.nodeId); } @Override @@ -166,7 +167,8 @@ public class TopologySerializers { long epoch = in.readLong(); Shard[] shards = ArraySerializers.deserializeArray(in, version, shard, Shard[]::new); - return new Topology(epoch, shards); + Set staleIds = CollectionSerializers.deserializeSet(in, version, TopologySerializers.nodeId); + return new Topology(epoch, staleIds, shards); } @Override @@ -175,6 +177,7 @@ public class TopologySerializers long size = 0; size += TypeSizes.LONG_SIZE; // epoch size += ArraySerializers.serializedArraySize(topology.unsafeGetShards(), version, shard); + size += CollectionSerializers.serializedCollectionSize(topology.staleIds(), version, TopologySerializers.nodeId); return size; } }; diff --git a/src/java/org/apache/cassandra/tcm/ClusterMetadata.java b/src/java/org/apache/cassandra/tcm/ClusterMetadata.java index 95c3e1ba2c..9c497e2d15 100644 --- a/src/java/org/apache/cassandra/tcm/ClusterMetadata.java +++ b/src/java/org/apache/cassandra/tcm/ClusterMetadata.java @@ -21,6 +21,7 @@ package org.apache.cassandra.tcm; import java.io.IOException; import java.util.ArrayList; import java.util.Collection; +import java.util.Collections; import java.util.HashMap; import java.util.HashSet; import java.util.Iterator; @@ -57,6 +58,8 @@ import org.apache.cassandra.schema.KeyspaceMetadata; import org.apache.cassandra.schema.Keyspaces; import org.apache.cassandra.schema.ReplicationParams; import org.apache.cassandra.schema.TableId; +import org.apache.cassandra.service.accord.AccordStaleReplicas; +import org.apache.cassandra.service.accord.AccordTopology; import org.apache.cassandra.service.consensus.migration.ConsensusMigrationState; import org.apache.cassandra.service.consensus.migration.TableMigrationState; import org.apache.cassandra.tcm.extensions.ExtensionKey; @@ -100,6 +103,7 @@ public class ClusterMetadata public final InProgressSequences inProgressSequences; public final ConsensusMigrationState consensusMigrationState; public final ImmutableMap, ExtensionValue> extensions; + public final AccordStaleReplicas accordStaleReplicas; // This isn't serialized as part of ClusterMetadata it's really just a view over the Directory. public final Locator locator; @@ -135,7 +139,8 @@ public class ClusterMetadata LockedRanges.EMPTY, InProgressSequences.EMPTY, ConsensusMigrationState.EMPTY, - ImmutableMap.of()); + ImmutableMap.of(), + AccordStaleReplicas.EMPTY); } public ClusterMetadata(Epoch epoch, @@ -148,7 +153,8 @@ public class ClusterMetadata LockedRanges lockedRanges, InProgressSequences inProgressSequences, ConsensusMigrationState consensusMigrationState, - Map, ExtensionValue> extensions) + Map, ExtensionValue> extensions, + AccordStaleReplicas accordStaleReplicas) { this(EMPTY_METADATA_IDENTIFIER, epoch, @@ -161,7 +167,8 @@ public class ClusterMetadata lockedRanges, inProgressSequences, consensusMigrationState, - extensions); + extensions, + accordStaleReplicas); } private ClusterMetadata(int metadataIdentifier, @@ -175,7 +182,8 @@ public class ClusterMetadata LockedRanges lockedRanges, InProgressSequences inProgressSequences, ConsensusMigrationState consensusMigrationState, - Map, ExtensionValue> extensions) + Map, ExtensionValue> extensions, + AccordStaleReplicas accordStaleReplicas) { // TODO: token map is a feature of the specific placement strategy, and so may not be a relevant component of // ClusterMetadata in the long term. We need to consider how the actual components of metadata can be evolved @@ -194,16 +202,17 @@ public class ClusterMetadata this.consensusMigrationState = consensusMigrationState; this.extensions = ImmutableMap.copyOf(extensions); this.locator = Locator.usingDirectory(directory); + this.accordStaleReplicas = accordStaleReplicas; } public ClusterMetadata withDirectory(Directory directory) { - return new ClusterMetadata(epoch, partitioner, schema, directory, tokenMap, placements, accordFastPath, lockedRanges, inProgressSequences, consensusMigrationState, extensions); + return new ClusterMetadata(epoch, partitioner, schema, directory, tokenMap, placements, accordFastPath, lockedRanges, inProgressSequences, consensusMigrationState, extensions, accordStaleReplicas); } public ClusterMetadata withPlacements(DataPlacements placements) { - return new ClusterMetadata(epoch, partitioner, schema, directory, tokenMap, placements, accordFastPath, lockedRanges, inProgressSequences, consensusMigrationState, extensions); + return new ClusterMetadata(epoch, partitioner, schema, directory, tokenMap, placements, accordFastPath, lockedRanges, inProgressSequences, consensusMigrationState, extensions, accordStaleReplicas); } public Set fullCMSMembers() @@ -259,7 +268,8 @@ public class ClusterMetadata capLastModified(lockedRanges, epoch), capLastModified(inProgressSequences, epoch), capLastModified(consensusMigrationState, epoch), - capLastModified(extensions, epoch)); + capLastModified(extensions, epoch), + capLastModified(accordStaleReplicas, epoch)); } public ClusterMetadata initializeClusterIdentifier(int clusterIdentifier) @@ -281,7 +291,8 @@ public class ClusterMetadata lockedRanges, inProgressSequences, consensusMigrationState, - extensions); + extensions, + accordStaleReplicas); } private static Map, ExtensionValue> capLastModified(Map, ExtensionValue> original, Epoch maxEpoch) @@ -410,6 +421,7 @@ public class ClusterMetadata private ConsensusMigrationState consensusMigrationState; private final Map, ExtensionValue> extensions; private final Set modifiedKeys; + private AccordStaleReplicas accordStaleReplicas; private Transformer(ClusterMetadata metadata, Epoch epoch) { @@ -426,6 +438,7 @@ public class ClusterMetadata this.consensusMigrationState = metadata.consensusMigrationState; extensions = new HashMap<>(metadata.extensions); modifiedKeys = new HashSet<>(); + accordStaleReplicas = metadata.accordStaleReplicas; } public Epoch epoch() @@ -456,6 +469,11 @@ public class ClusterMetadata directory = directory.withoutRackAndDC(nodeId).without(nodeId); if (!tokenMap.tokens(nodeId).isEmpty()) tokenMap = tokenMap.unassignTokens(nodeId); + + Node.Id accordId = AccordTopology.tcmIdToAccord(nodeId); + if (accordStaleReplicas.contains(accordId)) + accordStaleReplicas = accordStaleReplicas.without(Collections.singleton(accordId)); + return this; } @@ -523,6 +541,11 @@ public class ClusterMetadata directory = directory.without(replaced) .withRackAndDC(replacement) .withNodeState(replacement, NodeState.JOINED); + + Node.Id accordId = AccordTopology.tcmIdToAccord(replaced); + if (accordStaleReplicas.contains(accordId)) + accordStaleReplicas = accordStaleReplicas.without(Collections.singleton(accordId)); + return this; } @@ -551,6 +574,18 @@ public class ClusterMetadata accordFastPath = accordFastPath.withNodeStatusSince(node, status, updateTimeMillis, updateDelayMillis); return this; } + + public Transformer markStaleReplicas(Set ids) + { + accordStaleReplicas = accordStaleReplicas.withNodeIds(ids); + return this; + } + + public Transformer unmarkStaleReplicas(Set ids) + { + accordStaleReplicas = accordStaleReplicas.without(ids); + return this; + } public Transformer with(LockedRanges lockedRanges) { @@ -677,6 +712,12 @@ public class ClusterMetadata modifiedKeys.add(MetadataKeys.ACCORD_FAST_PATH); accordFastPath = accordFastPath.withLastModified(epoch); } + + if (accordStaleReplicas != base.accordStaleReplicas) + { + modifiedKeys.add(MetadataKeys.ACCORD_STALE_REPLICAS); + accordStaleReplicas = accordStaleReplicas.withLastModified(epoch); + } if (lockedRanges != base.lockedRanges) { @@ -712,7 +753,8 @@ public class ClusterMetadata lockedRanges, inProgressSequences, consensusMigrationState, - extensions), + extensions, + accordStaleReplicas), ImmutableSet.copyOf(modifiedKeys)); } @@ -729,7 +771,8 @@ public class ClusterMetadata lockedRanges, inProgressSequences, consensusMigrationState, - extensions); + extensions, + accordStaleReplicas); } @Override @@ -861,6 +904,7 @@ public class ClusterMetadata lockedRanges.equals(that.lockedRanges) && inProgressSequences.equals(that.inProgressSequences) && consensusMigrationState.equals(that.consensusMigrationState) && + accordStaleReplicas.equals(that.accordStaleReplicas) && extensions.equals(that.extensions); } @@ -909,7 +953,7 @@ public class ClusterMetadata @Override public int hashCode() { - return Objects.hash(epoch, schema, directory, tokenMap, placements, accordFastPath, lockedRanges, inProgressSequences, consensusMigrationState, extensions); + return Objects.hash(epoch, schema, directory, tokenMap, placements, accordFastPath, lockedRanges, inProgressSequences, consensusMigrationState, accordStaleReplicas, extensions); } public static ClusterMetadata current() @@ -990,7 +1034,9 @@ public class ClusterMetadata { AccordFastPath.serializer.serialize(metadata.accordFastPath, out, version); ConsensusMigrationState.serializer.serialize(metadata.consensusMigrationState, out, version); + AccordStaleReplicas.serializer.serialize(metadata.accordStaleReplicas, out, version); } + LockedRanges.serializer.serialize(metadata.lockedRanges, out, version); InProgressSequences.serializer.serialize(metadata.inProgressSequences, out, version); out.writeInt(metadata.extensions.size()); @@ -1027,18 +1073,24 @@ public class ClusterMetadata Directory dir = Directory.serializer.deserialize(in, version); TokenMap tokenMap = TokenMap.serializer.deserialize(in, version); DataPlacements placements = DataPlacements.serializer.deserialize(in, version); + AccordFastPath accordFastPath; ConsensusMigrationState consensusMigrationState; + AccordStaleReplicas staleReplicas; + if (version.isAtLeast(V2)) { accordFastPath = AccordFastPath.serializer.deserialize(in, version); consensusMigrationState = ConsensusMigrationState.serializer.deserialize(in, version); + staleReplicas = AccordStaleReplicas.serializer.deserialize(in, version); } else { accordFastPath = AccordFastPath.EMPTY; consensusMigrationState = ConsensusMigrationState.EMPTY; + staleReplicas = AccordStaleReplicas.EMPTY; } + LockedRanges lockedRanges = LockedRanges.serializer.deserialize(in, version); InProgressSequences ips = InProgressSequences.serializer.deserialize(in, version); int items = in.readInt(); @@ -1061,7 +1113,8 @@ public class ClusterMetadata lockedRanges, ips, consensusMigrationState, - extensions); + extensions, + staleReplicas); } @Override @@ -1085,7 +1138,8 @@ public class ClusterMetadata if (version.isAtLeast(V2)) { size += AccordFastPath.serializer.serializedSize(metadata.accordFastPath, version) + - ConsensusMigrationState.serializer.serializedSize(metadata.consensusMigrationState, version); + ConsensusMigrationState.serializer.serializedSize(metadata.consensusMigrationState, version) + + AccordStaleReplicas.serializer.serializedSize(metadata.accordStaleReplicas, version); } size += LockedRanges.serializer.serializedSize(metadata.lockedRanges, version) + diff --git a/src/java/org/apache/cassandra/tcm/MetadataKeys.java b/src/java/org/apache/cassandra/tcm/MetadataKeys.java index 68306ce313..0aed60581e 100644 --- a/src/java/org/apache/cassandra/tcm/MetadataKeys.java +++ b/src/java/org/apache/cassandra/tcm/MetadataKeys.java @@ -40,6 +40,7 @@ public class MetadataKeys public static final MetadataKey TOKEN_MAP = make(CORE_NS, "ownership", "token_map"); public static final MetadataKey DATA_PLACEMENTS = make(CORE_NS, "ownership", "data_placements"); public static final MetadataKey ACCORD_FAST_PATH = make(CORE_NS, "ownership", "accord_fast_path"); + public static final MetadataKey ACCORD_STALE_REPLICAS = make(CORE_NS, "ownership", "accord_stale_replicas"); public static final MetadataKey LOCKED_RANGES = make(CORE_NS, "sequences", "locked_ranges"); public static final MetadataKey IN_PROGRESS_SEQUENCES = make(CORE_NS, "sequences", "in_progress"); public static final MetadataKey CONSENSUS_MIGRATION_STATE = make(CORE_NS, "consensus", "migration_state"); @@ -49,6 +50,7 @@ public class MetadataKeys TOKEN_MAP, DATA_PLACEMENTS, ACCORD_FAST_PATH, + ACCORD_STALE_REPLICAS, LOCKED_RANGES, IN_PROGRESS_SEQUENCES, CONSENSUS_MIGRATION_STATE); diff --git a/src/java/org/apache/cassandra/tcm/StubClusterMetadataService.java b/src/java/org/apache/cassandra/tcm/StubClusterMetadataService.java index 8a69f9acc4..16cf324799 100644 --- a/src/java/org/apache/cassandra/tcm/StubClusterMetadataService.java +++ b/src/java/org/apache/cassandra/tcm/StubClusterMetadataService.java @@ -29,6 +29,7 @@ import org.apache.cassandra.schema.DistributedSchema; import org.apache.cassandra.schema.KeyspaceMetadata; import org.apache.cassandra.schema.Keyspaces; import org.apache.cassandra.service.accord.AccordFastPath; +import org.apache.cassandra.service.accord.AccordStaleReplicas; import org.apache.cassandra.service.consensus.migration.ConsensusMigrationState; import org.apache.cassandra.tcm.Commit.Replicator; import org.apache.cassandra.tcm.log.Entry; @@ -184,7 +185,8 @@ public class StubClusterMetadataService extends ClusterMetadataService LockedRanges.EMPTY, InProgressSequences.EMPTY, ConsensusMigrationState.EMPTY, - ImmutableMap.of()); + ImmutableMap.of(), + AccordStaleReplicas.EMPTY); } return new StubClusterMetadataService(new UniformRangePlacement(), snapshots != null ? snapshots : MetadataSnapshots.NO_OP, diff --git a/src/java/org/apache/cassandra/tcm/Transformation.java b/src/java/org/apache/cassandra/tcm/Transformation.java index 1ccdb683e9..b8ce1cbc9a 100644 --- a/src/java/org/apache/cassandra/tcm/Transformation.java +++ b/src/java/org/apache/cassandra/tcm/Transformation.java @@ -38,6 +38,8 @@ import org.apache.cassandra.tcm.sequences.LockedRanges; import org.apache.cassandra.tcm.serialization.AsymmetricMetadataSerializer; import org.apache.cassandra.tcm.serialization.VerboseMetadataSerializer; import org.apache.cassandra.tcm.serialization.Version; +import org.apache.cassandra.tcm.transformations.AccordMarkRejoining; +import org.apache.cassandra.tcm.transformations.AccordMarkStale; import org.apache.cassandra.tcm.transformations.AlterSchema; import org.apache.cassandra.tcm.transformations.AlterTopology; import org.apache.cassandra.tcm.transformations.Assassinate; @@ -242,6 +244,8 @@ public interface Transformation BEGIN_CONSENSUS_MIGRATION_FOR_TABLE_AND_RANGE(37, () -> BeginConsensusMigrationForTableAndRange.serializer), MAYBE_FINISH_CONSENSUS_MIGRATION_FOR_TABLE_AND_RANGE(38, () -> MaybeFinishConsensusMigrationForTableAndRange.serializer), + ACCORD_MARK_STALE(39, () -> AccordMarkStale.serializer), + ACCORD_MARK_REJOINING(40, () -> AccordMarkRejoining.serializer), ; private final Supplier> serializer; diff --git a/src/java/org/apache/cassandra/tcm/compatibility/GossipHelper.java b/src/java/org/apache/cassandra/tcm/compatibility/GossipHelper.java index e82572f53b..f569a9d1cb 100644 --- a/src/java/org/apache/cassandra/tcm/compatibility/GossipHelper.java +++ b/src/java/org/apache/cassandra/tcm/compatibility/GossipHelper.java @@ -54,6 +54,7 @@ import org.apache.cassandra.schema.Keyspaces; import org.apache.cassandra.schema.SchemaConstants; import org.apache.cassandra.schema.SchemaKeyspace; import org.apache.cassandra.service.StorageService; +import org.apache.cassandra.service.accord.AccordStaleReplicas; import org.apache.cassandra.service.consensus.migration.ConsensusMigrationState; import org.apache.cassandra.tcm.ClusterMetadata; import org.apache.cassandra.tcm.Epoch; @@ -300,7 +301,8 @@ public class GossipHelper LockedRanges.EMPTY, InProgressSequences.EMPTY, ConsensusMigrationState.EMPTY, - Collections.emptyMap()); + Collections.emptyMap(), + AccordStaleReplicas.EMPTY); } public static ClusterMetadata fromEndpointStates(DistributedSchema schema, Map epStates) @@ -389,7 +391,8 @@ public class GossipHelper LockedRanges.EMPTY, InProgressSequences.EMPTY, ConsensusMigrationState.EMPTY, - extensions); + extensions, + AccordStaleReplicas.EMPTY); DataPlacements placements = new UniformRangePlacement().calculatePlacements(Epoch.UPGRADE_GOSSIP, forPlacementCalculation, schema.getKeyspaces()); @@ -403,7 +406,8 @@ public class GossipHelper LockedRanges.EMPTY, InProgressSequences.EMPTY, ConsensusMigrationState.EMPTY, - extensions); + extensions, + AccordStaleReplicas.EMPTY); } public static boolean isValidForClusterMetadata(Map epstates) diff --git a/src/java/org/apache/cassandra/tcm/serialization/Version.java b/src/java/org/apache/cassandra/tcm/serialization/Version.java index da99e726a0..245ef32bbc 100644 --- a/src/java/org/apache/cassandra/tcm/serialization/Version.java +++ b/src/java/org/apache/cassandra/tcm/serialization/Version.java @@ -37,6 +37,7 @@ public enum Version * - Added version to PlacementForRange serializer * - Serialize MemtableParams when serializing TableParams * - Added AccordFastPath + * - Added AccordStaleReplicas */ V2(2), /** diff --git a/src/java/org/apache/cassandra/tcm/transformations/AccordMarkRejoining.java b/src/java/org/apache/cassandra/tcm/transformations/AccordMarkRejoining.java new file mode 100644 index 0000000000..402d05cf50 --- /dev/null +++ b/src/java/org/apache/cassandra/tcm/transformations/AccordMarkRejoining.java @@ -0,0 +1,123 @@ +/* + * 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.transformations; + +import java.io.IOException; +import java.util.Objects; +import java.util.Set; +import java.util.stream.Collectors; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import accord.local.Node; +import org.apache.cassandra.io.util.DataInputPlus; +import org.apache.cassandra.io.util.DataOutputPlus; +import org.apache.cassandra.service.accord.AccordTopology; +import org.apache.cassandra.tcm.ClusterMetadata; +import org.apache.cassandra.tcm.Transformation; +import org.apache.cassandra.tcm.membership.NodeId; +import org.apache.cassandra.tcm.sequences.LockedRanges; +import org.apache.cassandra.tcm.serialization.AsymmetricMetadataSerializer; +import org.apache.cassandra.tcm.serialization.Version; +import org.apache.cassandra.utils.CollectionSerializers; + +import static org.apache.cassandra.exceptions.ExceptionCode.INVALID; + +public class AccordMarkRejoining implements Transformation +{ + private static final Logger logger = LoggerFactory.getLogger(AccordMarkRejoining.class); + + private final Set ids; + + public AccordMarkRejoining(Set ids) + { + this.ids = ids; + } + + @Override + public Kind kind() + { + return Kind.ACCORD_MARK_REJOINING; + } + + @Override + public Result execute(ClusterMetadata prev) + { + for (NodeId id : ids) + if (!prev.directory.peerIds().contains(id)) + return new Rejected(INVALID, String.format("Can not unmark node %s as it is not present in the directory.", id)); + + Set accordIds = ids.stream().map(AccordTopology::tcmIdToAccord).collect(Collectors.toSet()); + + for (Node.Id id : accordIds) + if (!prev.accordStaleReplicas.contains(id)) + return new Rejected(INVALID, String.format("Can not unmark node %s as it is not stale.", id)); + + logger.info("Unmarking " + ids + ". They will now participate in durability status coordination..."); + ClusterMetadata.Transformer next = prev.transformer().unmarkStaleReplicas(accordIds); + return Transformation.success(next, LockedRanges.AffectedRanges.EMPTY); + } + + @Override + public String toString() + { + return "AccordMarkRejoining{ids=" + ids + '}'; + } + + @Override + public boolean equals(Object o) + { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + AccordMarkRejoining that = (AccordMarkRejoining) o; + return Objects.equals(ids, that.ids); + } + + @Override + public int hashCode() + { + return Objects.hash(ids); + } + + public static final AsymmetricMetadataSerializer serializer = new AsymmetricMetadataSerializer<>() + { + @Override + public void serialize(Transformation t, DataOutputPlus out, Version version) throws IOException + { + assert t instanceof AccordMarkRejoining; + AccordMarkRejoining mark = (AccordMarkRejoining) t; + CollectionSerializers.serializeCollection(mark.ids, out, version, NodeId.serializer); + } + + @Override + public AccordMarkRejoining deserialize(DataInputPlus in, Version version) throws IOException + { + return new AccordMarkRejoining(CollectionSerializers.deserializeSet(in, version, NodeId.serializer)); + } + + @Override + public long serializedSize(Transformation t, Version version) + { + assert t instanceof AccordMarkRejoining; + AccordMarkRejoining mark = (AccordMarkRejoining) t; + return CollectionSerializers.serializedCollectionSize(mark.ids, version, NodeId.serializer); + } + }; +} diff --git a/src/java/org/apache/cassandra/tcm/transformations/AccordMarkStale.java b/src/java/org/apache/cassandra/tcm/transformations/AccordMarkStale.java new file mode 100644 index 0000000000..261e0a4eb2 --- /dev/null +++ b/src/java/org/apache/cassandra/tcm/transformations/AccordMarkStale.java @@ -0,0 +1,151 @@ +/* + * 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.transformations; + +import java.io.IOException; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Objects; +import java.util.Set; +import java.util.stream.Collectors; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import accord.local.Node; +import accord.topology.Shard; +import org.apache.cassandra.io.util.DataInputPlus; +import org.apache.cassandra.io.util.DataOutputPlus; +import org.apache.cassandra.schema.KeyspaceMetadata; +import org.apache.cassandra.schema.SchemaConstants; +import org.apache.cassandra.service.accord.AccordTopology; +import org.apache.cassandra.tcm.ClusterMetadata; +import org.apache.cassandra.tcm.Transformation; +import org.apache.cassandra.tcm.membership.NodeId; +import org.apache.cassandra.tcm.sequences.LockedRanges; +import org.apache.cassandra.tcm.serialization.AsymmetricMetadataSerializer; +import org.apache.cassandra.tcm.serialization.Version; +import org.apache.cassandra.utils.CollectionSerializers; + +import static org.apache.cassandra.exceptions.ExceptionCode.INVALID; + +public class AccordMarkStale implements Transformation +{ + private static final Logger logger = LoggerFactory.getLogger(AccordMarkStale.class); + + private final Set ids; + + public AccordMarkStale(Set ids) + { + this.ids = ids; + } + + @Override + public Kind kind() + { + return Kind.ACCORD_MARK_STALE; + } + + @Override + public Result execute(ClusterMetadata prev) + { + for (NodeId id : ids) + if (!prev.directory.peerIds().contains(id)) + return new Rejected(INVALID, String.format("Can not mark node %s stale as it is not present in the directory.", id)); + + Set accordIds = ids.stream().map(AccordTopology::tcmIdToAccord).collect(Collectors.toSet()); + + for (Node.Id id : accordIds) + if (prev.accordStaleReplicas.contains(id)) + return new Rejected(INVALID, String.format("Can not mark node %s stale as it already is.", id)); + + for (KeyspaceMetadata keyspace : prev.schema.getKeyspaces().without(SchemaConstants.REPLICATED_SYSTEM_KEYSPACE_NAMES)) + { + List shards = AccordTopology.KeyspaceShard.forKeyspace(keyspace, prev.placements, prev.directory); + + for (AccordTopology.KeyspaceShard shard : shards) + { + // We're trying to mark a node in this shard stale... + if (!Collections.disjoint(shard.nodes(), accordIds)) + { + int quorumSize = Shard.slowPathQuorumSize(shard.nodes().size()); + Set nonStaleNodes = new HashSet<>(shard.nodes()); + nonStaleNodes.removeAll(accordIds); + nonStaleNodes.removeAll(prev.accordStaleReplicas.ids()); + + // ...but reject the transformation if this would bring us below quorum. + if (nonStaleNodes.size() < quorumSize) + return new Rejected(INVALID, String.format("Can not mark nodes %s stale as that would leave fewer than a quorum of nodes active for range %s in keyspace '%s'.", + accordIds, shard.range(), keyspace.name)); + } + } + } + + logger.info("Marking " + ids + " stale. They will no longer participate in durability status coordination..."); + ClusterMetadata.Transformer next = prev.transformer().markStaleReplicas(accordIds); + return Transformation.success(next, LockedRanges.AffectedRanges.EMPTY); + } + + @Override + public String toString() + { + return "AccordMarkStale{ids=" + ids + '}'; + } + + @Override + public boolean equals(Object o) + { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + AccordMarkStale that = (AccordMarkStale) o; + return Objects.equals(ids, that.ids); + } + + @Override + public int hashCode() + { + return Objects.hash(ids); + } + + public static final AsymmetricMetadataSerializer serializer = new AsymmetricMetadataSerializer<>() + { + @Override + public void serialize(Transformation t, DataOutputPlus out, Version version) throws IOException + { + assert t instanceof AccordMarkStale; + AccordMarkStale mark = (AccordMarkStale) t; + CollectionSerializers.serializeCollection(mark.ids, out, version, NodeId.serializer); + } + + @Override + public AccordMarkStale deserialize(DataInputPlus in, Version version) throws IOException + { + return new AccordMarkStale(CollectionSerializers.deserializeSet(in, version, NodeId.serializer)); + } + + @Override + public long serializedSize(Transformation t, Version version) + { + assert t instanceof AccordMarkStale; + AccordMarkStale mark = (AccordMarkStale) t; + return CollectionSerializers.serializedCollectionSize(mark.ids, version, NodeId.serializer); + } + }; +} diff --git a/src/java/org/apache/cassandra/tools/NodeProbe.java b/src/java/org/apache/cassandra/tools/NodeProbe.java index 9c8fd50391..e733667b1e 100644 --- a/src/java/org/apache/cassandra/tools/NodeProbe.java +++ b/src/java/org/apache/cassandra/tools/NodeProbe.java @@ -120,6 +120,8 @@ import org.apache.cassandra.service.GCInspectorMXBean; import org.apache.cassandra.service.StorageProxy; import org.apache.cassandra.service.StorageProxyMBean; import org.apache.cassandra.service.StorageServiceMBean; +import org.apache.cassandra.service.accord.AccordOperations; +import org.apache.cassandra.service.accord.AccordOperationsMBean; import org.apache.cassandra.streaming.StreamManagerMBean; import org.apache.cassandra.streaming.StreamState; import org.apache.cassandra.streaming.management.StreamStateCompositeData; @@ -154,6 +156,7 @@ public class NodeProbe implements AutoCloseable protected StorageServiceMBean ssProxy; protected SnapshotManagerMBean snapshotProxy; protected CMSOperationsMBean cmsProxy; + protected AccordOperationsMBean accordProxy; protected GossiperMBean gossProxy; protected MemoryMXBean memProxy; protected GCInspectorMXBean gcProxy; @@ -270,6 +273,8 @@ public class NodeProbe implements AutoCloseable snapshotProxy = JMX.newMBeanProxy(mbeanServerConn, name, SnapshotManagerMBean.class); name = new ObjectName(CMSOperations.MBEAN_OBJECT_NAME); cmsProxy = JMX.newMBeanProxy(mbeanServerConn, name, CMSOperationsMBean.class); + name = new ObjectName(AccordOperations.MBEAN_OBJECT_NAME); + accordProxy = JMX.newMBeanProxy(mbeanServerConn, name, AccordOperationsMBean.class); name = new ObjectName(MessagingService.MBEAN_NAME); msProxy = JMX.newMBeanProxy(mbeanServerConn, name, MessagingServiceMBean.class); name = new ObjectName(StreamManagerMBean.OBJECT_NAME); @@ -1296,6 +1301,11 @@ public class NodeProbe implements AutoCloseable return cmsProxy; } + public AccordOperationsMBean getAccordOperationsProxy() + { + return accordProxy; + } + public GossiperMBean getGossProxy() { return gossProxy; diff --git a/src/java/org/apache/cassandra/tools/NodeTool.java b/src/java/org/apache/cassandra/tools/NodeTool.java index 00f2e66dfa..d821735586 100644 --- a/src/java/org/apache/cassandra/tools/NodeTool.java +++ b/src/java/org/apache/cassandra/tools/NodeTool.java @@ -278,6 +278,13 @@ public class NodeTool .withCommands(ConsensusMigrationAdmin.ListCmd.class) .withCommands(ConsensusMigrationAdmin.FinishMigration.class); + builder.withGroup("accord") + .withDescription("Manage the operation of Accord") + .withDefaultCommand(AccordAdmin.Describe.class) + .withCommand(AccordAdmin.Describe.class) + .withCommand(AccordAdmin.MarkStale.class) + .withCommand(AccordAdmin.MarkRejoining.class); + Cli parser = builder.build(); int status = 0; diff --git a/src/java/org/apache/cassandra/tools/nodetool/AccordAdmin.java b/src/java/org/apache/cassandra/tools/nodetool/AccordAdmin.java new file mode 100644 index 0000000000..ff7e88ca9f --- /dev/null +++ b/src/java/org/apache/cassandra/tools/nodetool/AccordAdmin.java @@ -0,0 +1,69 @@ +/* + * 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.tools.nodetool; + +import java.util.List; +import java.util.Map; + +import io.airlift.airline.Arguments; +import io.airlift.airline.Command; +import org.apache.cassandra.tools.NodeProbe; +import org.apache.cassandra.tools.NodeTool; + +public abstract class AccordAdmin extends NodeTool.NodeToolCmd +{ + @Command(name = "describe", description = "Describe current cluster metadata relating to Accord") + public static class Describe extends NodeTool.NodeToolCmd + { + @Override + protected void execute(NodeProbe probe) + { + Map info = probe.getAccordOperationsProxy().describe(); + output.out.printf("Accord Service:%n"); + output.out.printf("Epoch: %s%n", info.get("EPOCH")); + output.out.printf("Stale Replicas: %s%n", info.get("STALE_REPLICAS")); + } + } + + @Command(name = "mark_stale", description = "Mark a replica as being stale and no longer able to participate in durability status coordination") + public static class MarkStale extends AccordAdmin + { + @Arguments(required = true, description = "One or more node IDs to mark stale", usage = "+") + public List nodeIds; + + @Override + protected void execute(NodeProbe probe) + { + probe.getAccordOperationsProxy().accordMarkStale(nodeIds); + } + } + + @Command(name = "mark_rejoining", description = "Mark a stale replica as being allowed to participate in durability status coordination again") + public static class MarkRejoining extends AccordAdmin + { + @Arguments(required = true, description = "One or more node IDs to mark no longer stale", usage = "+") + public List nodeIds; + + @Override + protected void execute(NodeProbe probe) + { + probe.getAccordOperationsProxy().accordMarkRejoining(nodeIds); + } + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/mock/nodetool/InternalNodeProbe.java b/test/distributed/org/apache/cassandra/distributed/mock/nodetool/InternalNodeProbe.java index bf6a0879ce..1ab4f843b7 100644 --- a/test/distributed/org/apache/cassandra/distributed/mock/nodetool/InternalNodeProbe.java +++ b/test/distributed/org/apache/cassandra/distributed/mock/nodetool/InternalNodeProbe.java @@ -46,6 +46,7 @@ import org.apache.cassandra.service.CacheServiceMBean; import org.apache.cassandra.service.GCInspector; import org.apache.cassandra.service.StorageProxy; import org.apache.cassandra.service.StorageService; +import org.apache.cassandra.service.accord.AccordOperations; import org.apache.cassandra.service.snapshot.SnapshotManager; import org.apache.cassandra.streaming.StreamManager; import org.apache.cassandra.tcm.CMSOperations; @@ -74,6 +75,7 @@ public class InternalNodeProbe extends NodeProbe ssProxy = StorageService.instance; snapshotProxy = SnapshotManager.instance; cmsProxy = CMSOperations.instance; + accordProxy = AccordOperations.instance; msProxy = MessagingService.instance(); streamProxy = StreamManager.instance; compactionProxy = CompactionManager.instance; diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordNodetoolTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordNodetoolTest.java new file mode 100644 index 0000000000..9af94b783d --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordNodetoolTest.java @@ -0,0 +1,99 @@ +/* + * 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.accord; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; + +import com.google.common.collect.ImmutableSet; +import org.junit.Test; + +import accord.local.Node; +import org.apache.cassandra.tcm.ClusterMetadata; +import org.apache.cassandra.distributed.Cluster; +import org.apache.cassandra.distributed.api.Feature; +import org.apache.cassandra.distributed.test.TestBaseImpl; + + +import static org.junit.Assert.assertEquals; +import static org.apache.cassandra.distributed.shared.ClusterUtils.getNodeId; + +public class AccordNodetoolTest extends TestBaseImpl +{ + @Test + public void testMarkSingleNode() throws Throwable + { + try (Cluster cluster = init(builder().withNodes(3).withConfig((config) -> config.with(Feature.NETWORK, Feature.GOSSIP)).start())) + { + cluster.get(1).nodetoolResult("accord", "mark_stale", "1").asserts().success(); + cluster.get(1).runOnInstance(() -> assertEquals(ImmutableSet.of(new Node.Id(1)), ClusterMetadata.current().accordStaleReplicas.ids())); + cluster.get(1).nodetoolResult("accord", "describe").asserts().stdoutContains("Stale Replicas: 1"); + + // Reject the operation if the target node is already stale: + cluster.get(1).nodetoolResult("accord", "mark_stale", "1").asserts().failure().errorContains("it already is"); + + // Reject the operation if marking the node stale brings us below a quorum of non-stale nodes: + cluster.get(1).nodetoolResult("accord", "mark_stale", "2").asserts().failure().errorContains("that would leave fewer than a quorum"); + + // Reject the operation if the target node doesn't exist: + cluster.get(1).nodetoolResult("accord", "mark_stale", "4").asserts().failure().errorContains("not present in the directory"); + + cluster.get(1).nodetoolResult("accord", "mark_rejoining", "1").asserts().success(); + cluster.get(1).runOnInstance(() -> assertEquals(Collections.emptySet(), ClusterMetadata.current().accordStaleReplicas.ids())); + + cluster.get(1).nodetoolResult("accord", "mark_rejoining", "1").asserts().failure().errorContains("it is not stale"); + cluster.get(1).nodetoolResult("accord", "mark_rejoining", "4").asserts().failure().errorContains("not present in the directory"); + } + } + + @Test + public void testMarkMultipleNodes() throws Throwable + { + try (Cluster cluster = init(builder().withNodes(5).withConfig((config) -> config.with(Feature.NETWORK, Feature.GOSSIP)).start())) + { + // Reject the operation if marking the node stale brings us below a quorum of non-stale nodes: + cluster.get(1).nodetoolResult("accord", "mark_stale", "1", "2", "3").asserts().failure().errorContains("that would leave fewer than a quorum"); + + cluster.get(1).nodetoolResult("accord", "mark_stale", "1", "2").asserts().success(); + cluster.get(1).runOnInstance(() -> assertEquals(ImmutableSet.of(new Node.Id(1), new Node.Id(2)), ClusterMetadata.current().accordStaleReplicas.ids())); + cluster.get(1).nodetoolResult("accord", "describe").asserts().stdoutContains("Stale Replicas: 1,2"); + + // Reject the operation if a target node is already stale: + cluster.get(1).nodetoolResult("accord", "mark_stale", "1", "2").asserts().failure().errorContains("it already is"); + + // Reject the operation if a target node doesn't exist: + cluster.get(1).nodetoolResult("accord", "mark_stale", "4", "6").asserts().failure().errorContains("not present in the directory"); + + Map nodeIdToNode = new HashMap<>(); + for (int i = 1; i <= 5; i++) + nodeIdToNode.put(getNodeId(cluster.get(i)).id(), i); + + // Remove the second stale node, and ensure the set of stale replicas is updated: + cluster.get(nodeIdToNode.get(2)).shutdown().get(); + cluster.get(1).nodetoolResult("removenode", "2", "--force").asserts().success(); + cluster.get(1).nodetoolResult("cms", "unregister", "2").asserts().success(); + cluster.get(1).runOnInstance(() -> assertEquals(ImmutableSet.of(new Node.Id(1)), ClusterMetadata.current().accordStaleReplicas.ids())); + + cluster.get(1).nodetoolResult("accord", "mark_rejoining", "1", "3").asserts().failure().errorContains("it is not stale"); + cluster.get(1).nodetoolResult("accord", "mark_rejoining", "1", "6").asserts().failure().errorContains("not present in the directory"); + cluster.get(1).runOnInstance(() -> assertEquals(ImmutableSet.of(new Node.Id(1)), ClusterMetadata.current().accordStaleReplicas.ids())); + } + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/test/log/ClusterMetadataTestHelper.java b/test/distributed/org/apache/cassandra/distributed/test/log/ClusterMetadataTestHelper.java index 1481e2a3be..69d4001bd1 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/log/ClusterMetadataTestHelper.java +++ b/test/distributed/org/apache/cassandra/distributed/test/log/ClusterMetadataTestHelper.java @@ -58,6 +58,7 @@ import org.apache.cassandra.schema.Schema; import org.apache.cassandra.schema.SchemaConstants; import org.apache.cassandra.schema.SchemaTransformation; import org.apache.cassandra.service.ClientState; +import org.apache.cassandra.service.accord.AccordStaleReplicas; import org.apache.cassandra.tcm.AtomicLongBackedProcessor; import org.apache.cassandra.tcm.ClusterMetadata; import org.apache.cassandra.tcm.ClusterMetadataService; @@ -158,7 +159,8 @@ public class ClusterMetadataTestHelper LockedRanges.EMPTY, InProgressSequences.EMPTY, null, - ImmutableMap.of()); + ImmutableMap.of(), + AccordStaleReplicas.EMPTY); } public static ClusterMetadata minimalForTesting(IPartitioner partitioner) @@ -173,7 +175,8 @@ public class ClusterMetadataTestHelper null, null, null, - ImmutableMap.of()); + ImmutableMap.of(), + AccordStaleReplicas.EMPTY); } public static ClusterMetadata minimalForTesting(Keyspaces keyspaces) @@ -188,7 +191,8 @@ public class ClusterMetadataTestHelper null, null, null, - ImmutableMap.of()); + ImmutableMap.of(), + AccordStaleReplicas.EMPTY); } public static ClusterMetadataService syncInstanceForTest() diff --git a/test/unit/org/apache/cassandra/locator/MetaStrategyTest.java b/test/unit/org/apache/cassandra/locator/MetaStrategyTest.java index 2b22b0be3b..35ab27d426 100644 --- a/test/unit/org/apache/cassandra/locator/MetaStrategyTest.java +++ b/test/unit/org/apache/cassandra/locator/MetaStrategyTest.java @@ -26,14 +26,15 @@ import java.util.Map; import java.util.Set; import com.google.common.collect.ImmutableMap; -import org.apache.cassandra.service.accord.AccordFastPath; -import org.apache.cassandra.service.consensus.migration.ConsensusMigrationState; import org.junit.Assert; import org.junit.Test; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.dht.Murmur3Partitioner; import org.apache.cassandra.schema.DistributedSchema; +import org.apache.cassandra.service.accord.AccordFastPath; +import org.apache.cassandra.service.accord.AccordStaleReplicas; +import org.apache.cassandra.service.consensus.migration.ConsensusMigrationState; import org.apache.cassandra.tcm.ClusterMetadata; import org.apache.cassandra.tcm.Epoch; import org.apache.cassandra.tcm.membership.Directory; @@ -93,7 +94,8 @@ public class MetaStrategyTest LockedRanges.EMPTY, InProgressSequences.EMPTY, ConsensusMigrationState.EMPTY, - ImmutableMap.of()); + ImmutableMap.of(), + AccordStaleReplicas.EMPTY); } @Test @@ -159,4 +161,4 @@ public class MetaStrategyTest { return new Location(dc, rack); } -} \ No newline at end of file +} diff --git a/test/unit/org/apache/cassandra/service/accord/AccordFastPathCoordinatorTest.java b/test/unit/org/apache/cassandra/service/accord/AccordFastPathCoordinatorTest.java index 143f63a26c..4bec25b4d8 100644 --- a/test/unit/org/apache/cassandra/service/accord/AccordFastPathCoordinatorTest.java +++ b/test/unit/org/apache/cassandra/service/accord/AccordFastPathCoordinatorTest.java @@ -18,6 +18,15 @@ package org.apache.cassandra.service.accord; +import java.util.ArrayList; +import java.util.List; +import java.util.Objects; +import java.util.concurrent.TimeUnit; + +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.Test; + import accord.local.Node; import accord.topology.Shard; import accord.topology.Topology; @@ -28,14 +37,6 @@ import org.apache.cassandra.dht.Murmur3Partitioner; import org.apache.cassandra.schema.*; import org.apache.cassandra.service.accord.AccordFastPath.Status; import org.apache.cassandra.tcm.ClusterMetadata; -import org.junit.Assert; -import org.junit.BeforeClass; -import org.junit.Test; - -import java.util.ArrayList; -import java.util.List; -import java.util.Objects; -import java.util.concurrent.TimeUnit; import static org.apache.cassandra.service.accord.AccordTestUtils.*; diff --git a/test/unit/org/apache/cassandra/service/accord/AccordStaleReplicasTest.java b/test/unit/org/apache/cassandra/service/accord/AccordStaleReplicasTest.java new file mode 100644 index 0000000000..83eced9b66 --- /dev/null +++ b/test/unit/org/apache/cassandra/service/accord/AccordStaleReplicasTest.java @@ -0,0 +1,54 @@ +/* + * 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.service.accord; + +import java.util.HashSet; +import java.util.Set; + +import org.junit.Test; + +import accord.local.Node; +import accord.utils.AccordGens; +import accord.utils.Gen; +import accord.utils.Gens; +import org.apache.cassandra.io.util.DataOutputBuffer; +import org.apache.cassandra.tcm.Epoch; +import org.apache.cassandra.tcm.serialization.AsymmetricMetadataSerializers; +import org.apache.cassandra.tcm.serialization.Version; + +import static accord.utils.Property.qt; + +public class AccordStaleReplicasTest +{ + @Test + public void serde() + { + try (DataOutputBuffer buffer = new DataOutputBuffer()) + { + Gen> nodesGen = Gens.lists(AccordGens.nodes()).unique().ofSizeBetween(0, 9).map(nodes -> new HashSet<>(nodes)); + Gen epochGen = AccordGens.epochs().map(Epoch::create); + + qt().check(rs -> { + Epoch epoch = epochGen.next(rs); + Set nodes = nodesGen.next(rs); + AsymmetricMetadataSerializers.testSerde(buffer, AccordStaleReplicas.serializer, new AccordStaleReplicas(nodes, epoch), Version.V2); + }); + } + } +} diff --git a/test/unit/org/apache/cassandra/tcm/ClusterMetadataTransformationTest.java b/test/unit/org/apache/cassandra/tcm/ClusterMetadataTransformationTest.java index 68f02de123..1219502e98 100644 --- a/test/unit/org/apache/cassandra/tcm/ClusterMetadataTransformationTest.java +++ b/test/unit/org/apache/cassandra/tcm/ClusterMetadataTransformationTest.java @@ -301,6 +301,8 @@ public class ClusterMetadataTransformationTest return metadata.accordFastPath; else if (key == CONSENSUS_MIGRATION_STATE) return metadata.consensusMigrationState; + else if (key == ACCORD_STALE_REPLICAS) + return metadata.accordStaleReplicas; throw new IllegalArgumentException("Unknown metadata key " + key); } diff --git a/test/unit/org/apache/cassandra/tcm/serialization/AsymmetricMetadataSerializers.java b/test/unit/org/apache/cassandra/tcm/serialization/AsymmetricMetadataSerializers.java new file mode 100644 index 0000000000..bd3cd4547e --- /dev/null +++ b/test/unit/org/apache/cassandra/tcm/serialization/AsymmetricMetadataSerializers.java @@ -0,0 +1,42 @@ +/* + * 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.serialization; + +import java.io.IOException; + +import org.assertj.core.api.Assertions; + +import accord.utils.LazyToString; +import accord.utils.ReflectionUtils; +import org.apache.cassandra.io.util.DataInputBuffer; +import org.apache.cassandra.io.util.DataOutputBuffer; + +public class AsymmetricMetadataSerializers +{ + public static void testSerde(DataOutputBuffer output, AsymmetricMetadataSerializer serializer, In input, Version version) throws IOException + { + output.clear(); + long expectedSize = serializer.serializedSize(input, version); + serializer.serialize(input, output, version); + Assertions.assertThat(output.getLength()).describedAs("The serialized size and bytes written do not match").isEqualTo(expectedSize); + DataInputBuffer in = new DataInputBuffer(output.unsafeGetBufferAndFlip(), false); + Out read = serializer.deserialize(in, version); + Assertions.assertThat(read).describedAs("The deserialized output does not match the serialized input; difference %s", new LazyToString(() -> ReflectionUtils.recursiveEquals(read, input).toString())).isEqualTo(input); + } +} diff --git a/test/unit/org/apache/cassandra/tcm/transformations/AccordMarkRejoiningTest.java b/test/unit/org/apache/cassandra/tcm/transformations/AccordMarkRejoiningTest.java new file mode 100644 index 0000000000..032c8dd1ec --- /dev/null +++ b/test/unit/org/apache/cassandra/tcm/transformations/AccordMarkRejoiningTest.java @@ -0,0 +1,56 @@ +/* + * 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.transformations; + +import java.io.IOException; +import java.util.Collections; + +import com.google.common.collect.ImmutableSet; +import org.junit.Test; + +import org.apache.cassandra.io.util.DataOutputBuffer; +import org.apache.cassandra.tcm.membership.NodeId; +import org.apache.cassandra.tcm.serialization.AsymmetricMetadataSerializers; +import org.apache.cassandra.tcm.serialization.Version; + +public class AccordMarkRejoiningTest +{ + @Test + public void shouldSerializeEmpty() throws IOException + { + DataOutputBuffer buffer = new DataOutputBuffer(); + AsymmetricMetadataSerializers.testSerde(buffer, AccordMarkRejoining.serializer, new AccordMarkRejoining(Collections.emptySet()), Version.V2); + } + + @Test + public void shouldSerializeSingleton() throws IOException + { + DataOutputBuffer buffer = new DataOutputBuffer(); + AccordMarkRejoining markStale = new AccordMarkRejoining(Collections.singleton(NodeId.fromString("1"))); + AsymmetricMetadataSerializers.testSerde(buffer, AccordMarkRejoining.serializer, markStale, Version.V2); + } + + @Test + public void shouldSerializeMulti() throws IOException + { + DataOutputBuffer buffer = new DataOutputBuffer(); + AccordMarkRejoining markStale = new AccordMarkRejoining(ImmutableSet.of(NodeId.fromString("1"), NodeId.fromString("2"))); + AsymmetricMetadataSerializers.testSerde(buffer, AccordMarkRejoining.serializer, markStale, Version.V2); + } +} diff --git a/test/unit/org/apache/cassandra/tcm/transformations/AccordMarkStaleTest.java b/test/unit/org/apache/cassandra/tcm/transformations/AccordMarkStaleTest.java new file mode 100644 index 0000000000..d794b3a2a9 --- /dev/null +++ b/test/unit/org/apache/cassandra/tcm/transformations/AccordMarkStaleTest.java @@ -0,0 +1,56 @@ +/* + * 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.transformations; + +import java.io.IOException; +import java.util.Collections; + +import com.google.common.collect.ImmutableSet; +import org.junit.Test; + +import org.apache.cassandra.io.util.DataOutputBuffer; +import org.apache.cassandra.tcm.membership.NodeId; +import org.apache.cassandra.tcm.serialization.AsymmetricMetadataSerializers; +import org.apache.cassandra.tcm.serialization.Version; + +public class AccordMarkStaleTest +{ + @Test + public void shouldSerializeEmpty() throws IOException + { + DataOutputBuffer buffer = new DataOutputBuffer(); + AsymmetricMetadataSerializers.testSerde(buffer, AccordMarkStale.serializer, new AccordMarkStale(Collections.emptySet()), Version.V2); + } + + @Test + public void shouldSerializeSingleton() throws IOException + { + DataOutputBuffer buffer = new DataOutputBuffer(); + AccordMarkStale markStale = new AccordMarkStale(Collections.singleton(NodeId.fromString("1"))); + AsymmetricMetadataSerializers.testSerde(buffer, AccordMarkStale.serializer, markStale, Version.V2); + } + + @Test + public void shouldSerializeMulti() throws IOException + { + DataOutputBuffer buffer = new DataOutputBuffer(); + AccordMarkStale markStale = new AccordMarkStale(ImmutableSet.of(NodeId.fromString("1"), NodeId.fromString("2"))); + AsymmetricMetadataSerializers.testSerde(buffer, AccordMarkStale.serializer, markStale, Version.V2); + } +}