[cassandra] branch cassandra-3.0 updated: ConnectionLimitHandler may leaks connection count if remote connection drops

[jo...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

jonmeredith pushed a commit to branch cassandra-3.0
in repository https://gitbox.apache.org/repos/asf/cassandra.git


The following commit(s) were added to refs/heads/cassandra-3.0 by this push:
     new 647ec92  ConnectionLimitHandler may leaks connection count if remote connection drops
647ec92 is described below

commit 647ec92db17e1b1a3677e7880849a29c0c30adde
Author: Jon Meredith <jo...@apache.org>
AuthorDate: Fri Mar 18 17:17:17 2022 -0600

    ConnectionLimitHandler may leaks connection count if remote connection drops
    
    patch by Jon Meredith; reviewed by Caleb Rackliffe for CASSANDRA-17252
---
 CHANGES.txt                                        |  1 +
 .../transport/ConnectionLimitHandler.java          | 44 ++++++++++++++++++++--
 2 files changed, 41 insertions(+), 4 deletions(-)

diff --git a/CHANGES.txt b/CHANGES.txt
index 36d6064..28fe6ec 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
 3.0.27
+ * ConnectionLimitHandler may leaks connection count if remote connection drops (CASSANDRA-17252)
  * Require ant >= 1.10 (CASSANDRA-17428)
  * Disallow CONTAINS for UPDATE and DELETE (CASSANDRA-15266)
  * filter out NULL_VERSION entries from peers table in ConfiguredLimit (CASSANDRA-16518)
diff --git a/src/java/org/apache/cassandra/transport/ConnectionLimitHandler.java b/src/java/org/apache/cassandra/transport/ConnectionLimitHandler.java
index 7bcf280..271004d 100644
--- a/src/java/org/apache/cassandra/transport/ConnectionLimitHandler.java
+++ b/src/java/org/apache/cassandra/transport/ConnectionLimitHandler.java
@@ -18,15 +18,19 @@
 package org.apache.cassandra.transport;
 
 
+import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.util.Attribute;
+import io.netty.util.AttributeKey;
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
+import java.net.SocketAddress;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.atomic.AtomicLong;
@@ -40,9 +44,37 @@ import java.util.concurrent.atomic.AtomicLong;
 final class ConnectionLimitHandler extends ChannelInboundHandlerAdapter
 {
     private static final Logger logger = LoggerFactory.getLogger(ConnectionLimitHandler.class);
+    private static final AttributeKey<InetAddress> addressAttributeKey = AttributeKey.valueOf("address");
     private final ConcurrentMap<InetAddress, AtomicLong> connectionsPerClient = new ConcurrentHashMap<>();
     private final AtomicLong counter = new AtomicLong(0);
 
+    // Keep the remote address as a channel attribute.  The channel inactive callback needs
+    // to know the entry into the connetionsPerClient map and depending on the state of the remote
+    // an exception may be thrown trying to retrieve the address. Make sure the same address used
+    // to increment is used for decrement.
+    private static InetAddress setRemoteAddressAttribute(Channel channel)
+    {
+        Attribute<InetAddress> addressAttribute = channel.attr(addressAttributeKey);
+        SocketAddress remoteAddress = channel.remoteAddress();
+        if (remoteAddress instanceof InetSocketAddress)
+        {
+            addressAttribute.setIfAbsent(((InetSocketAddress) remoteAddress).getAddress());
+        }
+        else
+        {
+            logger.warn("Remote address of unknown type: {}, skipping per-IP connection limits",
+                              remoteAddress.getClass());
+        }
+        return addressAttribute.get();
+    }
+
+    private static InetAddress getRemoteAddressAttribute(Channel channel)
+    {
+        Attribute<InetAddress> addressAttribute = channel.attr(addressAttributeKey);
+        return addressAttribute.get();
+    }
+
+
     @Override
     public void channelActive(ChannelHandlerContext ctx) throws Exception
     {
@@ -64,8 +96,12 @@ final class ConnectionLimitHandler extends ChannelInboundHandlerAdapter
             long perIpLimit = DatabaseDescriptor.getNativeTransportMaxConcurrentConnectionsPerIp();
             if (perIpLimit > 0)
             {
-                InetAddress address = ((InetSocketAddress) ctx.channel().remoteAddress()).getAddress();
-
+                InetAddress address = setRemoteAddressAttribute(ctx.channel());
+                if (address == null)
+                {
+                    ctx.close();
+                    return;
+                }
                 AtomicLong perIpCount = connectionsPerClient.get(address);
                 if (perIpCount == null)
                 {
@@ -93,9 +129,9 @@ final class ConnectionLimitHandler extends ChannelInboundHandlerAdapter
     public void channelInactive(ChannelHandlerContext ctx) throws Exception
     {
         counter.decrementAndGet();
-        InetAddress address = ((InetSocketAddress) ctx.channel().remoteAddress()).getAddress();
+        InetAddress address = getRemoteAddressAttribute(ctx.channel());
 
-        AtomicLong count = connectionsPerClient.get(address);
+        AtomicLong count = address == null ? null : connectionsPerClient.get(address);
         if (count != null)
         {
             if (count.decrementAndGet() <= 0)

---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org