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

Offline tool for working with cluster metadata dump files 

The following subcommands have been added:

* abortbootstrap        Cancel and undo an inflight bootstrap
* abortmove             Cancel and undo an inflight token move
* abortdecommission     Cancel and undo an inflight decommission
* assassinate           Remove a node from cluster metadata
* resetcms              Redefine the CMS membership
* move                  Modify a node's (single) owned token
* describe              Output a summary of the cluster metadata
* forcejoin             Force a registered or joining node into a JOINED state
* print                 Output a full rendering of cluster metadata
* printdirectory        Output a full rendering of the directory
* printdataplacements   Output the full data placements

Patch by Venkata Harikrishna Nukala; reviewed by Alex Petrov and Sam Tunnicliffe
for CASSANDRA-19151
This commit is contained in:
nvharikrishna 2025-05-08 13:26:32 +05:30 committed by Sam Tunnicliffe
parent 4930ae0e6e
commit c9d60a61e3
9 changed files with 2968 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
6.0-alpha2
* Add an offline cluster metadata tool (CASSANDRA-19151)
* Accord: Tail Latency Improvements (CASSANDRA-21361)
* Artificial Latency Injection (CASSANDRA-17024)
* Accord: Clean Shutdown/Restart, Rebootstrap, et al (CASSANDRA-21355)

825
src/java/org/apache/cassandra/tools/CMSOfflineTool.java Normal file
View File

@ -0,0 +1,825 @@
/*
* 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;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Collection;
import java.util.EnumSet;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.io.util.File;
import org.apache.cassandra.io.util.FileInputStreamPlus;
import org.apache.cassandra.io.util.FileOutputStreamPlus;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.locator.MetaStrategy;
import org.apache.cassandra.locator.Replica;
import org.apache.cassandra.schema.KeyspaceMetadata;
import org.apache.cassandra.schema.ReplicationParams;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.ClusterMetadataService;
import org.apache.cassandra.tcm.MultiStepOperation;
import org.apache.cassandra.tcm.membership.Directory;
import org.apache.cassandra.tcm.membership.Location;
import org.apache.cassandra.tcm.membership.NodeAddresses;
import org.apache.cassandra.tcm.membership.NodeId;
import org.apache.cassandra.tcm.membership.NodeState;
import org.apache.cassandra.tcm.membership.NodeVersion;
import org.apache.cassandra.tcm.ownership.DataPlacement;
import org.apache.cassandra.tcm.ownership.ReplicaGroups;
import org.apache.cassandra.tcm.ownership.UniformRangePlacement;
import org.apache.cassandra.tcm.sequences.BootstrapAndJoin;
import org.apache.cassandra.tcm.sequences.Move;
import org.apache.cassandra.tcm.sequences.ReconfigureCMS;
import org.apache.cassandra.tcm.serialization.VerboseMetadataSerializer;
import org.apache.cassandra.tcm.serialization.Version;
import org.apache.cassandra.tcm.transformations.Assassinate;
import org.apache.cassandra.tcm.transformations.CancelInProgressSequence;
import org.apache.cassandra.tcm.transformations.PrepareMove;
import org.apache.cassandra.tcm.transformations.Unregister;
import org.apache.cassandra.tcm.transformations.UnsafeJoin;
import org.apache.cassandra.tcm.transformations.cms.PrepareCMSReconfiguration;
import org.apache.cassandra.utils.FBUtilities;
import picocli.CommandLine;
import picocli.CommandLine.Command;
import picocli.CommandLine.Option;
import static com.google.common.base.Throwables.getStackTraceAsString;
import static org.apache.cassandra.tcm.transformations.cms.PrepareCMSReconfiguration.needsReconfiguration;
/**
* Offline tool to print or update cluster metadata stored in a dump file.
* <p>
* The tool operates entirely offline: it reads a metadata dump file produced by
* {@code nodetool cms dump} (or equivalent), applies the requested transformation,
* and writes the result to a new file. The original dump file is never modified.
* <p>
* Run without a subcommand to print usage information.
*/
@SuppressWarnings({ "unused", "DefaultAnnotationParam", "UseOfSystemOutOrSystemErr" })
@Command(name = "cmsofflinetool",
mixinStandardHelpOptions = true,
description = "Offline tool to print or update cluster metadata dump.",
subcommands = { CMSOfflineTool.AbortBootstrap.class,
CMSOfflineTool.AbortDecommission.class,
CMSOfflineTool.AbortMove.class,
CMSOfflineTool.AssassinateNode.class,
CMSOfflineTool.Describe.class,
CMSOfflineTool.ForceJoin.class,
CommandLine.HelpCommand.class,
CMSOfflineTool.MoveToken.class,
CMSOfflineTool.Print.class,
CMSOfflineTool.PrintDataPlacements.class,
CMSOfflineTool.PrintDirectoryCmd.class,
CMSOfflineTool.ResetCMS.class })
public class CMSOfflineTool implements Runnable
{
private final Output output;
public CMSOfflineTool(Output output)
{
this.output = output;
}
public static void main(String[] args) throws IOException
{
CMSOfflineTool tool = new CMSOfflineTool(new Output(System.out, System.err));
CommandLine cli = new CommandLine(tool)
.setColorScheme(CommandLine.Help.defaultColorScheme(CommandLine.Help.Ansi.OFF))
.setExecutionExceptionHandler((ex, cmd, parseResult) -> {
cmd.getErr().println("Error: " + ex.getMessage());
cmd.getErr().println("-- StackTrace --");
cmd.getErr().println(getStackTraceAsString(ex));
return 2;
});
int status = cli.execute(args);
System.exit(status);
}
@Override
public void run()
{
CommandLine.usage(this, output.out, CommandLine.Help.Ansi.OFF);
}
public static abstract class ClusterMetadataToolCmd implements Runnable
{
@Option(names = { "-f", "--file" }, description = "Cluster metadata dump file path.", required = true)
protected String metadataDumpFile;
@Option(names = { "-sv", "--serialization-version" }, description = "Serialization version to use.")
private Version serializationVersion;
@CommandLine.ParentCommand
private CMSOfflineTool parent;
@Override
public void run()
{
try
{
execute(parent.output);
parent.output.out.flush();
parent.output.err.flush();
}
catch (IOException e)
{
throw new RuntimeException(e);
}
}
protected abstract void execute(Output output) throws IOException;
public ClusterMetadata parseClusterMetadata() throws IOException
{
File file = new File(metadataDumpFile);
if (!file.exists())
{
throw new IllegalArgumentException("Cluster metadata dump file " + metadataDumpFile + " does not exist.");
}
// Make sure the partitioner we use to manipulate the metadata is the same one used to generate it
IPartitioner partitioner;
try (FileInputStreamPlus fisp = new FileInputStreamPlus(metadataDumpFile))
{
int x = fisp.readUnsignedVInt32();
Version version = Version.fromInt(x);
partitioner = ClusterMetadata.Serializer.getPartitioner(fisp, version);
}
DatabaseDescriptor.toolInitialization();
DatabaseDescriptor.setPartitionerUnsafe(partitioner);
ClusterMetadataService.initializeForTools(false);
return ClusterMetadataService.deserializeClusterMetadata(metadataDumpFile);
}
public void writeMetadata(Output output, ClusterMetadata metadata, String outputFilePath) throws IOException
{
Version serializationVersion = getSerializationVersion(metadata);
Path p = outputFilePath != null
? Files.createFile(Path.of(outputFilePath))
: Files.createTempFile("clustermetadata", ".dump");
try (FileOutputStreamPlus out = new FileOutputStreamPlus(p))
{
VerboseMetadataSerializer.serialize(ClusterMetadata.serializer, metadata, out, serializationVersion);
output.out.println("Updated cluster metadata written to file " + p.toAbsolutePath());
}
}
Version getSerializationVersion(ClusterMetadata metadata)
{
// Step 1: Default to the serialization version from the cluster metadata
Version finalVersion = metadata.directory.commonSerializationVersion;
// Step 2: User-specified version takes precedence
if (serializationVersion != null)
{
// Warn if user-specified version is older than what the metadata was written with
if (serializationVersion.isBefore(finalVersion))
{
parent.output.err.printf("WARNING: Given serialization version %s is older than " +
"the version in cluster metadata (%s). Proceeding as requested.%n",
serializationVersion, finalVersion);
}
finalVersion = serializationVersion;
}
// Step 3: Current binary version must be able to handle the finalized version
Version currentVersion = NodeVersion.CURRENT.serializationVersion();
if (currentVersion.isBefore(finalVersion))
{
throw new IllegalArgumentException("Current version " + currentVersion +
" is older than the target serialization version " +
finalVersion + ". Cannot proceed further. " +
"Try modifying cluster metadata using binaries that support " +
"minimum serialization version: " + finalVersion + '.');
}
return finalVersion;
}
}
/**
* Base class for commands that cancel an in-progress sequence for a given node.
* Subclasses specify which sequence kinds they handle via {@link #supportedKinds()} and
* may apply additional transformations after cancellation via {@link #postCancel}.
*/
abstract static class AbstractAbortSequence extends ClusterMetadataToolCmd
{
@CommandLine.ArgGroup(exclusive = true, multiplicity = "1")
NodeIdentifierOption nodeIdentifierOption;
@Option(names = { "-o", "--output-file" },
description = "Output file path for storing the updated Cluster Metadata.")
String outputFilePath;
protected abstract EnumSet<MultiStepOperation.Kind> supportedKinds();
protected ClusterMetadata postCancel(ClusterMetadata metadata, NodeId nodeId)
{
return metadata;
}
@Override
protected void execute(Output output) throws IOException
{
ClusterMetadata metadata = parseClusterMetadata();
NodeId nodeId = nodeIdentifierOption.getNodeId(metadata);
MultiStepOperation<?> multiStepOperation = metadata.inProgressSequences.get(nodeId);
if (multiStepOperation == null)
{
throw new IllegalArgumentException("No transformation sequence is in progress for " +
nodeIdentifierOption.getNodeIpOrId() + '.');
}
if (!supportedKinds().contains(multiStepOperation.kind()))
{
throw new IllegalArgumentException("Sequence of kind " + multiStepOperation.kind() +
" is in progress for node " + nodeIdentifierOption.getNodeIpOrId() +
". Cannot proceed with this operation.");
}
CancelInProgressSequence cancelSequence = new CancelInProgressSequence(nodeId);
ClusterMetadata updatedMetadata = cancelSequence.execute(metadata).success().metadata;
updatedMetadata = postCancel(updatedMetadata, nodeId);
writeMetadata(output, updatedMetadata, outputFilePath);
}
}
/**
* Cancels a JOIN or REPLACE bootstrap sequence for the given node and unregisters it
* from the cluster. Use this when a node is stuck in bootstrapping or replacement.
* Fails if no in-progress sequence exists, or if the sequence is not of kind JOIN or REPLACE.
*/
@Command(name = "abortbootstrap",
description = "Aborts bootstrap for given node if in progress.")
public static class AbortBootstrap extends AbstractAbortSequence
{
@Override
protected EnumSet<MultiStepOperation.Kind> supportedKinds()
{
return EnumSet.of(MultiStepOperation.Kind.JOIN, MultiStepOperation.Kind.REPLACE);
}
@Override
protected ClusterMetadata postCancel(ClusterMetadata metadata, NodeId nodeId)
{
// Cancelling the sequence is not enough, we need to unregister as well
Unregister unregister = new Unregister(nodeId, EnumSet.of(NodeState.REGISTERED),
ClusterMetadataService.instance().placementProvider());
return unregister.execute(metadata).success().metadata;
}
}
/**
* Cancels an in-progress MOVE sequence for the given node, returning it to its
* pre-move token assignment. Fails if no in-progress sequence exists, or if the
* sequence is not of kind MOVE.
*/
@Command(name = "abortmove", description = "Aborts in progress move sequence for given node.")
static class AbortMove extends AbstractAbortSequence
{
@Override
protected EnumSet<MultiStepOperation.Kind> supportedKinds()
{
return EnumSet.of(MultiStepOperation.Kind.MOVE);
}
}
/**
* Cancels an in-progress LEAVE (decommission) sequence for the given node, keeping
* it as an active member of the ring. Fails if no in-progress sequence exists, or if
* the sequence is not of kind LEAVE.
*/
@Command(name = "abortdecommission", description = "Aborts in progress decommission sequence for given node.")
static class AbortDecommission extends AbstractAbortSequence
{
@Override
protected EnumSet<MultiStepOperation.Kind> supportedKinds()
{
return EnumSet.of(MultiStepOperation.Kind.LEAVE);
}
}
/**
* Removes a node from cluster metadata by applying the {@link Assassinate} transformation.
* If a MOVE or LEAVE sequence is in progress for the node, it is cancelled first.
* If the node is a CMS member, it is removed from CMS before assassination.
* Fails if the node is in a JOIN or REPLACE sequence; use {@code abortbootstrap} instead.
*/
@Command(name = "assassinate", description = "Assassinates given node from Cluster metadata.")
static class AssassinateNode extends ClusterMetadataToolCmd
{
private final EnumSet<MultiStepOperation.Kind> supportedCancelSequences =
EnumSet.of(MultiStepOperation.Kind.MOVE, MultiStepOperation.Kind.LEAVE);
@CommandLine.ArgGroup(exclusive = true, multiplicity = "1")
NodeIdentifierOption nodeIdentifierOption;
@Option(names = { "-o", "--output-file" },
description = "Output file path for storing the updated Cluster Metadata.")
private String outputFilePath;
@Override
protected void execute(Output output) throws IOException
{
ClusterMetadata metadata = parseClusterMetadata();
NodeId nodeId = nodeIdentifierOption.getNodeId(metadata);
// Check if there are any in-progress sequences for given node
// If any, then cancel the sequence and then assassinate it
if (metadata.inProgressSequences.contains(nodeId))
{
MultiStepOperation<?> multiStepOperation = metadata.inProgressSequences.get(nodeId);
MultiStepOperation.Kind sequenceKind = multiStepOperation.kind();
if (!supportedCancelSequences.contains(sequenceKind))
{
if (sequenceKind == MultiStepOperation.Kind.JOIN || sequenceKind == MultiStepOperation.Kind.REPLACE)
{
throw new IllegalArgumentException("Cannot assassinate the node when sequence of kind " +
sequenceKind + " is in progress. " +
"Run abortbootstrap instead.");
}
else
{
throw new IllegalArgumentException("Cannot assassinate the node when sequence of kind " +
sequenceKind + " is in progress.");
}
}
output.out.printf("Cancelling in-progress sequence of kind %s before assassinating node.\n",
metadata.inProgressSequences.get(nodeId).kind());
metadata = new CancelInProgressSequence(nodeId).execute(metadata).success().metadata;
}
metadata = maybeReconfigureCMS(metadata, nodeId);
Assassinate transformation = new Assassinate(nodeId, ClusterMetadataService.instance().placementProvider());
ClusterMetadata updatedMetadata = transformation.execute(metadata).success().metadata;
writeMetadata(output, updatedMetadata, outputFilePath);
}
ClusterMetadata maybeReconfigureCMS(ClusterMetadata metadata, NodeId nodeId)
{
InetAddressAndPort addressAndPort = metadata.directory.endpoint(nodeId);
if (!metadata.isCMSMember(addressAndPort))
{
return metadata;
}
// Ref: org.apache.cassandra.tcm.sequences.ReconfigureCMS.maybeReconfigureCMS
PrepareCMSReconfiguration.Simple transformation = new PrepareCMSReconfiguration.Simple(nodeId, Set.of());
ClusterMetadata updatedMetadata = transformation.execute(metadata).success().metadata;
updatedMetadata = updatedMetadata.inProgressSequences.get(ReconfigureCMS.SequenceKey.instance)
.applyTo(updatedMetadata).success().metadata;
if (updatedMetadata.isCMSMember(addressAndPort))
{
throw new IllegalStateException("Could not remove node " + nodeIdentifierOption.getNodeIpOrId() +
" from CMS.");
}
return updatedMetadata;
}
}
/**
* Replaces the entire CMS membership with a single node. All existing CMS replicas are
* removed and the specified node becomes the sole CMS member. Useful for recovering from
* a state where the CMS is unreachable.
*/
@Command(name = "resetcms",
description = "Replaces all CMS members with the specified node. WARNING: all existing CMS replicas are removed.")
static class ResetCMS extends ClusterMetadataToolCmd
{
@CommandLine.ArgGroup(exclusive = true, multiplicity = "1")
NodeIdentifierOption nodeIdentifierOption;
@Option(names = { "-o", "--output-file" },
description = "Output file path for storing the updated Cluster Metadata.")
private String outputFilePath;
@Override
protected void execute(Output output) throws IOException
{
ClusterMetadata metadata = parseClusterMetadata();
NodeId nodeId = nodeIdentifierOption.getNodeId(metadata);
metadata = resetCMS(metadata, nodeId);
writeMetadata(output, metadata, outputFilePath);
}
ClusterMetadata resetCMS(ClusterMetadata metadata, NodeId nodeId)
{
NodeState nodeState = metadata.directory.peerState(nodeId);
if (nodeState != NodeState.JOINED)
{
throw new IllegalArgumentException("Node " + nodeIdentifierOption.getNodeIpOrId() + " is in " +
nodeState + " state. Only a JOINED node can be set as CMS member.");
}
InetAddressAndPort endpoint = metadata.directory.getNodeAddresses(nodeId).broadcastAddress;
ReplicationParams metaParams = ReplicationParams.meta(metadata);
Iterable<Replica> currentReplicas = metadata.placements.get(metaParams).writes.byEndpoint().flattenValues();
DataPlacement.Builder placementBuilder = metadata.placements.get(metaParams).unbuild();
for (Replica replica : currentReplicas)
{
placementBuilder.withoutReadReplica(metadata.epoch, replica)
.withoutWriteReplica(metadata.epoch, replica);
}
Replica newCMS = MetaStrategy.replica(endpoint);
placementBuilder.withReadReplica(metadata.epoch, newCMS)
.withWriteReplica(metadata.epoch, newCMS);
return metadata.transformer()
.with(metadata.placements.unbuild()
.with(metaParams, placementBuilder.build())
.build())
.build().metadata;
}
}
/**
* Moves a node to a new token. Only supports single-token nodes.
* If a MOVE sequence is already in progress for the node, it is completed rather than started fresh;
* in that case the provided token must match the token of the in-progress sequence.
*/
@Command(name = "move",
description = "Moves node to given token. Works only for cluster having nodes with single token.")
static class MoveToken extends ClusterMetadataToolCmd
{
@CommandLine.ArgGroup(exclusive = true, multiplicity = "1")
NodeIdentifierOption nodeIdentifierOption;
@Option(names = { "-t", "--token" },
description = "Token to assign.")
private String token;
@Option(names = { "-o", "--output-file" },
description = "Output file path for storing the updated Cluster Metadata.")
private String outputFilePath;
@Override
protected void execute(Output output) throws IOException
{
// Took the reference from org.apache.cassandra.tcm.sequences.SingleNodeSequences.move
ClusterMetadata metadata = parseClusterMetadata();
NodeId nodeId = nodeIdentifierOption.getNodeId(metadata);
if (metadata.inProgressSequences.contains(nodeId))
{
ClusterMetadata updatedMetadata = finishInProgressSequence(nodeId, token, metadata);
writeMetadata(output, updatedMetadata, outputFilePath);
return;
}
if (null == token)
{
throw new IllegalArgumentException("Token required when no MOVE sequence is in progress.");
}
metadata.partitioner.getTokenFactory().validate(token);
Token toToken = metadata.partitioner.getTokenFactory().fromString(token);
if (metadata.tokenMap.tokens().contains(toToken))
{
NodeId tokenOwnerId = metadata.tokenMap.owner(toToken);
throw new IllegalArgumentException("Target token " + toToken + " is already owned by node " + tokenOwnerId.id());
}
if (metadata.tokenMap.tokens(nodeId).size() > 1)
{
throw new UnsupportedOperationException("This node has more than one token and cannot be moved thusly.");
}
PrepareMove prepareMove = new PrepareMove(nodeId, Set.of(toToken), new UniformRangePlacement(), false);
ClusterMetadata updatedMetadata = prepareMove.execute(metadata).success().metadata;
updatedMetadata = updatedMetadata.inProgressSequences.get(nodeId).applyTo(updatedMetadata).success().metadata;
writeMetadata(output, updatedMetadata, outputFilePath);
}
ClusterMetadata finishInProgressSequence(NodeId nodeId, String token, ClusterMetadata metadata)
{
MultiStepOperation<?> multiStepOperation = metadata.inProgressSequences.get(nodeId);
if (multiStepOperation.kind() != MultiStepOperation.Kind.MOVE)
{
throw new IllegalArgumentException("Another sequence of kind " + multiStepOperation.kind() +
" is in progress for node " + nodeIdentifierOption.getNodeIpOrId() +
". Cannot proceed with move.");
}
Move moveInProgress = (Move) multiStepOperation;
Collection<Token> inProgressTokens = moveInProgress.tokens;
Collection<Token> givenTokenSet;
if (token == null)
{
givenTokenSet = Set.of();
}
else
{
metadata.partitioner.getTokenFactory().validate(token);
givenTokenSet = Set.of(metadata.partitioner.getTokenFactory().fromString(token));
}
if (givenTokenSet.isEmpty() || new HashSet<>(moveInProgress.tokens).equals(givenTokenSet))
{
return moveInProgress.applyTo(metadata).success().metadata;
}
throw new IllegalArgumentException("Move in progress for another token(s) " + inProgressTokens + '.');
}
}
/**
* Identifies a target node using either its IP address or its integer/UUID node ID.
* Exactly one of {@code -ip} or {@code -id} must be provided.
*/
static class NodeIdentifierOption
{
@Option(names = { "-ip", "--ip-address" }, required = true,
description = "IP address of the target endpoint. Port can be optionally specified " +
"using a colon after the IP address (e.g., 127.0.0.1:9042).")
private String ip;
@Option(names = { "-id", "--node-id" }, required = true,
description = "Node ID. It can be integer ID assigned to node or the node uuid.")
private String id;
String getNodeIpOrId()
{
return ip != null ? ip : id;
}
NodeId getNodeId(ClusterMetadata metadata)
{
if (id != null)
{
NodeId nodeId = NodeId.fromString(id);
if (!metadata.directory.peerIds().contains(nodeId))
{
throw new IllegalArgumentException("No node present with id " + id +
" in the given cluster metadata.");
}
return nodeId;
}
else if (ip != null)
{
InetAddressAndPort nodeAddress = InetAddressAndPort.getByNameUnchecked(ip);
NodeId nodeId = metadata.directory.peerId(nodeAddress);
if (null == nodeId)
{
throw new IllegalArgumentException("No node present with ip address " + ip +
" in given cluster metadata.");
}
return nodeId;
}
throw new IllegalArgumentException("Neither node id nor ip address specified to fetch NodeId from metadata.");
}
}
/**
* Prints a high-level summary of the CMS state: members, epoch, replication factor,
* service state, and whether reconfiguration is needed.
*/
@Command(name = "describe", description = "Describes the cluster metadata.")
static class Describe extends ClusterMetadataToolCmd
{
@Override
protected void execute(Output output) throws IOException
{
ClusterMetadata metadata = parseClusterMetadata();
String members = metadata.fullCMSMembers()
.stream()
.sorted()
.map(Object::toString)
.collect(Collectors.joining(","));
output.out.printf("Cluster Metadata Service:%n");
output.out.printf("Members: %s%n", members);
output.out.printf("Needs reconfiguration: %s%n", needsReconfiguration(metadata));
output.out.printf("Service State: %s%n", ClusterMetadataService.state(metadata));
output.out.printf("Epoch: %s%n", metadata.epoch.getEpoch());
output.out.printf("Replication factor: %s%n", ReplicationParams.meta(metadata).toString());
}
}
/**
* Forces a node directly to JOINED state, bypassing the normal bootstrap sequence.
* <ul>
* <li>If the node has an in-progress JOIN sequence, it is completed; provided tokens
* must match the sequence tokens (or be omitted to use the sequence tokens).</li>
* <li>If no sequence is in progress, tokens must be explicitly provided and
* an {@link UnsafeJoin} is applied.</li>
* </ul>
* Fails if the node is already JOINED, or if a non-JOIN sequence is in progress.
*/
@Command(name = "forcejoin", description = "Forces a node to move to JOINED state.")
static class ForceJoin extends ClusterMetadataToolCmd
{
@SuppressWarnings("MismatchedQueryAndUpdateOfCollection")
@Option(names = { "-t", "--token" },
description = "Token to assign. Pass it multiple times to assign multiple tokens to node.")
private final List<String> tokens = new ArrayList<>();
@CommandLine.ArgGroup(exclusive = true, multiplicity = "1")
NodeIdentifierOption nodeIdentifierOption;
@Option(names = { "-o", "--output-file" },
description = "Output file path for storing the updated Cluster Metadata.")
private String outputFilePath;
@Override
protected void execute(Output output) throws IOException
{
ClusterMetadata metadata = parseClusterMetadata();
NodeId nodeId = nodeIdentifierOption.getNodeId(metadata);
Set<Token> tokenSet = new HashSet<>(tokens.size());
Token.TokenFactory tokenFactory = metadata.partitioner.getTokenFactory();
tokens.forEach(t -> tokenSet.add(tokenFactory.fromString(t)));
NodeState nodeState = metadata.directory.peerState(nodeId);
if (nodeState == NodeState.JOINED)
{
throw new IllegalArgumentException("Node " + nodeIdentifierOption.getNodeIpOrId() +
" is already in JOINED state.");
}
ClusterMetadata updatedMetadata;
if (metadata.inProgressSequences.get(nodeId) != null)
{
MultiStepOperation<?> multiStepOperation = metadata.inProgressSequences.get(nodeId);
if (multiStepOperation.kind() != MultiStepOperation.Kind.JOIN)
{
throw new IllegalArgumentException("Another sequence of kind " + multiStepOperation.kind() +
" is in progress for node " + nodeIdentifierOption.getNodeIpOrId() +
". Cannot proceed with force join.");
}
BootstrapAndJoin bootstrapAndJoin = (BootstrapAndJoin) multiStepOperation;
Set<Token> sequenceTokens = bootstrapAndJoin.finishJoin.tokens;
if (tokens.isEmpty()
|| (tokenSet.size() == sequenceTokens.size() && sequenceTokens.containsAll(tokenSet)))
{
updatedMetadata = bootstrapAndJoin.applyTo(metadata).success().metadata;
}
else
{
// If tokens are provided, then it should match with the in-progress sequence tokens
throw new IllegalArgumentException("The tokens provided " + tokens + " do not match with " +
" in progress BootstrapAndJoin sequence tokens " +
sequenceTokens + ". Cannot proceed further.");
}
}
else
{
// There are no in-progress sequences, force join by using UnsafeJoin transformation
if (tokenSet.isEmpty())
{
throw new IllegalArgumentException("Tokens must be provided to force join a node.");
}
UnsafeJoin unsafeJoin = new UnsafeJoin(nodeId, tokenSet, new UniformRangePlacement());
updatedMetadata = unsafeJoin.execute(metadata).success().metadata;
}
writeMetadata(output, updatedMetadata, outputFilePath);
}
}
/**
* Prints the full {@code toString()} representation of the cluster metadata to stdout.
* Useful for a quick human-readable overview of the entire metadata state.
* <p>
* <b>Note:</b> The output format is not stable and may change between versions.
* Do not rely on it for programmatic parsing.
*/
@Command(name = "print", description = "Prints string output of the cluster metadata. " +
"Output format is subject to change and should not be relied on for parsing.")
static class Print extends ClusterMetadataToolCmd
{
@Override
protected void execute(Output output) throws IOException
{
// It supports only toString output of the ClusterMetadata for now
ClusterMetadata metadata = parseClusterMetadata();
output.out.println(metadata);
}
}
/**
* Prints per-node directory information: addresses, ports, rack, DC, state,
* serialization version, and CMS membership status.
*/
@Command(name = "printdirectory", description = "Prints directory information in cluster metadata file.")
static class PrintDirectoryCmd extends ClusterMetadataToolCmd
{
@Override
protected void execute(Output output) throws IOException
{
ClusterMetadata metadata = parseClusterMetadata();
Directory directory = metadata.directory;
Set<NodeId> nodeIdList = directory.peerIds();
for (NodeId nodeId : nodeIdList)
{
NodeAddresses nodeAddresses = directory.getNodeAddresses(nodeId);
Location location = directory.location(nodeId);
output.out.println("NodeId: " + nodeId.id());
String format = " %-22s%s\n";
output.out.printf(format, "rack", location.rack);
output.out.printf(format, "local_port", nodeAddresses.localAddress.getPort());
output.out.printf(format, "broadcast_port", nodeAddresses.broadcastAddress.getPort());
output.out.printf(format, "host_id", nodeId.toUUID());
output.out.printf(format, "broadcast_address", nodeAddresses.broadcastAddress.getAddress().toString());
output.out.printf(format, "native_address", nodeAddresses.nativeAddress.getAddress().toString());
output.out.printf(format, "native_port", nodeAddresses.nativeAddress.getPort());
output.out.printf(format, "local_address", nodeAddresses.localAddress.getAddress().toString());
output.out.printf(format, "state", directory.peerState(nodeId));
output.out.printf(format, "serialization_version", directory.version(nodeId).serializationVersion());
output.out.printf(format, "cassandra_version", directory.version(nodeId).cassandraVersion);
output.out.printf(format, "dc", location.datacenter);
output.out.printf(format, "is_cms_member", metadata.isCMSMember(nodeAddresses.broadcastAddress));
}
}
}
/**
* Prints read and write replica placements for a specific keyspace, sorted by token range.
* Requires the {@code -ks} option to specify the target keyspace.
*/
@Command(name = "printdataplacements", description = "Prints data placements in cluster metadata file.")
static class PrintDataPlacements extends ClusterMetadataToolCmd
{
@Option(names = { "-ks", "--keyspace" }, required = true,
description = "Keyspace to use for printing data placements.")
private String keyspace;
@Override
protected void execute(Output output) throws IOException
{
ClusterMetadata metadata = parseClusterMetadata();
KeyspaceMetadata keyspaceMetadata = metadata.schema.getKeyspaces().getNullable(keyspace);
if (keyspaceMetadata == null)
{
throw new IllegalArgumentException("Keyspace " + keyspace + " not found in cluster metadata.");
}
DataPlacement placement = metadata.placements.get(keyspaceMetadata.params.replication);
List<Object[]> rows = new ArrayList<>();
rows.addAll(replicaGroupsToRows(placement.reads, "read"));
rows.addAll(replicaGroupsToRows(placement.writes, "write"));
rows.sort((o1, o2) -> {
Range<Token> left = (Range<Token>) o1[0];
Range<Token> right = (Range<Token>) o2[0];
return left.compareTo(right);
});
int rangeMaxLength = 0;
for (Object[] objects : rows)
{
rangeMaxLength = Math.max(rangeMaxLength, objects[0].toString().length());
}
String rowFormat = String.format("%%-%ds %%-7s %%s\n", (rangeMaxLength + 2));
output.out.printf(rowFormat, "Token Range", "Type", "Endpoints");
rows.forEach(row -> output.out.printf(rowFormat, row));
}
List<Object[]> replicaGroupsToRows(ReplicaGroups replicaGroups, String replicaGroupType)
{
List<Object[]> rows = new ArrayList<>();
replicaGroups.forEach(((tokenRange, forRange) -> {
List<String> endpoints = new ArrayList<>(forRange.get().size());
forRange.get().forEach(replica -> endpoints.add(replica.endpoint().toString()));
String addresses = String.join(", ", endpoints);
rows.add(new Object[]{ tokenRange, replicaGroupType, addresses });
}));
return rows;
}
}
static
{
FBUtilities.preventIllegalAccessWarnings();
}
}

57
src/java/org/apache/cassandra/tools/nodetool/CMSAdmin.java
View File

@ -18,6 +18,7 @@
package org.apache.cassandra.tools.nodetool;
import java.io.IOException;
import java.io.PrintStream;
import java.util.ArrayList;
import java.util.Comparator;
@ -28,9 +29,11 @@ import java.util.Map;
import com.google.common.collect.ImmutableList;
import org.apache.cassandra.tcm.Epoch;
import org.apache.cassandra.tcm.serialization.Version;
import org.apache.cassandra.tools.NodeProbe;
import org.apache.cassandra.tools.nodetool.layout.CassandraUsage;
import picocli.CommandLine.ArgGroup;
import picocli.CommandLine.Command;
import picocli.CommandLine.Option;
import picocli.CommandLine.Parameters;
@ -53,6 +56,7 @@ import static org.apache.cassandra.tcm.CMSOperations.SERVICE_STATE;
CMSAdmin.Snapshot.class,
CMSAdmin.Unregister.class,
CMSAdmin.AbortInitialization.class,
CMSAdmin.DumpClusterMetadata.class,
CMSAdmin.DumpDirectory.class,
CMSAdmin.DumpLog.class,
CMSAdmin.ResumeDropAccordTable.class })
@ -164,7 +168,7 @@ public class CMSAdmin extends AbstractCommand
if (!rf.contains(":"))
{
if (args.size() > 1)
throw new IllegalArgumentException("Simple placement can only specify a single replication factor accross all data centers");
throw new IllegalArgumentException("Simple placement can only specify a single replication factor across all data centers");
int parsedRf;
try
{
@ -295,4 +299,55 @@ public class CMSAdmin extends AbstractCommand
probe.getCMSOperationsProxy().resumeDropAccordTable(tableId);
}
}
@Command(name = "dump", description = "Dumps cluster metadata into a file")
public static class DumpClusterMetadata extends AbstractCommand
{
@ArgGroup(exclusive = false, multiplicity = "0..1")
DumpOptions dumpOptions;
static class DumpOptions
{
@Option(names = { "-e", "--epoch" }, paramLabel = "epoch",
description = "Epoch at which cluster metadata should be dumped", required = true)
Long epoch;
@Option(names = { "-te", "--transform-epoch" }, paramLabel = "transform_epoch",
description = "Force metadata to given X epoch while dumping", required = true)
Long transformEpoch;
@Option(names = { "-sv", "--serialization-version" }, paramLabel = "serialization_version",
description = "Serialization Version", required = true)
Version version;
}
@Override
protected void execute(NodeProbe probe)
{
try
{
if (dumpOptions == null)
{
String fileLocation = probe.getCMSOperationsProxy().dumpClusterMetadata();
printCMSDumpLocation(probe, fileLocation);
}
else
{
String fileLocation = probe.getCMSOperationsProxy().dumpClusterMetadata(dumpOptions.epoch,
dumpOptions.transformEpoch,
dumpOptions.version.name());
printCMSDumpLocation(probe, fileLocation);
}
}
catch (IOException e)
{
throw new RuntimeException(e);
}
}
private void printCMSDumpLocation(NodeProbe probe, String fileLocation)
{
probe.output().out.println("Cluster Metadata dump available at " + fileLocation);
}
}
}

58
test/distributed/org/apache/cassandra/distributed/test/log/ClusterMetadataDumpTest.java
View File

@ -25,7 +25,10 @@ import org.junit.Test;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.NodeToolResult;
import org.apache.cassandra.distributed.test.TestBaseImpl;
import org.apache.cassandra.io.util.File;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.ClusterMetadataService;
import org.apache.cassandra.tcm.membership.NodeVersion;
import org.apache.cassandra.tcm.transformations.CustomTransformation;
import static org.junit.Assert.assertEquals;
@ -113,4 +116,59 @@ public class ClusterMetadataDumpTest extends TestBaseImpl
assertEquals(3, tokensFound);
}
}
@Test
public void dumpClusterMetadataTest() throws IOException
{
try (Cluster cluster = init(builder().withNodes(3)
.start()))
{
NodeToolResult res = cluster.get(1).nodetoolResult("cms", "dump");
res.asserts().success();
String stdout = res.getStdout();
String expectedMsgPrefix = "Cluster Metadata dump available at ";
assertTrue(stdout.contains(expectedMsgPrefix));
int index = stdout.indexOf(expectedMsgPrefix);
String dumpFile = stdout.substring(index + expectedMsgPrefix.length()).trim();
assertTrue(new File(dumpFile).exists());
}
}
@Test
public void dumpClusterMetadataWithParamsTest() throws IOException
{
try (Cluster cluster = init(builder().withNodes(3)
.start()))
{
long currentEpoch = cluster.get(1).callOnInstance(() -> ClusterMetadata.current().epoch.getEpoch());
String serVersion = NodeVersion.CURRENT.serializationVersion().toString();
NodeToolResult res = cluster.get(1).nodetoolResult("cms", "dump",
"--epoch", "1",
"--transform-epoch", String.valueOf(currentEpoch),
"--serialization-version", serVersion);
res.asserts().success();
String stdout = res.getStdout();
String expectedMsgPrefix = "Cluster Metadata dump available at ";
assertTrue(stdout.contains(expectedMsgPrefix));
int index = stdout.indexOf(expectedMsgPrefix);
String dumpFile = stdout.substring(index + expectedMsgPrefix.length()).trim();
assertTrue(new File(dumpFile).exists());
}
}
@Test
public void dumpClusterMetadataWithPartialParamsFailsTest() throws IOException
{
try (Cluster cluster = init(builder().withNodes(3)
.start()))
{
// Providing only --epoch without --transform-epoch and --serialization-version
// should be rejected by picocli since all three are required together via @ArgGroup
NodeToolResult res = cluster.get(1).nodetoolResult("cms", "dump", "--epoch", "1");
res.asserts().failure();
}
}
}

17
test/resources/nodetool/help/cms
View File

@ -41,6 +41,15 @@ SYNOPSIS
[(-u <username> | --username <username>)] cms abortinitialization
[--initiator <initiator>]
nodetool [(-h <host> | --host <host>)] [(-p <port> | --port <port>)]
[(-pw <password> | --password <password>)]
[(-pwf <passwordFilePath> | --password-file <passwordFilePath>)]
[(-u <username> | --username <username>)] cms dump
[(-e <epoch> | --epoch <epoch>)]
[(-sv <serialization_version> | --serialization-version
<serialization_version>)]
[(-te <transform_epoch> | --transform-epoch <transform_epoch>)]
nodetool [(-h <host> | --host <host>)] [(-p <port> | --port <port>)]
[(-pw <password> | --password <password>)]
[(-pwf <passwordFilePath> | --password-file <passwordFilePath>)]
@ -102,6 +111,14 @@ COMMANDS
With --initiator option, The address of the node where `cms initialize` was
run.
dump
Dumps cluster metadata into a file
With --epoch option, Epoch at which cluster metadata should be dumped
With --transform-epoch option, Force metadata to given X epoch while dumping
With --serialization-version option, Serialization Version
dumpdirectory
Dump the directory from the current ClusterMetadata

38
test/resources/nodetool/help/cms$dump Normal file
View File

@ -0,0 +1,38 @@
NAME
nodetool cms dump - Dumps cluster metadata into a file
SYNOPSIS
nodetool [(-h <host> | --host <host>)] [(-p <port> | --port <port>)]
[(-pw <password> | --password <password>)]
[(-pwf <passwordFilePath> | --password-file <passwordFilePath>)]
[(-u <username> | --username <username>)] cms dump
[(-e <epoch> | --epoch <epoch>)]
[(-sv <serialization_version> | --serialization-version
<serialization_version>)]
[(-te <transform_epoch> | --transform-epoch <transform_epoch>)]
OPTIONS
-e <epoch>, --epoch <epoch>
Epoch at which cluster metadata should be dumped
-h <host>, --host <host>
Node hostname or ip address
-p <port>, --port <port>
Remote jmx agent port number
-pw <password>, --password <password>
Remote jmx agent password
-pwf <passwordFilePath>, --password-file <passwordFilePath>
Path to the JMX password file
-sv <serialization_version>, --serialization-version
<serialization_version>
Serialization Version
-te <transform_epoch>, --transform-epoch <transform_epoch>
Force metadata to given X epoch while dumping
-u <username>, --username <username>
Remote jmx agent username

1967
test/unit/org/apache/cassandra/tools/CMSOfflineToolTest.java Normal file
View File

File diff suppressed because it is too large Load Diff

5
test/unit/org/apache/cassandra/tools/OfflineToolUtils.java
View File

@ -108,7 +108,9 @@ public abstract class OfflineToolUtils
Collections.addAll(allowedThreadNames, EXTRA_JDK_THREADS);
Collections.addAll(allowedThreadNames, optionalThreadNames);
if (allowNonDefaultMemtableThreads && DatabaseDescriptor.getMemtableConfigurations().containsKey("default"))
if (allowNonDefaultMemtableThreads
&& (DatabaseDescriptor.getMemtableConfigurations() != null
&& DatabaseDescriptor.getMemtableConfigurations().containsKey("default")))
Collections.addAll(allowedThreadNames, NON_DEFAULT_MEMTABLE_THREADS);
var allowedRegexes = allowedThreadNames.stream()
@ -119,6 +121,7 @@ public abstract class OfflineToolUtils
var badThreads = Arrays.stream(threads.getThreadInfo(threads.getAllThreadIds()))
.filter(Objects::nonNull)
.filter(threadInfo -> allowedRegexes.stream().noneMatch(pattern -> pattern.matcher(threadInfo.getThreadName()).matches()))
.filter(threadInfo -> !allowedThreadNames.contains(threadInfo.getThreadName()))
.collect(Collectors.toSet());
if (!badThreads.isEmpty())

4
tools/bin/addtocmstool → tools/bin/cmsofflinetool
View File

@ -43,7 +43,7 @@ fi
"$JAVA" $JAVA_AGENT -ea -cp "$CLASSPATH" $JVM_OPTS -Xmx$MAX_HEAP_SIZE \
-Dcassandra.storagedir="$cassandra_storagedir" \
-Dlog4j.configurationFile=log4j2-tools.xml \
org.apache.cassandra.tools.TransformClusterMetadataHelper "$@"
-Dlogback.configurationFile=logback-tools.xml \
org.apache.cassandra.tools.CMSOfflineTool "$@"
# vi:ai sw=4 ts=4 tw=0 et