mirror of https://github.com/apache/cassandra
Show the progress of data streaming and index build
patch by Stefan Miklosovic; reviewed by Benjamin Lerer, Berenguer Blasi, David Capwell for CASSANDRA-15406
This commit is contained in:
Stefan Miklosovic
David Capwell
parent
68aca1064b
commit
b936d3e95d
|
|
@ -12,6 +12,7 @@
|
|||
* Make Table/Keyspace Metric Names Consistent With Each Other (CASSANDRA-15909)
|
||||
* Mutating sstable component may race with entire-sstable-streaming(ZCS) causing checksum validation failure (CASSANDRA-15861)
|
||||
* NPE thrown while updating speculative execution time if keyspace is removed during task execution (CASSANDRA-15949)
|
||||
* Show the progress of data streaming and index build (CASSANDRA-15406)
|
||||
Merged from 3.11:
|
||||
* Use IF NOT EXISTS for index and UDT create statements in snapshot schema files (CASSANDRA-13935)
|
||||
* Make sure LCS handles duplicate sstable added/removed notifications correctly (CASSANDRA-14103)
|
||||
|
|
|
|||
|
|
@ -97,7 +97,7 @@ public class CassandraCompressedStreamReader extends CassandraStreamReader
|
|||
{
|
||||
writePartition(deserializer, writer);
|
||||
// when compressed, report total bytes of compressed chunks read since remoteFile.size is the sum of chunks transferred
|
||||
session.progress(filename, ProgressInfo.Direction.IN, cis.chunkBytesRead(), totalSize);
|
||||
session.progress(filename + '-' + fileSeqNum, ProgressInfo.Direction.IN, cis.chunkBytesRead(), totalSize);
|
||||
}
|
||||
assert in.getBytesRead() == sectionLength;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -124,7 +124,7 @@ public class CassandraStreamReader implements IStreamReader
|
|||
{
|
||||
writePartition(deserializer, writer);
|
||||
// TODO move this to BytesReadTracker
|
||||
session.progress(writer.getFilename(), ProgressInfo.Direction.IN, in.getBytesRead(), totalSize);
|
||||
session.progress(writer.getFilename() + '-' + fileSeqNum, ProgressInfo.Direction.IN, in.getBytesRead(), totalSize);
|
||||
}
|
||||
logger.debug("[Stream #{}] Finished receiving file #{} from {} readBytes = {}, totalSize = {}",
|
||||
session.planId(), fileSeqNum, session.peer, FBUtilities.prettyPrintMemory(in.getBytesRead()), FBUtilities.prettyPrintMemory(totalSize));
|
||||
|
|
|
|||
|
|
@ -110,7 +110,7 @@ public class ProgressInfo implements Serializable
|
|||
{
|
||||
StringBuilder sb = new StringBuilder(fileName);
|
||||
sb.append(" ").append(currentBytes);
|
||||
sb.append("/").append(totalBytes).append(" bytes");
|
||||
sb.append("/").append(totalBytes).append(" bytes ");
|
||||
sb.append("(").append(currentBytes*100/totalBytes).append("%) ");
|
||||
sb.append(direction == Direction.OUT ? "sent to " : "received from ");
|
||||
sb.append("idx:").append(sessionIndex);
|
||||
|
|
|
|||
|
|
@ -85,8 +85,8 @@ public class StreamReceiveTask extends StreamTask
|
|||
remoteStreamsReceived += stream.getNumFiles();
|
||||
bytesReceived += stream.getSize();
|
||||
Preconditions.checkArgument(tableId.equals(stream.getTableId()));
|
||||
logger.debug("received {} of {} total files {} of total bytes {}", remoteStreamsReceived, totalStreams,
|
||||
bytesReceived, totalSize);
|
||||
logger.debug("received {} of {} total files, {} of total bytes {}", remoteStreamsReceived, totalStreams,
|
||||
bytesReceived, stream.getSize());
|
||||
|
||||
receiver.received(stream);
|
||||
|
||||
|
|
|
|||
|
|
@ -59,10 +59,29 @@ public class NetStats extends NodeToolCmd
|
|||
System.out.printf("%n");
|
||||
if (!info.receivingSummaries.isEmpty())
|
||||
{
|
||||
long totalFilesToReceive = info.getTotalFilesToReceive();
|
||||
long totalBytesToReceive = info.getTotalSizeToReceive();
|
||||
long totalFilesReceived = info.getTotalFilesReceived();
|
||||
long totalSizeReceived = info.getTotalSizeReceived();
|
||||
double percentageFilesReceived = ((double) totalFilesReceived / totalFilesToReceive) * 100;
|
||||
double percentageSizesReceived = ((double) totalSizeReceived / totalBytesToReceive) * 100;
|
||||
|
||||
if (humanReadable)
|
||||
System.out.printf(" Receiving %d files, %s total. Already received %d files, %s total%n", info.getTotalFilesToReceive(), FileUtils.stringifyFileSize(info.getTotalSizeToReceive()), info.getTotalFilesReceived(), FileUtils.stringifyFileSize(info.getTotalSizeReceived()));
|
||||
System.out.printf(" Receiving %d files, %s total. Already received %d files (%.2f%%), %s total (%.2f%%)%n",
|
||||
totalFilesToReceive,
|
||||
FileUtils.stringifyFileSize(totalBytesToReceive),
|
||||
totalFilesReceived,
|
||||
percentageFilesReceived,
|
||||
FileUtils.stringifyFileSize(totalSizeReceived),
|
||||
percentageSizesReceived);
|
||||
else
|
||||
System.out.printf(" Receiving %d files, %d bytes total. Already received %d files, %d bytes total%n", info.getTotalFilesToReceive(), info.getTotalSizeToReceive(), info.getTotalFilesReceived(), info.getTotalSizeReceived());
|
||||
System.out.printf(" Receiving %d files, %d bytes total. Already received %d files (%.2f%%), %d bytes total (%.2f%%)%n",
|
||||
totalFilesToReceive,
|
||||
totalBytesToReceive,
|
||||
totalFilesReceived,
|
||||
percentageFilesReceived,
|
||||
totalSizeReceived,
|
||||
percentageSizesReceived);
|
||||
for (ProgressInfo progress : info.getReceivingFiles())
|
||||
{
|
||||
System.out.printf(" %s%n", progress.toString(printPort));
|
||||
|
|
@ -70,10 +89,29 @@ public class NetStats extends NodeToolCmd
|
|||
}
|
||||
if (!info.sendingSummaries.isEmpty())
|
||||
{
|
||||
long totalFilesToSend = info.getTotalFilesToSend();
|
||||
long totalSizeToSend = info.getTotalSizeToSend();
|
||||
long totalFilesSent = info.getTotalFilesSent();
|
||||
long totalSizeSent = info.getTotalSizeSent();
|
||||
double percentageFilesSent = ((double) totalFilesSent / totalFilesToSend) * 100;
|
||||
double percentageSizeSent = ((double) totalSizeSent / totalSizeToSend) * 100;
|
||||
|
||||
if (humanReadable)
|
||||
System.out.printf(" Sending %d files, %s total. Already sent %d files, %s total%n", info.getTotalFilesToSend(), FileUtils.stringifyFileSize(info.getTotalSizeToSend()), info.getTotalFilesSent(), FileUtils.stringifyFileSize(info.getTotalSizeSent()));
|
||||
System.out.printf(" Sending %d files, %s total. Already sent %d files (%.2f%%), %s total (%.2f%%)%n",
|
||||
totalFilesToSend,
|
||||
FileUtils.stringifyFileSize(totalSizeToSend),
|
||||
totalFilesSent,
|
||||
percentageFilesSent,
|
||||
FileUtils.stringifyFileSize(totalSizeSent),
|
||||
percentageSizeSent);
|
||||
else
|
||||
System.out.printf(" Sending %d files, %d bytes total. Already sent %d files, %d bytes total%n", info.getTotalFilesToSend(), info.getTotalSizeToSend(), info.getTotalFilesSent(), info.getTotalSizeSent());
|
||||
System.out.printf(" Sending %d files, %d bytes total. Already sent %d files (%.2f%%), %d bytes total (%.2f%%) %n",
|
||||
totalFilesToSend,
|
||||
totalSizeToSend,
|
||||
totalFilesSent,
|
||||
percentageFilesSent,
|
||||
totalSizeSent,
|
||||
percentageSizeSent);
|
||||
for (ProgressInfo progress : info.getSendingFiles())
|
||||
{
|
||||
System.out.printf(" %s%n", progress.toString(printPort));
|
||||
|
|
@ -98,35 +136,35 @@ public class NetStats extends NodeToolCmd
|
|||
long dropped;
|
||||
|
||||
pending = 0;
|
||||
for (int n : ms.getLargeMessagePendingTasks().values())
|
||||
for (int n : ms.getLargeMessagePendingTasksWithPort().values())
|
||||
pending += n;
|
||||
completed = 0;
|
||||
for (long n : ms.getLargeMessageCompletedTasks().values())
|
||||
for (long n : ms.getLargeMessageCompletedTasksWithPort().values())
|
||||
completed += n;
|
||||
dropped = 0;
|
||||
for (long n : ms.getLargeMessageDroppedTasks().values())
|
||||
for (long n : ms.getLargeMessageDroppedTasksWithPort().values())
|
||||
dropped += n;
|
||||
System.out.printf("%-25s%10s%10s%15s%10s%n", "Large messages", "n/a", pending, completed, dropped);
|
||||
|
||||
pending = 0;
|
||||
for (int n : ms.getSmallMessagePendingTasks().values())
|
||||
for (int n : ms.getSmallMessagePendingTasksWithPort().values())
|
||||
pending += n;
|
||||
completed = 0;
|
||||
for (long n : ms.getSmallMessageCompletedTasks().values())
|
||||
for (long n : ms.getSmallMessageCompletedTasksWithPort().values())
|
||||
completed += n;
|
||||
dropped = 0;
|
||||
for (long n : ms.getSmallMessageDroppedTasks().values())
|
||||
for (long n : ms.getSmallMessageDroppedTasksWithPort().values())
|
||||
dropped += n;
|
||||
System.out.printf("%-25s%10s%10s%15s%10s%n", "Small messages", "n/a", pending, completed, dropped);
|
||||
|
||||
pending = 0;
|
||||
for (int n : ms.getGossipMessagePendingTasks().values())
|
||||
for (int n : ms.getGossipMessagePendingTasksWithPort().values())
|
||||
pending += n;
|
||||
completed = 0;
|
||||
for (long n : ms.getGossipMessageCompletedTasks().values())
|
||||
for (long n : ms.getGossipMessageCompletedTasksWithPort().values())
|
||||
completed += n;
|
||||
dropped = 0;
|
||||
for (long n : ms.getGossipMessageDroppedTasks().values())
|
||||
for (long n : ms.getGossipMessageDroppedTasksWithPort().values())
|
||||
dropped += n;
|
||||
System.out.printf("%-25s%10s%10s%15s%10s%n", "Gossip messages", "n/a", pending, completed, dropped);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -605,18 +605,28 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
|
|||
}).call();
|
||||
}
|
||||
|
||||
private static class DTestNodeTool extends NodeTool {
|
||||
public static class DTestNodeTool extends NodeTool {
|
||||
private final StorageServiceMBean storageProxy;
|
||||
private final CollectingNotificationListener notifications = new CollectingNotificationListener();
|
||||
|
||||
private Throwable latestError;
|
||||
|
||||
DTestNodeTool(boolean withNotifications) {
|
||||
public DTestNodeTool(boolean withNotifications) {
|
||||
super(new InternalNodeProbeFactory(withNotifications));
|
||||
storageProxy = new InternalNodeProbe(withNotifications).getStorageService();
|
||||
storageProxy.addNotificationListener(notifications, null, null);
|
||||
}
|
||||
|
||||
public List<Notification> getNotifications()
|
||||
{
|
||||
return new ArrayList<>(notifications.notifications);
|
||||
}
|
||||
|
||||
public Throwable getLatestError()
|
||||
{
|
||||
return latestError;
|
||||
}
|
||||
|
||||
public int execute(String... args)
|
||||
{
|
||||
try
|
||||
|
|
|
|||
|
|
@ -0,0 +1,48 @@
|
|||
/*
|
||||
* 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.shared;
|
||||
|
||||
import java.io.ByteArrayOutputStream;
|
||||
|
||||
import org.apache.cassandra.distributed.api.NodeToolResult;
|
||||
|
||||
public class NodeToolResultWithOutput
|
||||
{
|
||||
private final NodeToolResult result;
|
||||
private final ByteArrayOutputStream stdout;
|
||||
private final ByteArrayOutputStream stderr;
|
||||
|
||||
public NodeToolResultWithOutput(NodeToolResult result, ByteArrayOutputStream stdout, ByteArrayOutputStream stderr) {
|
||||
this.result = result;
|
||||
this.stdout = stdout;
|
||||
this.stderr = stderr;
|
||||
}
|
||||
|
||||
public NodeToolResult getResult() {
|
||||
return this.result;
|
||||
}
|
||||
|
||||
public String getStdout() {
|
||||
return this.stdout.toString();
|
||||
}
|
||||
|
||||
public String getStderr() {
|
||||
return this.stderr.toString();
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,85 @@
|
|||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one
|
||||
* or more contributor license agreements. See the NOTICE file
|
||||
* distributed with this work for additional information
|
||||
* regarding copyright ownership. The ASF licenses this file
|
||||
* to you under the Apache License, Version 2.0 (the
|
||||
* "License"); you may not use this file except in compliance
|
||||
* with the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.apache.cassandra.distributed.test;
|
||||
|
||||
import java.util.concurrent.Future;
|
||||
|
||||
import org.apache.cassandra.distributed.Cluster;
|
||||
import org.apache.cassandra.distributed.api.IInstanceConfig;
|
||||
import org.apache.cassandra.distributed.api.IInvokableInstance;
|
||||
import org.apache.cassandra.distributed.api.TokenSupplier;
|
||||
import org.apache.cassandra.distributed.shared.NetworkTopology;
|
||||
|
||||
import static java.util.concurrent.TimeUnit.MINUTES;
|
||||
import static org.apache.cassandra.distributed.api.Feature.GOSSIP;
|
||||
import static org.apache.cassandra.distributed.api.Feature.NATIVE_PROTOCOL;
|
||||
import static org.apache.cassandra.distributed.api.Feature.NETWORK;
|
||||
|
||||
public abstract class AbstractNetstatsBootstrapStreaming extends AbstractNetstatsStreaming
|
||||
{
|
||||
protected void executeTest(final boolean streamEntireSSTables,
|
||||
final boolean compressionEnabled) throws Exception
|
||||
{
|
||||
final Cluster.Builder builder = builder().withNodes(1)
|
||||
.withTokenSupplier(TokenSupplier.evenlyDistributedTokens(2))
|
||||
.withNodeIdTopology(NetworkTopology.singleDcNetworkTopology(2, "dc0", "rack0"))
|
||||
.withConfig(config -> config.with(NETWORK, GOSSIP, NATIVE_PROTOCOL)
|
||||
.set("stream_throughput_outbound_megabits_per_sec", 1)
|
||||
.set("compaction_throughput_mb_per_sec", 1)
|
||||
.set("stream_entire_sstables", streamEntireSSTables));
|
||||
|
||||
try (final Cluster cluster = builder.withNodes(1).start())
|
||||
{
|
||||
// populate data only against 1 node first
|
||||
|
||||
createTable(cluster, 1, compressionEnabled);
|
||||
|
||||
cluster.get(1).nodetoolResult("disableautocompaction", "netstats_test").asserts().success();
|
||||
|
||||
if (compressionEnabled)
|
||||
{
|
||||
populateData(true);
|
||||
}
|
||||
else
|
||||
{
|
||||
populateData(false);
|
||||
}
|
||||
|
||||
cluster.get(1).flush("netstats_test");
|
||||
|
||||
// then bootstrap the second one, upon joining,
|
||||
// we should see that netstats shows how SSTables are being streamed on the first node
|
||||
|
||||
final IInstanceConfig config = cluster.newInstanceConfig();
|
||||
config.set("auto_bootstrap", true);
|
||||
|
||||
IInvokableInstance secondNode = cluster.bootstrap(config);
|
||||
|
||||
final Future<?> startupRunnable = executorService.submit((Runnable) secondNode::startup);
|
||||
final Future<AbstractNetstatsStreaming.NetstatResults> netstatsFuture = executorService.submit(new NetstatsCallable(cluster.get(1)));
|
||||
|
||||
final AbstractNetstatsStreaming.NetstatResults results = netstatsFuture.get(1, MINUTES);
|
||||
startupRunnable.get(2, MINUTES);
|
||||
|
||||
results.assertSuccessful();
|
||||
|
||||
AbstractNetstatsStreaming.NetstatsOutputParser.validate(AbstractNetstatsStreaming.NetstatsOutputParser.parse(results));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,548 @@
|
|||
/*
|
||||
* 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;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.LinkedHashSet;
|
||||
import java.util.List;
|
||||
import java.util.Set;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.Callable;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.regex.Matcher;
|
||||
import java.util.regex.Pattern;
|
||||
import java.util.stream.Stream;
|
||||
|
||||
import org.junit.After;
|
||||
import org.junit.Assert;
|
||||
import org.junit.Before;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import com.datastax.driver.core.Session;
|
||||
import org.apache.cassandra.distributed.Cluster;
|
||||
import org.apache.cassandra.distributed.api.IInvokableInstance;
|
||||
import org.apache.cassandra.distributed.shared.NodeToolResultWithOutput;
|
||||
import org.apache.cassandra.distributed.util.NodetoolUtils;
|
||||
import org.apache.cassandra.utils.Pair;
|
||||
|
||||
import static java.util.stream.Collectors.toList;
|
||||
|
||||
public abstract class AbstractNetstatsStreaming extends TestBaseImpl
|
||||
{
|
||||
protected static final Logger logger = LoggerFactory.getLogger(AbstractNetstatsStreaming.class);
|
||||
|
||||
protected ExecutorService executorService;
|
||||
|
||||
@Before
|
||||
public void setup()
|
||||
{
|
||||
executorService = Executors.newCachedThreadPool();
|
||||
}
|
||||
|
||||
@After
|
||||
public void teardown() throws Exception
|
||||
{
|
||||
try
|
||||
{
|
||||
executorService.shutdownNow();
|
||||
|
||||
if (!executorService.isShutdown())
|
||||
{
|
||||
if (!executorService.awaitTermination(1, TimeUnit.MINUTES))
|
||||
{
|
||||
throw new IllegalStateException("Unable to shutdown executor for invoking netstat commands.");
|
||||
}
|
||||
}
|
||||
}
|
||||
finally
|
||||
{
|
||||
executorService = null;
|
||||
}
|
||||
}
|
||||
|
||||
protected void changeReplicationFactor()
|
||||
{
|
||||
try (com.datastax.driver.core.Cluster c = com.datastax.driver.core.Cluster.builder().addContactPoint("127.0.0.1").build();
|
||||
Session s = c.connect())
|
||||
{
|
||||
s.execute("ALTER KEYSPACE netstats_test WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 2 };");
|
||||
}
|
||||
}
|
||||
|
||||
protected void createTable(Cluster cluster, int replicationFactor, boolean compressionEnabled)
|
||||
{
|
||||
// replication factor is 1
|
||||
cluster.schemaChange("CREATE KEYSPACE netstats_test WITH replication = {'class': 'SimpleStrategy', 'replication_factor': " + replicationFactor + "};");
|
||||
|
||||
if (compressionEnabled)
|
||||
{
|
||||
cluster.schemaChange("CREATE TABLE netstats_test.test_table (id uuid primary key) WITH compression = {'enabled':'true', 'class': 'LZ4Compressor'};");
|
||||
}
|
||||
else
|
||||
{
|
||||
cluster.schemaChange("CREATE TABLE netstats_test.test_table (id uuid primary key) WITH compression = {'enabled':'false'};");
|
||||
}
|
||||
}
|
||||
|
||||
protected void populateData(boolean forCompressedTest)
|
||||
{
|
||||
try (com.datastax.driver.core.Cluster c = com.datastax.driver.core.Cluster.builder().addContactPoint("127.0.0.1").build();
|
||||
Session s = c.connect("netstats_test"))
|
||||
{
|
||||
int records = forCompressedTest ? 100_000 : 70_000;
|
||||
|
||||
for (int i = 0; i < records; i++)
|
||||
{
|
||||
s.execute("INSERT INTO test_table (id) VALUES (" + UUID.randomUUID() + ")");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
protected static class NetstatsOutputParser
|
||||
{
|
||||
public static List<Pair<ReceivingStastistics, SendingStatistics>> parse(final NetstatResults results)
|
||||
{
|
||||
final Set<String> outputs = new LinkedHashSet<>();
|
||||
|
||||
results.netstatOutputs.stream()
|
||||
.map(NodeToolResultWithOutput::getStdout)
|
||||
.filter(output -> !output.contains("Not sending any streams"))
|
||||
.filter(output -> output.contains("Receiving") || output.contains("Sending"))
|
||||
.forEach(outputs::add);
|
||||
|
||||
final List<Pair<ReceivingStastistics, SendingStatistics>> parsed = new ArrayList<>();
|
||||
|
||||
for (final String output : outputs)
|
||||
{
|
||||
boolean processingReceiving = false;
|
||||
boolean processingSending = false;
|
||||
|
||||
final ReceivingStastistics receivingStastistics = new ReceivingStastistics();
|
||||
final SendingStatistics sendingStatistics = new SendingStatistics();
|
||||
|
||||
final List<String> sanitisedOutput = Stream.of(output.split("\n"))
|
||||
.map(String::trim)
|
||||
.filter(line -> !line.isEmpty())
|
||||
// sometimes logs are mangled into output
|
||||
.filter(line -> Stream.of("DEBUG", "INFO", "ERROR", "WARN").noneMatch(line::contains))
|
||||
.filter(line -> Stream.of("Mode:", "Read", "Attempted", "Mismatch", "Pool", "Large", "Small", "Gossip").noneMatch(line::startsWith))
|
||||
.collect(toList());
|
||||
|
||||
for (final String outputLine : sanitisedOutput)
|
||||
{
|
||||
if (outputLine.startsWith("Receiving"))
|
||||
{
|
||||
processingReceiving = true;
|
||||
processingSending = false;
|
||||
|
||||
receivingStastistics.parseHeader(outputLine);
|
||||
}
|
||||
else if (outputLine.startsWith("Sending"))
|
||||
{
|
||||
processingSending = true;
|
||||
processingReceiving = false;
|
||||
|
||||
sendingStatistics.parseHeader(outputLine);
|
||||
}
|
||||
else if (processingReceiving)
|
||||
{
|
||||
receivingStastistics.parseTable(outputLine);
|
||||
}
|
||||
else if (processingSending)
|
||||
{
|
||||
sendingStatistics.parseTable(outputLine);
|
||||
}
|
||||
}
|
||||
|
||||
parsed.add(Pair.create(receivingStastistics, sendingStatistics));
|
||||
}
|
||||
|
||||
return parsed;
|
||||
}
|
||||
|
||||
public static void validate(List<Pair<ReceivingStastistics, SendingStatistics>> result)
|
||||
{
|
||||
List<SendingStatistics> sendingStatistics = result.stream().map(pair -> pair.right).collect(toList());
|
||||
|
||||
if (sendingStatistics.size() >= 2)
|
||||
{
|
||||
for (int i = 0; i < sendingStatistics.size() - 1; i++)
|
||||
{
|
||||
SendingStatistics.SendingHeader header1 = sendingStatistics.get(i).sendingHeader;
|
||||
SendingStatistics.SendingHeader header2 = sendingStatistics.get(i + 1).sendingHeader;
|
||||
|
||||
if (header1 != null && header2 != null)
|
||||
{
|
||||
Assert.assertTrue(header1.compareTo(header2) <= 0);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for (SendingStatistics sending : sendingStatistics)
|
||||
{
|
||||
if (sending.sendingHeader != null)
|
||||
{
|
||||
Assert.assertEquals(sending.sendingHeader.bytesTotalSoFar, (long) sending.sendingSSTable.stream().map(table -> table.bytesSent).reduce(Long::sum).orElseGet(() -> 0L));
|
||||
Assert.assertTrue(sending.sendingHeader.bytesTotal >= sending.sendingSSTable.stream().map(table -> table.bytesInTotal).reduce(Long::sum).orElseGet(() -> 0L));
|
||||
|
||||
if (sending.sendingHeader.bytesTotalSoFar != 0)
|
||||
{
|
||||
double progress = (double) sending.sendingSSTable.stream().map(table -> table.bytesSent).reduce(Long::sum).orElse(0L) / (double) sending.sendingHeader.bytesTotal;
|
||||
|
||||
Assert.assertTrue((int) sending.sendingHeader.progressBytes >= (int) (progress * 100));
|
||||
|
||||
Assert.assertTrue((double) sending.sendingHeader.bytesTotal >= (double) sending.sendingSSTable.stream().map(table -> table.bytesInTotal).reduce(Long::sum).orElse(0L));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
List<ReceivingStastistics> receivingStastistics = result.stream().map(pair -> pair.left).collect(toList());
|
||||
|
||||
for (ReceivingStastistics receiving : receivingStastistics)
|
||||
{
|
||||
if (receiving.receivingHeader != null)
|
||||
{
|
||||
Assert.assertTrue(receiving.receivingHeader.bytesTotal >= receiving.receivingTables.stream().map(table -> table.receivedSoFar).reduce(Long::sum).orElse(0L));
|
||||
Assert.assertEquals(receiving.receivingHeader.bytesTotalSoFar, (long) receiving.receivingTables.stream().map(table -> table.receivedSoFar).reduce(Long::sum).orElse(0L));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public static class ReceivingStastistics
|
||||
{
|
||||
public ReceivingHeader receivingHeader;
|
||||
public List<ReceivingTable> receivingTables = new ArrayList<>();
|
||||
|
||||
public void parseHeader(String header)
|
||||
{
|
||||
receivingHeader = ReceivingHeader.parseHeader(header);
|
||||
}
|
||||
|
||||
public void parseTable(String table)
|
||||
{
|
||||
receivingTables.add(ReceivingTable.parseTable(table));
|
||||
}
|
||||
|
||||
public String toString()
|
||||
{
|
||||
return "ReceivingStastistics{" +
|
||||
"receivingHeader=" + receivingHeader +
|
||||
", receivingTables=" + receivingTables +
|
||||
'}';
|
||||
}
|
||||
|
||||
public static class ReceivingHeader
|
||||
{
|
||||
private static final Pattern receivingHeaderPattern = Pattern.compile(
|
||||
"Receiving (.*) files, (.*) bytes total. Already received (.*) files \\((.*)%\\), (.*) bytes total \\((.*)%\\)"
|
||||
);
|
||||
|
||||
int totalReceiving = 0;
|
||||
long bytesTotal = 0;
|
||||
int alreadyReceived = 0;
|
||||
double progressFiles = 0.0f;
|
||||
long bytesTotalSoFar = 0;
|
||||
double progressBytes = 0.0f;
|
||||
|
||||
public static ReceivingHeader parseHeader(String header)
|
||||
{
|
||||
final Matcher matcher = receivingHeaderPattern.matcher(header);
|
||||
|
||||
if (matcher.matches())
|
||||
{
|
||||
final ReceivingHeader receivingHeader = new ReceivingHeader();
|
||||
|
||||
receivingHeader.totalReceiving = Integer.parseInt(matcher.group(1));
|
||||
receivingHeader.bytesTotal = Long.parseLong(matcher.group(2));
|
||||
receivingHeader.alreadyReceived = Integer.parseInt(matcher.group(3));
|
||||
receivingHeader.progressFiles = Double.parseDouble(matcher.group(4));
|
||||
receivingHeader.bytesTotalSoFar = Long.parseLong(matcher.group(5));
|
||||
receivingHeader.progressBytes = Double.parseDouble(matcher.group(6));
|
||||
|
||||
return receivingHeader;
|
||||
}
|
||||
|
||||
throw new IllegalStateException("Header does not match - " + header);
|
||||
}
|
||||
|
||||
public String toString()
|
||||
{
|
||||
return "ReceivingHeader{" +
|
||||
"totalReceiving=" + totalReceiving +
|
||||
", bytesTotal=" + bytesTotal +
|
||||
", alreadyReceived=" + alreadyReceived +
|
||||
", progressFiles=" + progressFiles +
|
||||
", bytesTotalSoFar=" + bytesTotalSoFar +
|
||||
", progressBytes=" + progressBytes +
|
||||
'}';
|
||||
}
|
||||
}
|
||||
|
||||
public static class ReceivingTable
|
||||
{
|
||||
long receivedSoFar = 0;
|
||||
long toReceive = 0;
|
||||
double progress = 0.0;
|
||||
|
||||
private static final Pattern recievingFilePattern = Pattern.compile("(.*) (.*)/(.*) bytes \\((.*)%\\) received from (.*)");
|
||||
|
||||
public static ReceivingTable parseTable(String table)
|
||||
{
|
||||
final Matcher matcher = recievingFilePattern.matcher(table);
|
||||
|
||||
if (matcher.matches())
|
||||
{
|
||||
final ReceivingTable receivingTable = new ReceivingTable();
|
||||
|
||||
receivingTable.receivedSoFar = Long.parseLong(matcher.group(2));
|
||||
receivingTable.toReceive = Long.parseLong(matcher.group(3));
|
||||
receivingTable.progress = Double.parseDouble(matcher.group(4));
|
||||
|
||||
return receivingTable;
|
||||
}
|
||||
|
||||
throw new IllegalStateException("Table line does not match - " + table);
|
||||
}
|
||||
|
||||
public String toString()
|
||||
{
|
||||
return "ReceivingTable{" +
|
||||
"receivedSoFar=" + receivedSoFar +
|
||||
", toReceive=" + toReceive +
|
||||
", progress=" + progress +
|
||||
'}';
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public static class SendingStatistics
|
||||
{
|
||||
public SendingHeader sendingHeader;
|
||||
public List<SendingSSTable> sendingSSTable = new ArrayList<>();
|
||||
|
||||
public void parseHeader(String outputLine)
|
||||
{
|
||||
this.sendingHeader = SendingHeader.parseHeader(outputLine);
|
||||
}
|
||||
|
||||
public void parseTable(String table)
|
||||
{
|
||||
sendingSSTable.add(SendingSSTable.parseTable(table));
|
||||
}
|
||||
|
||||
public String toString()
|
||||
{
|
||||
return "SendingStatistics{" +
|
||||
"sendingHeader=" + sendingHeader +
|
||||
", sendingSSTable=" + sendingSSTable +
|
||||
'}';
|
||||
}
|
||||
|
||||
public static class SendingHeader implements Comparable<SendingHeader>
|
||||
{
|
||||
private static final Pattern sendingHeaderPattern = Pattern.compile(
|
||||
"Sending (.*) files, (.*) bytes total. Already sent (.*) files \\((.*)%\\), (.*) bytes total \\((.*)%\\)"
|
||||
);
|
||||
|
||||
int totalSending = 0;
|
||||
long bytesTotal = 0;
|
||||
int alreadySent = 0;
|
||||
double progressFiles = 0.0f;
|
||||
long bytesTotalSoFar = 0;
|
||||
double progressBytes = 0.0f;
|
||||
|
||||
public static SendingHeader parseHeader(String header)
|
||||
{
|
||||
final Matcher matcher = sendingHeaderPattern.matcher(header);
|
||||
|
||||
if (matcher.matches())
|
||||
{
|
||||
final SendingHeader sendingHeader = new SendingHeader();
|
||||
|
||||
sendingHeader.totalSending = Integer.parseInt(matcher.group(1));
|
||||
sendingHeader.bytesTotal = Long.parseLong(matcher.group(2));
|
||||
sendingHeader.alreadySent = Integer.parseInt(matcher.group(3));
|
||||
sendingHeader.progressFiles = Double.parseDouble(matcher.group(4));
|
||||
sendingHeader.bytesTotalSoFar = Long.parseLong(matcher.group(5));
|
||||
sendingHeader.progressBytes = Double.parseDouble(matcher.group(6));
|
||||
|
||||
return sendingHeader;
|
||||
}
|
||||
|
||||
throw new IllegalStateException("Header does not match - " + header);
|
||||
}
|
||||
|
||||
public String toString()
|
||||
{
|
||||
return "SendingHeader{" +
|
||||
"totalSending=" + totalSending +
|
||||
", bytesTotal=" + bytesTotal +
|
||||
", alreadySent=" + alreadySent +
|
||||
", progressFiles=" + progressFiles +
|
||||
", bytesTotalSoFar=" + bytesTotalSoFar +
|
||||
", progressBytes=" + progressBytes +
|
||||
'}';
|
||||
}
|
||||
|
||||
|
||||
public int compareTo(SendingHeader o)
|
||||
{
|
||||
// progress on bytes has to be strictly lower,
|
||||
// even alreadySent and progressFiles and progressBytes are same,
|
||||
// bytesTotalSoFar has to be lower, bigger or same
|
||||
|
||||
if (alreadySent <= o.alreadySent
|
||||
&& progressFiles <= o.progressFiles
|
||||
&& bytesTotalSoFar <= o.bytesTotalSoFar
|
||||
&& progressBytes <= o.progressBytes)
|
||||
{
|
||||
return -1;
|
||||
}
|
||||
else if (alreadySent == o.alreadySent
|
||||
&& progressFiles == o.progressFiles
|
||||
&& bytesTotalSoFar == o.bytesTotalSoFar
|
||||
&& progressBytes == o.progressBytes)
|
||||
{
|
||||
return 0;
|
||||
}
|
||||
else if (alreadySent >= o.alreadySent
|
||||
&& progressFiles >= o.progressFiles
|
||||
&& bytesTotalSoFar > o.bytesTotalSoFar
|
||||
&& progressBytes >= o.progressBytes)
|
||||
{
|
||||
return 1;
|
||||
}
|
||||
else
|
||||
{
|
||||
throw new IllegalStateException(String.format("Could not compare arguments %s and %s", this, o));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public static class SendingSSTable
|
||||
{
|
||||
private static final Pattern sendingFilePattern = Pattern.compile("(.*) (.*)/(.*) bytes \\((.*)%\\) sent to (.*)");
|
||||
|
||||
long bytesSent = 0;
|
||||
long bytesInTotal = 0;
|
||||
double progress = 0.0f;
|
||||
|
||||
public static SendingSSTable parseTable(String table)
|
||||
{
|
||||
final Matcher matcher = sendingFilePattern.matcher(table);
|
||||
|
||||
if (matcher.matches())
|
||||
{
|
||||
final SendingSSTable sendingSSTable = new SendingSSTable();
|
||||
|
||||
sendingSSTable.bytesSent = Long.parseLong(matcher.group(2));
|
||||
sendingSSTable.bytesInTotal = Long.parseLong(matcher.group(3));
|
||||
sendingSSTable.progress = Double.parseDouble(matcher.group(4));
|
||||
|
||||
return sendingSSTable;
|
||||
}
|
||||
|
||||
throw new IllegalStateException("Table does not match - " + table);
|
||||
}
|
||||
|
||||
public String toString()
|
||||
{
|
||||
return "SendingSSTable{" +
|
||||
"bytesSent=" + bytesSent +
|
||||
", bytesInTotal=" + bytesInTotal +
|
||||
", progress=" + progress +
|
||||
'}';
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
protected static final class NetstatResults
|
||||
{
|
||||
private final List<NodeToolResultWithOutput> netstatOutputs = new ArrayList<>();
|
||||
|
||||
public void add(NodeToolResultWithOutput result)
|
||||
{
|
||||
netstatOutputs.add(result);
|
||||
}
|
||||
|
||||
public void assertSuccessful()
|
||||
{
|
||||
for (final NodeToolResultWithOutput result : netstatOutputs)
|
||||
{
|
||||
Assert.assertEquals(result.getResult().getRc(), 0);
|
||||
Assert.assertTrue(result.getStderr().isEmpty());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
protected static class NetstatsCallable implements Callable<NetstatResults>
|
||||
{
|
||||
private final IInvokableInstance node;
|
||||
|
||||
public NetstatsCallable(final IInvokableInstance node)
|
||||
{
|
||||
this.node = node;
|
||||
}
|
||||
|
||||
public NetstatResults call() throws Exception
|
||||
{
|
||||
final NetstatResults results = new NetstatResults();
|
||||
|
||||
boolean sawAnyStreamingOutput = false;
|
||||
|
||||
while (true)
|
||||
{
|
||||
try
|
||||
{
|
||||
final NodeToolResultWithOutput result = NodetoolUtils.nodetool(node, false, "netstats");
|
||||
|
||||
logger.info(node.broadcastAddress().toString() + " " + result.getStdout());
|
||||
|
||||
if (!sawAnyStreamingOutput)
|
||||
{
|
||||
if (result.getStdout().contains("Receiving") || result.getStdout().contains("Sending"))
|
||||
{
|
||||
sawAnyStreamingOutput = true;
|
||||
}
|
||||
}
|
||||
|
||||
if (sawAnyStreamingOutput && (!result.getStdout().contains("Receiving") && !result.getStdout().contains("Sending")))
|
||||
{
|
||||
break;
|
||||
}
|
||||
|
||||
results.add(result);
|
||||
|
||||
Thread.currentThread().sleep(500);
|
||||
}
|
||||
catch (final Exception ex)
|
||||
{
|
||||
System.out.println(ex.getMessage());
|
||||
Thread.currentThread().sleep(500);
|
||||
}
|
||||
}
|
||||
|
||||
return results;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,36 @@
|
|||
/*
|
||||
* 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;
|
||||
|
||||
import org.junit.Test;
|
||||
|
||||
public class NetstatsBootstrapWithEntireSSTablesCompressionStreamingTest extends AbstractNetstatsBootstrapStreaming
|
||||
{
|
||||
@Test
|
||||
public void testWithStreamingEntireSSTablesWithCompression() throws Exception
|
||||
{
|
||||
executeTest(true, true);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testWithStreamingEntireSSTablesWithoutCompression() throws Exception
|
||||
{
|
||||
executeTest(true, false);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,36 @@
|
|||
/*
|
||||
* 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;
|
||||
|
||||
import org.junit.Test;
|
||||
|
||||
public class NetstatsBootstrapWithoutEntireSSTablesCompressionStreamingTest extends AbstractNetstatsBootstrapStreaming
|
||||
{
|
||||
@Test
|
||||
public void testWithoutStreamingEntireSSTablesWithCompression() throws Exception
|
||||
{
|
||||
executeTest(false, true);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testWithoutStreamingEntireSSTablesWithoutCompression() throws Exception
|
||||
{
|
||||
executeTest(false, false);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,88 @@
|
|||
/*
|
||||
* 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;
|
||||
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.Future;
|
||||
|
||||
import org.junit.Test;
|
||||
|
||||
import org.apache.cassandra.distributed.Cluster;
|
||||
import org.apache.cassandra.distributed.api.IInvokableInstance;
|
||||
import org.apache.cassandra.distributed.shared.NetworkTopology;
|
||||
|
||||
import static java.util.concurrent.TimeUnit.MINUTES;
|
||||
import static org.apache.cassandra.distributed.api.Feature.GOSSIP;
|
||||
import static org.apache.cassandra.distributed.api.Feature.NATIVE_PROTOCOL;
|
||||
import static org.apache.cassandra.distributed.api.Feature.NETWORK;
|
||||
|
||||
public class NetstatsRepairStreamingTest extends AbstractNetstatsStreaming
|
||||
{
|
||||
@Test
|
||||
public void testWithCompressionEnabled() throws Exception
|
||||
{
|
||||
executeTest(true);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testWithCompressionDisabled() throws Exception
|
||||
{
|
||||
executeTest(false);
|
||||
}
|
||||
|
||||
private void executeTest(boolean compressionEnabled) throws Exception
|
||||
{
|
||||
final ExecutorService executorService = Executors.newFixedThreadPool(1);
|
||||
|
||||
try (final Cluster cluster = Cluster.build()
|
||||
.withNodeIdTopology(NetworkTopology.singleDcNetworkTopology(2, "dc0", "rack0"))
|
||||
.withConfig(config -> config.with(NETWORK, GOSSIP, NATIVE_PROTOCOL)
|
||||
.set("stream_throughput_outbound_megabits_per_sec", 1)
|
||||
.set("compaction_throughput_mb_per_sec", 1)
|
||||
.set("stream_entire_sstables", false)).start())
|
||||
{
|
||||
final IInvokableInstance node1 = cluster.get(1);
|
||||
final IInvokableInstance node2 = cluster.get(2);
|
||||
|
||||
createTable(cluster, 1, compressionEnabled);
|
||||
|
||||
node1.nodetoolResult("disableautocompaction", "netstats_test").asserts().success();
|
||||
node2.nodetoolResult("disableautocompaction", "netstats_test").asserts().success();
|
||||
|
||||
populateData(compressionEnabled);
|
||||
|
||||
node1.flush("netstats_test");
|
||||
node2.flush("netstats_test");
|
||||
|
||||
//change RF from 1 to 2 so we need to repair it, repairing will causes streaming shown in netstats
|
||||
changeReplicationFactor();
|
||||
|
||||
final Future<NetstatResults> resultsFuture1 = executorService.submit(new NetstatsCallable(node1));
|
||||
|
||||
node1.nodetoolResult("repair", "netstats_test").asserts().success();
|
||||
|
||||
final NetstatResults results = resultsFuture1.get(1, MINUTES);
|
||||
|
||||
results.assertSuccessful();
|
||||
|
||||
NetstatsOutputParser.validate(NetstatsOutputParser.parse(results));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,68 @@
|
|||
/*
|
||||
* 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.util;
|
||||
|
||||
import java.io.ByteArrayOutputStream;
|
||||
import java.io.PrintStream;
|
||||
|
||||
import org.apache.cassandra.distributed.api.IInvokableInstance;
|
||||
import org.apache.cassandra.distributed.api.NodeToolResult;
|
||||
import org.apache.cassandra.distributed.impl.Instance;
|
||||
import org.apache.cassandra.distributed.shared.NodeToolResultWithOutput;
|
||||
|
||||
public final class NodetoolUtils
|
||||
{
|
||||
private NodetoolUtils()
|
||||
{
|
||||
|
||||
}
|
||||
|
||||
public static NodeToolResultWithOutput nodetool(IInvokableInstance inst, String... args)
|
||||
{
|
||||
return nodetool(inst, true, args);
|
||||
}
|
||||
|
||||
public static NodeToolResultWithOutput nodetool(IInvokableInstance inst, boolean withNotifications, String... args)
|
||||
{
|
||||
return inst.callOnInstance(() -> {
|
||||
PrintStream originalSysOut = System.out;
|
||||
PrintStream originalSysErr = System.err;
|
||||
originalSysOut.flush();
|
||||
originalSysErr.flush();
|
||||
ByteArrayOutputStream toolOut = new ByteArrayOutputStream();
|
||||
ByteArrayOutputStream toolErr = new ByteArrayOutputStream();
|
||||
|
||||
try (PrintStream newOut = new PrintStream(toolOut);
|
||||
PrintStream newErr = new PrintStream(toolErr))
|
||||
{
|
||||
System.setOut(newOut);
|
||||
System.setErr(newErr);
|
||||
Instance.DTestNodeTool nodetool = new Instance.DTestNodeTool(withNotifications);
|
||||
int rc = nodetool.execute(args);
|
||||
NodeToolResult result = new NodeToolResult(args, rc, nodetool.getNotifications(), nodetool.getLatestError());
|
||||
return new NodeToolResultWithOutput(result, toolOut, toolErr);
|
||||
}
|
||||
finally
|
||||
{
|
||||
System.setOut(originalSysOut);
|
||||
System.setErr(originalSysErr);
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue