Limit count of HTTP channels with tracked stats (#77303)

Today we expire the client stats for HTTP channels 5 minutes after they
close. It's possible to open a very large number of HTTP channels in 5
minutes, possibly inadvertently, and the stats for those channels can be
overwhelming.

This commit introduces a limit on the number of channels tracked by each
node which applies in addition to the age limit, and makes these limits
configurable via static settings. It drops the pruning of old stats when
starting to track a new channel and instead uses a queue to expire the
oldest stats when each channel closes if necessary to respect the count
limit; it only performs age-based expiry when retrieving the stats,
since the count limit now bounds the memory needed. Finally, it
tightents up some missing synchronization and makes sure that we expose
only immutable objects to the stats subsystem.

docs/reference/modules/http.asciidoc

@@ -190,3 +190,14 @@ Defaults to `network.tcp.receive_buffer_size`.
 `http.client_stats.enabled`::
 (<<dynamic-cluster-setting,Dynamic>>)
 Enable or disable collection of HTTP client stats. Defaults to `true`.
+
+`http.client_stats.closed_channels.max_count`::
+(<<static-cluster-setting,Static>>)
+When `http.client_stats.enabled` is `true`, sets the maximum number of closed
+HTTP channels for which {es} reports statistics. Defaults to `10000`.
+
+`http.client_stats.closed_channels.max_age`::
+(<<static-cluster-setting,Static>>)
+When `http.client_stats.enabled` is `true`, sets the maximum length of time
+after closing a HTTP channel that {es} will report that channel's statistics.
+Defaults to `5m`.

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

@@ -16,6 +16,7 @@
 import org.elasticsearch.transport.netty4.Netty4TcpChannel;
 
 import java.net.InetSocketAddress;
+import java.net.SocketAddress;
 
 public class Netty4HttpChannel implements HttpChannel {
 
@@ -34,12 +35,20 @@ public void sendResponse(HttpResponse response, ActionListener<Void> listener) {
 
     @Override
     public InetSocketAddress getLocalAddress() {
-        return (InetSocketAddress) channel.localAddress();
+        return castAddressOrNull(channel.localAddress());
     }
 
     @Override
     public InetSocketAddress getRemoteAddress() {
-        return (InetSocketAddress) channel.remoteAddress();
+        return castAddressOrNull(channel.remoteAddress());
+    }
+
+    private static InetSocketAddress castAddressOrNull(SocketAddress socketAddress) {
+        if (socketAddress instanceof InetSocketAddress) {
+            return (InetSocketAddress) socketAddress;
+        } else {
+            return null;
+        }
     }
 
     @Override

server/src/main/java/org/elasticsearch/common/settings/ClusterSettings.java

@@ -275,6 +275,8 @@ public void apply(Settings value, Settings current, Settings previous) {
             HttpTransportSettings.SETTING_HTTP_TRACE_LOG_INCLUDE,
             HttpTransportSettings.SETTING_HTTP_TRACE_LOG_EXCLUDE,
             HttpTransportSettings.SETTING_HTTP_CLIENT_STATS_ENABLED,
+            HttpTransportSettings.SETTING_HTTP_CLIENT_STATS_MAX_CLOSED_CHANNEL_AGE,
+            HttpTransportSettings.SETTING_HTTP_CLIENT_STATS_MAX_CLOSED_CHANNEL_COUNT,
             HierarchyCircuitBreakerService.USE_REAL_MEMORY_USAGE_SETTING,
             HierarchyCircuitBreakerService.TOTAL_CIRCUIT_BREAKER_LIMIT_SETTING,
             HierarchyCircuitBreakerService.FIELDDATA_CIRCUIT_BREAKER_LIMIT_SETTING,

server/src/main/java/org/elasticsearch/http/HttpStats.java

@@ -16,7 +16,6 @@
 
 import java.io.IOException;
 import java.util.List;
-import java.util.concurrent.atomic.LongAdder;
 
 public class HttpStats implements Writeable, ToXContentFragment {
 
@@ -93,28 +92,36 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
     }
 
     public static class ClientStats implements Writeable, ToXContentFragment {
-        final int id;
-        String agent;
-        String localAddress;
-        String remoteAddress;
-        String lastUri;
-        String forwardedFor;
-        String opaqueId;
-        long openedTimeMillis;
-        long closedTimeMillis = -1;
-        volatile long lastRequestTimeMillis = -1;
-        final LongAdder requestCount = new LongAdder();
-        final LongAdder requestSizeBytes = new LongAdder();
-
-        ClientStats(long openedTimeMillis) {
-            this.id = System.identityHashCode(this);
-            this.openedTimeMillis = openedTimeMillis;
-        }
+        public static final long NOT_CLOSED = -1L;
 
-        // visible for testing
-        public ClientStats(String agent, String localAddress, String remoteAddress, String lastUri, String forwardedFor, String opaqueId,
-            long openedTimeMillis, long closedTimeMillis, long lastRequestTimeMillis, long requestCount, long requestSizeBytes) {
-            this.id = System.identityHashCode(this);
+        final int id;
+        final String agent;
+        final String localAddress;
+        final String remoteAddress;
+        final String lastUri;
+        final String forwardedFor;
+        final String opaqueId;
+        final long openedTimeMillis;
+        final long closedTimeMillis;
+        final long lastRequestTimeMillis;
+        final long requestCount;
+        final long requestSizeBytes;
+
+        public ClientStats(
+            int id,
+            String agent,
+            String localAddress,
+            String remoteAddress,
+            String lastUri,
+            String forwardedFor,
+            String opaqueId,
+            long openedTimeMillis,
+            long closedTimeMillis,
+            long lastRequestTimeMillis,
+            long requestCount,
+            long requestSizeBytes
+        ) {
+            this.id = id;
             this.agent = agent;
             this.localAddress = localAddress;
             this.remoteAddress = remoteAddress;
@@ -124,8 +131,8 @@ public ClientStats(String agent, String localAddress, String remoteAddress, Stri
             this.openedTimeMillis = openedTimeMillis;
             this.closedTimeMillis = closedTimeMillis;
             this.lastRequestTimeMillis = lastRequestTimeMillis;
-            this.requestCount.add(requestCount);
-            this.requestSizeBytes.add(requestSizeBytes);
+            this.requestCount = requestCount;
+            this.requestSizeBytes = requestSizeBytes;
         }
 
         ClientStats(StreamInput in) throws IOException {
@@ -139,8 +146,8 @@ public ClientStats(String agent, String localAddress, String remoteAddress, Stri
             this.openedTimeMillis = in.readLong();
             this.closedTimeMillis = in.readLong();
             this.lastRequestTimeMillis = in.readLong();
-            this.requestCount.add(in.readLong());
-            this.requestSizeBytes.add(in.readLong());
+            this.requestCount = in.readLong();
+            this.requestSizeBytes = in.readLong();
         }
 
         @Override public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
@@ -165,12 +172,12 @@ public ClientStats(String agent, String localAddress, String remoteAddress, Stri
                 builder.field(Fields.CLIENT_OPAQUE_ID, opaqueId);
             }
             builder.field(Fields.CLIENT_OPENED_TIME_MILLIS, openedTimeMillis);
-            if (closedTimeMillis != -1) {
+            if (closedTimeMillis != NOT_CLOSED) {
                 builder.field(Fields.CLIENT_CLOSED_TIME_MILLIS, closedTimeMillis);
             }
             builder.field(Fields.CLIENT_LAST_REQUEST_TIME_MILLIS, lastRequestTimeMillis);
-            builder.field(Fields.CLIENT_REQUEST_COUNT, requestCount.longValue());
-            builder.field(Fields.CLIENT_REQUEST_SIZE_BYTES, requestSizeBytes.longValue());
+            builder.field(Fields.CLIENT_REQUEST_COUNT, requestCount);
+            builder.field(Fields.CLIENT_REQUEST_SIZE_BYTES, requestSizeBytes);
             builder.endObject();
             return builder;
         }
@@ -187,8 +194,8 @@ public void writeTo(StreamOutput out) throws IOException {
             out.writeLong(openedTimeMillis);
             out.writeLong(closedTimeMillis);
             out.writeLong(lastRequestTimeMillis);
-            out.writeLong(requestCount.longValue());
-            out.writeLong(requestSizeBytes.longValue());
+            out.writeLong(requestCount);
+            out.writeLong(requestSizeBytes);
         }
     }
 }

server/src/main/java/org/elasticsearch/http/HttpTransportSettings.java

@@ -114,6 +114,10 @@ public final class HttpTransportSettings {
 
     public static final Setting<Boolean> SETTING_HTTP_CLIENT_STATS_ENABLED =
         boolSetting("http.client_stats.enabled", true, Property.Dynamic, Property.NodeScope);
+    public static final Setting<Integer> SETTING_HTTP_CLIENT_STATS_MAX_CLOSED_CHANNEL_COUNT =
+        intSetting("http.client_stats.closed_channels.max_count", 10000, Property.NodeScope);
+    public static final Setting<TimeValue> SETTING_HTTP_CLIENT_STATS_MAX_CLOSED_CHANNEL_AGE =
+        Setting.timeSetting("http.client_stats.closed_channels.max_age", TimeValue.timeValueMinutes(5), Property.NodeScope);
 
     private HttpTransportSettings() {
     }

server/src/test/java/org/elasticsearch/action/admin/cluster/node/stats/NodeStatsTests.java

@@ -522,6 +522,7 @@ public static NodeStats createNodeStats() {
             List<HttpStats.ClientStats> clientStats = new ArrayList<>(numClients);
             for (int k = 0; k < numClients; k++) {
                 var cs = new HttpStats.ClientStats(
+                    randomInt(),
                     randomAlphaOfLength(6),
                     randomAlphaOfLength(6),
                     randomAlphaOfLength(6),

server/src/test/java/org/elasticsearch/http/AbstractHttpServerTransportTests.java

@@ -462,6 +462,7 @@ protected void stopInternal() {
                 .withPath("/internal/stats_test")
                 .withHeaders(Map.of(Task.X_OPAQUE_ID, Collections.singletonList(opaqueId)))
                 .build();
+            transport.serverAcceptedChannel(fakeRestRequest.getHttpChannel());
             transport.incomingRequest(fakeRestRequest.getHttpRequest(), fakeRestRequest.getHttpChannel());
 
             HttpStats httpStats = transport.stats();
@@ -478,6 +479,7 @@ protected void stopInternal() {
                 .withPath("/internal/stats_test2")
                 .withHeaders(Map.of(Task.X_OPAQUE_ID.toUpperCase(Locale.ROOT), Collections.singletonList(opaqueId)))
                 .build();
+            transport.serverAcceptedChannel(fakeRestRequest.getHttpChannel());
             transport.incomingRequest(fakeRestRequest.getHttpRequest(), fakeRestRequest.getHttpChannel());
             httpStats = transport.stats();
             assertThat(httpStats.getClientStats().size(), equalTo(2));
@@ -532,6 +534,7 @@ public void testDisablingHttpClientStats() {
                     .withPath("/internal/stats_test")
                     .withHeaders(Map.of(Task.X_OPAQUE_ID, Collections.singletonList(opaqueId)))
                     .build();
+            transport.serverAcceptedChannel(fakeRestRequest.getHttpChannel());
             transport.incomingRequest(fakeRestRequest.getHttpRequest(), fakeRestRequest.getHttpChannel());
 
             // HTTP client stats should default to enabled
@@ -556,6 +559,7 @@ public void testDisablingHttpClientStats() {
                     .withPath("/internal/stats_test")
                     .withHeaders(Map.of(Task.X_OPAQUE_ID, Collections.singletonList(opaqueId)))
                     .build();
+            transport.serverAcceptedChannel(fakeRestRequest.getHttpChannel());
             transport.incomingRequest(fakeRestRequest.getHttpRequest(), fakeRestRequest.getHttpChannel());
             httpStats = transport.stats();
             assertThat(httpStats.getClientStats().size(), equalTo(0));
@@ -573,6 +577,7 @@ public void testDisablingHttpClientStats() {
                     .withPath("/internal/stats_test")
                     .withHeaders(Map.of(Task.X_OPAQUE_ID, Collections.singletonList(opaqueId)))
                     .build();
+            transport.serverAcceptedChannel(fakeRestRequest.getHttpChannel());
             transport.incomingRequest(fakeRestRequest.getHttpRequest(), fakeRestRequest.getHttpChannel());
             httpStats = transport.stats();
             assertThat(httpStats.getClientStats().size(), equalTo(1));