diff --git a/CHANGES.txt b/CHANGES.txt index d0d021c6db..6244eefe63 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 4.1 + * Add native transport rate limiter options to example cassandra.yaml, and expose metric for dispatch rate (CASSANDRA-17423) * Add diagnostic events for guardrails (CASSANDRA-17197) * Increase cqlsh version (CASSANDRA-17432) * Update SUPPORTED_UPGRADE_PATHS to include 3.0 and 3.x to 4.1 paths and remove obsolete tests (CASSANDRA-17362) diff --git a/NEWS.txt b/NEWS.txt index 5cad3e6ae8..5cd9fa224b 100644 --- a/NEWS.txt +++ b/NEWS.txt @@ -56,6 +56,8 @@ using the provided 'sstableupgrade' tool. New features ------------ + - Support for native transport rate limiting via native_transport_rate_limiting_enabled and + native_transport_max_requests_per_second in cassandra.yaml. - Expose all client options via system_views.clients and nodetool clientstats. - Support for String concatenation has been added through the + operator. - New configuration max_hints_size_per_host to limit the size of local hints files per host in mebibytes. Setting to diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml index a2c4db8e12..9da466360e 100644 --- a/conf/cassandra.yaml +++ b/conf/cassandra.yaml @@ -800,6 +800,16 @@ native_transport_allow_older_protocols: true # Idle connection timeouts are disabled by default. # native_transport_idle_timeout: 60000ms +# When enabled, limits the number of native transport requests dispatched for processing per second. +# Behavior once the limit has been breached depends on the value of THROW_ON_OVERLOAD specified in +# the STARTUP message sent by the client during connection establishment. (See section "4.1.1. STARTUP" +# in "CQL BINARY PROTOCOL v5".) With the THROW_ON_OVERLOAD flag enabled, messages that breach the limit +# are dropped, and an OverloadedException is thrown for the client to handle. When the flag is not +# enabled, the server will stop consuming messages from the channel/socket, putting backpressure on +# the client while already dispatched messages are processed. +# native_transport_rate_limiting_enabled: false +# native_transport_max_requests_per_second: 1000000 + # The address or interface to bind the native transport server to. # # Set rpc_address OR rpc_interface, not both. diff --git a/src/java/org/apache/cassandra/metrics/ClientMetrics.java b/src/java/org/apache/cassandra/metrics/ClientMetrics.java index 29ae693c93..4e3eb4f8ec 100644 --- a/src/java/org/apache/cassandra/metrics/ClientMetrics.java +++ b/src/java/org/apache/cassandra/metrics/ClientMetrics.java @@ -46,6 +46,7 @@ public final class ClientMetrics private Gauge pausedConnectionsGauge; private Meter requestDiscarded; + private Meter requestDispatched; private Meter protocolException; private Meter unknownException; @@ -68,6 +69,7 @@ public final class ClientMetrics public void unpauseConnection() { pausedConnections.decrementAndGet(); } public void markRequestDiscarded() { requestDiscarded.mark(); } + public void markRequestDispatched() { requestDispatched.mark(); } public List allConnectedClients() { @@ -119,6 +121,7 @@ public final class ClientMetrics pausedConnections = new AtomicInteger(); pausedConnectionsGauge = registerGauge("PausedConnections", pausedConnections::get); requestDiscarded = registerMeter("RequestDiscarded"); + requestDispatched = registerMeter("RequestDispatched"); protocolException = registerMeter("ProtocolException"); unknownException = registerMeter("UnknownException"); diff --git a/src/java/org/apache/cassandra/transport/Dispatcher.java b/src/java/org/apache/cassandra/transport/Dispatcher.java index 3aff2d22e0..45addcf7ab 100644 --- a/src/java/org/apache/cassandra/transport/Dispatcher.java +++ b/src/java/org/apache/cassandra/transport/Dispatcher.java @@ -24,6 +24,7 @@ import java.util.concurrent.TimeUnit; import java.util.function.Consumer; import com.google.common.base.Predicate; +import org.apache.cassandra.metrics.ClientMetrics; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -79,6 +80,7 @@ public class Dispatcher public void dispatch(Channel channel, Message.Request request, FlushItemConverter forFlusher, Overload backpressure) { requestExecutor.submit(() -> processRequest(channel, request, forFlusher, backpressure)); + ClientMetrics.instance.markRequestDispatched(); } /** diff --git a/test/unit/org/apache/cassandra/transport/RateLimitingTest.java b/test/unit/org/apache/cassandra/transport/RateLimitingTest.java index 215542d25c..fda392b898 100644 --- a/test/unit/org/apache/cassandra/transport/RateLimitingTest.java +++ b/test/unit/org/apache/cassandra/transport/RateLimitingTest.java @@ -19,14 +19,17 @@ package org.apache.cassandra.transport; import java.util.Collection; +import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; import java.util.stream.Collectors; +import com.codahale.metrics.Meter; import com.google.common.base.Ticker; import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.metrics.CassandraMetricsRegistry; import org.apache.cassandra.service.StorageService; import org.junit.Before; import org.junit.BeforeClass; @@ -43,6 +46,7 @@ import org.apache.cassandra.utils.Throwables; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; import static org.apache.cassandra.Util.spinAssertEquals; import static org.apache.cassandra.transport.ProtocolVersion.V4; @@ -254,7 +258,9 @@ public class RateLimitingTest extends CQLTester private void testThrowOnOverload(int payloadSize, SimpleClient client) { // The first query takes the one available permit... + long dispatchedPrior = getRequestDispatchedMeter().getCount(); client.execute(queryMessage(payloadSize)); + assertEquals(dispatchedPrior + 1, getRequestDispatchedMeter().getCount()); try { @@ -266,10 +272,15 @@ public class RateLimitingTest extends CQLTester assertTrue(Throwables.anyCauseMatches(e, cause -> cause instanceof OverloadedException)); } + // The last request attempt was rejected and therefore not dispatched. + assertEquals(dispatchedPrior + 1, getRequestDispatchedMeter().getCount()); + // Advance the timeline and verify that we can take a permit again. // (Note that we don't take one when we throw on overload.) tick.addAndGet(ClientResourceLimits.GLOBAL_REQUEST_LIMITER.getIntervalNanos()); client.execute(queryMessage(payloadSize)); + + assertEquals(dispatchedPrior + 2, getRequestDispatchedMeter().getCount()); } private QueryMessage queryMessage(int length) @@ -305,4 +316,13 @@ public class RateLimitingTest extends CQLTester version, KEYSPACE); } + + protected static Meter getRequestDispatchedMeter() + { + String metricName = "org.apache.cassandra.metrics.Client.RequestDispatched"; + Map metrics = CassandraMetricsRegistry.Metrics.getMeters((name, metric) -> name.equals(metricName)); + if (metrics.size() != 1) + fail(String.format("Expected a single registered metric for request dispatched, found %s",metrics.size())); + return metrics.get(metricName); + } }