Ensure correct access log handling for HTTP/1.1 pipelining (#2667)

- Move AccessLogHandler after HttpTrafficHandler, as the latter handles HTTP/1.1 pipelining
- When draining cached requests ensure exceptions are handled as decoding failures
- AccessLogArgProviderH1 now depends on HttpServerRequest, as the latter holds
information for the time when the requests were received on the server

Fixes #2628

Author: violetagg

reactor-netty-core/src/testFixtures/java/reactor/netty/LogTracker.java

@@ -20,6 +20,8 @@
 import ch.qos.logback.core.AppenderBase;
 import org.slf4j.LoggerFactory;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import java.util.stream.Stream;
 
@@ -30,39 +32,52 @@
  */
 public final class LogTracker extends AppenderBase<ILoggingEvent> implements AutoCloseable {
 
+	public final List<ILoggingEvent> actualMessages = new ArrayList<>();
 	public final CountDownLatch latch;
 
 	private final Logger logger;
-	private final String[] messages;
+	private final String[] expectedMessages;
 
 	/**
 	 * Creates a new {@link LogTracker}.
 	 *
 	 * @param className the logger name
-	 * @param messages the expected messages
+	 * @param expectedMessages the expected messages
 	 */
-	public LogTracker(Class<?> className, String... messages) {
-		this(className.getName(), messages);
+	public LogTracker(Class<?> className, String... expectedMessages) {
+		this(className.getName(), expectedMessages);
 	}
 
 	/**
 	 * Creates a new {@link LogTracker}.
 	 *
 	 * @param loggerName the logger name
-	 * @param messages the expected messages
+	 * @param expectedMessages the expected messages
 	 */
-	public LogTracker(String loggerName, String... messages) {
+	public LogTracker(String loggerName, String... expectedMessages) {
+		this(loggerName, expectedMessages.length, expectedMessages);
+	}
+
+	/**
+	 * Creates a new {@link LogTracker}.
+	 *
+	 * @param loggerName the logger name
+	 * @param expectedCount the number of times the expected messages will appear in the logs
+	 * @param expectedMessages the expected messages
+	 */
+	public LogTracker(String loggerName, int expectedCount, String... expectedMessages) {
 		this.logger = (Logger) LoggerFactory.getLogger(loggerName);
-		this.messages = messages;
-		this.latch = new CountDownLatch(messages.length);
+		this.expectedMessages = expectedMessages;
+		this.latch = new CountDownLatch(expectedCount);
 
 		start();
 		this.logger.addAppender(this);
 	}
 
 	@Override
 	protected void append(ILoggingEvent eventObject) {
-		if (Stream.of(messages).anyMatch(msg -> eventObject.getFormattedMessage().contains(msg))) {
+		if (Stream.of(expectedMessages).anyMatch(msg -> eventObject.getFormattedMessage().contains(msg))) {
+			actualMessages.add(eventObject);
 			latch.countDown();
 		}
 	}

reactor-netty-http/build.gradle

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2020-2022 VMware, Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2020-2023 VMware, Inc. or its affiliates, All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -235,6 +235,8 @@ task japicmp(type: JapicmpTask) {
 
 	compatibilityChangeExcludes = [ "METHOD_NEW_DEFAULT" ]
 	methodExcludes = [
+			'reactor.netty.http.server.HttpServerRequest#protocol()',
+			'reactor.netty.http.server.HttpServerRequest#timestamp()'
 	]
 }
 

reactor-netty-http/src/main/java/reactor/netty/http/server/Http2StreamBridgeServerHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2018-2022 VMware, Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2018-2023 VMware, Inc. or its affiliates, All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,6 +16,7 @@
 package reactor.netty.http.server;
 
 import java.net.SocketAddress;
+import java.time.ZonedDateTime;
 import java.util.Optional;
 import java.util.function.BiFunction;
 import java.util.function.BiPredicate;
@@ -39,6 +40,7 @@
 import reactor.core.publisher.Mono;
 import reactor.netty.Connection;
 import reactor.netty.ConnectionObserver;
+import reactor.netty.ReactorNetty;
 import reactor.netty.http.logging.HttpMessageArgProviderFactory;
 import reactor.netty.http.logging.HttpMessageLogFactory;
 import reactor.util.annotation.Nullable;
@@ -113,6 +115,7 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) {
 		if (msg instanceof HttpRequest) {
 			HttpRequest request = (HttpRequest) msg;
 			HttpServerOperations ops;
+			ZonedDateTime timestamp = ZonedDateTime.now(ReactorNetty.ZONE_ID_SYSTEM);
 			try {
 				pendingResponse = true;
 				ops = new HttpServerOperations(Connection.from(ctx.channel()),
@@ -129,12 +132,13 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) {
 						formDecoderProvider,
 						httpMessageLogFactory,
 						mapHandle,
-						secured);
+						secured,
+						timestamp);
 			}
 			catch (RuntimeException e) {
 				pendingResponse = false;
 				request.setDecoderResult(DecoderResult.failure(e.getCause() != null ? e.getCause() : e));
-				HttpServerOperations.sendDecodingFailures(ctx, listener, secured, e, msg, httpMessageLogFactory);
+				HttpServerOperations.sendDecodingFailures(ctx, listener, secured, e, msg, httpMessageLogFactory, timestamp);
 				return;
 			}
 			ops.bind();

reactor-netty-http/src/main/java/reactor/netty/http/server/HttpServerConfig.java

@@ -588,7 +588,7 @@ static void configureHttp11OrH2CleartextPipeline(ChannelPipeline p,
 		                    forwardedHeaderHandler, httpMessageLogFactory, idleTimeout, listener, mapHandle, maxKeepAliveRequests));
 
 		if (accessLogEnabled) {
-			p.addBefore(NettyPipeline.HttpTrafficHandler, NettyPipeline.AccessLogHandler, AccessLogHandlerFactory.H1.create(accessLog));
+			p.addAfter(NettyPipeline.HttpTrafficHandler, NettyPipeline.AccessLogHandler, AccessLogHandlerFactory.H1.create(accessLog));
 		}
 
 		boolean alwaysCompress = compressPredicate == null && minCompressionSize == 0;
@@ -640,7 +640,7 @@ static void configureHttp11Pipeline(ChannelPipeline p,
 		                    forwardedHeaderHandler, httpMessageLogFactory, idleTimeout, listener, mapHandle, maxKeepAliveRequests));
 
 		if (accessLogEnabled) {
-			p.addAfter(NettyPipeline.HttpCodec, NettyPipeline.AccessLogHandler, AccessLogHandlerFactory.H1.create(accessLog));
+			p.addAfter(NettyPipeline.HttpTrafficHandler, NettyPipeline.AccessLogHandler, AccessLogHandlerFactory.H1.create(accessLog));
 		}
 
 		boolean alwaysCompress = compressPredicate == null && minCompressionSize == 0;

reactor-netty-http/src/main/java/reactor/netty/http/server/HttpServerOperations.java

@@ -20,6 +20,7 @@
 import java.net.SocketAddress;
 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.time.ZonedDateTime;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Locale;
@@ -80,6 +81,7 @@
 import reactor.netty.FutureMono;
 import reactor.netty.NettyOutbound;
 import reactor.netty.NettyPipeline;
+import reactor.netty.ReactorNetty;
 import reactor.netty.channel.AbortedException;
 import reactor.netty.channel.ChannelOperations;
 import reactor.netty.http.HttpOperations;
@@ -117,6 +119,7 @@ class HttpServerOperations extends HttpOperations<HttpServerRequest, HttpServerR
 	final HttpResponse nettyResponse;
 	final HttpHeaders responseHeaders;
 	final String scheme;
+	final ZonedDateTime timestamp;
 
 	BiPredicate<HttpServerRequest, HttpServerResponse> compressionPredicate;
 	Function<? super String, Map<String, String>> paramsResolver;
@@ -142,6 +145,7 @@ class HttpServerOperations extends HttpOperations<HttpServerRequest, HttpServerR
 		this.path = replaced.path;
 		this.responseHeaders = replaced.responseHeaders;
 		this.scheme = replaced.scheme;
+		this.timestamp = replaced.timestamp;
 		this.trailerHeadersConsumer = replaced.trailerHeadersConsumer;
 	}
 
@@ -153,9 +157,10 @@ class HttpServerOperations extends HttpOperations<HttpServerRequest, HttpServerR
 			HttpServerFormDecoderProvider formDecoderProvider,
 			HttpMessageLogFactory httpMessageLogFactory,
 			@Nullable BiFunction<? super Mono<Void>, ? super Connection, ? extends Mono<Void>> mapHandle,
-			boolean secured) {
+			boolean secured,
+			ZonedDateTime timestamp) {
 		this(c, listener, nettyRequest, compressionPredicate, connectionInfo, decoder, encoder, formDecoderProvider,
-				httpMessageLogFactory, mapHandle, true, secured);
+				httpMessageLogFactory, mapHandle, true, secured, timestamp);
 	}
 
 	HttpServerOperations(Connection c, ConnectionObserver listener, HttpRequest nettyRequest,
@@ -167,7 +172,8 @@ class HttpServerOperations extends HttpOperations<HttpServerRequest, HttpServerR
 			HttpMessageLogFactory httpMessageLogFactory,
 			@Nullable BiFunction<? super Mono<Void>, ? super Connection, ? extends Mono<Void>> mapHandle,
 			boolean resolvePath,
-			boolean secured) {
+			boolean secured,
+			ZonedDateTime timestamp) {
 		super(c, listener, httpMessageLogFactory);
 		this.compressionPredicate = compressionPredicate;
 		this.configuredCompressionPredicate = compressionPredicate;
@@ -189,6 +195,7 @@ class HttpServerOperations extends HttpOperations<HttpServerRequest, HttpServerR
 		this.responseHeaders = nettyResponse.headers();
 		this.responseHeaders.set(HttpHeaderNames.TRANSFER_ENCODING, HttpHeaderValues.CHUNKED);
 		this.scheme = secured ? "https" : "http";
+		this.timestamp = timestamp;
 	}
 
 	@Override
@@ -472,6 +479,16 @@ public HttpHeaders responseHeaders() {
 		return responseHeaders;
 	}
 
+	@Override
+	public String protocol() {
+		return nettyRequest.protocolVersion().text();
+	}
+
+	@Override
+	public ZonedDateTime timestamp() {
+		return timestamp;
+	}
+
 	@Override
 	public Mono<Void> send() {
 		return FutureMono.deferFuture(() -> markSentHeaderAndBody() ?
@@ -767,7 +784,8 @@ static void sendDecodingFailures(
 			boolean secure,
 			Throwable t,
 			Object msg,
-			HttpMessageLogFactory httpMessageLogFactory) {
+			HttpMessageLogFactory httpMessageLogFactory,
+			@Nullable ZonedDateTime timestamp) {
 
 		Throwable cause = t.getCause() != null ? t.getCause() : t;
 
@@ -798,7 +816,8 @@ else if (cause instanceof TooLongHttpHeaderException) {
 		if (ops == null) {
 			Connection conn = Connection.from(ctx.channel());
 			if (msg instanceof HttpRequest) {
-				ops = new FailedHttpServerRequest(conn, listener, (HttpRequest) msg, response, httpMessageLogFactory, secure);
+				ops = new FailedHttpServerRequest(conn, listener, (HttpRequest) msg, response, httpMessageLogFactory, secure,
+						timestamp == null ? ZonedDateTime.now(ReactorNetty.ZONE_ID_SYSTEM) : timestamp);
 				ops.bind();
 			}
 			else {
@@ -967,9 +986,10 @@ static final class FailedHttpServerRequest extends HttpServerOperations {
 				HttpRequest nettyRequest,
 				HttpResponse nettyResponse,
 				HttpMessageLogFactory httpMessageLogFactory,
-				boolean secure) {
+				boolean secure,
+				ZonedDateTime timestamp) {
 			super(c, listener, nettyRequest, null, null, ServerCookieDecoder.STRICT, ServerCookieEncoder.STRICT,
-					DEFAULT_FORM_DECODER_SPEC, httpMessageLogFactory, null, false, secure);
+					DEFAULT_FORM_DECODER_SPEC, httpMessageLogFactory, null, false, secure, timestamp);
 			this.customResponse = nettyResponse;
 			String tempPath = "";
 			try {

reactor-netty-http/src/main/java/reactor/netty/http/server/HttpServerRequest.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2011-2022 VMware, Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2011-2023 VMware, Inc. or its affiliates, All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,6 +16,7 @@
 package reactor.netty.http.server;
 
 import java.net.InetSocketAddress;
+import java.time.ZonedDateTime;
 import java.util.Map;
 import java.util.function.Consumer;
 import java.util.function.Function;
@@ -142,4 +143,19 @@ default Flux<HttpContent> receiveContent() {
 	 */
 	HttpHeaders requestHeaders();
 
+	/**
+	 * Returns the inbound protocol and version.
+	 *
+	 * @return the inbound protocol and version
+	 * @since 1.0.28
+	 */
+	String protocol();
+
+	/**
+	 * Returns the time when the request was received.
+	 *
+	 * @return the time when the request was received
+	 * @since 1.0.28
+	 */
+	ZonedDateTime timestamp();
 }