New approach

Author: ywelsch

distribution/docker/src/test/resources/rest-api-spec/test/20_settings.yml

@@ -89,19 +89,21 @@ example above:
    starting to send TCP keepalive probes. A value of `-1` means not to set
    this option at the socket level but to use the system default instead.
    Only applicable on Linux and Mac, and requires Java 11 or newer.
-   Defaults to 60 on applicable configurations, and -1 otherwise.
+   Defaults to -1. On applicable configurations, this value is set to 5 minutes
+   if the system default or user-configured setting is higher than that.
 * `tcp.keep_interval`: Configures the `TCP_KEEPINTVL` option for this socket,
    which determines the time in seconds between sending TCP keepalive probes.
    A value of `-1` means not to set this option at the socket level but to
    use the system default instead.
    Only applicable on Linux and Mac, and requires Java 11 or newer.
-   Defaults to 10 on applicable configurations, and -1 otherwise.
+   Defaults to -1. On applicable configurations, this value is set to 5 minutes
+   if the system default or user-configured setting is higher than that.
 * `tcp.keep_count`: Configures the `TCP_KEEPCNT` option for this socket, which
    determines the number of unacknowledged TCP keepalive probes that may be
    sent on a connection before it is dropped. A value of `-1` means not to set
    this option at the socket level but to use the system default instead.
    Only applicable on Linux and Mac, and requires Java 11 or newer.
-   Defaults to 3 on applicable configurations, and -1 otherwise.
+   Defaults to -1.
 * `tcp.reuse_address`: Configures the `SO_REUSEADDR` option for this socket
 * `tcp.send_buffer_size`: Configures the send buffer size of the socket
 * `tcp.receive_buffer_size`: Configures the receive buffer size of the socket

docs/reference/modules/transport.asciidoc

@@ -19,8 +19,13 @@
 
 package org.elasticsearch.core.internal.net;
 
+import java.io.IOException;
 import java.lang.reflect.Field;
 import java.net.SocketOption;
+import java.net.StandardSocketOptions;
+import java.nio.channels.ClosedChannelException;
+import java.nio.channels.NetworkChannel;
+import java.util.Arrays;
 
 /**
  * Utilities for network-related methods.
@@ -59,4 +64,45 @@ private static <T> SocketOption<T> getExtendedSocketOptionOrNull(String fieldNam
             return null;
         }
     }
+
+    /**
+     * If SO_KEEPALIVE is enabled (default), this method ensures sane default values for the extended socket options
+     * TCP_KEEPIDLE and TCP_KEEPINTERVAL. The default value for TCP_KEEPIDLE is system dependent, but is typically 2 hours.
+     * Such a high value can result in firewalls eagerly closing these connections. To tell any intermediate devices that
+     * the connection remains alive, we explicitly set these options to 5 minutes if the defaults are higher than that.
+     */
+    public static void setSaneDefaultKeepAliveOptions(NetworkChannel socketChannel) {
+        assert socketChannel != null;
+        try {
+            if (socketChannel.supportedOptions().contains(StandardSocketOptions.SO_KEEPALIVE)) {
+                final Boolean keepalive = socketChannel.getOption(StandardSocketOptions.SO_KEEPALIVE);
+                if (keepalive != null && keepalive.booleanValue()) {
+                    for (SocketOption<Integer> option : Arrays.asList(
+                        NetUtils.getTcpKeepIdleSocketOptionOrNull(),
+                        NetUtils.getTcpKeepIntervalSocketOptionOrNull())) {
+                        setMinValueForSocketOption(socketChannel, option, 300);
+                    }
+                }
+            }
+        } catch (Exception e) {
+            assert e instanceof IOException || e instanceof ClosedChannelException :
+                "unexpected exception when setting channel option: " + e.getClass() + ": " + e.getMessage();
+        }
+    }
+
+    private static void setMinValueForSocketOption(NetworkChannel socketChannel, SocketOption<Integer> option, int minValue) {
+        if (option != null && socketChannel.supportedOptions().contains(option)) {
+            try {
+                final Integer currentIdleVal = socketChannel.getOption(option);
+                if (currentIdleVal != null && currentIdleVal.intValue() > minValue) {
+                    socketChannel.setOption(option, minValue);
+                }
+            } catch (Exception e) {
+                // Getting an exception here should be ok when concurrently closing the channel
+                // An UnsupportedOperationException or IllegalArgumentException, however, should not happen
+                assert e instanceof IOException || e instanceof ClosedChannelException :
+                    "unexpected exception when setting channel option: " + e.getClass() + ": " + e.getMessage();
+            }
+        }
+    }
 }

libs/core/src/main/java/org/elasticsearch/core/internal/net/NetUtils.java

@@ -350,6 +350,7 @@ private void configureSocket(Socket socket, boolean isConnectComplete) throws IO
                     }
                 }
             }
+            NetUtils.setSaneDefaultKeepAliveOptions(socket.getChannel());
             socket.setTcpNoDelay(socketConfig.tcpNoDelay());
             int tcpSendBufferSize = socketConfig.tcpSendBufferSize();
             if (tcpSendBufferSize > 0) {

libs/nio/src/main/java/org/elasticsearch/nio/SocketChannelContext.java

@@ -59,7 +59,7 @@
  * local buffer with a defined size.
  */
 @SuppressForbidden(reason = "Channel#write")
-public class CopyBytesSocketChannel extends NioSocketChannel {
+public class CopyBytesSocketChannel extends Netty4NioSocketChannel {
 
     private static final int MAX_BYTES_PER_WRITE = StrictMath.toIntExact(ByteSizeValue.parseBytesSizeValue(
         System.getProperty("es.transport.buffer.size", "1m"), "es.transport.buffer.size").getBytes());

modules/transport-netty4/src/main/java/org/elasticsearch/transport/CopyBytesSocketChannel.java

@@ -0,0 +1,45 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch 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.elasticsearch.transport;
+
+import io.netty.channel.Channel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+
+import java.nio.channels.SocketChannel;
+
+/**
+ * Helper class to expose {@link #javaChannel()} method
+ */
+public class Netty4NioSocketChannel extends NioSocketChannel {
+
+    public Netty4NioSocketChannel() {
+        super();
+    }
+
+    public Netty4NioSocketChannel(Channel parent, SocketChannel socket) {
+        super(parent, socket);
+    }
+
+    @Override
+    public SocketChannel javaChannel() {
+        return super.javaChannel();
+    }
+
+}

modules/transport-netty4/src/main/java/org/elasticsearch/transport/Netty4NioSocketChannel.java

@@ -68,7 +68,7 @@ public static Class<? extends Channel> getChannelType() {
         if (ALLOCATOR instanceof NoDirectBuffers) {
             return CopyBytesSocketChannel.class;
         } else {
-            return NioSocketChannel.class;
+            return Netty4NioSocketChannel.class;
         }
     }
 

modules/transport-netty4/src/main/java/org/elasticsearch/transport/NettyAllocator.java

@@ -49,12 +49,12 @@
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.util.PageCacheRecycler;
 import org.elasticsearch.common.util.concurrent.EsExecutors;
-import org.elasticsearch.core.internal.io.IOUtils;
 import org.elasticsearch.core.internal.net.NetUtils;
 import org.elasticsearch.indices.breaker.CircuitBreakerService;
 import org.elasticsearch.threadpool.ThreadPool;
-import org.elasticsearch.transport.SharedGroupFactory;
+import org.elasticsearch.transport.Netty4NioSocketChannel;
 import org.elasticsearch.transport.NettyAllocator;
+import org.elasticsearch.transport.SharedGroupFactory;
 import org.elasticsearch.transport.TcpTransport;
 import org.elasticsearch.transport.TransportSettings;
 
@@ -143,6 +143,7 @@ private Bootstrap createClientBootstrap(SharedGroupFactory.SharedGroup sharedGro
         bootstrap.group(sharedGroup.getLowLevelGroup());
 
         // NettyAllocator will return the channel type designed to work with the configured allocator
+        assert Netty4NioSocketChannel.class.isAssignableFrom(NettyAllocator.getChannelType());
         bootstrap.channel(NettyAllocator.getChannelType());
         bootstrap.option(ChannelOption.ALLOCATOR, NettyAllocator.getAllocator());
 
@@ -277,7 +278,6 @@ protected Netty4TcpChannel initiateChannel(DiscoveryNode node) throws IOExceptio
             ExceptionsHelper.maybeDieOnAnotherThread(connectFuture.cause());
             throw new IOException(connectFuture.cause());
         }
-        addClosedExceptionLogger(channel);
 
         Netty4TcpChannel nettyChannel = new Netty4TcpChannel(channel, false, "default", connectFuture);
         channel.attr(CHANNEL_KEY).set(nettyChannel);
@@ -307,6 +307,11 @@ protected class ClientChannelInitializer extends ChannelInitializer<Channel> {
 
         @Override
         protected void initChannel(Channel ch) throws Exception {
+            addClosedExceptionLogger(ch);
+            assert ch instanceof Netty4NioSocketChannel;
+            if (ch instanceof Netty4NioSocketChannel) {
+                NetUtils.setSaneDefaultKeepAliveOptions(((Netty4NioSocketChannel) ch).javaChannel());
+            }
             ch.pipeline().addLast("logging", new ESLoggingHandler());
             // using a dot as a prefix means this cannot come from any settings parsed
             ch.pipeline().addLast("dispatcher", new Netty4MessageChannelHandler(pageCacheRecycler, Netty4Transport.this));
@@ -330,6 +335,10 @@ protected ServerChannelInitializer(String name) {
         @Override
         protected void initChannel(Channel ch) throws Exception {
             addClosedExceptionLogger(ch);
+            assert ch instanceof Netty4NioSocketChannel;
+            if (ch instanceof Netty4NioSocketChannel) {
+                NetUtils.setSaneDefaultKeepAliveOptions(((Netty4NioSocketChannel) ch).javaChannel());
+            }
             Netty4TcpChannel nettyTcpChannel = new Netty4TcpChannel(ch, true, name, ch.newSucceededFuture());
             ch.attr(CHANNEL_KEY).set(nettyTcpChannel);
             ch.pipeline().addLast("logging", new ESLoggingHandler());

modules/transport-netty4/src/main/java/org/elasticsearch/transport/netty4/Netty4Transport.java

@@ -21,28 +21,41 @@
 
 import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.bootstrap.JavaVersion;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.network.NetworkService;
 import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.transport.TransportAddress;
 import org.elasticsearch.common.util.PageCacheRecycler;
+import org.elasticsearch.core.internal.io.IOUtils;
+import org.elasticsearch.core.internal.net.NetUtils;
 import org.elasticsearch.indices.breaker.NoneCircuitBreakerService;
+import org.elasticsearch.test.transport.MockTransportService;
+import org.elasticsearch.test.transport.StubbableTransport;
 import org.elasticsearch.transport.AbstractSimpleTransportTestCase;
 import org.elasticsearch.transport.ConnectTransportException;
 import org.elasticsearch.transport.ConnectionProfile;
+import org.elasticsearch.transport.Netty4NioSocketChannel;
 import org.elasticsearch.transport.SharedGroupFactory;
 import org.elasticsearch.transport.TcpChannel;
+import org.elasticsearch.transport.TcpTransport;
+import org.elasticsearch.transport.TestProfiles;
 import org.elasticsearch.transport.Transport;
 
+import java.io.IOException;
 import java.net.InetAddress;
 import java.net.UnknownHostException;
+import java.nio.channels.SocketChannel;
 import java.util.Collections;
 
 import static java.util.Collections.emptyMap;
 import static java.util.Collections.emptySet;
 import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.hasItem;
+import static org.hamcrest.Matchers.instanceOf;
+import static org.hamcrest.Matchers.lessThanOrEqualTo;
 
 public class SimpleNetty4TransportTests extends AbstractSimpleTransportTestCase {
 
@@ -75,4 +88,50 @@ public void testConnectException() throws UnknownHostException {
             assertThat(e.getMessage(), containsString("[127.0.0.1:9876]"));
         }
     }
+
+    public void testDefaultKeepAliveSettings() throws IOException {
+        assumeTrue("setting default keepalive options not supported on this platform",
+            (IOUtils.LINUX || IOUtils.MAC_OS_X) &&
+                JavaVersion.current().compareTo(JavaVersion.parse("11")) >= 0);
+        try (MockTransportService serviceC = buildService("TS_C", Version.CURRENT, Settings.EMPTY);
+            MockTransportService serviceD = buildService("TS_D", Version.CURRENT, Settings.EMPTY)) {
+            serviceC.start();
+            serviceC.acceptIncomingRequests();
+            serviceD.start();
+            serviceD.acceptIncomingRequests();
+
+            try (Transport.Connection connection = openConnection(serviceC, serviceD.getLocalDiscoNode(), TestProfiles.LIGHT_PROFILE)) {
+                assertThat(connection, instanceOf(StubbableTransport.WrappedConnection.class));
+                Transport.Connection conn = ((StubbableTransport.WrappedConnection) connection).getConnection();
+                assertThat(conn, instanceOf(TcpTransport.NodeChannels.class));
+                TcpTransport.NodeChannels nodeChannels = (TcpTransport.NodeChannels) conn;
+                for (TcpChannel channel : nodeChannels.getChannels()) {
+                    assertFalse(channel.isServerChannel());
+                    checkDefaultKeepAliveOptions(channel);
+                }
+
+                assertThat(serviceD.getOriginalTransport(), instanceOf(TcpTransport.class));
+                for (TcpChannel channel : getAcceptedChannels((TcpTransport) serviceD.getOriginalTransport())) {
+                    assertTrue(channel.isServerChannel());
+                    checkDefaultKeepAliveOptions(channel);
+                }
+            }
+        }
+    }
+
+    private void checkDefaultKeepAliveOptions(TcpChannel channel) throws IOException {
+        assertThat(channel, instanceOf(Netty4TcpChannel.class));
+        Netty4TcpChannel nettyChannel = (Netty4TcpChannel) channel;
+        assertThat(nettyChannel.getNettyChannel(), instanceOf(Netty4NioSocketChannel.class));
+        Netty4NioSocketChannel netty4NioSocketChannel = (Netty4NioSocketChannel) nettyChannel.getNettyChannel();
+        SocketChannel socketChannel = netty4NioSocketChannel.javaChannel();
+        assertThat(socketChannel.supportedOptions(), hasItem(NetUtils.getTcpKeepIdleSocketOptionOrNull()));
+        Integer keepIdle = socketChannel.getOption(NetUtils.getTcpKeepIdleSocketOptionOrNull());
+        assertNotNull(keepIdle);
+        assertThat(keepIdle, lessThanOrEqualTo(500));
+        assertThat(socketChannel.supportedOptions(), hasItem(NetUtils.getTcpKeepIntervalSocketOptionOrNull()));
+        Integer keepInterval = socketChannel.getOption(NetUtils.getTcpKeepIntervalSocketOptionOrNull());
+        assertNotNull(keepInterval);
+        assertThat(keepInterval, lessThanOrEqualTo(500));
+    }
 }