Unify http channels and exception handling (#31379)

This is a general cleanup of channels and exception handling in http.
This commit introduces a CloseableChannel that is a superclass of
TcpChannel and HttpChannel. This allows us to unify the closing logic
between tcp and http transports. Additionally, the normal http channels
are extracted to the abstract server transport.

Finally, this commit (mostly) unifies the exception handling between nio
and netty4 http server transports.

Author: Tim-Brooks

modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpChannel.java

@@ -22,6 +22,7 @@
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelPromise;
 import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.common.concurrent.CompletableContext;
 import org.elasticsearch.http.HttpChannel;
 import org.elasticsearch.http.HttpResponse;
 import org.elasticsearch.transport.netty4.Netty4Utils;
@@ -31,9 +32,23 @@
 public class Netty4HttpChannel implements HttpChannel {
 
     private final Channel channel;
+    private final CompletableContext<Void> closeContext = new CompletableContext<>();
 
     Netty4HttpChannel(Channel channel) {
         this.channel = channel;
+        this.channel.closeFuture().addListener(f -> {
+            if (f.isSuccess()) {
+                closeContext.complete(null);
+            } else {
+                Throwable cause = f.cause();
+                if (cause instanceof Error) {
+                    Netty4Utils.maybeDie(cause);
+                    closeContext.completeExceptionally(new Exception(cause));
+                } else {
+                    closeContext.completeExceptionally((Exception) cause);
+                }
+            }
+        });
     }
 
     @Override
@@ -65,6 +80,16 @@ public InetSocketAddress getRemoteAddress() {
         return (InetSocketAddress) channel.remoteAddress();
     }
 
+    @Override
+    public void addCloseListener(ActionListener<Void> listener) {
+        closeContext.addListener(ActionListener.toBiConsumer(listener));
+    }
+
+    @Override
+    public boolean isOpen() {
+        return channel.isOpen();
+    }
+
     @Override
     public void close() {
         channel.close();
@@ -73,4 +98,12 @@ public void close() {
     public Channel getNettyChannel() {
         return channel;
     }
+
+    @Override
+    public String toString() {
+        return "Netty4HttpChannel{" +
+            "localAddress=" + getLocalAddress() +
+            ", remoteAddress=" + getRemoteAddress() +
+            '}';
+    }
 }

modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpRequestHandler.java

@@ -29,6 +29,8 @@
 import org.elasticsearch.http.HttpPipelinedRequest;
 import org.elasticsearch.transport.netty4.Netty4Utils;
 
+import static org.elasticsearch.http.netty4.Netty4HttpServerTransport.HTTP_CHANNEL_KEY;
+
 @ChannelHandler.Sharable
 class Netty4HttpRequestHandler extends SimpleChannelInboundHandler<HttpPipelinedRequest<FullHttpRequest>> {
 
@@ -40,7 +42,7 @@ class Netty4HttpRequestHandler extends SimpleChannelInboundHandler<HttpPipelined
 
     @Override
     protected void channelRead0(ChannelHandlerContext ctx, HttpPipelinedRequest<FullHttpRequest> msg) throws Exception {
-        Netty4HttpChannel channel = ctx.channel().attr(Netty4HttpServerTransport.HTTP_CHANNEL_KEY).get();
+        Netty4HttpChannel channel = ctx.channel().attr(HTTP_CHANNEL_KEY).get();
         FullHttpRequest request = msg.getRequest();
 
         try {
@@ -75,7 +77,12 @@ protected void channelRead0(ChannelHandlerContext ctx, HttpPipelinedRequest<Full
     @Override
     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
         Netty4Utils.maybeDie(cause);
-        serverTransport.exceptionCaught(ctx, cause);
+        Netty4HttpChannel httpChannel = ctx.channel().attr(HTTP_CHANNEL_KEY).get();
+        if (cause instanceof Error) {
+            serverTransport.onException(httpChannel, new Exception(cause));
+        } else {
+            serverTransport.onException(httpChannel, (Exception) cause);
+        }
     }
 
 }

modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpServerTransport.java

@@ -40,15 +40,13 @@
 import io.netty.handler.timeout.ReadTimeoutException;
 import io.netty.handler.timeout.ReadTimeoutHandler;
 import io.netty.util.AttributeKey;
-import org.apache.logging.log4j.message.ParameterizedMessage;
-import org.apache.logging.log4j.util.Supplier;
 import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.network.CloseableChannel;
 import org.elasticsearch.common.network.NetworkAddress;
 import org.elasticsearch.common.network.NetworkService;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Setting.Property;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.common.transport.NetworkExceptionHelper;
 import org.elasticsearch.common.transport.TransportAddress;
 import org.elasticsearch.common.unit.ByteSizeUnit;
 import org.elasticsearch.common.unit.ByteSizeValue;
@@ -57,14 +55,14 @@
 import org.elasticsearch.common.xcontent.NamedXContentRegistry;
 import org.elasticsearch.http.AbstractHttpServerTransport;
 import org.elasticsearch.http.BindHttpException;
+import org.elasticsearch.http.HttpChannel;
 import org.elasticsearch.http.HttpHandlingSettings;
 import org.elasticsearch.http.HttpStats;
 import org.elasticsearch.http.netty4.cors.Netty4CorsConfig;
 import org.elasticsearch.http.netty4.cors.Netty4CorsConfigBuilder;
 import org.elasticsearch.http.netty4.cors.Netty4CorsHandler;
 import org.elasticsearch.rest.RestUtils;
 import org.elasticsearch.threadpool.ThreadPool;
-import org.elasticsearch.transport.netty4.Netty4OpenChannelsHandler;
 import org.elasticsearch.transport.netty4.Netty4Utils;
 
 import java.io.IOException;
@@ -171,10 +169,6 @@ public class Netty4HttpServerTransport extends AbstractHttpServerTransport {
 
     protected final List<Channel> serverChannels = new ArrayList<>();
 
-    // package private for testing
-    Netty4OpenChannelsHandler serverOpenChannels;
-
-
     private final Netty4CorsConfig corsConfig;
 
     public Netty4HttpServerTransport(Settings settings, NetworkService networkService, BigArrays bigArrays, ThreadPool threadPool,
@@ -216,8 +210,6 @@ public Settings settings() {
     protected void doStart() {
         boolean success = false;
         try {
-            this.serverOpenChannels = new Netty4OpenChannelsHandler(logger);
-
             serverBootstrap = new ServerBootstrap();
 
             serverBootstrap.group(new NioEventLoopGroup(workerCount, daemonThreadFactory(settings,
@@ -281,10 +273,9 @@ static Netty4CorsConfig buildCorsConfig(Settings settings) {
             builder.allowCredentials();
         }
         String[] strMethods = Strings.tokenizeToStringArray(SETTING_CORS_ALLOW_METHODS.get(settings), ",");
-        HttpMethod[] methods = Arrays.asList(strMethods)
-            .stream()
+        HttpMethod[] methods = Arrays.stream(strMethods)
             .map(HttpMethod::valueOf)
-            .toArray(size -> new HttpMethod[size]);
+            .toArray(HttpMethod[]::new);
         return builder.allowedRequestMethods(methods)
             .maxAge(SETTING_CORS_MAX_AGE.get(settings))
             .allowedRequestHeaders(Strings.tokenizeToStringArray(SETTING_CORS_ALLOW_HEADERS.get(settings), ","))
@@ -327,15 +318,21 @@ protected void doStop() {
                     Netty4Utils.closeChannels(serverChannels);
                 } catch (IOException e) {
                     logger.trace("exception while closing channels", e);
+                } finally {
+                    serverChannels.clear();
                 }
-                serverChannels.clear();
             }
         }
 
-        if (serverOpenChannels != null) {
-            serverOpenChannels.close();
-            serverOpenChannels = null;
+        // TODO: Move all of channel closing to abstract class once server channels are handled
+        try {
+            CloseableChannel.closeChannels(new ArrayList<>(httpChannels), true);
+        } catch (Exception e) {
+            logger.warn("unexpected exception while closing http channels", e);
         }
+        httpChannels.clear();
+
+
 
         if (serverBootstrap != null) {
             serverBootstrap.config().group().shutdownGracefully(0, 5, TimeUnit.SECONDS).awaitUninterruptibly();
@@ -349,38 +346,18 @@ protected void doClose() {
 
     @Override
     public HttpStats stats() {
-        Netty4OpenChannelsHandler channels = serverOpenChannels;
-        return new HttpStats(channels == null ? 0 : channels.numberOfOpenChannels(), channels == null ? 0 : channels.totalChannels());
+        return new HttpStats(httpChannels.size(), totalChannelsAccepted.get());
     }
 
-    public Netty4CorsConfig getCorsConfig() {
-        return corsConfig;
-    }
-
-    protected void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+    @Override
+    protected void onException(HttpChannel channel, Exception cause) {
         if (cause instanceof ReadTimeoutException) {
             if (logger.isTraceEnabled()) {
-                logger.trace("Read timeout [{}]", ctx.channel().remoteAddress());
+                logger.trace("Http read timeout {}", channel);
             }
-            ctx.channel().close();
+            CloseableChannel.closeChannel(channel);;
         } else {
-            if (!lifecycle.started()) {
-                // ignore
-                return;
-            }
-            if (!NetworkExceptionHelper.isCloseConnectionException(cause)) {
-                logger.warn(
-                    (Supplier<?>) () -> new ParameterizedMessage(
-                        "caught exception while handling client http traffic, closing connection {}", ctx.channel()),
-                    cause);
-                ctx.channel().close();
-            } else {
-                logger.debug(
-                    (Supplier<?>) () -> new ParameterizedMessage(
-                        "caught exception while handling client http traffic, closing connection {}", ctx.channel()),
-                    cause);
-                ctx.channel().close();
-            }
+            super.onException(channel, cause);
         }
     }
 
@@ -404,9 +381,8 @@ protected HttpChannelHandler(final Netty4HttpServerTransport transport, final Ht
 
         @Override
         protected void initChannel(Channel ch) throws Exception {
-            Netty4HttpChannel nettyTcpChannel = new Netty4HttpChannel(ch);
-            ch.attr(HTTP_CHANNEL_KEY).set(nettyTcpChannel);
-            ch.pipeline().addLast("openChannels", transport.serverOpenChannels);
+            Netty4HttpChannel nettyHttpChannel = new Netty4HttpChannel(ch);
+            ch.attr(HTTP_CHANNEL_KEY).set(nettyHttpChannel);
             ch.pipeline().addLast("read_timeout", new ReadTimeoutHandler(transport.readTimeoutMillis, TimeUnit.MILLISECONDS));
             final HttpRequestDecoder decoder = new HttpRequestDecoder(
                 handlingSettings.getMaxInitialLineLength(),
@@ -423,10 +399,11 @@ protected void initChannel(Channel ch) throws Exception {
                 ch.pipeline().addLast("encoder_compress", new HttpContentCompressor(handlingSettings.getCompressionLevel()));
             }
             if (handlingSettings.isCorsEnabled()) {
-                ch.pipeline().addLast("cors", new Netty4CorsHandler(transport.getCorsConfig()));
+                ch.pipeline().addLast("cors", new Netty4CorsHandler(transport.corsConfig));
             }
             ch.pipeline().addLast("pipelining", new Netty4HttpPipeliningHandler(transport.logger, transport.pipeliningMaxEvents));
             ch.pipeline().addLast("handler", requestHandler);
+            transport.serverAcceptedChannel(nettyHttpChannel);
         }
 
         @Override

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

@@ -22,6 +22,7 @@
 import org.elasticsearch.Version;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
+import org.elasticsearch.common.network.CloseableChannel;
 import org.elasticsearch.common.network.NetworkService;
 import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.Settings;
@@ -91,7 +92,7 @@ protected void closeConnectionChannel(Transport transport, Transport.Connection
         final Netty4Transport t = (Netty4Transport) transport;
         @SuppressWarnings("unchecked")
         final TcpTransport.NodeChannels channels = (TcpTransport.NodeChannels) connection;
-        TcpChannel.closeChannels(channels.getChannels().subList(0, randomIntBetween(1, channels.getChannels().size())), true);
+        CloseableChannel.closeChannels(channels.getChannels().subList(0, randomIntBetween(1, channels.getChannels().size())), true);
     }
 
     public void testConnectException() throws UnknownHostException {

modules/transport-netty4/src/test/java/org/elasticsearch/transport/netty4/SimpleNetty4TransportTests.java

@@ -36,4 +36,17 @@ public class NioHttpChannel extends NioSocketChannel implements HttpChannel {
     public void sendResponse(HttpResponse response, ActionListener<Void> listener) {
         getContext().sendMessage(response, ActionListener.toBiConsumer(listener));
     }
+
+    @Override
+    public void addCloseListener(ActionListener<Void> listener) {
+        addCloseListener(ActionListener.toBiConsumer(listener));
+    }
+
+    @Override
+    public String toString() {
+        return "NioHttpChannel{" +
+            "localAddress=" + getLocalAddress() +
+            ", remoteAddress=" + getRemoteAddress() +
+            '}';
+    }
 }