Add native transport rate limiter options to example cassandra.yaml, and expose metric for dispatch rate

patch by Caleb Rackliffe; reviewed by Josh McKenzie for CASSANDRA-17423

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)

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

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.

src/java/org/apache/cassandra/metrics/ClientMetrics.java

@@ -46,6 +46,7 @@ public final class ClientMetrics
     private Gauge<Integer> 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<ConnectedClient> 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");

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();
     }
 
     /**

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<String, Meter> 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);
+    }
 }