You are viewing a plain text version of this content. The canonical link for it is here.
Posted to pr@cassandra.apache.org by GitBox <gi...@apache.org> on 2020/08/06 18:05:54 UTC
[GitHub] [cassandra] smiklosovic opened a new pull request #711: CASSANDRA-15406 - added progress of streaming in percents in netstats
smiklosovic opened a new pull request #711:
URL: https://github.com/apache/cassandra/pull/711
… command
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
[GitHub] [cassandra] smiklosovic commented on a change in pull request #711: CASSANDRA-15406 - added progress of streaming in percents in netstats
Posted by GitBox <gi...@apache.org>.
smiklosovic commented on a change in pull request #711:
URL: https://github.com/apache/cassandra/pull/711#discussion_r466594435
##########
File path: src/java/org/apache/cassandra/streaming/SessionInfo.java
##########
@@ -149,6 +149,14 @@ public long getTotalSizeToReceive()
return getTotalSizes(receivingSummaries);
}
+ public long getTotalBytesToReceive()
Review comment:
without this method, the progress was not computed correctly.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
[GitHub] [cassandra] dcapwell commented on a change in pull request #711: CASSANDRA-15406 - added progress of streaming in percents in netstats
Posted by GitBox <gi...@apache.org>.
dcapwell commented on a change in pull request #711:
URL: https://github.com/apache/cassandra/pull/711#discussion_r474828428
##########
File path: test/distributed/org/apache/cassandra/distributed/test/AbstractNetstatsStreaming.java
##########
@@ -0,0 +1,533 @@
+/*
+ * 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.Assert;
+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 final ExecutorService executorService = Executors.newFixedThreadPool(2);
Review comment:
should move into the test method or have the ownership be junit BeforeClass/AfterClass methods (tests no longer clean up).
##########
File path: test/distributed/org/apache/cassandra/distributed/test/NetstatsBootstrapWithEntireSSTablesCompressionStreamingTest.java
##########
@@ -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
Review comment:
does this pass if you run the class? executeTest relies on the shared executor pool and executeTest shuts it down after the first test runs.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
[GitHub] [cassandra] smiklosovic commented on a change in pull request #711: CASSANDRA-15406 - added progress of streaming in percents in netstats
Posted by GitBox <gi...@apache.org>.
smiklosovic commented on a change in pull request #711:
URL: https://github.com/apache/cassandra/pull/711#discussion_r474859681
##########
File path: test/distributed/org/apache/cassandra/distributed/test/NetstatsBootstrapWithEntireSSTablesCompressionStreamingTest.java
##########
@@ -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
Review comment:
yes it does pass, maybe just a glitch, gonna fix that
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
[GitHub] [cassandra] dcapwell commented on a change in pull request #711: CASSANDRA-15406 - added progress of streaming in percents in netstats
Posted by GitBox <gi...@apache.org>.
dcapwell commented on a change in pull request #711:
URL: https://github.com/apache/cassandra/pull/711#discussion_r472516969
##########
File path: src/java/org/apache/cassandra/tools/nodetool/NetStats.java
##########
@@ -98,35 +136,35 @@ public void execute(NodeProbe probe)
long dropped;
pending = 0;
- for (int n : ms.getLargeMessagePendingTasks().values())
+ for (int n : ms.getLargeMessagePendingTasksWithPort().values())
Review comment:
note: looks related to https://issues.apache.org/jira/browse/CASSANDRA-15937
##########
File path: src/java/org/apache/cassandra/streaming/SessionInfo.java
##########
@@ -149,6 +149,14 @@ public long getTotalSizeToReceive()
return getTotalSizes(receivingSummaries);
}
+ public long getTotalBytesToReceive()
Review comment:
Can you add a java doc and put in the doc why this method != `getTotalSizeToReceive`?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
[GitHub] [cassandra] dcapwell commented on a change in pull request #711: CASSANDRA-15406 - added progress of streaming in percents in netstats
Posted by GitBox <gi...@apache.org>.
dcapwell commented on a change in pull request #711:
URL: https://github.com/apache/cassandra/pull/711#discussion_r482652463
##########
File path: src/java/org/apache/cassandra/db/streaming/CassandraCompressedStreamReader.java
##########
@@ -97,7 +97,7 @@ public SSTableMultiWriter read(DataInputPlus inputPlus) throws IOException
{
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);
Review comment:
When writer is `org.apache.cassandra.io.sstable.format.RangeAwareSSTableWriter` getFileName is
```
return String.join("/", cfs.keyspace.getName(), cfs.getTableName());
```
else it is
```
return descriptor.filenameFor(Component.DATA);
```
But when entire file streaming is used, then the reader is `CassandraEntireSSTableStreamReader` which uses `session.progress(writer.descriptor.filenameFor(component), ProgressInfo.Direction.IN, length, length);`
So when `RangeAwareSSTableWriter` is used we can't use a better name as it writes to different sstables
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
[GitHub] [cassandra] smiklosovic commented on a change in pull request #711: CASSANDRA-15406 - added progress of streaming in percents in netstats
Posted by GitBox <gi...@apache.org>.
smiklosovic commented on a change in pull request #711:
URL: https://github.com/apache/cassandra/pull/711#discussion_r471739268
##########
File path: test/distributed/org/apache/cassandra/distributed/mock/nodetool/NetstatsTest.java
##########
@@ -0,0 +1,607 @@
+/*
+ * 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.mock.nodetool;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+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.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Stream;
+
+import org.junit.Assert;
+import org.junit.Test;
+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.api.IIsolatedExecutor;
+import org.apache.cassandra.distributed.impl.InstanceAwareNodetool;
+import org.apache.cassandra.distributed.shared.AbstractBuilder;
+import org.apache.cassandra.distributed.shared.NetworkTopology;
+import org.apache.cassandra.distributed.test.TestBaseImpl;
+import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.utils.Pair;
+
+import static java.util.stream.Collectors.toList;
+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 NetstatsTest extends TestBaseImpl
+{
+
+ private static final Logger logger = LoggerFactory.getLogger(NetstatsTest.class);
+
+ @Test
+ public void testNetstatsStreamingProgress() throws Exception
+ {
+ final ExecutorService executorService = Executors.newFixedThreadPool(2);
+
+ try (final InstanceAwareNodetool nodetool = new InstanceAwareNodetool(getCluster()))
+ {
+ final Cluster cluster = nodetool.getCluster();
+ final IInvokableInstance node1 = cluster.get(1);
+ final IInvokableInstance node2 = cluster.get(2);
+
+ createTable(cluster);
+
+ node1.flush("netstats_test");
+ node2.flush("netstats_test");
+
+ disableCompaction(node1);
+ disableCompaction(node2);
+
+ populateData();
+
+ // 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(nodetool, node1));
+ final Future<NetstatResults> resultsFuture2 = executorService.submit(new NetstatsCallable(nodetool, node2));
+
+ final IIsolatedExecutor.CallableNoExcept<?> repairCallable = node1.asyncCallsOnInstance(() -> {
+ try
+ {
+ StorageService.instance.repair("netstats_test", Collections.emptyMap(), Collections.emptyList()).right.get();
+
+ return null;
+ }
+ catch (final Exception ex)
+ {
+ throw new RuntimeException(ex);
+ }
+ });
+
+ repairCallable.call();
+
+ logger.info("Waiting for netstats callables to stop ...");
+
+ final NetstatResults results1 = resultsFuture1.get(1, TimeUnit.MINUTES);
+ final NetstatResults results2 = resultsFuture2.get(1, TimeUnit.MINUTES);
+
+ results1.assertSuccessful();
+ results2.assertSuccessful();
+
+ NetstatsOutputParser.validate(NetstatsOutputParser.parse(results1));
+ NetstatsOutputParser.validate(NetstatsOutputParser.parse(results2));
+ }
+ catch (final Throwable t)
+ {
+ t.printStackTrace();
+ Assert.fail("Test has failed: " + t.getMessage());
+ }
+ finally
+ {
+ logger.info("Shutting down executor service");
+
+ executorService.shutdownNow();
+
+ if (!executorService.isShutdown())
+ {
+ if (!executorService.awaitTermination(1, TimeUnit.MINUTES))
+ {
+ throw new IllegalStateException("Unable to shutdown executor for invoking netstat commands.");
+ }
+ }
+ }
+ }
+
+ // helpers
+
+ private static class NetstatsOutputParser
+ {
+ public static List<Pair<ReceivingStastistics, SendingStatistics>> parse(final NetstatResults results)
+ {
+ final Set<String> outputs = new LinkedHashSet<>();
+
+ results.netstatOutputs.stream()
+ .map(result -> result[0])
+ .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("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++)
+ {
+ Assert.assertTrue(sendingStatistics.get(i).sendingHeader.compareTo(sendingStatistics.get(i + 1).sendingHeader) <= 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).orElseGet(() -> 0L) / (double) sending.sendingHeader.bytesTotal;
+
+ Assert.assertEquals((int) sending.sendingHeader.progressBytes, (int) (progress * 100));
+
+ Assert.assertTrue((double) sending.sendingHeader.bytesTotal > (double) sending.sendingSSTable.stream().map(table -> table.bytesInTotal).reduce(Long::sum).orElseGet(() -> 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).orElseGet(() -> 0L));
+ Assert.assertEquals(receiving.receivingHeader.bytesTotalSoFar, (long) receiving.receivingTables.stream().map(table -> table.receivedSoFar).reduce(Long::sum).orElseGet(() -> 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 +
+ '}';
+ }
+ }
+ }
+ }
+
+ private static final class NetstatResults
+ {
+ private final List<String[]> netstatOutputs = new ArrayList<>();
+
+ public void add(String[] result)
+ {
+ netstatOutputs.add(result);
+ }
+
+ public void assertSuccessful()
+ {
+ for (final String[] result : netstatOutputs)
+ {
+ InstanceAwareNodetool.assertSuccessful(result);
+ }
+ }
+ }
+
+ private static class NetstatsCallable implements Callable<NetstatResults>
+ {
+ private final IInvokableInstance node;
+ private final InstanceAwareNodetool nodetool;
+
+ public NetstatsCallable(final InstanceAwareNodetool nodetool,
+ final IInvokableInstance node)
+ {
+ this.nodetool = nodetool;
+ this.node = node;
+ }
+
+ public NetstatResults call() throws Exception
+ {
+ final NetstatResults results = new NetstatResults();
+
+ boolean sawAnyStreamingOutput = false;
+
+ while (true)
+ {
+ String[] result = nodetool.execute(node, "netstats");
+
+ if (!sawAnyStreamingOutput)
+ {
+ if (result[0].contains("Receiving") || result[0].contains("Sending"))
+ {
+ sawAnyStreamingOutput = true;
+ }
+ }
+
+ if (sawAnyStreamingOutput && (!result[0].contains("Receiving") && !result[0].contains("Sending")))
+ {
+ break;
+ }
+
+ results.add(result);
+
+ Thread.currentThread().sleep(1000);
+ }
+
+ return results;
+ }
+ }
+
+ private 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};");
+ }
+ }
+
+ private void createTable(Cluster cluster)
+ {
+ // replication factor is 1
+ cluster.schemaChange("CREATE KEYSPACE netstats_test WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1};");
+ cluster.schemaChange("CREATE TABLE netstats_test.test_table (id uuid primary key);");
+ }
+
+ private void populateData()
+ {
+ 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"))
+ {
+
+ for (int i = 0; i < 100000; i++)
+ {
+ s.execute("INSERT INTO test_table (id) VALUES (" + UUID.randomUUID() + ")");
+ }
+ }
+ }
+
+ private void disableCompaction(IInvokableInstance invokableInstance)
+ {
+ invokableInstance.runOnInstance(() -> {
Review comment:
@dcapwell interesting enough, if I remember correctly, this has not worked. I dont know why but sstables still got compacted, or maybe it was instance.disableCompaction() or something like that ... it just didnt work
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
[GitHub] [cassandra] dcapwell commented on a change in pull request #711: CASSANDRA-15406 - added progress of streaming in percents in netstats
Posted by GitBox <gi...@apache.org>.
dcapwell commented on a change in pull request #711:
URL: https://github.com/apache/cassandra/pull/711#discussion_r482631860
##########
File path: src/java/org/apache/cassandra/db/streaming/CassandraStreamHeader.java
##########
@@ -86,6 +86,10 @@ private CassandraStreamHeader(Builder builder)
isEntireSSTable = builder.isEntireSSTable;
componentManifest = builder.componentManifest;
firstKey = builder.firstKey;
+
+ if (compressionMetadata != null && compressionInfo == null)
Review comment:
we should go with the approach in https://issues.apache.org/jira/browse/CASSANDRA-15861 that keeps this lazy and fixes `calculateSize` to deal with this case
##########
File path: src/java/org/apache/cassandra/db/streaming/CassandraOutgoingFile.java
##########
@@ -166,17 +166,17 @@ public void write(StreamSession session, DataOutputStreamPlus out, int version)
CassandraStreamHeader.serializer.serialize(header, out, version);
out.flush();
- if (shouldStreamEntireSSTable && out instanceof AsyncStreamingOutputPlus)
Review comment:
should keep `out instanceof AsyncStreamingOutputPlus` for now.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
[GitHub] [cassandra] dcapwell commented on pull request #711: CASSANDRA-15406 - added progress of streaming in percents in netstats
Posted by GitBox <gi...@apache.org>.
dcapwell commented on pull request #711:
URL: https://github.com/apache/cassandra/pull/711#issuecomment-678403569
Did a pass, but need to look closer at the writer changes to see if they are safe.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
[GitHub] [cassandra] smiklosovic commented on a change in pull request #711: CASSANDRA-15406 - added progress of streaming in percents in netstats
Posted by GitBox <gi...@apache.org>.
smiklosovic commented on a change in pull request #711:
URL: https://github.com/apache/cassandra/pull/711#discussion_r466594306
##########
File path: src/java/org/apache/cassandra/streaming/ProgressInfo.java
##########
@@ -110,7 +110,7 @@ public String toString(boolean withPorts)
{
StringBuilder sb = new StringBuilder(fileName);
sb.append(" ").append(currentBytes);
- sb.append("/").append(totalBytes).append(" bytes");
+ sb.append("/").append(totalBytes).append(" bytes ");
Review comment:
looks like this
```
netstats_test/test_table 528363/528363 bytes (100%) received from idx:0/127.0.0.1
```
instead of
```
netstats_test/test_table 528363/528363 bytes(100%) received from idx:0/127.0.0.1
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
[GitHub] [cassandra] smiklosovic commented on a change in pull request #711: CASSANDRA-15406 - added progress of streaming in percents in netstats
Posted by GitBox <gi...@apache.org>.
smiklosovic commented on a change in pull request #711:
URL: https://github.com/apache/cassandra/pull/711#discussion_r466594142
##########
File path: src/java/org/apache/cassandra/db/streaming/CassandraCompressedStreamReader.java
##########
@@ -97,7 +97,7 @@ public SSTableMultiWriter read(DataInputPlus inputPlus) throws IOException
{
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);
Review comment:
This is needed here because without it, the output in netstats look like this, which is wrong:
```
Repair ec3d7e50-d667-11ea-b3d0-8f5940ba25d7
/127.0.0.1
Receiving 3 files, 1890453 bytes total. Already received 2 files (66.67%), 528363 bytes total (53.09%)
netstats_test/test_table 528363/528363 bytes (100%) received from idx:0/127.0.0.1
```
New output looks like this:
```
Repair ec3d7e50-d667-11ea-b3d0-8f5940ba25d7
/127.0.0.1
Receiving 3 files, 1890453 bytes total. Already received 2 files (66.67%), 1003673 bytes total (53.09%)
netstats_test/test_table-1 528363/528363 bytes (100%) received from idx:0/127.0.0.1
netstats_test/test_table-0 221727/221727 bytes (100%) received from idx:0/127.0.0.1
netstats_test/test_table-2 253583/529137 bytes (47%) received from idx:0/127.0.0.1
```
The problem with the original code is that `netstats_test/test_table` is periodically put into `SessionInfo#updateProgress` but since we are putting there same "file name", it is replaced so there is ever just one entry in that map.
"file name" is not completely clear here, it is "logical file name" because it is not fully known yet where that will be be actually saved, the original "file name" does not make sense anyway ... I am appeding the sequence number just for the sake of having it unique.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
[GitHub] [cassandra] dcapwell commented on a change in pull request #711: CASSANDRA-15406 - added progress of streaming in percents in netstats
Posted by GitBox <gi...@apache.org>.
dcapwell commented on a change in pull request #711:
URL: https://github.com/apache/cassandra/pull/711#discussion_r471722112
##########
File path: test/distributed/org/apache/cassandra/distributed/mock/nodetool/NetstatsTest.java
##########
@@ -0,0 +1,607 @@
+/*
+ * 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.mock.nodetool;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+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.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Stream;
+
+import org.junit.Assert;
+import org.junit.Test;
+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.api.IIsolatedExecutor;
+import org.apache.cassandra.distributed.impl.InstanceAwareNodetool;
+import org.apache.cassandra.distributed.shared.AbstractBuilder;
+import org.apache.cassandra.distributed.shared.NetworkTopology;
+import org.apache.cassandra.distributed.test.TestBaseImpl;
+import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.utils.Pair;
+
+import static java.util.stream.Collectors.toList;
+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 NetstatsTest extends TestBaseImpl
+{
+
+ private static final Logger logger = LoggerFactory.getLogger(NetstatsTest.class);
+
+ @Test
+ public void testNetstatsStreamingProgress() throws Exception
+ {
+ final ExecutorService executorService = Executors.newFixedThreadPool(2);
+
+ try (final InstanceAwareNodetool nodetool = new InstanceAwareNodetool(getCluster()))
+ {
+ final Cluster cluster = nodetool.getCluster();
+ final IInvokableInstance node1 = cluster.get(1);
+ final IInvokableInstance node2 = cluster.get(2);
+
+ createTable(cluster);
+
+ node1.flush("netstats_test");
+ node2.flush("netstats_test");
+
+ disableCompaction(node1);
+ disableCompaction(node2);
+
+ populateData();
+
+ // 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(nodetool, node1));
+ final Future<NetstatResults> resultsFuture2 = executorService.submit(new NetstatsCallable(nodetool, node2));
+
+ final IIsolatedExecutor.CallableNoExcept<?> repairCallable = node1.asyncCallsOnInstance(() -> {
+ try
+ {
+ StorageService.instance.repair("netstats_test", Collections.emptyMap(), Collections.emptyList()).right.get();
+
+ return null;
+ }
+ catch (final Exception ex)
+ {
+ throw new RuntimeException(ex);
+ }
+ });
+
+ repairCallable.call();
Review comment:
can simplify and just do
```
node1.asyncCallsOnInstance(() -> { ... }).call();
```
Looks like the API is missing asyncCallOnInstance which would do that for you... =(
##########
File path: test/distributed/org/apache/cassandra/distributed/impl/InstanceAwareNodetool.java
##########
@@ -0,0 +1,264 @@
+/*
+ * 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.impl;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Assert;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.api.IIsolatedExecutor;
+import org.apache.cassandra.distributed.mock.nodetool.InternalNodeProbeFactory;
+import org.apache.cassandra.distributed.shared.AbstractBuilder;
+import org.apache.cassandra.tools.INodeProbeFactory;
+import org.apache.cassandra.tools.NodeTool;
+import org.apache.cassandra.tools.ToolRunner;
+
+/**
+ * Handy class for invoking nodetool against an instance, the difference between this
+ * and nodetool method invocation on IInstance is that we can have an output from this one which is not
+ * possible on IInstance nodetool invocation.
+ * <p>
+ * This is the gateway for testing the output of nodetool command and we can do various asserts on that, e.g.
+ * if the output is of some format and that is not broken (even accidentally).
+ * <p>
+ * The actual usage is like:
+ *
+ * <pre>
+ * {@code
+ * public class GossipinfoTest extends TestBaseImpl
+ * {
+ * @Test
+ * public void testGossipinfo() throws Exception
+ * {
+ * try (final InstanceAwareNodetool nodetool = new InstanceAwareNodetool(getCluster()))
+ * {
+ * String[] results = nodetool.execute("gossipinfo");
+ * }
+ * }
+ * }
+ * }</pre>
+ */
+public class InstanceAwareNodetool implements IIsolatedExecutor.SerializableCallable<String[]>, AutoCloseable
Review comment:
mixed feelings about this class. For one its practical and solves the problem at hand, but on the other hand it doesn't compose with the existing nodetool testing feature...
I would love to see org.apache.cassandra.distributed.api.NodeToolResult updated with stdout/stderr. Looking closer at `NodetoolWrapper` since it will just run in-process and this does basically the same as the existing logic, the stdout/err logs will have logs from the daemon as well, so simply reading stdout/err should be enough.
I feel like this class could go away in favor of the following
```
class NodeToolResultsWithOutput {
private final NodeToolResult result;
private final ByteArrayOutputStream stdout;
private final ByteArrayOutputStream stderr;
public NodeToolResultsWithOutput(NodeToolResult result, ByteArrayOutputStream stdout, ByteArrayOutputStream stderr) {
this.result = result;
this.stdout = stdout;
this.stderr = stderr;
}
public NodeToolResult getResult() { return result; }
public String getStdout() { return stdout.toString(); }
public String getStderr() { return stderr.toString(); }
}
public static NodeToolResultsWithOutput nodetool(IInvokeableInstanc inst, boolean withNotifications, String... commandAndArgs) {
return inst.callOnInstance(() -> {
PrintStream originalSysOut = System.out;
PrintStream originalSysErr = System.err;
InputStream originalSysIn = System.in;
originalSysOut.flush();
originalSysErr.flush();
ByteArrayOutputStream toolOut = new ByteArrayOutputStream();
ByteArrayOutputStream toolErr = new ByteArrayOutputStream();
System.setIn(stdin == null ? originalSysIn : stdin);
try (PrintStream newOut = new PrintStream(toolOut); PrintStream newErr = new PrintStream(toolErr);)
{
System.setOut(newOut);
System.setErr(newErr);
DTestNodeTool nodetool = new DTestNodeTool(withNotifications);
int rc = nodetool.execute(commandAndArgs);
NodeToolResult results = NodeToolResult(commandAndArgs, rc, new ArrayList<>(nodetool.notifications.notifications), nodetool.latestError);
return new NodeToolResultsWithOutput(results, toolOut, toolErr);
} finally {
System.setOut(originalSysOut);
System.setErr(originalSysErr);
System.setIn(originalSysIn);
}
});
}
```
TL;DR - long term I feel this logic should be pulled into instance.nodetoolWithResults, short term can we do something like the above so the logic can be used in other places?
##########
File path: test/distributed/org/apache/cassandra/distributed/mock/nodetool/NetstatsTest.java
##########
@@ -0,0 +1,607 @@
+/*
+ * 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.mock.nodetool;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+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.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Stream;
+
+import org.junit.Assert;
+import org.junit.Test;
+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.api.IIsolatedExecutor;
+import org.apache.cassandra.distributed.impl.InstanceAwareNodetool;
+import org.apache.cassandra.distributed.shared.AbstractBuilder;
+import org.apache.cassandra.distributed.shared.NetworkTopology;
+import org.apache.cassandra.distributed.test.TestBaseImpl;
+import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.utils.Pair;
+
+import static java.util.stream.Collectors.toList;
+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 NetstatsTest extends TestBaseImpl
+{
+
+ private static final Logger logger = LoggerFactory.getLogger(NetstatsTest.class);
+
+ @Test
+ public void testNetstatsStreamingProgress() throws Exception
+ {
+ final ExecutorService executorService = Executors.newFixedThreadPool(2);
+
+ try (final InstanceAwareNodetool nodetool = new InstanceAwareNodetool(getCluster()))
+ {
+ final Cluster cluster = nodetool.getCluster();
+ final IInvokableInstance node1 = cluster.get(1);
+ final IInvokableInstance node2 = cluster.get(2);
+
+ createTable(cluster);
+
+ node1.flush("netstats_test");
+ node2.flush("netstats_test");
+
+ disableCompaction(node1);
+ disableCompaction(node2);
+
+ populateData();
+
+ // 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(nodetool, node1));
+ final Future<NetstatResults> resultsFuture2 = executorService.submit(new NetstatsCallable(nodetool, node2));
+
+ final IIsolatedExecutor.CallableNoExcept<?> repairCallable = node1.asyncCallsOnInstance(() -> {
+ try
+ {
+ StorageService.instance.repair("netstats_test", Collections.emptyMap(), Collections.emptyList()).right.get();
+
+ return null;
+ }
+ catch (final Exception ex)
+ {
+ throw new RuntimeException(ex);
+ }
+ });
+
+ repairCallable.call();
+
+ logger.info("Waiting for netstats callables to stop ...");
+
+ final NetstatResults results1 = resultsFuture1.get(1, TimeUnit.MINUTES);
+ final NetstatResults results2 = resultsFuture2.get(1, TimeUnit.MINUTES);
+
+ results1.assertSuccessful();
+ results2.assertSuccessful();
+
+ NetstatsOutputParser.validate(NetstatsOutputParser.parse(results1));
+ NetstatsOutputParser.validate(NetstatsOutputParser.parse(results2));
+ }
+ catch (final Throwable t)
+ {
+ t.printStackTrace();
+ Assert.fail("Test has failed: " + t.getMessage());
+ }
+ finally
+ {
+ logger.info("Shutting down executor service");
+
+ executorService.shutdownNow();
+
+ if (!executorService.isShutdown())
+ {
+ if (!executorService.awaitTermination(1, TimeUnit.MINUTES))
+ {
+ throw new IllegalStateException("Unable to shutdown executor for invoking netstat commands.");
+ }
+ }
+ }
+ }
+
+ // helpers
+
+ private static class NetstatsOutputParser
+ {
+ public static List<Pair<ReceivingStastistics, SendingStatistics>> parse(final NetstatResults results)
+ {
+ final Set<String> outputs = new LinkedHashSet<>();
+
+ results.netstatOutputs.stream()
+ .map(result -> result[0])
+ .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("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++)
+ {
+ Assert.assertTrue(sendingStatistics.get(i).sendingHeader.compareTo(sendingStatistics.get(i + 1).sendingHeader) <= 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).orElseGet(() -> 0L) / (double) sending.sendingHeader.bytesTotal;
+
+ Assert.assertEquals((int) sending.sendingHeader.progressBytes, (int) (progress * 100));
+
+ Assert.assertTrue((double) sending.sendingHeader.bytesTotal > (double) sending.sendingSSTable.stream().map(table -> table.bytesInTotal).reduce(Long::sum).orElseGet(() -> 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).orElseGet(() -> 0L));
+ Assert.assertEquals(receiving.receivingHeader.bytesTotalSoFar, (long) receiving.receivingTables.stream().map(table -> table.receivedSoFar).reduce(Long::sum).orElseGet(() -> 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 +
+ '}';
+ }
+ }
+ }
+ }
+
+ private static final class NetstatResults
+ {
+ private final List<String[]> netstatOutputs = new ArrayList<>();
+
+ public void add(String[] result)
+ {
+ netstatOutputs.add(result);
+ }
+
+ public void assertSuccessful()
+ {
+ for (final String[] result : netstatOutputs)
+ {
+ InstanceAwareNodetool.assertSuccessful(result);
+ }
+ }
+ }
+
+ private static class NetstatsCallable implements Callable<NetstatResults>
+ {
+ private final IInvokableInstance node;
+ private final InstanceAwareNodetool nodetool;
+
+ public NetstatsCallable(final InstanceAwareNodetool nodetool,
+ final IInvokableInstance node)
+ {
+ this.nodetool = nodetool;
+ this.node = node;
+ }
+
+ public NetstatResults call() throws Exception
+ {
+ final NetstatResults results = new NetstatResults();
+
+ boolean sawAnyStreamingOutput = false;
+
+ while (true)
+ {
+ String[] result = nodetool.execute(node, "netstats");
+
+ if (!sawAnyStreamingOutput)
+ {
+ if (result[0].contains("Receiving") || result[0].contains("Sending"))
+ {
+ sawAnyStreamingOutput = true;
+ }
+ }
+
+ if (sawAnyStreamingOutput && (!result[0].contains("Receiving") && !result[0].contains("Sending")))
+ {
+ break;
+ }
+
+ results.add(result);
+
+ Thread.currentThread().sleep(1000);
+ }
+
+ return results;
+ }
+ }
+
+ private 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};");
+ }
+ }
+
+ private void createTable(Cluster cluster)
+ {
+ // replication factor is 1
+ cluster.schemaChange("CREATE KEYSPACE netstats_test WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1};");
+ cluster.schemaChange("CREATE TABLE netstats_test.test_table (id uuid primary key);");
+ }
+
+ private void populateData()
+ {
+ 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"))
+ {
+
+ for (int i = 0; i < 100000; i++)
+ {
+ s.execute("INSERT INTO test_table (id) VALUES (" + UUID.randomUUID() + ")");
+ }
+ }
+ }
+
+ private void disableCompaction(IInvokableInstance invokableInstance)
+ {
+ invokableInstance.runOnInstance(() -> {
Review comment:
can do
```
invokableInstance.nodetoolResult("disableautocompaction", "netstats_test").asserts().success()
```
##########
File path: test/distributed/org/apache/cassandra/distributed/mock/nodetool/NetstatsTest.java
##########
@@ -0,0 +1,607 @@
+/*
+ * 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.mock.nodetool;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+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.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Stream;
+
+import org.junit.Assert;
+import org.junit.Test;
+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.api.IIsolatedExecutor;
+import org.apache.cassandra.distributed.impl.InstanceAwareNodetool;
+import org.apache.cassandra.distributed.shared.AbstractBuilder;
+import org.apache.cassandra.distributed.shared.NetworkTopology;
+import org.apache.cassandra.distributed.test.TestBaseImpl;
+import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.utils.Pair;
+
+import static java.util.stream.Collectors.toList;
+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 NetstatsTest extends TestBaseImpl
+{
+
+ private static final Logger logger = LoggerFactory.getLogger(NetstatsTest.class);
+
+ @Test
+ public void testNetstatsStreamingProgress() throws Exception
+ {
+ final ExecutorService executorService = Executors.newFixedThreadPool(2);
+
+ try (final InstanceAwareNodetool nodetool = new InstanceAwareNodetool(getCluster()))
+ {
+ final Cluster cluster = nodetool.getCluster();
+ final IInvokableInstance node1 = cluster.get(1);
+ final IInvokableInstance node2 = cluster.get(2);
+
+ createTable(cluster);
+
+ node1.flush("netstats_test");
+ node2.flush("netstats_test");
+
+ disableCompaction(node1);
+ disableCompaction(node2);
+
+ populateData();
+
+ // 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(nodetool, node1));
+ final Future<NetstatResults> resultsFuture2 = executorService.submit(new NetstatsCallable(nodetool, node2));
+
+ final IIsolatedExecutor.CallableNoExcept<?> repairCallable = node1.asyncCallsOnInstance(() -> {
+ try
+ {
+ StorageService.instance.repair("netstats_test", Collections.emptyMap(), Collections.emptyList()).right.get();
+
+ return null;
+ }
+ catch (final Exception ex)
+ {
+ throw new RuntimeException(ex);
+ }
+ });
+
+ repairCallable.call();
+
+ logger.info("Waiting for netstats callables to stop ...");
+
+ final NetstatResults results1 = resultsFuture1.get(1, TimeUnit.MINUTES);
+ final NetstatResults results2 = resultsFuture2.get(1, TimeUnit.MINUTES);
+
+ results1.assertSuccessful();
+ results2.assertSuccessful();
+
+ NetstatsOutputParser.validate(NetstatsOutputParser.parse(results1));
+ NetstatsOutputParser.validate(NetstatsOutputParser.parse(results2));
+ }
+ catch (final Throwable t)
+ {
+ t.printStackTrace();
+ Assert.fail("Test has failed: " + t.getMessage());
+ }
+ finally
+ {
+ logger.info("Shutting down executor service");
+
+ executorService.shutdownNow();
+
+ if (!executorService.isShutdown())
+ {
+ if (!executorService.awaitTermination(1, TimeUnit.MINUTES))
+ {
+ throw new IllegalStateException("Unable to shutdown executor for invoking netstat commands.");
+ }
+ }
+ }
+ }
+
+ // helpers
+
+ private static class NetstatsOutputParser
+ {
+ public static List<Pair<ReceivingStastistics, SendingStatistics>> parse(final NetstatResults results)
+ {
+ final Set<String> outputs = new LinkedHashSet<>();
+
+ results.netstatOutputs.stream()
+ .map(result -> result[0])
+ .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("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++)
+ {
+ Assert.assertTrue(sendingStatistics.get(i).sendingHeader.compareTo(sendingStatistics.get(i + 1).sendingHeader) <= 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).orElseGet(() -> 0L) / (double) sending.sendingHeader.bytesTotal;
+
+ Assert.assertEquals((int) sending.sendingHeader.progressBytes, (int) (progress * 100));
+
+ Assert.assertTrue((double) sending.sendingHeader.bytesTotal > (double) sending.sendingSSTable.stream().map(table -> table.bytesInTotal).reduce(Long::sum).orElseGet(() -> 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).orElseGet(() -> 0L));
+ Assert.assertEquals(receiving.receivingHeader.bytesTotalSoFar, (long) receiving.receivingTables.stream().map(table -> table.receivedSoFar).reduce(Long::sum).orElseGet(() -> 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 +
+ '}';
+ }
+ }
+ }
+ }
+
+ private static final class NetstatResults
+ {
+ private final List<String[]> netstatOutputs = new ArrayList<>();
+
+ public void add(String[] result)
+ {
+ netstatOutputs.add(result);
+ }
+
+ public void assertSuccessful()
+ {
+ for (final String[] result : netstatOutputs)
+ {
+ InstanceAwareNodetool.assertSuccessful(result);
+ }
+ }
+ }
+
+ private static class NetstatsCallable implements Callable<NetstatResults>
+ {
+ private final IInvokableInstance node;
+ private final InstanceAwareNodetool nodetool;
+
+ public NetstatsCallable(final InstanceAwareNodetool nodetool,
+ final IInvokableInstance node)
+ {
+ this.nodetool = nodetool;
+ this.node = node;
+ }
+
+ public NetstatResults call() throws Exception
Review comment:
FYI there is a class `org.apache.cassandra.utils.Retry` which can do retries for you, the main benefit is it will gradually backoff rather than retry every 1s.
In this case I am ok with this, just giving as an FYI
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
[GitHub] [cassandra] smiklosovic commented on a change in pull request #711: CASSANDRA-15406 - added progress of streaming in percents in netstats
Posted by GitBox <gi...@apache.org>.
smiklosovic commented on a change in pull request #711:
URL: https://github.com/apache/cassandra/pull/711#discussion_r471760739
##########
File path: test/distributed/org/apache/cassandra/distributed/impl/InstanceAwareNodetool.java
##########
@@ -0,0 +1,264 @@
+/*
+ * 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.impl;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Assert;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.api.IIsolatedExecutor;
+import org.apache.cassandra.distributed.mock.nodetool.InternalNodeProbeFactory;
+import org.apache.cassandra.distributed.shared.AbstractBuilder;
+import org.apache.cassandra.tools.INodeProbeFactory;
+import org.apache.cassandra.tools.NodeTool;
+import org.apache.cassandra.tools.ToolRunner;
+
+/**
+ * Handy class for invoking nodetool against an instance, the difference between this
+ * and nodetool method invocation on IInstance is that we can have an output from this one which is not
+ * possible on IInstance nodetool invocation.
+ * <p>
+ * This is the gateway for testing the output of nodetool command and we can do various asserts on that, e.g.
+ * if the output is of some format and that is not broken (even accidentally).
+ * <p>
+ * The actual usage is like:
+ *
+ * <pre>
+ * {@code
+ * public class GossipinfoTest extends TestBaseImpl
+ * {
+ * @Test
+ * public void testGossipinfo() throws Exception
+ * {
+ * try (final InstanceAwareNodetool nodetool = new InstanceAwareNodetool(getCluster()))
+ * {
+ * String[] results = nodetool.execute("gossipinfo");
+ * }
+ * }
+ * }
+ * }</pre>
+ */
+public class InstanceAwareNodetool implements IIsolatedExecutor.SerializableCallable<String[]>, AutoCloseable
Review comment:
@dcapwell the amount of changes in order to rewrite it like you want is just not worth it. I would need to abstract this into some kind of wrapper which would call tool repeatedly, gather results etc, it is just additional work to rewrite it just for ... what exactly?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
[GitHub] [cassandra] smiklosovic closed pull request #711: CASSANDRA-15406 - added progress of streaming in percents in netstats
Posted by GitBox <gi...@apache.org>.
smiklosovic closed pull request #711:
URL: https://github.com/apache/cassandra/pull/711
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
[GitHub] [cassandra] dcapwell commented on a change in pull request #711: CASSANDRA-15406 - added progress of streaming in percents in netstats
Posted by GitBox <gi...@apache.org>.
dcapwell commented on a change in pull request #711:
URL: https://github.com/apache/cassandra/pull/711#discussion_r472361071
##########
File path: test/distributed/org/apache/cassandra/distributed/impl/InstanceAwareNodetool.java
##########
@@ -0,0 +1,264 @@
+/*
+ * 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.impl;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Assert;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.api.IIsolatedExecutor;
+import org.apache.cassandra.distributed.mock.nodetool.InternalNodeProbeFactory;
+import org.apache.cassandra.distributed.shared.AbstractBuilder;
+import org.apache.cassandra.tools.INodeProbeFactory;
+import org.apache.cassandra.tools.NodeTool;
+import org.apache.cassandra.tools.ToolRunner;
+
+/**
+ * Handy class for invoking nodetool against an instance, the difference between this
+ * and nodetool method invocation on IInstance is that we can have an output from this one which is not
+ * possible on IInstance nodetool invocation.
+ * <p>
+ * This is the gateway for testing the output of nodetool command and we can do various asserts on that, e.g.
+ * if the output is of some format and that is not broken (even accidentally).
+ * <p>
+ * The actual usage is like:
+ *
+ * <pre>
+ * {@code
+ * public class GossipinfoTest extends TestBaseImpl
+ * {
+ * @Test
+ * public void testGossipinfo() throws Exception
+ * {
+ * try (final InstanceAwareNodetool nodetool = new InstanceAwareNodetool(getCluster()))
+ * {
+ * String[] results = nodetool.execute("gossipinfo");
+ * }
+ * }
+ * }
+ * }</pre>
+ */
+public class InstanceAwareNodetool implements IIsolatedExecutor.SerializableCallable<String[]>, AutoCloseable
Review comment:
spoke in slack, gave what I was asking for and tested it works.
The issue I see here is that
1) doesn't handle notifications, so any test which wants logs + notifications can't use
2) returns String[] which isn't clear how to use without reading the code; returning a structured object solves this
3) close closes cluster, but multiple nodetool commands are common.
4) half the methods duplicate NodetoolResult, so can remove if returning that.
5) timeout doesn't do anything since you are using a class for ToolRunner, so it blocks waiting for it to exit, so will never timeout
> it is just additional work to rewrite it just for ... what exactly?
nodetool stdout is a common need, so making this accessible to other use cases is a big win. Also, what I was asking for is smaller than this (27 lines of code (plus 33 for the pojo) vs 264 lines.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
[GitHub] [cassandra] dcapwell commented on a change in pull request #711: CASSANDRA-15406 - added progress of streaming in percents in netstats
Posted by GitBox <gi...@apache.org>.
dcapwell commented on a change in pull request #711:
URL: https://github.com/apache/cassandra/pull/711#discussion_r472502992
##########
File path: test/distributed/org/apache/cassandra/distributed/mock/nodetool/NetstatsTest.java
##########
@@ -0,0 +1,599 @@
+/*
+ * 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.mock.nodetool;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+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.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Stream;
+
+import org.junit.Assert;
+import org.junit.Test;
+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.AbstractBuilder;
+import org.apache.cassandra.distributed.shared.NetworkTopology;
+import org.apache.cassandra.distributed.shared.NodeToolResultWithOutput;
+import org.apache.cassandra.distributed.test.TestBaseImpl;
+import org.apache.cassandra.distributed.util.NodetoolUtils;
+import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.utils.Pair;
+
+import static java.util.stream.Collectors.toList;
+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 NetstatsTest extends TestBaseImpl
+{
+
+ private static final Logger logger = LoggerFactory.getLogger(NetstatsTest.class);
+
+ @Test
+ public void testNetstatsStreamingProgress() throws Exception
+ {
+ final ExecutorService executorService = Executors.newFixedThreadPool(2);
+
+ try (final Cluster cluster = getCluster().start())
+ {
+ final IInvokableInstance node1 = cluster.get(1);
+ final IInvokableInstance node2 = cluster.get(2);
+
+ createTable(cluster);
+
+ node1.flush("netstats_test");
+ node2.flush("netstats_test");
+
+ disableCompaction(node1);
+ disableCompaction(node2);
+
+ populateData();
+
+ //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.asyncCallsOnInstance(() -> {
Review comment:
should replace with
```
node1.nodetoolResult("repair", "netstats_test").asserts().success();
```
This will block the main thread but removes the timeout from the result future if the repair takes longer than 1m.
##########
File path: test/distributed/org/apache/cassandra/distributed/mock/nodetool/NetstatsTest.java
##########
@@ -0,0 +1,599 @@
+/*
+ * 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.mock.nodetool;
Review comment:
can you move the class from `org.apache.cassandra.distributed.mock.nodetool.NetstatsTest` to `org.apache.cassandra.distributed.test.NetstatsTest`
##########
File path: test/distributed/org/apache/cassandra/distributed/mock/nodetool/NetstatsTest.java
##########
@@ -0,0 +1,599 @@
+/*
+ * 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.mock.nodetool;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+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.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Stream;
+
+import org.junit.Assert;
+import org.junit.Test;
+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.AbstractBuilder;
+import org.apache.cassandra.distributed.shared.NetworkTopology;
+import org.apache.cassandra.distributed.shared.NodeToolResultWithOutput;
+import org.apache.cassandra.distributed.test.TestBaseImpl;
+import org.apache.cassandra.distributed.util.NodetoolUtils;
+import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.utils.Pair;
+
+import static java.util.stream.Collectors.toList;
+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 NetstatsTest extends TestBaseImpl
+{
+
+ private static final Logger logger = LoggerFactory.getLogger(NetstatsTest.class);
+
+ @Test
+ public void testNetstatsStreamingProgress() throws Exception
+ {
+ final ExecutorService executorService = Executors.newFixedThreadPool(2);
+
+ try (final Cluster cluster = getCluster().start())
+ {
+ final IInvokableInstance node1 = cluster.get(1);
+ final IInvokableInstance node2 = cluster.get(2);
+
+ createTable(cluster);
+
+ node1.flush("netstats_test");
+ node2.flush("netstats_test");
+
+ disableCompaction(node1);
Review comment:
nit: better to do `node1.nodetoolResult("disableautocompaction", "netstats_test").asserts().success();` rather than exec into the instance.
##########
File path: test/distributed/org/apache/cassandra/distributed/mock/nodetool/NetstatsTest.java
##########
@@ -0,0 +1,599 @@
+/*
+ * 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.mock.nodetool;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+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.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Stream;
+
+import org.junit.Assert;
+import org.junit.Test;
+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.AbstractBuilder;
+import org.apache.cassandra.distributed.shared.NetworkTopology;
+import org.apache.cassandra.distributed.shared.NodeToolResultWithOutput;
+import org.apache.cassandra.distributed.test.TestBaseImpl;
+import org.apache.cassandra.distributed.util.NodetoolUtils;
+import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.utils.Pair;
+
+import static java.util.stream.Collectors.toList;
+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 NetstatsTest extends TestBaseImpl
+{
+
+ private static final Logger logger = LoggerFactory.getLogger(NetstatsTest.class);
+
+ @Test
+ public void testNetstatsStreamingProgress() throws Exception
+ {
+ final ExecutorService executorService = Executors.newFixedThreadPool(2);
+
+ try (final Cluster cluster = getCluster().start())
+ {
+ final IInvokableInstance node1 = cluster.get(1);
+ final IInvokableInstance node2 = cluster.get(2);
+
+ createTable(cluster);
+
+ node1.flush("netstats_test");
+ node2.flush("netstats_test");
+
+ disableCompaction(node1);
+ disableCompaction(node2);
+
+ populateData();
+
+ //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.asyncCallsOnInstance(() -> {
+ try
+ {
+ StorageService.instance.repair("netstats_test", Collections.emptyMap(), Collections.emptyList()).right.get();
+
+ return null;
+ }
+ catch (final Exception ex)
+ {
+ throw new RuntimeException(ex);
+ }
+ }).call();
+
+ logger.info("Waiting for netstats callables to stop ...");
+
+ final NetstatResults results = resultsFuture1.get(1, TimeUnit.MINUTES);
+
+ results.assertSuccessful();
+
+ NetstatsOutputParser.validate(NetstatsOutputParser.parse(results));
+ }
+ catch (final Throwable t)
+ {
+ t.printStackTrace();
+ Assert.fail("Test has failed: " + t.getMessage());
+ }
+ finally
+ {
+ logger.info("Shutting down executor service");
+
+ executorService.shutdownNow();
+
+ if (!executorService.isShutdown())
+ {
+ if (!executorService.awaitTermination(1, TimeUnit.MINUTES))
+ {
+ throw new IllegalStateException("Unable to shutdown executor for invoking netstat commands.");
+ }
+ }
+ }
+ }
+
+ // helpers
+
+ private 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++)
+ {
+ Assert.assertTrue(sendingStatistics.get(i).sendingHeader.compareTo(sendingStatistics.get(i + 1).sendingHeader) <= 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).orElseGet(() -> 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).orElseGet(() -> 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).orElseGet(() -> 0L));
+ Assert.assertEquals(receiving.receivingHeader.bytesTotalSoFar, (long) receiving.receivingTables.stream().map(table -> table.receivedSoFar).reduce(Long::sum).orElseGet(() -> 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 +
+ '}';
+ }
+ }
+ }
+ }
+
+ private 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());
+ }
+ }
+ }
+
+ private 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)
+ {
+ final NodeToolResultWithOutput result = NodetoolUtils.nodetool(node, false, "netstats");
+
+ System.out.println(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(1000);
+ }
+
+ return results;
+ }
+ }
+
+ private 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};");
+ }
+ }
+
+ private void createTable(Cluster cluster)
+ {
+ // replication factor is 1
+ cluster.schemaChange("CREATE KEYSPACE netstats_test WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1};");
+ cluster.schemaChange("CREATE TABLE netstats_test.test_table (id uuid primary key);");
+ }
+
+ private void populateData()
+ {
+ 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"))
+ {
+
+ for (int i = 0; i < 500000; i++)
Review comment:
this causes periodic timeouts as the test takes 5m30s and the timeout for the class is 6m. in testing I find that `100_000` (100k) is around 3m30s to 4m so should be safe in CI.
##########
File path: test/distributed/org/apache/cassandra/distributed/mock/nodetool/NetstatsTest.java
##########
@@ -0,0 +1,599 @@
+/*
+ * 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.mock.nodetool;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+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.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Stream;
+
+import org.junit.Assert;
+import org.junit.Test;
+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.AbstractBuilder;
+import org.apache.cassandra.distributed.shared.NetworkTopology;
+import org.apache.cassandra.distributed.shared.NodeToolResultWithOutput;
+import org.apache.cassandra.distributed.test.TestBaseImpl;
+import org.apache.cassandra.distributed.util.NodetoolUtils;
+import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.utils.Pair;
+
+import static java.util.stream.Collectors.toList;
+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 NetstatsTest extends TestBaseImpl
+{
+
+ private static final Logger logger = LoggerFactory.getLogger(NetstatsTest.class);
+
+ @Test
+ public void testNetstatsStreamingProgress() throws Exception
+ {
+ final ExecutorService executorService = Executors.newFixedThreadPool(2);
+
+ try (final Cluster cluster = getCluster().start())
+ {
+ final IInvokableInstance node1 = cluster.get(1);
+ final IInvokableInstance node2 = cluster.get(2);
+
+ createTable(cluster);
+
+ node1.flush("netstats_test");
+ node2.flush("netstats_test");
+
+ disableCompaction(node1);
+ disableCompaction(node2);
+
+ populateData();
+
+ //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.asyncCallsOnInstance(() -> {
+ try
+ {
+ StorageService.instance.repair("netstats_test", Collections.emptyMap(), Collections.emptyList()).right.get();
+
+ return null;
+ }
+ catch (final Exception ex)
+ {
+ throw new RuntimeException(ex);
+ }
+ }).call();
+
+ logger.info("Waiting for netstats callables to stop ...");
+
+ final NetstatResults results = resultsFuture1.get(1, TimeUnit.MINUTES);
+
+ results.assertSuccessful();
+
+ NetstatsOutputParser.validate(NetstatsOutputParser.parse(results));
+ }
+ catch (final Throwable t)
Review comment:
shouldn't catch, we get better error reports from unit if we pass through
##########
File path: test/distributed/org/apache/cassandra/distributed/mock/nodetool/NetstatsTest.java
##########
@@ -0,0 +1,599 @@
+/*
+ * 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.mock.nodetool;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+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.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Stream;
+
+import org.junit.Assert;
+import org.junit.Test;
+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.AbstractBuilder;
+import org.apache.cassandra.distributed.shared.NetworkTopology;
+import org.apache.cassandra.distributed.shared.NodeToolResultWithOutput;
+import org.apache.cassandra.distributed.test.TestBaseImpl;
+import org.apache.cassandra.distributed.util.NodetoolUtils;
+import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.utils.Pair;
+
+import static java.util.stream.Collectors.toList;
+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 NetstatsTest extends TestBaseImpl
+{
+
+ private static final Logger logger = LoggerFactory.getLogger(NetstatsTest.class);
+
+ @Test
+ public void testNetstatsStreamingProgress() throws Exception
+ {
+ final ExecutorService executorService = Executors.newFixedThreadPool(2);
+
+ try (final Cluster cluster = getCluster().start())
+ {
+ final IInvokableInstance node1 = cluster.get(1);
+ final IInvokableInstance node2 = cluster.get(2);
+
+ createTable(cluster);
+
+ node1.flush("netstats_test");
+ node2.flush("netstats_test");
+
+ disableCompaction(node1);
+ disableCompaction(node2);
+
+ populateData();
+
+ //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.asyncCallsOnInstance(() -> {
+ try
+ {
+ StorageService.instance.repair("netstats_test", Collections.emptyMap(), Collections.emptyList()).right.get();
+
+ return null;
+ }
+ catch (final Exception ex)
+ {
+ throw new RuntimeException(ex);
+ }
+ }).call();
+
+ logger.info("Waiting for netstats callables to stop ...");
+
+ final NetstatResults results = resultsFuture1.get(1, TimeUnit.MINUTES);
+
+ results.assertSuccessful();
+
+ NetstatsOutputParser.validate(NetstatsOutputParser.parse(results));
+ }
+ catch (final Throwable t)
+ {
+ t.printStackTrace();
+ Assert.fail("Test has failed: " + t.getMessage());
+ }
+ finally
+ {
+ logger.info("Shutting down executor service");
+
+ executorService.shutdownNow();
+
+ if (!executorService.isShutdown())
+ {
+ if (!executorService.awaitTermination(1, TimeUnit.MINUTES))
+ {
+ throw new IllegalStateException("Unable to shutdown executor for invoking netstat commands.");
+ }
+ }
+ }
+ }
+
+ // helpers
+
+ private 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++)
+ {
+ Assert.assertTrue(sendingStatistics.get(i).sendingHeader.compareTo(sendingStatistics.get(i + 1).sendingHeader) <= 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).orElseGet(() -> 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).orElseGet(() -> 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).orElseGet(() -> 0L));
+ Assert.assertEquals(receiving.receivingHeader.bytesTotalSoFar, (long) receiving.receivingTables.stream().map(table -> table.receivedSoFar).reduce(Long::sum).orElseGet(() -> 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 +
+ '}';
+ }
+ }
+ }
+ }
+
+ private 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());
+ }
+ }
+ }
+
+ private 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)
+ {
+ final NodeToolResultWithOutput result = NodetoolUtils.nodetool(node, false, "netstats");
+
+ System.out.println(result.getStdout());
Review comment:
should use `logger.info(result.getStdout());`
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org