diff --git a/CHANGES.txt b/CHANGES.txt index d8cab35210..0eb7ed5f7e 100644 --- a/CHANGES.txt +++ b/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) diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml index 835a8f4fa6..2f11843921 100644 --- a/conf/cassandra.yaml +++ b/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, diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java index 995aaa69d2..57388eed68 100644 --- a/src/java/org/apache/cassandra/config/Config.java +++ b/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; diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index d425e53807..bf96a3f885 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/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); diff --git a/src/java/org/apache/cassandra/transport/Dispatcher.java b/src/java/org/apache/cassandra/transport/Dispatcher.java index 05b55e8010..e1318d1afe 100644 --- a/src/java/org/apache/cassandra/transport/Dispatcher.java +++ b/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 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(); - } + requestExecutor.shutdown(); + authExecutor.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 diff --git a/test/unit/org/apache/cassandra/transport/MessageDispatcherTest.java b/test/unit/org/apache/cassandra/transport/MessageDispatcherTest.java new file mode 100644 index 0000000000..244cd2f4fe --- /dev/null +++ b/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 check) throws Exception + { + return tryAuth(check, AUTH_RESPONSE_REQUEST); + } + + @SuppressWarnings("UnstableApiUsage") + public long tryAuth(Callable 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 + } + } +}