Capture HTTP/2 server active stream metrics

This is a backport for PR #2357

Fixes #2356

docs/asciidoc/http-server.adoc

@@ -560,6 +560,7 @@ The following table provides information for the HTTP server metrics:
 [width="100%",options="header"]
 |=======
 | metric name | type | description
+| reactor.netty.http.server.streams.active | Gauge | The number of active HTTP/2 streams.
 | reactor.netty.http.server.connections.active | Gauge | The number of http connections currently processing requests
 | reactor.netty.http.server.connections.total | Gauge | The number of all opened connections
 | reactor.netty.http.server.data.received | DistributionSummary | Amount of the data received, in bytes

reactor-netty-core/src/main/java/reactor/netty/Metrics.java

@@ -245,6 +245,11 @@ public class Metrics {
 	 */
 	public static final String PENDING_TASKS = ".pending.tasks";
 
+	// HttpServer Metrics
+	/**
+	 * The number of active HTTP/2 streams
+	 */
+	public static final String STREAMS_ACTIVE = ".streams.active";
 
 	// Tags
 	public static final String LOCAL_ADDRESS = "local.address";

reactor-netty-http/build.gradle

@@ -190,6 +190,10 @@ task japicmp(type: JapicmpTask) {
 	ignoreMissingClasses = true
 	includeSynthetic = true
 	onlyIf { "$compatibleVersion" != "SKIP" }
+	methodExcludes = [
+			"reactor.netty.http.server.HttpServerMetricsRecorder#recordStreamClosed(java.net.SocketAddress)",
+			"reactor.netty.http.server.HttpServerMetricsRecorder#recordStreamOpened(java.net.SocketAddress)"
+	]
 }
 
 tasks.japicmp.dependsOn(downloadBaseline)

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

@@ -45,12 +45,10 @@ abstract class AbstractHttpServerMetricsHandler extends ChannelDuplexHandler {
 
 	long dataSent;
 
-
 	long dataReceivedTime;
 
 	long dataSentTime;
 
-
 	final Function<String, String> uriTagValue;
 
 	protected AbstractHttpServerMetricsHandler(@Nullable Function<String, String> uriTagValue) {
@@ -123,12 +121,16 @@ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
 							log.warn("Exception caught while recording metrics.", e);
 							// Allow request-response exchange to continue, unaffected by metrics problem
 						}
-						if (!ops.isHttp2() && ops.hostAddress() != null) {
-							// This metric is not applicable for HTTP/2
-							// ops.hostAddress() == null when request decoding failed, in this case
-							// we do not report active connection, so we do not report inactive connection
+						// ops.hostAddress() == null when request decoding failed, in this case
+						// we do not report active connection, so we do not report inactive connection
+						if (ops.hostAddress() != null) {
 							try {
-								recordInactiveConnection(ops);
+								if (ops.isHttp2()) {
+									recordClosedStream(ops);
+								}
+								else {
+									recordInactiveConnection(ops);
+								}
 							}
 							catch (RuntimeException e) {
 								log.warn("Exception caught while recording metrics.", e);
@@ -159,8 +161,10 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) {
 				ChannelOperations<?, ?> channelOps = ChannelOperations.get(ctx.channel());
 				if (channelOps instanceof HttpServerOperations) {
 					HttpServerOperations ops = (HttpServerOperations) channelOps;
-					if (!ops.isHttp2()) {
-						// This metric is not applicable for HTTP/2
+					if (ops.isHttp2()) {
+						recordOpenStream(ops);
+					}
+					else {
 						recordActiveConnection(ops);
 					}
 				}
@@ -250,4 +254,12 @@ protected void recordActiveConnection(HttpServerOperations ops) {
 	protected void recordInactiveConnection(HttpServerOperations ops) {
 		recorder().recordServerConnectionInactive(ops.hostAddress());
 	}
+
+	protected void recordOpenStream(HttpServerOperations ops) {
+		recorder().recordStreamOpened(ops.hostAddress());
+	}
+
+	protected void recordClosedStream(HttpServerOperations ops) {
+		recorder().recordStreamClosed(ops.hostAddress());
+	}
 }

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

@@ -71,4 +71,20 @@ default void recordServerConnectionActive(SocketAddress localAddress) { }
 	 * @since 1.0.15
 	 */
 	default void recordServerConnectionInactive(SocketAddress localAddress) { }
+
+	/**
+	 * Record an opened HTTP/2 stream
+	 *
+	 * @param localAddress the local server address
+	 * @since 1.0.21
+	 */
+	default void recordStreamOpened(SocketAddress localAddress) { }
+
+	/**
+	 * Record a closed HTTP/2 stream
+	 *
+	 * @param localAddress the local server address
+	 * @since 1.0.21
+	 */
+	default void recordStreamClosed(SocketAddress localAddress) { }
 }

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

@@ -22,6 +22,7 @@
 import reactor.netty.channel.MeterKey;
 import reactor.netty.http.MicrometerHttpMetricsRecorder;
 import reactor.netty.internal.util.MapUtils;
+import reactor.util.annotation.Nullable;
 
 import java.net.SocketAddress;
 import java.time.Duration;
@@ -41,6 +42,7 @@
 import static reactor.netty.Metrics.REGISTRY;
 import static reactor.netty.Metrics.RESPONSE_TIME;
 import static reactor.netty.Metrics.STATUS;
+import static reactor.netty.Metrics.STREAMS_ACTIVE;
 import static reactor.netty.Metrics.URI;
 
 /**
@@ -52,8 +54,11 @@ final class MicrometerHttpServerMetricsRecorder extends MicrometerHttpMetricsRec
 	final static MicrometerHttpServerMetricsRecorder INSTANCE = new MicrometerHttpServerMetricsRecorder();
 	private final static String PROTOCOL_VALUE_HTTP = "http";
 	private final static String ACTIVE_CONNECTIONS_DESCRIPTION = "The number of http connections currently processing requests";
+	private final static String ACTIVE_STREAMS_DESCRIPTION = "The number of HTTP/2 streams currently active on the server";
 	private final LongAdder activeConnectionsAdder = new LongAdder();
+	private final LongAdder activeStreamsAdder = new LongAdder();
 	private final ConcurrentMap<String, LongAdder> activeConnectionsCache = new ConcurrentHashMap<>();
+	private final ConcurrentMap<String, LongAdder> activeStreamsCache = new ConcurrentHashMap<>();
 	private final ConcurrentMap<String, DistributionSummary> dataReceivedCache = new ConcurrentHashMap<>();
 	private final ConcurrentMap<String, DistributionSummary> dataSentCache = new ConcurrentHashMap<>();
 	private final ConcurrentMap<String, Counter> errorsCache = new ConcurrentHashMap<>();
@@ -154,6 +159,22 @@ public void recordServerConnectionInactive(SocketAddress localAddress) {
 		}
 	}
 
+	@Override
+	public void recordStreamOpened(SocketAddress localAddress) {
+		LongAdder adder = getActiveStreamsAdder(localAddress);
+		if (adder != null) {
+			adder.increment();
+		}
+	}
+
+	@Override
+	public void recordStreamClosed(SocketAddress localAddress) {
+		LongAdder adder = getActiveStreamsAdder(localAddress);
+		if (adder != null) {
+			adder.decrement();
+		}
+	}
+
 	@Override
 	public void recordDataReceived(SocketAddress remoteAddress, long bytes) {
 		// noop
@@ -184,6 +205,21 @@ public void recordResolveAddressTime(SocketAddress remoteAddress, Duration time,
 		throw new UnsupportedOperationException();
 	}
 
+	@Nullable
+	private LongAdder getActiveStreamsAdder(SocketAddress localAddress) {
+		String address = reactor.netty.Metrics.formatSocketAddress(localAddress);
+		return MapUtils.computeIfAbsent(activeStreamsCache, address,
+				key -> {
+					Gauge gauge = filter(
+							Gauge.builder(name() + STREAMS_ACTIVE, activeStreamsAdder, LongAdder::longValue)
+							     .tags(URI, PROTOCOL_VALUE_HTTP, LOCAL_ADDRESS, address)
+							     .description(ACTIVE_STREAMS_DESCRIPTION)
+							     .register(REGISTRY));
+					return gauge != null ? activeStreamsAdder : null;
+				});
+	}
+
+	@Nullable
 	private LongAdder getServerConnectionAdder(SocketAddress localAddress) {
 		String address = reactor.netty.Metrics.formatSocketAddress(localAddress);
 		return MapUtils.computeIfAbsent(activeConnectionsCache, address,

reactor-netty-http/src/test/java/reactor/netty/http/HttpMetricsHandlerTests.java

@@ -83,6 +83,7 @@
 import static reactor.netty.Metrics.REMOTE_ADDRESS;
 import static reactor.netty.Metrics.RESPONSE_TIME;
 import static reactor.netty.Metrics.STATUS;
+import static reactor.netty.Metrics.STREAMS_ACTIVE;
 import static reactor.netty.Metrics.TLS_HANDSHAKE_TIME;
 import static reactor.netty.Metrics.URI;
 import static reactor.netty.Metrics.formatSocketAddress;
@@ -600,6 +601,7 @@ void testServerConnectionsMicrometer(HttpProtocol[] serverProtocols, HttpProtoco
 		}
 		else {
 			checkGauge(SERVER_CONNECTIONS_TOTAL, true, 1, URI, HTTP, LOCAL_ADDRESS, address);
+			checkGauge(SERVER_STREAMS_ACTIVE, true, 0, URI, HTTP, LOCAL_ADDRESS, address);
 		}
 
 		// These metrics are meant only for the servers,
@@ -734,7 +736,10 @@ private void checkServerConnectionsMicrometer(HttpServerRequest request) {
 		String address = formatSocketAddress(request.hostAddress());
 		boolean isHttp2 = request.requestHeaders().contains(HttpConversionUtil.ExtensionHeaderNames.SCHEME.text());
 		checkGauge(SERVER_CONNECTIONS_TOTAL, true, 1, URI, HTTP, LOCAL_ADDRESS, address);
-		if (!isHttp2) {
+		if (isHttp2) {
+			checkGauge(SERVER_STREAMS_ACTIVE, true, 1, URI, HTTP, LOCAL_ADDRESS, address);
+		}
+		else {
 			checkGauge(SERVER_CONNECTIONS_ACTIVE, true, 1, URI, HTTP, LOCAL_ADDRESS, address);
 		}
 	}
@@ -936,6 +941,7 @@ static Stream<Arguments> httpCompatibleProtocols() {
 
 	private static final String SERVER_CONNECTIONS_ACTIVE = HTTP_SERVER_PREFIX + CONNECTIONS_ACTIVE;
 	private static final String SERVER_CONNECTIONS_TOTAL = HTTP_SERVER_PREFIX + CONNECTIONS_TOTAL;
+	private static final String SERVER_STREAMS_ACTIVE = HTTP_SERVER_PREFIX + STREAMS_ACTIVE;
 	private static final String SERVER_RESPONSE_TIME = HTTP_SERVER_PREFIX + RESPONSE_TIME;
 	private static final String SERVER_DATA_SENT_TIME = HTTP_SERVER_PREFIX + DATA_SENT_TIME;
 	private static final String SERVER_DATA_RECEIVED_TIME = HTTP_SERVER_PREFIX + DATA_RECEIVED_TIME;