Ensure HttpServer metrics connections.total/active are correct when multiple local addresses

reactor-netty-core/src/main/java/reactor/netty/channel/MicrometerChannelMetricsRecorder.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2019-2022 VMware, Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2019-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.
@@ -72,7 +72,7 @@ public class MicrometerChannelMetricsRecorder implements ChannelMetricsRecorder
 	final ConcurrentMap<MeterKey, Timer> addressResolverTimeCache = new ConcurrentHashMap<>();
 
 	final ConcurrentMap<String, LongAdder> totalConnectionsCache = new ConcurrentHashMap<>();
-	final LongAdder totalConnectionsAdder = new LongAdder();
+
 	final String name;
 	final String protocol;
 
@@ -199,10 +199,11 @@ protected String protocol() {
 	}
 
 	@Nullable
-	private LongAdder getTotalConnectionsAdder(SocketAddress serverAddress) {
+	LongAdder getTotalConnectionsAdder(SocketAddress serverAddress) {
 		String address = reactor.netty.Metrics.formatSocketAddress(serverAddress);
 		return MapUtils.computeIfAbsent(totalConnectionsCache, address,
 				key -> {
+					LongAdder totalConnectionsAdder = new LongAdder();
 					Gauge gauge = filter(Gauge.builder(name + CONNECTIONS_TOTAL, totalConnectionsAdder, LongAdder::longValue)
 							.description(TOTAL_CONNECTIONS_DESCRIPTION)
 							.tags(URI, protocol, LOCAL_ADDRESS, address)

reactor-netty-core/src/test/java/reactor/netty/channel/MicrometerChannelMetricsRecorderTests.java

@@ -0,0 +1,40 @@
+/*
+ * Copyright (c) 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.
+ * You may obtain a copy of the License at
+ *
+ *   https://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 reactor.netty.channel;
+
+import org.junit.jupiter.api.Test;
+import reactor.netty.transport.AddressUtils;
+
+import java.net.InetSocketAddress;
+import java.util.concurrent.atomic.LongAdder;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class MicrometerChannelMetricsRecorderTests {
+	static final InetSocketAddress ADDRESS_1 = AddressUtils.createUnresolved("127.0.0.1", 80);
+	static final InetSocketAddress ADDRESS_2 = AddressUtils.createUnresolved("0:0:0:0:0:0:0:1", 80);
+
+	@Test
+	void testGetTotalConnectionsAdder() {
+		MicrometerChannelMetricsRecorder recorder = new MicrometerChannelMetricsRecorder("test", "test");
+
+		LongAdder longAdder1 = recorder.getTotalConnectionsAdder(ADDRESS_1);
+
+		LongAdder longAdder2 = recorder.getTotalConnectionsAdder(ADDRESS_2);
+
+		assertThat(longAdder1).isNotSameAs(longAdder2);
+	}
+}

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

@@ -55,8 +55,6 @@ final class MicrometerHttpServerMetricsRecorder extends MicrometerHttpMetricsRec
 	private static final String PROTOCOL_VALUE_HTTP = "http";
 	private static final String ACTIVE_CONNECTIONS_DESCRIPTION = "The number of http connections currently processing requests";
 	private static final 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<>();
@@ -206,10 +204,11 @@ public void recordResolveAddressTime(SocketAddress remoteAddress, Duration time,
 	}
 
 	@Nullable
-	private LongAdder getActiveStreamsAdder(SocketAddress localAddress) {
+	LongAdder getActiveStreamsAdder(SocketAddress localAddress) {
 		String address = reactor.netty.Metrics.formatSocketAddress(localAddress);
 		return MapUtils.computeIfAbsent(activeStreamsCache, address,
 				key -> {
+					LongAdder activeStreamsAdder = new LongAdder();
 					Gauge gauge = filter(
 							Gauge.builder(name() + STREAMS_ACTIVE, activeStreamsAdder, LongAdder::longValue)
 							     .tags(URI, PROTOCOL_VALUE_HTTP, LOCAL_ADDRESS, address)
@@ -220,10 +219,11 @@ private LongAdder getActiveStreamsAdder(SocketAddress localAddress) {
 	}
 
 	@Nullable
-	private LongAdder getServerConnectionAdder(SocketAddress localAddress) {
+	LongAdder getServerConnectionAdder(SocketAddress localAddress) {
 		String address = reactor.netty.Metrics.formatSocketAddress(localAddress);
 		return MapUtils.computeIfAbsent(activeConnectionsCache, address,
 				key -> {
+					LongAdder activeConnectionsAdder = new LongAdder();
 					Gauge gauge = filter(Gauge.builder(reactor.netty.Metrics.HTTP_SERVER_PREFIX + CONNECTIONS_ACTIVE,
 							activeConnectionsAdder, LongAdder::longValue)
 							.tags(URI, PROTOCOL_VALUE_HTTP, LOCAL_ADDRESS, address)

reactor-netty-http/src/test/java/reactor/netty/http/server/MicrometerHttpServerMetricsRecorderTests.java

@@ -0,0 +1,47 @@
+/*
+ * Copyright (c) 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.
+ * You may obtain a copy of the License at
+ *
+ *   https://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 reactor.netty.http.server;
+
+import org.junit.jupiter.api.Test;
+import reactor.netty.transport.AddressUtils;
+
+import java.net.InetSocketAddress;
+import java.util.concurrent.atomic.LongAdder;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class MicrometerHttpServerMetricsRecorderTests {
+	static final InetSocketAddress ADDRESS_1 = AddressUtils.createUnresolved("127.0.0.1", 80);
+	static final InetSocketAddress ADDRESS_2 = AddressUtils.createUnresolved("0:0:0:0:0:0:0:1", 80);
+
+	@Test
+	void testGetServerConnectionAdder() {
+		LongAdder longAdder1 = MicrometerHttpServerMetricsRecorder.INSTANCE.getServerConnectionAdder(ADDRESS_1);
+
+		LongAdder longAdder2 = MicrometerHttpServerMetricsRecorder.INSTANCE.getServerConnectionAdder(ADDRESS_2);
+
+		assertThat(longAdder1).isNotSameAs(longAdder2);
+	}
+
+	@Test
+	void testGetActiveStreamsAdder() {
+		LongAdder longAdder1 = MicrometerHttpServerMetricsRecorder.INSTANCE.getActiveStreamsAdder(ADDRESS_1);
+
+		LongAdder longAdder2 = MicrometerHttpServerMetricsRecorder.INSTANCE.getActiveStreamsAdder(ADDRESS_2);
+
+		assertThat(longAdder1).isNotSameAs(longAdder2);
+	}
+}