HCD-247: Port CASSANDRA-19483 (#2170)

Ports CASSANDRA-19483 which added an option to group invalid legacy
protocol magic exception with an no-spam logger.

The test was slightly modified, it contains the timeout for log watching
and ensure the logs start from the mark that happens before the
exception. It ensures the test does not hang without the fix it covers.

The motivation behind the port is to fix
https://datastax.jira.com/browse/DSP-24639

Author: szymon-miezal

CHANGES.txt

@@ -12,6 +12,7 @@ Merged from 4.0:
  * Add new concurrent_merkle_tree_requests config property to prevent OOM during multi-range and/or multi-table repairs (CASSANDRA-19336)
 
 4.1
+ * Optionally skip exception logging on invalid legacy protocol magic exception (CASSANDRA-19483)
  * Request-Based Native Transport Rate-Limiting (CASSANDRA-16663)
  * Add information whether sstables are dropped to SchemaChangeListener (CASSANDRA-17582)
  * Save sstable id as string in activity table (CASSANDRA-17585)

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

@@ -635,6 +635,7 @@ public static void setClientMode(boolean clientMode)
     }
 
     public volatile int consecutive_message_errors_threshold = 1;
+    public volatile boolean invalid_legacy_protocol_magic_no_spam_enabled = false;
 
     /**
      * Default compaction configuration, used if a table does not specify any.

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

@@ -3707,4 +3707,9 @@ public static void setAnnBruteForceExpenseFactor(double factor)
         Preconditions.checkArgument(factor <= StorageAttachedIndexOptions.MAXIMUM_ANN_BRUTE_FORCE_FACTOR, "ANN brute force expense factor must be at most " + StorageAttachedIndexOptions.MAXIMUM_ANN_BRUTE_FORCE_FACTOR);
         conf.sai_options.ann_brute_force_factor = factor;
     }
+
+    public static boolean getInvalidLegacyProtocolMagicNoSpamEnabled()
+    {
+        return conf.invalid_legacy_protocol_magic_no_spam_enabled;
+    }
 }

src/java/org/apache/cassandra/net/InboundConnectionInitiator.java

@@ -22,9 +22,11 @@
 import java.net.SocketAddress;
 import java.util.List;
 import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Throwables;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -44,12 +46,14 @@
 import io.netty.handler.logging.LoggingHandler;
 import io.netty.handler.ssl.SslContext;
 import io.netty.handler.ssl.SslHandler;
+import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.config.EncryptionOptions;
 import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.locator.InetAddressAndPort;
 import org.apache.cassandra.net.OutboundConnectionSettings.Framing;
 import org.apache.cassandra.security.SSLFactory;
 import org.apache.cassandra.streaming.async.StreamingInboundHandler;
+import org.apache.cassandra.utils.NoSpamLogger;
 import org.apache.cassandra.utils.memory.BufferPools;
 
 import static java.lang.Math.max;
@@ -67,6 +71,8 @@ public class InboundConnectionInitiator
 {
     private static final Logger logger = LoggerFactory.getLogger(InboundConnectionInitiator.class);
 
+    private static final NoSpamLogger noSpam5m = NoSpamLogger.getLogger(logger, 5, TimeUnit.MINUTES);
+
     private static class Initializer extends ChannelInitializer<SocketChannel>
     {
         private final InboundConnectionSettings settings;
@@ -399,7 +405,11 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
 
         private void exceptionCaught(Channel channel, Throwable cause)
         {
-            logger.error("Failed to properly handshake with peer {}. Closing the channel.", channel.remoteAddress(), cause);
+            if (cause != null && Throwables.getRootCause(cause) instanceof Message.InvalidLegacyProtocolMagic && DatabaseDescriptor.getInvalidLegacyProtocolMagicNoSpamEnabled())
+                noSpam5m.warn("Failed to properly handshake with peer {}. Closing the channel. Invalid legacy protocol magic.", ((InetSocketAddress) channel.remoteAddress()).getHostName());
+            else
+                logger.error("Failed to properly handshake with peer {}. Closing the channel.", channel.remoteAddress(), cause);
+
             try
             {
                 failHandshake(channel);

test/distributed/org/apache/cassandra/distributed/test/InternodeErrorExclusionTest.java

@@ -0,0 +1,69 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed 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.distributed.test;
+
+import java.time.Duration;
+
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.Feature;
+import org.apache.cassandra.transport.SimpleClient;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class InternodeErrorExclusionTest extends TestBaseImpl
+{
+    @BeforeClass
+    public static void beforeClass2()
+    {
+        DatabaseDescriptor.clientInitialization();
+    }
+
+    @Test
+    public void testNoSpammingInvalidLegacyProtocolMagicException() throws Throwable
+    {
+        try (Cluster cluster = Cluster.build(1)
+                                      .withConfig(c -> c
+                                                       .with(Feature.NETWORK)
+                                                       .set("invalid_legacy_protocol_magic_no_spam_enabled", true))
+                                      .start())
+        {
+            long logMark = cluster.get(1).logs().mark();
+            causeException();
+            causeException();
+            // we used no spam logger so the second message will not be emitted (the size is still 1).
+            assertThat(cluster.get(1).logs().watchFor(logMark, Duration.ofSeconds(30), "Failed to properly handshake with peer localhost. Closing the channel. Invalid legacy protocol magic.").getResult()).hasSize(1);
+        }
+    }
+
+    private void causeException()
+    {
+        try (SimpleClient client = SimpleClient.builder("127.0.0.1", 7012).build())
+        {
+            client.connect(true);
+            Assert.fail("Connection should fail");
+        }
+        catch (Exception e)
+        {
+            // expected
+        }
+    }
+}