Backport of CASSANDRA-17812: Rate-limit new client connection auth setup to avoid overwhelming bcrypt

This backport differs from CASSANDRA-17812 in such a way that by default the number
of auth request threads is set to 0. That will route all requests to request executor as before this change.
The patch in 5.0 and later sets the default number of auth request threads to 4.

patch by Josh McKenzie; reviewed by Chris Lohfink for CASSANDRA-20057

Co-authored-by: Stefan Miklosovic <smiklosovic@apache.org>

CHANGES.txt

@@ -1,4 +1,5 @@
 4.0.15
+ * Backport of CASSANDRA-17812: Rate-limit new client connection auth setup to avoid overwhelming bcrypt (CASSANDRA-20057)
  * Support UDTs and vectors as clustering keys in descending order (CASSANDRA-20050)
  * Fix CQL in snapshot's schema which did not contained UDTs used as reverse clustering columns (CASSANDRA-20036)
  * Add configurable batchlog endpoint strategies: random_remote, prefer_local, dynamic_remote, and dynamic (CASSANDRA-18120)

conf/cassandra.yaml

@@ -764,6 +764,9 @@ native_transport_port: 9042
 # The maximum threads for handling requests (note that idle threads are stopped
 # after 30 seconds so there is not corresponding minimum setting).
 # native_transport_max_threads: 128
+# The maximum threads for handling auth requests in a separate executor from main request executor.
+# When set to 0, main executor for requests is used.
+# native_transport_max_auth_threads: 0
 #
 # The maximum size of allowed frame. Frame (requests) larger than this will
 # be rejected as invalid. The default is 256MB. If you're changing this parameter,

src/java/org/apache/cassandra/config/Config.java

@@ -195,6 +195,8 @@ public class Config
     public Integer native_transport_port_ssl = null;
     public int native_transport_max_threads = 128;
     public int native_transport_max_frame_size_in_mb = 256;
+    /** 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;
     public volatile long native_transport_max_concurrent_connections_per_ip = -1L;
     public boolean native_transport_flush_in_batches_legacy = false;

src/java/org/apache/cassandra/config/DatabaseDescriptor.java

@@ -2259,6 +2259,22 @@ public class DatabaseDescriptor
         conf.native_transport_max_threads = max_threads;
     }
 
+    public static Integer getNativeTransportMaxAuthThreads()
+    {
+        return conf.native_transport_max_auth_threads;
+    }
+
+    /**
+     * If this value is set to <= 0 it will move auth requests to the standard request pool regardless of the current
+     * size of the authExecutor in {@link org.apache.cassandra.transport.Dispatcher}'s active size.
+     *
+     * see {@link org.apache.cassandra.transport.Dispatcher#dispatch} for executor selection
+     */
+    public static void setNativeTransportMaxAuthThreads(int threads)
+    {
+        conf.native_transport_max_auth_threads = threads;
+    }
+
     public static int getNativeTransportMaxFrameSize()
     {
         return (int) ByteUnit.MEBI_BYTES.toBytes(conf.native_transport_max_frame_size_in_mb);

src/java/org/apache/cassandra/transport/Dispatcher.java

@@ -22,6 +22,8 @@ import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.function.Consumer;
 
+import com.google.common.annotations.VisibleForTesting;
+
 import io.netty.channel.Channel;
 import io.netty.channel.EventLoop;
 import io.netty.util.AttributeKey;
@@ -39,11 +41,33 @@ import static org.apache.cassandra.concurrent.SharedExecutorPool.SHARED;
 
 public class Dispatcher
 {
-    private static final LocalAwareExecutorService requestExecutor = SHARED.newExecutor(DatabaseDescriptor.getNativeTransportMaxThreads(),
+    @VisibleForTesting
+    static final LocalAwareExecutorService requestExecutor = SHARED.newExecutor(DatabaseDescriptor.getNativeTransportMaxThreads(),
                                                                                         DatabaseDescriptor::setNativeTransportMaxThreads,
                                                                                         "transport",
                                                                                         "Native-Transport-Requests");
 
+    /** CASSANDRA-17812: Rate-limit new client connection setup to avoid overwhelming during bcrypt
+     *
+     * Backported by CASSANDRA-20057
+     *
+     * authExecutor is a separate thread pool for handling requests on connections that need to be authenticated.
+     * Calls to AUTHENTICATE can be expensive if the number of rounds for bcrypt is configured to a high value,
+     * so during a connection storm checking the password hash would starve existing connected clients for CPU and
+     * trigger timeouts if on the same thread pool as standard requests.
+     *
+     * Moving authentication requests to a small, separate pool prevents starvation handling all other
+     * requests. If the authExecutor pool backs up, it may cause authentication timeouts but the clients should
+     * back off and retry while the rest of the system continues to make progress.
+     *
+     * Setting less than 1 will service auth requests on the standard {@link Dispatcher#requestExecutor}
+     */
+    @VisibleForTesting
+    static final LocalAwareExecutorService authExecutor = SHARED.newExecutor(Math.max(1, DatabaseDescriptor.getNativeTransportMaxAuthThreads()),
+                                                                             DatabaseDescriptor::setNativeTransportMaxAuthThreads,
+                                                                             "transport",
+                                                                             "Native-Transport-Auth-Requests");
+
     private static final ConcurrentMap<EventLoop, Flusher> flusherLookup = new ConcurrentHashMap<>();
     private final boolean useLegacyFlusher;
 
@@ -67,7 +91,14 @@ public class Dispatcher
 
     public void dispatch(Channel channel, Message.Request request, FlushItemConverter forFlusher)
     {
-        requestExecutor.submit(() -> processRequest(channel, request, forFlusher));
+        // if native_transport_max_auth_threads is < 1, don't delegate to new pool on auth messages
+        boolean isAuthQuery = DatabaseDescriptor.getNativeTransportMaxAuthThreads() > 0 &&
+                              (request.type == Message.Type.AUTH_RESPONSE || request.type == Message.Type.CREDENTIALS);
+
+        // Importantly, the authExecutor will handle the AUTHENTICATE message which may be CPU intensive.
+        LocalAwareExecutorService executor = isAuthQuery ? authExecutor : requestExecutor;
+
+        executor.submit(() -> processRequest(channel, request, forFlusher));
     }
 
     /**
@@ -140,13 +171,10 @@ public class Dispatcher
 
     public static void shutdown()
     {
-        if (requestExecutor != null)
+        requestExecutor.shutdown();
-        {
+        authExecutor.shutdown();
-            requestExecutor.shutdown();
-        }
     }
 
-
     /**
      * Dispatcher for EventMessages. In {@link Server.ConnectionTracker#send(Event)}, the strategy
      * for delivering events to registered clients is dependent on protocol version and the configuration

test/unit/org/apache/cassandra/transport/MessageDispatcherTest.java

@@ -0,0 +1,180 @@
+/*
+ * 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.Collections;
+import java.util.concurrent.Callable;
+import java.util.concurrent.TimeUnit;
+
+import com.google.common.util.concurrent.Uninterruptibles;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import io.netty.channel.Channel;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.metrics.ClientMetrics;
+import org.apache.cassandra.service.QueryState;
+import org.apache.cassandra.transport.messages.AuthResponse;
+
+public class MessageDispatcherTest
+{
+    static final Message.Request AUTH_RESPONSE_REQUEST = new AuthResponse(new byte[0])
+    {
+        public Response execute(QueryState queryState, long queryStartNanoTime, boolean traceRequest)
+        {
+            return null;
+        }
+    };
+
+    private static AuthTestDispatcher dispatch;
+    private static int maxAuthThreadsBeforeTests;
+
+    @BeforeClass
+    public static void init() throws Exception
+    {
+        DatabaseDescriptor.daemonInitialization();
+        ClientMetrics.instance.init(Collections.emptyList());
+        maxAuthThreadsBeforeTests = DatabaseDescriptor.getNativeTransportMaxAuthThreads();
+        dispatch = new AuthTestDispatcher();
+    }
+
+    @AfterClass
+    public static void restoreAuthSize()
+    {
+        DatabaseDescriptor.setNativeTransportMaxAuthThreads(maxAuthThreadsBeforeTests);
+    }
+
+    @Test
+    public void testAuthRateLimiterTurnedOffByDefault() throws Exception
+    {
+        // All requests were executed on main request executor
+        Assert.assertEquals(0, tryAuth(this::completedAuth));
+        Assert.assertEquals(1, completedRequests());
+    }
+
+    @Test
+    public void testAuthRateLimiter() throws Exception
+    {
+        long startRequests = completedRequests();
+
+        DatabaseDescriptor.setNativeTransportMaxAuthThreads(1);
+        long auths = tryAuth(this::completedAuth);
+        Assert.assertEquals(auths, 1);
+
+        DatabaseDescriptor.setNativeTransportMaxAuthThreads(100);
+        auths = tryAuth(this::completedAuth);
+        Assert.assertEquals(auths, 1);
+
+        // Make sure no tasks executed on the regular pool
+        Assert.assertEquals(startRequests, completedRequests());
+    }
+
+    @Test
+    public void testAuthRateLimiterNotUsed() throws Exception
+    {
+        DatabaseDescriptor.setNativeTransportMaxAuthThreads(1);
+        for (Message.Type type : Message.Type.values())
+        {
+            if (type == Message.Type.AUTH_RESPONSE || type == Message.Type.CREDENTIALS || type.direction != Message.Direction.REQUEST)
+                continue;
+
+            long auths = completedAuth();
+            long requests = tryAuth(this::completedRequests, new Message.Request(type)
+            {
+                public Response execute(QueryState queryState, long queryStartNanoTime, boolean traceRequest)
+                {
+                    return null;
+                }
+            });
+            Assert.assertEquals(requests, 1);
+            Assert.assertEquals(completedAuth() - auths, 0);
+        }
+    }
+
+    @Test
+    public void testAuthRateLimiterDisabled() throws Exception
+    {
+        long startAuthRequests = completedAuth();
+
+        DatabaseDescriptor.setNativeTransportMaxAuthThreads(0);
+        long requests = tryAuth(this::completedRequests);
+        Assert.assertEquals(requests, 1);
+
+        DatabaseDescriptor.setNativeTransportMaxAuthThreads(-1);
+        requests = tryAuth(this::completedRequests);
+        Assert.assertEquals(requests, 1);
+
+        DatabaseDescriptor.setNativeTransportMaxAuthThreads(-1000);
+        requests = tryAuth(this::completedRequests);
+        Assert.assertEquals(requests, 1);
+
+        // Make sure no tasks executed on the auth pool
+        Assert.assertEquals(startAuthRequests, completedAuth());
+    }
+
+    private long completedRequests()
+    {
+        return Dispatcher.requestExecutor.getCompletedTaskCount();
+    }
+
+    private long completedAuth()
+    {
+        return Dispatcher.authExecutor.getCompletedTaskCount();
+    }
+
+    public long tryAuth(Callable<Long> check) throws Exception
+    {
+        return tryAuth(check, AUTH_RESPONSE_REQUEST);
+    }
+
+    @SuppressWarnings("UnstableApiUsage")
+    public long tryAuth(Callable<Long> check, Message.Request request) throws Exception
+    {
+        long start = check.call();
+        dispatch.dispatch(null, request, (channel,req,response) -> null);
+
+        // While this is timeout based, we should be *well below* a full second on any of this processing in any sane environment.
+        long timeout = System.currentTimeMillis();
+        while(start == check.call() && System.currentTimeMillis() - timeout < 1000)
+        {
+            Uninterruptibles.sleepUninterruptibly(10, TimeUnit.MILLISECONDS);
+        }
+        return check.call() - start;
+    }
+
+    public static class AuthTestDispatcher extends Dispatcher
+    {
+        public AuthTestDispatcher()
+        {
+            super(false);
+        }
+
+
+
+        @Override
+        void processRequest(Channel channel,
+                            Message.Request request,
+                            FlushItemConverter forFlusher)
+        {
+            // noop
+        }
+    }
+}