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

Enforce CQL message size limit on multiframe messages 

Patch by Dmitry Konstantinov; reviewed by Sam Tunnicliffe,
Caleb Rackliffe for CASSANDRA-20052
This commit is contained in:
Dmitry Konstantinov 2024-11-04 21:17:24 +00:00 committed by Sam Tunnicliffe
parent f9d4888c55
commit 6f90e962f5
15 changed files with 632 additions and 142 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
4.1.8
* Enforce CQL message size limit on multiframe messages (CASSANDRA-20052)
* Add nodetool checktokenmetadata command that checks TokenMetadata is insync with Gossip endpointState (CASSANDRA-18758)
* Backport Java 11 support for Simulator (CASSANDRA-17178/CASSANDRA-19935)
* Equality check for Paxos.Electorate should not depend on collection types (CASSANDRA-19935)

1
src/java/org/apache/cassandra/config/Config.java
View File

@ -269,6 +269,7 @@ public class Config
public int native_transport_max_threads = 128;
@Replaces(oldName = "native_transport_max_frame_size_in_mb", converter = Converters.MEBIBYTES_DATA_STORAGE_INT, deprecated = true)
public DataStorageSpec.IntMebibytesBound native_transport_max_frame_size = new DataStorageSpec.IntMebibytesBound("16MiB");
public volatile DataStorageSpec.LongBytesBound native_transport_max_message_size = null;
/** do bcrypt hashing in a limited pool to prevent cpu load spikes; 0 means that all requests will go to default request executor**/
public int native_transport_max_auth_threads = 0;
public volatile long native_transport_max_concurrent_connections = -1L;

52
src/java/org/apache/cassandra/config/DatabaseDescriptor.java
View File

@ -153,6 +153,9 @@ public class DatabaseDescriptor
private static long counterCacheSizeInMiB;
private static long indexSummaryCapacityInMiB;
private static volatile long nativeTransportMaxMessageSizeInBytes;
private static volatile boolean nativeTransportMaxMessageSizeConfiguredExplicitly;
private static String localDC;
private static Comparator<Replica> localComparator;
private static EncryptionContext encryptionContext;
@ -822,6 +825,23 @@ public class DatabaseDescriptor
else if (conf.commitlog_segment_size.toKibibytes() < 2 * conf.max_mutation_size.toKibibytes())
throw new ConfigurationException("commitlog_segment_size must be at least twice the size of max_mutation_size / 1024", false);
if (conf.native_transport_max_message_size == null)
{
conf.native_transport_max_message_size = new DataStorageSpec.LongBytesBound(calculateDefaultNativeTransportMaxMessageSizeInBytes());
}
else
{
nativeTransportMaxMessageSizeConfiguredExplicitly = true;
long maxCqlMessageSize = conf.native_transport_max_message_size.toBytes();
if (maxCqlMessageSize > conf.native_transport_max_request_data_in_flight.toBytes())
throw new ConfigurationException("native_transport_max_message_size must not exceed native_transport_max_request_data_in_flight", false);
if (maxCqlMessageSize > conf.native_transport_max_request_data_in_flight_per_ip.toBytes())
throw new ConfigurationException("native_transport_max_message_size must not exceed native_transport_max_request_data_in_flight_per_ip", false);
}
nativeTransportMaxMessageSizeInBytes = conf.native_transport_max_message_size.toBytes();
// native transport encryption options
if (conf.client_encryption_options != null)
{
@ -2748,6 +2768,30 @@ public class DatabaseDescriptor
return conf.native_transport_max_request_data_in_flight_per_ip.toBytes();
}
public static long getNativeTransportMaxMessageSizeInBytes()
{
// the value of native_transport_max_message_size in bytes is cached
// to avoid conversion overhead during a parsing of each incoming CQL message
return nativeTransportMaxMessageSizeInBytes;
}
@VisibleForTesting
public static void setNativeTransportMaxMessageSizeInBytes(long maxMessageSizeInBytes)
{
conf.native_transport_max_message_size = new DataStorageSpec.LongBytesBound(maxMessageSizeInBytes);
nativeTransportMaxMessageSizeInBytes = conf.native_transport_max_message_size.toBytes();
}
private static long calculateDefaultNativeTransportMaxMessageSizeInBytes()
{
return Math.min(conf.max_mutation_size.toBytes(),
Math.min(
conf.native_transport_max_request_data_in_flight.toBytes(),
conf.native_transport_max_request_data_in_flight_per_ip.toBytes()
)
);
}
public static Config.PaxosVariant getPaxosVariant()
{
return conf.paxos_variant;
@ -2874,6 +2918,10 @@ public class DatabaseDescriptor
maxRequestDataInFlightInBytes = Runtime.getRuntime().maxMemory() / 40;
conf.native_transport_max_request_data_in_flight_per_ip = new DataStorageSpec.LongBytesBound(maxRequestDataInFlightInBytes);
long newNativeTransportMaxMessageSizeInBytes = nativeTransportMaxMessageSizeConfiguredExplicitly
? Math.min(maxRequestDataInFlightInBytes, getNativeTransportMaxMessageSizeInBytes())
: calculateDefaultNativeTransportMaxMessageSizeInBytes();
setNativeTransportMaxMessageSizeInBytes(newNativeTransportMaxMessageSizeInBytes);
}
public static long getNativeTransportMaxRequestDataInFlightInBytes()
@ -2887,6 +2935,10 @@ public class DatabaseDescriptor
maxRequestDataInFlightInBytes = Runtime.getRuntime().maxMemory() / 10;
conf.native_transport_max_request_data_in_flight = new DataStorageSpec.LongBytesBound(maxRequestDataInFlightInBytes);
long newNativeTransportMaxMessageSizeInBytes = nativeTransportMaxMessageSizeConfiguredExplicitly
? Math.min(maxRequestDataInFlightInBytes, getNativeTransportMaxMessageSizeInBytes())
: calculateDefaultNativeTransportMaxMessageSizeInBytes();
setNativeTransportMaxMessageSizeInBytes(newNativeTransportMaxMessageSizeInBytes);
}
public static int getNativeTransportMaxRequestsPerSecond()

27
src/java/org/apache/cassandra/exceptions/OversizedCQLMessageException.java Normal file
View File

@ -0,0 +1,27 @@
/*
* 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.exceptions;
public class OversizedCQLMessageException extends InvalidRequestException
{
public OversizedCQLMessageException(String message)
{
super(message);
}
}

4
src/java/org/apache/cassandra/net/AbstractMessageHandler.java
View File

@ -562,7 +562,7 @@ public abstract class AbstractMessageHandler extends ChannelInboundHandlerAdapte
return size == received;
}
private void onIntactFrame(IntactFrame frame)
protected void onIntactFrame(IntactFrame frame)
{
boolean expires = approxTime.isAfter(expiresAtNanos);
if (!isExpired && !isCorrupt)
@ -578,7 +578,7 @@ public abstract class AbstractMessageHandler extends ChannelInboundHandlerAdapte
isExpired |= expires;
}
private void onCorruptFrame()
protected void onCorruptFrame()
{
if (!isExpired && !isCorrupt)
releaseBuffersAndCapacity(); // release resources once we transition from normal state to corrupt

110
src/java/org/apache/cassandra/transport/CQLMessageHandler.java
View File

@ -31,6 +31,7 @@ import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.exceptions.OverloadedException;
import org.apache.cassandra.exceptions.OversizedCQLMessageException;
import org.apache.cassandra.metrics.ClientMessageSizeMetrics;
import org.apache.cassandra.metrics.ClientMetrics;
import org.apache.cassandra.net.AbstractMessageHandler;
@ -81,6 +82,10 @@ public class CQLMessageHandler<M extends Message> extends AbstractMessageHandler
public static final int LARGE_MESSAGE_THRESHOLD = FrameEncoder.Payload.MAX_SIZE - 1;
public static final TimeUnit RATE_LIMITER_DELAY_UNIT = TimeUnit.NANOSECONDS;
static final String MULTI_FRAME_AUTH_ERROR_MESSAGE_PREFIX = "The connection is not yet in a valid state " +
"to process multi frame CQL Messages, usually this" +
"means that authentication is still pending. ";
private final QueueBackpressure queueBackpressure;
private final Envelope.Decoder envelopeDecoder;
private final Message.Decoder<M> messageDecoder;
@ -94,6 +99,8 @@ public class CQLMessageHandler<M extends Message> extends AbstractMessageHandler
long channelPayloadBytesInFlight;
private int consecutiveMessageErrors = 0;
private final ServerConnection serverConnection;
interface MessageConsumer<M extends Message>
{
void dispatch(Channel channel, M message, Dispatcher.FlushItemConverter toFlushItem, Overload backpressure);
@ -106,6 +113,7 @@ public class CQLMessageHandler<M extends Message> extends AbstractMessageHandler
}
CQLMessageHandler(Channel channel,
ServerConnection serverConnection,
ProtocolVersion version,
FrameDecoder decoder,
Envelope.Decoder envelopeDecoder,
@ -128,6 +136,7 @@ public class CQLMessageHandler<M extends Message> extends AbstractMessageHandler
resources.endpointWaitQueue(),
resources.globalWaitQueue(),
onClosed);
this.serverConnection = serverConnection;
this.envelopeDecoder = envelopeDecoder;
this.messageDecoder = messageDecoder;
this.payloadAllocator = payloadAllocator;
@ -518,10 +527,28 @@ public class CQLMessageHandler<M extends Message> extends AbstractMessageHandler
// max CQL message size defaults to 256mb, so should be safe to downcast
int messageSize = Ints.checkedCast(header.bodySizeInBytes);
receivedBytes += buf.remaining();
if (serverConnection != null && serverConnection.stage() != ConnectionStage.READY)
{
// Disallow any multiframe messages before the connection reaches the READY state.
// This guards against being swamped with oversize messages from unauthenticated
// clients. In this case, we raise a fatal error and close the connection so it does
// not make sense to continue processing subsequent frames
handleError(ProtocolException.toFatalException(new OversizedAuthMessageException(
MULTI_FRAME_AUTH_ERROR_MESSAGE_PREFIX +
"type = " + header.type + ", size = " + header.bodySizeInBytes)));
ClientMetrics.instance.markRequestDiscarded();
return false;
}
LargeMessage largeMessage = new LargeMessage(header);
if (throwOnOverload)
if (messageSize > DatabaseDescriptor.getNativeTransportMaxMessageSizeInBytes())
{
ClientMetrics.instance.markRequestDiscarded();
// Mark as too big so that discard the message after consuming any subsequent frames
largeMessage.markTooBig();
}
else if (throwOnOverload)
{
if (!acquireCapacity(header, endpointReserve, globalReserve))
{
@ -567,9 +594,9 @@ public class CQLMessageHandler<M extends Message> extends AbstractMessageHandler
}
}
}
else
else // throwOnOverload = false
{
if (acquireCapacity(header, endpointReserve, globalReserve))
if (acquireCapacityAndQueueOnFailure(header, endpointReserve, globalReserve))
{
long delay = -1;
Overload backpressure = Overload.NONE;
@ -605,7 +632,14 @@ public class CQLMessageHandler<M extends Message> extends AbstractMessageHandler
}
else
{
noSpamLogger.error("Could not aquire capacity while processing native protocol message");
// we checked previously that messageSize <= native_transport_max_message_size
// and native_transport_max_message_size <= native_transport_max_request_data_in_flight
// and native_transport_max_message_size <= native_transport_max_request_data_in_flight_per_ip
// so, a starvation is not possible for the following case:
// a connection is blocked forever if somebody tries to send a single too big message > total rate limiting capacity.
// Once other messages in the same or other CQL connections are processed and capacity is returned to the limits
// we have enough capacity to acquire it for the current large message.
return false;
}
}
@ -712,6 +746,7 @@ public class CQLMessageHandler<M extends Message> extends AbstractMessageHandler
private Overload overload = Overload.NONE;
private Overload backpressure = Overload.NONE;
private boolean tooBig = false;
private LargeMessage(Envelope.Header header)
{
@ -747,18 +782,75 @@ public class CQLMessageHandler<M extends Message> extends AbstractMessageHandler
this.backpressure = backpressure;
}
private void markTooBig()
{
this.tooBig = true;
}
@Override
protected void onIntactFrame(IntactFrame frame)
{
if (tooBig || overload != Overload.NONE)
// we do not want to add the frame to buffers (to not consume a lot of memory and throw it away later
// we also do not want to release capacity because we haven't accuired it
frame.consume();
else
super.onIntactFrame(frame);
}
@Override
protected void onCorruptFrame()
{
if (!isExpired && !isCorrupt && !tooBig)
{
releaseBuffers(); // release resources once we transition from normal state to corrupt
if (overload != Overload.BYTES_IN_FLIGHT)
releaseCapacity(size);
}
isCorrupt = true;
isExpired |= approxTime.isAfter(expiresAtNanos);
}
@Override
protected void onComplete()
{
if (overload != Overload.NONE)
handleErrorAndRelease(buildOverloadedException(endpointReserveCapacity, globalReserveCapacity, overload), header);
if (tooBig)
// we haven't accuired a capacity for too big messages to release it
handleError(buildOversizedCQLMessageException(header.bodySizeInBytes), header);
else if (overload != Overload.NONE)
if (overload == Overload.BYTES_IN_FLIGHT)
// we haven't accuired a capacity successfully to release it
handleError(buildOverloadedException(endpointReserveCapacity, globalReserveCapacity, overload), header);
else
handleErrorAndRelease(buildOverloadedException(endpointReserveCapacity, globalReserveCapacity, overload), header);
else if (!isCorrupt)
processRequest(assembleFrame(), backpressure);
}
@Override
protected void abort()
{
if (!isCorrupt)
releaseBuffersAndCapacity(); // release resources if in normal state when abort() is invoked
if (!isCorrupt && !tooBig && overload == Overload.NONE)
releaseBuffers();
if (overload == Overload.NONE || overload == Overload.BYTES_IN_FLIGHT)
releaseCapacity(size);
}
private OversizedCQLMessageException buildOversizedCQLMessageException(long messageBodySize)
{
return new OversizedCQLMessageException("CQL Message of size " + messageBodySize
+ " bytes exceeds allowed maximum of "
+ DatabaseDescriptor.getNativeTransportMaxMessageSizeInBytes() + " bytes");
}
}
static class OversizedAuthMessageException extends ProtocolException
{
OversizedAuthMessageException(String message)
{
super(message);
}
}
}

5
src/java/org/apache/cassandra/transport/ExceptionHandlers.java
View File

@ -38,6 +38,7 @@ import io.netty.channel.ChannelPromise;
import io.netty.channel.unix.Errors;
import org.apache.cassandra.exceptions.OverloadedException;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.exceptions.OversizedCQLMessageException;
import org.apache.cassandra.metrics.ClientMetrics;
import org.apache.cassandra.net.FrameEncoder;
import org.apache.cassandra.transport.messages.ErrorMessage;
@ -130,6 +131,10 @@ public class ExceptionHandlers
// Once the threshold for overload is breached, it will very likely spam the logs...
NoSpamLogger.log(logger, NoSpamLogger.Level.INFO, 1, TimeUnit.MINUTES, cause.getMessage());
}
else if (Throwables.anyCauseMatches(cause, t -> t instanceof OversizedCQLMessageException))
{
NoSpamLogger.log(logger, NoSpamLogger.Level.INFO, 1, TimeUnit.MINUTES, cause.getMessage());
}
else if (Throwables.anyCauseMatches(cause, t -> t instanceof Errors.NativeIoException))
{
ClientMetrics.instance.markUnknownException();

3
src/java/org/apache/cassandra/transport/InitialConnectionHandler.java
View File

@ -104,6 +104,7 @@ public class InitialConnectionHandler extends ByteToMessageDecoder
attrConn.set(connection);
}
assert connection instanceof ServerConnection;
ServerConnection serverConnection = (ServerConnection) connection;
StartupMessage startup = (StartupMessage) Message.Decoder.decodeMessage(ctx.channel(), inbound);
InetAddress remoteAddress = ((InetSocketAddress) ctx.channel().remoteAddress()).getAddress();
@ -120,7 +121,7 @@ public class InitialConnectionHandler extends ByteToMessageDecoder
if (future.isSuccess())
{
logger.trace("Response to STARTUP sent, configuring pipeline for {}", inbound.header.version);
configurator.configureModernPipeline(ctx, allocator, inbound.header.version, startup.options);
configurator.configureModernPipeline(ctx, serverConnection, allocator, inbound.header.version, startup.options);
allocator.release(inbound.header.bodySizeInBytes);
}
else

2
src/java/org/apache/cassandra/transport/PipelineConfigurator.java
View File

@ -270,6 +270,7 @@ public class PipelineConfigurator
}
public void configureModernPipeline(ChannelHandlerContext ctx,
ServerConnection serverConnection,
ClientResourceLimits.Allocator resourceAllocator,
ProtocolVersion version,
Map<String, String> options)
@ -311,6 +312,7 @@ public class PipelineConfigurator
CQLMessageHandler.MessageConsumer<Message.Request> messageConsumer = messageConsumer();
CQLMessageHandler<Message.Request> processor =
new CQLMessageHandler<>(ctx.channel(),
serverConnection,
version,
frameDecoder,
envelopeDecoder,

1
src/java/org/apache/cassandra/transport/SimpleClient.java
View File

@ -504,6 +504,7 @@ public class SimpleClient implements Closeable
CQLMessageHandler<Message.Response> processor =
new CQLMessageHandler<Message.Response>(ctx.channel(),
null,
version,
frameDecoder,
envelopeDecoder,

104
test/unit/org/apache/cassandra/transport/AuthMessageSizeLimitTest.java Normal file
View File

@ -0,0 +1,104 @@
/*
* 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.transport;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import com.datastax.driver.core.Authenticator;
import com.datastax.driver.core.EndPoint;
import com.datastax.driver.core.PlainTextAuthProvider;
import org.apache.cassandra.Util;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.net.FrameEncoder;
import org.apache.cassandra.transport.messages.AuthResponse;
import org.apache.cassandra.transport.messages.QueryMessage;
import org.assertj.core.api.Assertions;
public class AuthMessageSizeLimitTest extends NativeProtocolLimitsTestBase
{
private static final int TOO_BIG_MULTI_FRAME_AUTH_MESSAGE_SIZE = 2 * FrameEncoder.Payload.MAX_SIZE;
// set MAX_CQL_MESSAGE_SIZE bigger than TOO_BIG_MULTI_FRAME_AUTH_MESSAGE_SIZE to ensure what the auth message size check is more restrictive
private static final int MAX_CQL_MESSAGE_SIZE = TOO_BIG_MULTI_FRAME_AUTH_MESSAGE_SIZE * 2;
@BeforeClass
public static void setUp()
{
DatabaseDescriptor.setNativeTransportReceiveQueueCapacityInBytes(1);
DatabaseDescriptor.setNativeTransportMaxRequestDataInFlightPerIpInBytes(MAX_CQL_MESSAGE_SIZE);
DatabaseDescriptor.setNativeTransportConcurrentRequestDataInFlightInBytes(MAX_CQL_MESSAGE_SIZE);
DatabaseDescriptor.setNativeTransportMaxMessageSizeInBytes(MAX_CQL_MESSAGE_SIZE);
requireNetwork();
requireAuthentication();
}
@Before
public void setLimits()
{
ClientResourceLimits.setGlobalLimit(MAX_CQL_MESSAGE_SIZE);
ClientResourceLimits.setEndpointLimit(MAX_CQL_MESSAGE_SIZE);
}
@Test
public void sendSmallAuthMessage()
{
doTest((client) ->
{
AuthResponse authResponse = createAuthMessage("cassandra", "cassandra");
client.execute(authResponse);
createTable(client);
int valueLessThanMessageMaxSize = MAX_CQL_MESSAGE_SIZE - 500;
QueryMessage queryMessage = queryMessage(valueLessThanMessageMaxSize);
client.execute(queryMessage);
}
);
}
@Test
public void sendTooBigAuthMultiFrameMessage()
{
doTest((client) ->
{
AuthResponse authResponse = createAuthMessage("cassandra", createIncorrectLongPassword(TOO_BIG_MULTI_FRAME_AUTH_MESSAGE_SIZE));
Assertions.assertThatThrownBy(() -> client.execute(authResponse))
.hasCauseInstanceOf(ProtocolException.class)
.hasMessageContaining(CQLMessageHandler.MULTI_FRAME_AUTH_ERROR_MESSAGE_PREFIX);
Util.spinAssertEquals(false, () -> client.connection.channel().isOpen(), 10);
}
);
}
private static String createIncorrectLongPassword(int length)
{
StringBuilder password = new StringBuilder(length);
for (int i = 0; i < length; i++)
password.append('a');
return password.toString();
}
private AuthResponse createAuthMessage(String username, String password)
{
PlainTextAuthProvider authProvider = new PlainTextAuthProvider(username, password);
Authenticator authenticator = authProvider.newAuthenticator((EndPoint) null, null);
return new AuthResponse(authenticator.initialResponse());
}
}

106
test/unit/org/apache/cassandra/transport/ClientResourceLimitsTest.java
View File

@ -18,7 +18,6 @@
package org.apache.cassandra.transport;
import java.io.IOException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.TimeUnit;
@ -32,9 +31,6 @@ import org.apache.cassandra.service.StorageService;
import org.junit.*;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.cql3.QueryOptions;
import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.db.marshal.Int32Type;
import org.apache.cassandra.db.marshal.UTF8Type;
import org.apache.cassandra.db.virtual.*;
@ -50,21 +46,11 @@ import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
public class ClientResourceLimitsTest extends CQLTester
public class ClientResourceLimitsTest extends NativeProtocolLimitsTestBase
{
private static final long LOW_LIMIT = 600L;
private static final long HIGH_LIMIT = 5000000000L;
private static final QueryOptions V5_DEFAULT_OPTIONS =
QueryOptions.create(QueryOptions.DEFAULT.getConsistency(),
QueryOptions.DEFAULT.getValues(),
QueryOptions.DEFAULT.skipMetadata(),
QueryOptions.DEFAULT.getPageSize(),
QueryOptions.DEFAULT.getPagingState(),
QueryOptions.DEFAULT.getSerialConsistency(),
ProtocolVersion.V5,
KEYSPACE);
@BeforeClass
public static void setUp()
{
@ -89,54 +75,6 @@ public class ClientResourceLimitsTest extends CQLTester
ClientResourceLimits.setEndpointLimit(LOW_LIMIT);
}
@After
public void dropCreatedTable()
{
try
{
QueryProcessor.executeOnceInternal("DROP TABLE " + KEYSPACE + ".atable");
}
catch (Throwable t)
{
// ignore
}
}
@SuppressWarnings("resource")
private SimpleClient client(boolean throwOnOverload)
{
try
{
return SimpleClient.builder(nativeAddr.getHostAddress(), nativePort)
.protocolVersion(ProtocolVersion.V5)
.useBeta()
.build()
.connect(false, throwOnOverload);
}
catch (IOException e)
{
throw new RuntimeException("Error initializing client", e);
}
}
@SuppressWarnings({"resource", "SameParameterValue"})
private SimpleClient client(boolean throwOnOverload, int largeMessageThreshold)
{
try
{
return SimpleClient.builder(nativeAddr.getHostAddress(), nativePort)
.protocolVersion(ProtocolVersion.V5)
.useBeta()
.largeMessageThreshold(largeMessageThreshold)
.build()
.connect(false, throwOnOverload);
}
catch (IOException e)
{
throw new RuntimeException("Error initializing client", e);
}
}
@Test
public void testQueryExecutionWithThrowOnOverload()
{
@ -153,10 +91,8 @@ public class ClientResourceLimitsTest extends CQLTester
{
try (SimpleClient client = client(throwOnOverload))
{
QueryMessage queryMessage = new QueryMessage("CREATE TABLE atable (pk int PRIMARY KEY, v text)",
V5_DEFAULT_OPTIONS);
client.execute(queryMessage);
queryMessage = new QueryMessage("SELECT * FROM atable", V5_DEFAULT_OPTIONS);
createTable(client);
QueryMessage queryMessage = new QueryMessage("SELECT * FROM atable", queryOptions());
client.execute(queryMessage);
}
}
@ -189,7 +125,7 @@ public class ClientResourceLimitsTest extends CQLTester
{
// The first query does not trigger backpressure/pause the connection:
QueryMessage queryMessage =
new QueryMessage("CREATE TABLE atable (pk int PRIMARY KEY, v text)", V5_DEFAULT_OPTIONS);
new QueryMessage("CREATE TABLE atable (pk int PRIMARY KEY, v text)", queryOptions());
Message.Response belowThresholdResponse = client.execute(queryMessage);
assertEquals(0, getPausedConnectionsGauge().getValue().intValue());
assertNoWarningContains(belowThresholdResponse, "bytes in flight");
@ -261,7 +197,11 @@ public class ClientResourceLimitsTest extends CQLTester
{
// Bump the per-endpoint limit to make sure we exhaust the global
ClientResourceLimits.setEndpointLimit(HIGH_LIMIT);
testOverloadedException(() -> client(true, Ints.checkedCast(LOW_LIMIT / 2)));
// test message = 2/3 x
// emulated concurrent message = 2/3 x
// test message + emulated concurrent message = 4/3 x > x set as a global limit
emulateInFlightConcurrentMessage(LOW_LIMIT * 2 / 3);
testOverloadedException(() -> client(true, Ints.checkedCast(LOW_LIMIT / 2)), LOW_LIMIT * 2 / 3);
}
@Test
@ -269,18 +209,25 @@ public class ClientResourceLimitsTest extends CQLTester
{
// Make sure we can only exceed the per-endpoint limit
ClientResourceLimits.setGlobalLimit(HIGH_LIMIT);
testOverloadedException(() -> client(true, Ints.checkedCast(LOW_LIMIT / 2)));
// test message = 2/3 x
// emulated concurrent message = 2/3 x
// test message + emulated concurrent message = 4/3 x > x set as an endpoint limit
emulateInFlightConcurrentMessage(LOW_LIMIT * 2 / 3);
testOverloadedException(() -> client(true, Ints.checkedCast(LOW_LIMIT / 2)), LOW_LIMIT * 2 / 3);
}
private void testOverloadedException(Supplier<SimpleClient> clientSupplier)
{
testOverloadedException(clientSupplier, LOW_LIMIT * 2);
}
private void testOverloadedException(Supplier<SimpleClient> clientSupplier, long limit)
{
try (SimpleClient client = clientSupplier.get())
{
QueryMessage queryMessage = new QueryMessage("CREATE TABLE atable (pk int PRIMARY KEY, v text)",
V5_DEFAULT_OPTIONS);
client.execute(queryMessage);
createTable(client);
queryMessage = queryMessage();
QueryMessage queryMessage = queryMessage(limit);
try
{
client.execute(queryMessage);
@ -295,11 +242,7 @@ public class ClientResourceLimitsTest extends CQLTester
private QueryMessage queryMessage()
{
StringBuilder query = new StringBuilder("INSERT INTO atable (pk, v) VALUES (1, '");
for (int i=0; i < LOW_LIMIT * 2; i++)
query.append('a');
query.append("')");
return new QueryMessage(query.toString(), V5_DEFAULT_OPTIONS);
return queryMessage(LOW_LIMIT * 2);
}
@Test
@ -341,7 +284,7 @@ public class ClientResourceLimitsTest extends CQLTester
VirtualKeyspaceRegistry.instance.register(new VirtualKeyspace(table, ImmutableList.of(vt1)));
final QueryMessage queryMessage = new QueryMessage(String.format("SELECT * FROM %s.%s", table, table),
V5_DEFAULT_OPTIONS);
queryOptions());
try
{
Thread tester = new Thread(() -> client.execute(queryMessage));
@ -367,8 +310,9 @@ public class ClientResourceLimitsTest extends CQLTester
try
{
QueryMessage smallMessage = new QueryMessage(String.format("CREATE TABLE %s.atable (pk int PRIMARY KEY, v text)", KEYSPACE),
V5_DEFAULT_OPTIONS);
queryOptions());
client.execute(smallMessage);
createTable(client);
try
{
client.execute(queryMessage());

124
test/unit/org/apache/cassandra/transport/MessageSizeLimitTest.java Normal file
View File

@ -0,0 +1,124 @@
/*
* 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.transport;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.Assert;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.net.FrameEncoder;
import org.apache.cassandra.transport.messages.QueryMessage;
import org.assertj.core.api.Assertions;
public class MessageSizeLimitTest extends NativeProtocolLimitsTestBase
{
private static final int MAX_CQL_MESSAGE_SIZE = FrameEncoder.Payload.MAX_SIZE * 3;
private static final int TOO_BIG_MESSAGE_SIZE = MAX_CQL_MESSAGE_SIZE * 2;
private static final int NORMAL_MESSAGE_SIZE = MAX_CQL_MESSAGE_SIZE - 500;
@BeforeClass
public static void setUp()
{
DatabaseDescriptor.setNativeTransportReceiveQueueCapacityInBytes(1);
DatabaseDescriptor.setNativeTransportMaxRequestDataInFlightPerIpInBytes(MAX_CQL_MESSAGE_SIZE);
DatabaseDescriptor.setNativeTransportConcurrentRequestDataInFlightInBytes(MAX_CQL_MESSAGE_SIZE);
DatabaseDescriptor.setNativeTransportMaxMessageSizeInBytes(MAX_CQL_MESSAGE_SIZE);
requireNetwork();
}
@Before
public void setLimits()
{
ClientResourceLimits.setGlobalLimit(MAX_CQL_MESSAGE_SIZE);
ClientResourceLimits.setEndpointLimit(MAX_CQL_MESSAGE_SIZE);
}
@Test
public void sendMessageWithSizeMoreThanMaxMessageSize()
{
runClientLogic((client) ->
{
QueryMessage tooBigQueryMessage = queryMessage(TOO_BIG_MESSAGE_SIZE);
Assertions.assertThatThrownBy(() -> client.execute(tooBigQueryMessage))
.hasCauseInstanceOf(InvalidRequestException.class);
// InvalidRequestException: CQL Message of size 524362 bytes exceeds allowed maximum of 262144 bytes
// we send one more message to check that the server continues to process new messages in the opened connection
QueryMessage queryMessage = queryMessage(NORMAL_MESSAGE_SIZE);
client.execute(queryMessage);
}
);
}
@Test(timeout = 30_000)
public void checkThatThereIsNoStarvationForMultiFrameMessages() throws InterruptedException
{
runClientLogic((client) -> {}, true); // to create table
AtomicInteger completedSuccessfully = new AtomicInteger(0);
int threadsCount = 2;
List<Thread> threads = new ArrayList<>();
for (int i = 0; i < threadsCount; i++)
{
threads.add(new Thread(() -> runClientLogic((client) -> {
sendMessages(client, 100, NORMAL_MESSAGE_SIZE);
completedSuccessfully.incrementAndGet();
}, false))
);
}
for (Thread thread : threads)
thread.start();
for (Thread thread : threads)
thread.join();
Assert.assertEquals("not all messages were sent successfully by all threads",
threadsCount, completedSuccessfully.get());
}
private void sendMessages(SimpleClient client, int messagesCount, int messageSize)
{
for (int i = 0; i < messagesCount; i++)
{
QueryMessage queryMessage1 = queryMessage(messageSize);
client.execute(queryMessage1);
}
}
@Test
public void sendMessageWithSizeBelowLimit()
{
runClientLogic((client) ->
{
QueryMessage queryMessage = queryMessage(NORMAL_MESSAGE_SIZE);
client.execute(queryMessage);
// run one more time, to validate that the connection is still alive
queryMessage = queryMessage(NORMAL_MESSAGE_SIZE);
client.execute(queryMessage);
}
);
}
}

165
test/unit/org/apache/cassandra/transport/NativeProtocolLimitsTestBase.java Normal file
View File

@ -0,0 +1,165 @@
/*
* 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.transport;
import java.io.IOException;
import org.junit.After;
import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.cql3.QueryOptions;
import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.net.FrameEncoder;
import org.apache.cassandra.transport.messages.QueryMessage;
import org.apache.cassandra.utils.FBUtilities;
public abstract class NativeProtocolLimitsTestBase extends CQLTester
{
protected final ProtocolVersion version;
protected long emulatedUsedCapacity;
public NativeProtocolLimitsTestBase()
{
this(ProtocolVersion.V5);
}
public NativeProtocolLimitsTestBase(ProtocolVersion version)
{
this.version = version;
}
@After
public void dropCreatedTable()
{
if (emulatedUsedCapacity > 0)
{
releaseEmulatedCapacity(emulatedUsedCapacity);
}
try
{
QueryProcessor.executeOnceInternal("DROP TABLE " + KEYSPACE + ".atable");
}
catch (Throwable t)
{
// ignore
}
}
public QueryOptions queryOptions()
{
return QueryOptions.create(QueryOptions.DEFAULT.getConsistency(),
QueryOptions.DEFAULT.getValues(),
QueryOptions.DEFAULT.skipMetadata(),
QueryOptions.DEFAULT.getPageSize(),
QueryOptions.DEFAULT.getPagingState(),
QueryOptions.DEFAULT.getSerialConsistency(),
version,
KEYSPACE);
}
public SimpleClient client()
{
return client(false);
}
@SuppressWarnings({"resource", "SameParameterValue"})
public SimpleClient client(boolean throwOnOverload)
{
return client(throwOnOverload, FrameEncoder.Payload.MAX_SIZE);
}
@SuppressWarnings({"resource", "SameParameterValue"})
public SimpleClient client(boolean throwOnOverload, int largeMessageThreshold)
{
try
{
return SimpleClient.builder(nativeAddr.getHostAddress(), nativePort)
.protocolVersion(version)
.useBeta()
.largeMessageThreshold(largeMessageThreshold)
.build()
.connect(false, throwOnOverload);
}
catch (IOException e)
{
throw new RuntimeException("Error initializing client", e);
}
}
public void runClientLogic(ClientLogic clientLogic)
{
runClientLogic(clientLogic, true);
}
public void runClientLogic(ClientLogic clientLogic, boolean createTable)
{
try (SimpleClient client = client())
{
if (createTable)
createTable(client);
clientLogic.run(client);
}
}
public void createTable(SimpleClient client)
{
QueryMessage queryMessage = new QueryMessage("CREATE TABLE IF NOT EXISTS " +
KEYSPACE + ".atable (pk int PRIMARY KEY, v text)",
queryOptions());
client.execute(queryMessage);
}
public void doTest(ClientLogic testLogic)
{
try (SimpleClient client = client())
{
testLogic.run(client);
}
}
public interface ClientLogic
{
void run(SimpleClient simpleClient);
}
public QueryMessage queryMessage(long valueSize)
{
StringBuilder query = new StringBuilder("INSERT INTO " + KEYSPACE + ".atable (pk, v) VALUES (1, '");
for (int i = 0; i < valueSize; i++)
query.append('a');
query.append("')");
return new QueryMessage(query.toString(), queryOptions());
}
protected void emulateInFlightConcurrentMessage(long length)
{
ClientResourceLimits.Allocator allocator = ClientResourceLimits.getAllocatorForEndpoint(FBUtilities.getJustLocalAddress());
ClientResourceLimits.ResourceProvider resourceProvider = new ClientResourceLimits.ResourceProvider.Default(allocator);
resourceProvider.globalLimit().allocate(length);
resourceProvider.endpointLimit().allocate(length);
emulatedUsedCapacity += length;
}
protected void releaseEmulatedCapacity(long length)
{
ClientResourceLimits.Allocator allocator = ClientResourceLimits.getAllocatorForEndpoint(FBUtilities.getJustLocalAddress());
ClientResourceLimits.ResourceProvider resourceProvider = new ClientResourceLimits.ResourceProvider.Default(allocator);
resourceProvider.globalLimit().release(length);
resourceProvider.endpointLimit().release(length);
}
}

69
test/unit/org/apache/cassandra/transport/RateLimitingTest.java
View File

@ -28,7 +28,9 @@ import java.util.stream.Collectors;
import com.codahale.metrics.Meter;
import com.google.common.base.Ticker;
import org.awaitility.Awaitility;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
@ -36,12 +38,9 @@ import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.cql3.QueryOptions;
import org.apache.cassandra.exceptions.OverloadedException;
import org.apache.cassandra.metrics.CassandraMetricsRegistry;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.transport.messages.QueryMessage;
import org.apache.cassandra.utils.Throwables;
import static org.junit.Assert.assertEquals;
@ -54,7 +53,7 @@ import static org.apache.cassandra.transport.ProtocolVersion.V4;
@SuppressWarnings("UnstableApiUsage")
@RunWith(Parameterized.class)
public class RateLimitingTest extends CQLTester
public class RateLimitingTest extends NativeProtocolLimitsTestBase
{
public static final String BACKPRESSURE_WARNING_SNIPPET = "Request breached global limit";
@ -63,9 +62,6 @@ public class RateLimitingTest extends CQLTester
private static final long MAX_LONG_CONFIG_VALUE = Long.MAX_VALUE - 1;
@Parameterized.Parameter
public ProtocolVersion version;
@Parameterized.Parameters(name="{0}")
public static Collection<Object[]> versions()
{
@ -74,6 +70,11 @@ public class RateLimitingTest extends CQLTester
.collect(Collectors.toList());
}
public RateLimitingTest(ProtocolVersion version)
{
super(version);
}
private AtomicLong tick;
private Ticker ticker;
@ -105,6 +106,7 @@ public class RateLimitingTest extends CQLTester
ClientResourceLimits.setGlobalLimit(MAX_LONG_CONFIG_VALUE);
}
@Test
public void shouldThrowOnOverloadSmallMessages() throws Exception
{
@ -147,15 +149,19 @@ public class RateLimitingTest extends CQLTester
private void testBytesInFlightOverload(int payloadSize) throws Exception
{
try (SimpleClient client = client().connect(false, true))
int emulatedConcurrentMessageSize = payloadSize * 3 / 2;
try (SimpleClient client = client(true, LARGE_PAYLOAD_THRESHOLD_BYTES))
{
StorageService.instance.setNativeTransportRateLimitingEnabled(false);
QueryMessage queryMessage = new QueryMessage("CREATE TABLE IF NOT EXISTS " + KEYSPACE + ".atable (pk int PRIMARY KEY, v text)", queryOptions());
client.execute(queryMessage);
createTable(client);
StorageService.instance.setNativeTransportRateLimitingEnabled(true);
ClientResourceLimits.GLOBAL_REQUEST_LIMITER.setRate(OVERLOAD_PERMITS_PER_SECOND, ticker);
ClientResourceLimits.setGlobalLimit(1);
// test message = 1x
// emulated concurrent message = 1.5x
// test message + emulated concurrent message = 2.5x > 2x set as a global limit
ClientResourceLimits.setGlobalLimit(payloadSize * 2);
emulateInFlightConcurrentMessage(emulatedConcurrentMessageSize);
try
{
@ -170,18 +176,17 @@ public class RateLimitingTest extends CQLTester
finally
{
// Sanity check bytes in flight limiter.
Awaitility.await().untilAsserted(() -> assertEquals(0, ClientResourceLimits.getCurrentGlobalUsage()));
Awaitility.await().untilAsserted(() -> assertEquals(emulatedConcurrentMessageSize, ClientResourceLimits.getCurrentGlobalUsage()));
StorageService.instance.setNativeTransportRateLimitingEnabled(false);
}
}
private void testOverload(int payloadSize, boolean throwOnOverload) throws Exception
{
try (SimpleClient client = client().connect(false, throwOnOverload))
try (SimpleClient client = client(throwOnOverload, LARGE_PAYLOAD_THRESHOLD_BYTES))
{
StorageService.instance.setNativeTransportRateLimitingEnabled(false);
QueryMessage queryMessage = new QueryMessage("CREATE TABLE IF NOT EXISTS " + KEYSPACE + ".atable (pk int PRIMARY KEY, v text)", queryOptions());
client.execute(queryMessage);
createTable(client);
StorageService.instance.setNativeTransportRateLimitingEnabled(true);
ClientResourceLimits.GLOBAL_REQUEST_LIMITER.setRate(OVERLOAD_PERMITS_PER_SECOND, ticker);
@ -286,40 +291,6 @@ public class RateLimitingTest extends CQLTester
assertEquals(dispatchedPrior + 2, getRequestDispatchedMeter().getCount());
}
private QueryMessage queryMessage(int length)
{
StringBuilder query = new StringBuilder("INSERT INTO " + KEYSPACE + ".atable (pk, v) VALUES (1, '");
for (int i = 0; i < length; i++)
{
query.append('a');
}
query.append("')");
return new QueryMessage(query.toString(), queryOptions());
}
private SimpleClient client()
{
return SimpleClient.builder(nativeAddr.getHostAddress(), nativePort)
.protocolVersion(version)
.useBeta()
.largeMessageThreshold(LARGE_PAYLOAD_THRESHOLD_BYTES)
.build();
}
private QueryOptions queryOptions()
{
return QueryOptions.create(QueryOptions.DEFAULT.getConsistency(),
QueryOptions.DEFAULT.getValues(),
QueryOptions.DEFAULT.skipMetadata(),
QueryOptions.DEFAULT.getPageSize(),
QueryOptions.DEFAULT.getPagingState(),
QueryOptions.DEFAULT.getSerialConsistency(),
version,
KEYSPACE);
}
protected static Meter getRequestDispatchedMeter()
{
String metricName = "org.apache.cassandra.metrics.Client.RequestDispatched";