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

Merge branch 'cassandra-4.0' into trunk

This commit is contained in:
Andrés de la Peña 2021-10-01 11:59:09 +01:00
parent 97108ef0a7 339e8b74bf
commit 0ccca8dab2
8 changed files with 353 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -47,6 +47,7 @@ Merged from 3.11:
* Update Jackson from 2.9.10 to 2.12.5 (CASSANDRA-16851)
* Make assassinate more resilient to missing tokens (CASSANDRA-16847)
Merged from 3.0:
* Immediately apply stream throughput, considering negative values as unthrottled (CASSANDRA-16959)
* Do not release new SSTables in offline transactions (CASSANDRA-16975)
* ArrayIndexOutOfBoundsException in FunctionResource#fromName (CASSANDRA-16977, CASSANDRA-16995)
* CVE-2015-0886 Security vulnerability in jbcrypt is addressed (CASSANDRA-9384)

8
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -1488,8 +1488,10 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
public void setStreamThroughputMbPerSec(int value)
{
int oldValue = DatabaseDescriptor.getStreamThroughputOutboundMegabitsPerSec();
DatabaseDescriptor.setStreamThroughputOutboundMegabitsPerSec(value);
logger.info("setstreamthroughput: throttle set to {}", value);
StreamManager.StreamRateLimiter.updateThroughput();
logger.info("setstreamthroughput: throttle set to {} Mb/s (was {} Mb/s)", value, oldValue);
}
public int getStreamThroughputMbPerSec()
@ -1499,8 +1501,10 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
public void setInterDCStreamThroughputMbPerSec(int value)
{
int oldValue = DatabaseDescriptor.getInterDCStreamThroughputOutboundMegabitsPerSec();
DatabaseDescriptor.setInterDCStreamThroughputOutboundMegabitsPerSec(value);
logger.info("setinterdcstreamthroughput: throttle set to {}", value);
StreamManager.StreamRateLimiter.updateInterDCThroughput();
logger.info("setinterdcstreamthroughput: throttle set to {} Mb/s (was {} Mb/s)", value, oldValue);
}
public int getInterDCStreamThroughputMbPerSec()

58
src/java/org/apache/cassandra/streaming/StreamManager.java
View File

@ -26,6 +26,7 @@ import javax.management.NotificationFilter;
import javax.management.NotificationListener;
import javax.management.openmbean.CompositeData;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Function;
import com.google.common.collect.Iterables;
import com.google.common.collect.Sets;
@ -62,19 +63,13 @@ public class StreamManager implements StreamManagerMBean
public static class StreamRateLimiter
{
private static final double BYTES_PER_MEGABIT = (1024 * 1024) / 8; // from bits
private static final RateLimiter limiter = RateLimiter.create(Double.MAX_VALUE);
private static final RateLimiter interDCLimiter = RateLimiter.create(Double.MAX_VALUE);
public static final double BYTES_PER_MEGABIT = (1024 * 1024) / 8; // from bits
private static final RateLimiter limiter = RateLimiter.create(calculateRateInBytes());
private static final RateLimiter interDCLimiter = RateLimiter.create(calculateInterDCRateInBytes());
private final boolean isLocalDC;
public StreamRateLimiter(InetAddressAndPort peer)
{
double throughput = DatabaseDescriptor.getStreamThroughputOutboundMegabitsPerSec() * BYTES_PER_MEGABIT;
mayUpdateThroughput(throughput, limiter);
double interDCThroughput = DatabaseDescriptor.getInterDCStreamThroughputOutboundMegabitsPerSec() * BYTES_PER_MEGABIT;
mayUpdateThroughput(interDCThroughput, interDCLimiter);
if (DatabaseDescriptor.getLocalDataCenter() != null && DatabaseDescriptor.getEndpointSnitch() != null)
isLocalDC = DatabaseDescriptor.getLocalDataCenter().equals(
DatabaseDescriptor.getEndpointSnitch().getDatacenter(peer));
@ -82,21 +77,48 @@ public class StreamManager implements StreamManagerMBean
isLocalDC = true;
}
private void mayUpdateThroughput(double limit, RateLimiter rateLimiter)
{
// if throughput is set to 0, throttling is disabled
if (limit == 0)
limit = Double.MAX_VALUE;
if (rateLimiter.getRate() != limit)
rateLimiter.setRate(limit);
}
public void acquire(int toTransfer)
{
limiter.acquire(toTransfer);
if (!isLocalDC)
interDCLimiter.acquire(toTransfer);
}
public static void updateThroughput()
{
limiter.setRate(calculateRateInBytes());
}
public static void updateInterDCThroughput()
{
interDCLimiter.setRate(calculateInterDCRateInBytes());
}
private static double calculateRateInBytes()
{
return DatabaseDescriptor.getStreamThroughputOutboundMegabitsPerSec() > 0
? DatabaseDescriptor.getStreamThroughputOutboundMegabitsPerSec() * BYTES_PER_MEGABIT
: Double.MAX_VALUE; // if throughput is set to 0 or negative value, throttling is disabled
}
private static double calculateInterDCRateInBytes()
{
return DatabaseDescriptor.getInterDCStreamThroughputOutboundMegabitsPerSec() > 0
? DatabaseDescriptor.getInterDCStreamThroughputOutboundMegabitsPerSec() * BYTES_PER_MEGABIT
: Double.MAX_VALUE; // if throughput is set to 0 or negative value, throttling is disabled
}
@VisibleForTesting
public static double getRateLimiterRateInBytes()
{
return limiter.getRate();
}
@VisibleForTesting
public static double getInterDCRateLimiterRateInBytes()
{
return interDCLimiter.getRate();
}
}
private final StreamEventJMXNotifier notifier = new StreamEventJMXNotifier();

3
src/java/org/apache/cassandra/tools/nodetool/SetInterDCStreamThroughput.java
View File

@ -26,8 +26,9 @@ import org.apache.cassandra.tools.NodeTool.NodeToolCmd;
@Command(name = "setinterdcstreamthroughput", description = "Set the Mb/s throughput cap for inter-datacenter streaming in the system, or 0 to disable throttling")
public class SetInterDCStreamThroughput extends NodeToolCmd
{
@SuppressWarnings("UnusedDeclaration")
@Arguments(title = "inter_dc_stream_throughput", usage = "<value_in_mb>", description = "Value in Mb, 0 to disable throttling", required = true)
private Integer interDCStreamThroughput = null;
private int interDCStreamThroughput;
@Override
public void execute(NodeProbe probe)

3
src/java/org/apache/cassandra/tools/nodetool/SetStreamThroughput.java
View File

@ -26,8 +26,9 @@ import org.apache.cassandra.tools.NodeTool.NodeToolCmd;
@Command(name = "setstreamthroughput", description = "Set the Mb/s throughput cap for streaming in the system, or 0 to disable throttling")
public class SetStreamThroughput extends NodeToolCmd
{
@SuppressWarnings("UnusedDeclaration")
@Arguments(title = "stream_throughput", usage = "<value_in_mb>", description = "Value in Mb, 0 to disable throttling", required = true)
private Integer streamThroughput = null;
private int streamThroughput;
@Override
public void execute(NodeProbe probe)

91
test/unit/org/apache/cassandra/streaming/StreamManagerTest.java Normal file
View File

@ -0,0 +1,91 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.streaming;
import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.service.StorageService;
import static org.apache.cassandra.streaming.StreamManager.StreamRateLimiter;
import static org.apache.cassandra.streaming.StreamManager.StreamRateLimiter.BYTES_PER_MEGABIT;
import static org.junit.Assert.assertEquals;
public class StreamManagerTest
{
private static int defaultStreamThroughputMbPerSec;
private static int defaultInterDCStreamThroughputMbPerSec;
@BeforeClass
public static void setupClass()
{
Config c = DatabaseDescriptor.loadConfig();
defaultStreamThroughputMbPerSec = c.stream_throughput_outbound_megabits_per_sec;
defaultInterDCStreamThroughputMbPerSec = c.inter_dc_stream_throughput_outbound_megabits_per_sec;
DatabaseDescriptor.daemonInitialization(() -> c);
}
@Test
public void testUpdateStreamThroughput()
{
// Initialized value check
assertEquals(defaultStreamThroughputMbPerSec * BYTES_PER_MEGABIT, StreamRateLimiter.getRateLimiterRateInBytes(), 0);
// Positive value check
StorageService.instance.setStreamThroughputMbPerSec(500);
assertEquals(500.0d * BYTES_PER_MEGABIT, StreamRateLimiter.getRateLimiterRateInBytes(), 0);
// Max positive value check
StorageService.instance.setStreamThroughputMbPerSec(Integer.MAX_VALUE);
assertEquals(Integer.MAX_VALUE * BYTES_PER_MEGABIT, StreamRateLimiter.getRateLimiterRateInBytes(), 0);
// Zero value check
StorageService.instance.setStreamThroughputMbPerSec(0);
assertEquals(Double.MAX_VALUE, StreamRateLimiter.getRateLimiterRateInBytes(), 0);
// Negative value check
StorageService.instance.setStreamThroughputMbPerSec(-200);
assertEquals(Double.MAX_VALUE, StreamRateLimiter.getRateLimiterRateInBytes(), 0);
}
@Test
public void testUpdateInterDCStreamThroughput()
{
// Initialized value check
assertEquals(defaultInterDCStreamThroughputMbPerSec * BYTES_PER_MEGABIT, StreamRateLimiter.getInterDCRateLimiterRateInBytes(), 0);
// Positive value check
StorageService.instance.setInterDCStreamThroughputMbPerSec(200);
assertEquals(200.0d * BYTES_PER_MEGABIT, StreamRateLimiter.getInterDCRateLimiterRateInBytes(), 0);
// Max positive value check
StorageService.instance.setInterDCStreamThroughputMbPerSec(Integer.MAX_VALUE);
assertEquals(Integer.MAX_VALUE * BYTES_PER_MEGABIT, StreamRateLimiter.getInterDCRateLimiterRateInBytes(), 0);
// Zero value check
StorageService.instance.setInterDCStreamThroughputMbPerSec(0);
assertEquals(Double.MAX_VALUE, StreamRateLimiter.getInterDCRateLimiterRateInBytes(), 0);
// Negative value check
StorageService.instance.setInterDCStreamThroughputMbPerSec(-200);
assertEquals(Double.MAX_VALUE, StreamRateLimiter.getInterDCRateLimiterRateInBytes(), 0);
}
}

105
test/unit/org/apache/cassandra/tools/nodetool/SetGetInterDCStreamThroughputTest.java Normal file
View File

@ -0,0 +1,105 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.tools.nodetool;
import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.cql3.CQLTester;
import static org.apache.cassandra.streaming.StreamManager.StreamRateLimiter;
import static org.apache.cassandra.tools.ToolRunner.ToolResult;
import static org.apache.cassandra.tools.ToolRunner.invokeNodetool;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.withPrecision;
/**
* Tests for {@code nodetool setinterdcstreamthroughput} and {@code nodetool getinterdcstreamthroughput}.
*/
public class SetGetInterDCStreamThroughputTest extends CQLTester
{
@BeforeClass
public static void setup() throws Exception
{
startJMXServer();
}
@Test
public void testNull()
{
assertSetInvalidThroughput(null, "Required parameters are missing: inter_dc_stream_throughput");
}
@Test
public void testPositive()
{
assertSetGetValidThroughput(7, 7 * StreamRateLimiter.BYTES_PER_MEGABIT);
}
@Test
public void testMaxValue()
{
assertSetGetValidThroughput(Integer.MAX_VALUE, Integer.MAX_VALUE * StreamRateLimiter.BYTES_PER_MEGABIT);
}
@Test
public void testZero()
{
assertSetGetValidThroughput(0, Double.MAX_VALUE);
}
@Test
public void testNegative()
{
assertSetGetValidThroughput(-7, Double.MAX_VALUE);
}
@Test
public void testUnparseable()
{
assertSetInvalidThroughput("1.2", "inter_dc_stream_throughput: can not convert \"1.2\" to a int");
assertSetInvalidThroughput("value", "inter_dc_stream_throughput: can not convert \"value\" to a int");
}
private static void assertSetGetValidThroughput(int throughput, double rateInBytes)
{
ToolResult tool = invokeNodetool("setinterdcstreamthroughput", String.valueOf(throughput));
tool.assertOnCleanExit();
assertThat(tool.getStdout()).isEmpty();
assertGetThroughput(throughput);
assertThat(StreamRateLimiter.getInterDCRateLimiterRateInBytes()).isEqualTo(rateInBytes, withPrecision(0.01));
}
private static void assertSetInvalidThroughput(String throughput, String expectedErrorMessage)
{
ToolResult tool = throughput == null ? invokeNodetool("setinterdcstreamthroughput")
: invokeNodetool("setinterdcstreamthroughput", throughput);
assertThat(tool.getExitCode()).isEqualTo(1);
assertThat(tool.getStdout()).contains(expectedErrorMessage);
}
private static void assertGetThroughput(int expected)
{
ToolResult tool = invokeNodetool("getinterdcstreamthroughput");
tool.assertOnCleanExit();
assertThat(tool.getStdout()).contains("Current inter-datacenter stream throughput: " + expected + " Mb/s");
}
}

106
test/unit/org/apache/cassandra/tools/nodetool/SetGetStreamThroughputTest.java Normal file
View File

@ -0,0 +1,106 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.tools.nodetool;
import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.cql3.CQLTester;
import static org.assertj.core.api.Assertions.withPrecision;
import static org.apache.cassandra.streaming.StreamManager.StreamRateLimiter;
import static org.apache.cassandra.tools.ToolRunner.ToolResult;
import static org.apache.cassandra.tools.ToolRunner.invokeNodetool;
import static org.assertj.core.api.Assertions.assertThat;
/**
* Tests for {@code nodetool setstreamthroughput} and {@code nodetool getstreamthroughput}.
*/
public class SetGetStreamThroughputTest extends CQLTester
{
@BeforeClass
public static void setup() throws Exception
{
startJMXServer();
}
@Test
public void testNull()
{
assertSetInvalidThroughput(null, "Required parameters are missing: stream_throughput");
}
@Test
public void testPositive()
{
assertSetGetValidThroughput(7, 7 * StreamRateLimiter.BYTES_PER_MEGABIT);
}
@Test
public void testMaxValue()
{
assertSetGetValidThroughput(Integer.MAX_VALUE, Integer.MAX_VALUE * StreamRateLimiter.BYTES_PER_MEGABIT);
}
@Test
public void testZero()
{
assertSetGetValidThroughput(0, Double.MAX_VALUE);
}
@Test
public void testNegative()
{
assertSetGetValidThroughput(-7, Double.MAX_VALUE);
}
@Test
public void testUnparseable()
{
assertSetInvalidThroughput("1.2", "stream_throughput: can not convert \"1.2\" to a int");
assertSetInvalidThroughput("value", "stream_throughput: can not convert \"value\" to a int");
}
private static void assertSetGetValidThroughput(int throughput, double rateInBytes)
{
ToolResult tool = invokeNodetool("setstreamthroughput", String.valueOf(throughput));
tool.assertOnCleanExit();
assertThat(tool.getStdout()).isEmpty();
assertGetThroughput(throughput);
assertThat(StreamRateLimiter.getRateLimiterRateInBytes()).isEqualTo(rateInBytes, withPrecision(0.01));
}
private static void assertSetInvalidThroughput(String throughput, String expectedErrorMessage)
{
ToolResult tool = throughput == null ? invokeNodetool("setstreamthroughput")
: invokeNodetool("setstreamthroughput", throughput);
assertThat(tool.getExitCode()).isEqualTo(1);
assertThat(tool.getStdout()).contains(expectedErrorMessage);
}
private static void assertGetThroughput(int expected)
{
ToolResult tool = invokeNodetool("getstreamthroughput");
tool.assertOnCleanExit();
assertThat(tool.getStdout()).contains("Current stream throughput: " + expected + " Mb/s");
}
}