Collect HTTP/1 metrics in NettyNioHttpClient.

Author: millems

http-clients/apache-client/src/main/java/software/amazon/awssdk/http/apache/internal/NoOpMetricCollector.java renamed to core/metrics-spi/src/main/java/software/amazon/awssdk/metrics/NoOpMetricCollector.java

@@ -13,20 +13,20 @@
  * permissions and limitations under the License.
  */
 
-package software.amazon.awssdk.http.apache.internal;
+package software.amazon.awssdk.metrics;
 
-import software.amazon.awssdk.annotations.SdkInternalApi;
-import software.amazon.awssdk.metrics.MetricCollection;
-import software.amazon.awssdk.metrics.MetricCollector;
-import software.amazon.awssdk.metrics.SdkMetric;
+import software.amazon.awssdk.annotations.SdkPublicApi;
 
 /**
  * A metric collector that doesn't do anything.
  */
-@SdkInternalApi
+@SdkPublicApi
 public final class NoOpMetricCollector implements MetricCollector {
     private static final NoOpMetricCollector INSTANCE = new NoOpMetricCollector();
 
+    private NoOpMetricCollector() {
+    }
+
     @Override
     public String name() {
         return "NoOp";

core/metrics-spi/src/main/java/software/amazon/awssdk/metrics/internal/DefaultMetricCollector.java

@@ -27,14 +27,15 @@
 import software.amazon.awssdk.metrics.SdkMetric;
 import software.amazon.awssdk.utils.Validate;
 
+/**
+ * TODO: Before launch, we should iterate on the performance of this collector, because it's currently very naive.
+ */
 @SdkInternalApi
 public final class DefaultMetricCollector implements MetricCollector {
     private final String name;
-    private Map<SdkMetric<?>, List<MetricRecord<?>>> metrics = new LinkedHashMap<>();
+    private final Map<SdkMetric<?>, List<MetricRecord<?>>> metrics = new LinkedHashMap<>();
     private final List<MetricCollector> children = new ArrayList<>();
 
-    private MetricCollection collection;
-
     public DefaultMetricCollector(String name) {
         this.name = name;
     }
@@ -45,40 +46,29 @@ public String name() {
     }
 
     @Override
-    public <T> void reportMetric(SdkMetric<T> metric, T data) {
-        if (collected()) {
-            throw new IllegalStateException("This collector has already been closed");
-        }
+    public synchronized <T> void reportMetric(SdkMetric<T> metric, T data) {
         metrics.computeIfAbsent(metric, (m) -> new ArrayList<>())
-                .add(new DefaultMetricRecord<>(metric, data));
+               .add(new DefaultMetricRecord<>(metric, data));
     }
 
     @Override
-    public MetricCollector createChild(String name) {
+    public synchronized MetricCollector createChild(String name) {
         MetricCollector child = new DefaultMetricCollector(name);
         children.add(child);
         return child;
     }
 
     @Override
-    public MetricCollection collect() {
-        if (!collected()) {
-            List<MetricCollection> collectedChildren = children.stream()
-                    .map(MetricCollector::collect)
-                    .collect(Collectors.toList());
-
-            collection = new DefaultMetricCollection(name, metrics, collectedChildren);
-        }
+    public synchronized MetricCollection collect() {
+        List<MetricCollection> collectedChildren = children.stream()
+                .map(MetricCollector::collect)
+                .collect(Collectors.toList());
 
-        return collection;
+        return new DefaultMetricCollection(name, metrics, collectedChildren);
     }
 
     public static MetricCollector create(String name) {
         Validate.notEmpty(name, "name");
         return new DefaultMetricCollector(name);
     }
-
-    private boolean collected() {
-        return collection != null;
-    }
 }

core/metrics-spi/src/test/java/software/amazon/awssdk/metrics/internal/DefaultMetricCollectorTest.java

@@ -68,14 +68,4 @@ public void testCollect_returnedCollectionContainsAllChildren() {
         MetricCollection collected = parent.collect();
         assertThat(collected.children().stream().map(MetricCollection::name)).containsExactly(childNames);
     }
-
-    @Test
-    public void testReportMetric_collected_throws() {
-        thrown.expect(IllegalStateException.class);
-        thrown.expectMessage("This collector has already been closed");
-
-        MetricCollector collector = MetricCollector.create("collector");
-        collector.collect();
-        collector.reportMetric(M1, 42);
-    }
 }

http-client-spi/src/main/java/software/amazon/awssdk/http/HttpExecuteRequest.java

@@ -83,7 +83,7 @@ public interface Builder {
          * report metrics collected for this request.
          *
          * @param metricCollector The metric collector.
-         * @return This bilder for method chaining.
+         * @return This builder for method chaining.
          */
         Builder metricCollector(MetricCollector metricCollector);
 

http-client-spi/src/main/java/software/amazon/awssdk/http/async/AsyncExecuteRequest.java

@@ -15,8 +15,10 @@
 
 package software.amazon.awssdk.http.async;
 
+import java.util.Optional;
 import software.amazon.awssdk.annotations.SdkPublicApi;
 import software.amazon.awssdk.http.SdkHttpRequest;
+import software.amazon.awssdk.metrics.MetricCollector;
 
 /**
  * Request object containing the parameters necessary to make an asynchronous HTTP request.
@@ -28,12 +30,14 @@ public final class AsyncExecuteRequest {
     private final SdkHttpRequest request;
     private final SdkHttpContentPublisher requestContentPublisher;
     private final SdkAsyncHttpResponseHandler responseHandler;
+    private final MetricCollector metricCollector;
     private final boolean isFullDuplex;
 
     private AsyncExecuteRequest(BuilderImpl builder) {
         this.request = builder.request;
         this.requestContentPublisher = builder.requestContentPublisher;
         this.responseHandler = builder.responseHandler;
+        this.metricCollector = builder.metricCollector;
         this.isFullDuplex = builder.isFullDuplex;
     }
 
@@ -58,6 +62,13 @@ public SdkAsyncHttpResponseHandler responseHandler() {
         return responseHandler;
     }
 
+    /**
+     * @return The {@link MetricCollector}.
+     */
+    public Optional<MetricCollector> metricCollector() {
+        return Optional.ofNullable(metricCollector);
+    }
+
     /**
      * @return True if the operation this request belongs to is full duplex. Otherwise false.
      */
@@ -94,6 +105,15 @@ public interface Builder {
          */
         Builder responseHandler(SdkAsyncHttpResponseHandler responseHandler);
 
+        /**
+         * Set the {@link MetricCollector} to be used by the HTTP client to
+         * report metrics collected for this request.
+         *
+         * @param metricCollector The metric collector.
+         * @return This builder for method chaining.
+         */
+        Builder metricCollector(MetricCollector metricCollector);
+
         /**
          * Option to indicate if the request is for a full duplex operation ie., request and response are sent/received at
          * the same time.
@@ -112,6 +132,7 @@ private static class BuilderImpl implements Builder {
         private SdkHttpRequest request;
         private SdkHttpContentPublisher requestContentPublisher;
         private SdkAsyncHttpResponseHandler responseHandler;
+        private MetricCollector metricCollector;
         private boolean isFullDuplex;
 
         @Override
@@ -132,6 +153,12 @@ public Builder responseHandler(SdkAsyncHttpResponseHandler responseHandler) {
             return this;
         }
 
+        @Override
+        public Builder metricCollector(MetricCollector metricCollector) {
+            this.metricCollector = metricCollector;
+            return this;
+        }
+
         @Override
         public Builder fullDuplex(boolean fullDuplex) {
             isFullDuplex = fullDuplex;

http-clients/apache-client/src/main/java/software/amazon/awssdk/http/apache/ApacheHttpClient.java

@@ -78,7 +78,6 @@
 import software.amazon.awssdk.http.TlsTrustManagersProvider;
 import software.amazon.awssdk.http.apache.internal.ApacheHttpRequestConfig;
 import software.amazon.awssdk.http.apache.internal.DefaultConfiguration;
-import software.amazon.awssdk.http.apache.internal.NoOpMetricCollector;
 import software.amazon.awssdk.http.apache.internal.SdkProxyRoutePlanner;
 import software.amazon.awssdk.http.apache.internal.conn.ClientConnectionManagerFactory;
 import software.amazon.awssdk.http.apache.internal.conn.IdleConnectionReaper;
@@ -89,6 +88,7 @@
 import software.amazon.awssdk.http.apache.internal.impl.ConnectionManagerAwareHttpClient;
 import software.amazon.awssdk.http.apache.internal.utils.ApacheUtils;
 import software.amazon.awssdk.metrics.MetricCollector;
+import software.amazon.awssdk.metrics.NoOpMetricCollector;
 import software.amazon.awssdk.utils.AttributeMap;
 import software.amazon.awssdk.utils.Logger;
 import software.amazon.awssdk.utils.Validate;

http-clients/netty-nio-client/pom.xml

@@ -46,6 +46,11 @@
             <artifactId>utils</artifactId>
             <version>${awsjavasdk.version}</version>
         </dependency>
+        <dependency>
+            <groupId>software.amazon.awssdk</groupId>
+            <artifactId>metrics-spi</artifactId>
+            <version>${awsjavasdk.version}</version>
+        </dependency>
 
         <!--Netty dependencies-->
         <dependency>

http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/NettyNioAsyncHttpClient.java

@@ -15,6 +15,7 @@
 
 package software.amazon.awssdk.http.nio.netty;
 
+import static software.amazon.awssdk.http.HttpMetric.HTTP_CLIENT_NAME;
 import static software.amazon.awssdk.http.nio.netty.internal.NettyConfiguration.EVENTLOOP_SHUTDOWN_FUTURE_TIMEOUT_SECONDS;
 import static software.amazon.awssdk.http.nio.netty.internal.NettyConfiguration.EVENTLOOP_SHUTDOWN_QUIET_PERIOD_SECONDS;
 import static software.amazon.awssdk.http.nio.netty.internal.NettyConfiguration.EVENTLOOP_SHUTDOWN_TIMEOUT_SECONDS;
@@ -23,7 +24,6 @@
 
 import io.netty.channel.ChannelOption;
 import io.netty.channel.EventLoopGroup;
-import io.netty.channel.pool.ChannelPool;
 import io.netty.handler.ssl.SslContext;
 import io.netty.handler.ssl.SslProvider;
 import java.net.URI;
@@ -51,6 +51,7 @@
 import software.amazon.awssdk.http.nio.netty.internal.NonManagedEventLoopGroup;
 import software.amazon.awssdk.http.nio.netty.internal.RequestContext;
 import software.amazon.awssdk.http.nio.netty.internal.SdkChannelOptions;
+import software.amazon.awssdk.http.nio.netty.internal.SdkChannelPool;
 import software.amazon.awssdk.http.nio.netty.internal.SdkChannelPoolMap;
 import software.amazon.awssdk.http.nio.netty.internal.SharedSdkEventLoopGroup;
 import software.amazon.awssdk.utils.AttributeMap;
@@ -79,7 +80,7 @@ public final class NettyNioAsyncHttpClient implements SdkAsyncHttpClient {
                     .build();
 
     private final SdkEventLoopGroup sdkEventLoopGroup;
-    private final SdkChannelPoolMap<URI, ? extends ChannelPool> pools;
+    private final SdkChannelPoolMap<URI, ? extends SdkChannelPool> pools;
     private final NettyConfiguration configuration;
 
     private NettyNioAsyncHttpClient(DefaultBuilder builder, AttributeMap serviceDefaultsMap) {
@@ -107,7 +108,7 @@ private NettyNioAsyncHttpClient(DefaultBuilder builder, AttributeMap serviceDefa
 
     @SdkTestInternalApi
     NettyNioAsyncHttpClient(SdkEventLoopGroup sdkEventLoopGroup,
-                            SdkChannelPoolMap<URI, ? extends ChannelPool> pools,
+                            SdkChannelPoolMap<URI, ? extends SdkChannelPool> pools,
                             NettyConfiguration configuration) {
         this.sdkEventLoopGroup = sdkEventLoopGroup;
         this.pools = pools;
@@ -117,6 +118,7 @@ private NettyNioAsyncHttpClient(DefaultBuilder builder, AttributeMap serviceDefa
     @Override
     public CompletableFuture<Void> execute(AsyncExecuteRequest request) {
         RequestContext ctx = createRequestContext(request);
+        ctx.metricCollector().reportMetric(HTTP_CLIENT_NAME, clientName()); // TODO: Can't this be done in core?
         return new NettyRequestExecutor(ctx).execute();
     }
 
@@ -125,7 +127,7 @@ public static Builder builder() {
     }
 
     private RequestContext createRequestContext(AsyncExecuteRequest request) {
-        ChannelPool pool = pools.get(poolKey(request.request()));
+        SdkChannelPool pool = pools.get(poolKey(request.request()));
         return new RequestContext(pool, sdkEventLoopGroup.eventLoopGroup(), request, configuration);
     }
 

http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/AwaitCloseChannelPoolMap.java

@@ -150,7 +150,7 @@ protected SimpleChannelPoolAwareChannelPool newPool(URI key) {
             baseChannelPool = tcpChannelPool;
         }
 
-        ChannelPool wrappedPool = wrapBaseChannelPool(bootstrap, baseChannelPool);
+        SdkChannelPool wrappedPool = wrapBaseChannelPool(bootstrap, baseChannelPool);
 
         channelPoolRef.set(wrappedPool);
         return new SimpleChannelPoolAwareChannelPool(wrappedPool, tcpChannelPool);
@@ -231,33 +231,32 @@ private URI proxyAddress(URI remoteHost) {
         }
     }
 
-    private ChannelPool wrapBaseChannelPool(Bootstrap bootstrap, ChannelPool channelPool) {
+    private SdkChannelPool wrapBaseChannelPool(Bootstrap bootstrap, ChannelPool channelPool) {
 
         // Wrap the channel pool such that the ChannelAttributeKey.CLOSE_ON_RELEASE flag is honored.
         channelPool = new HonorCloseOnReleaseChannelPool(channelPool);
 
         // Wrap the channel pool such that HTTP 2 channels won't be released to the underlying pool while they're still in use.
-        channelPool = new HttpOrHttp2ChannelPool(channelPool,
-                                                 bootstrap.config().group(),
-                                                 configuration.maxConnections(),
-                                                 configuration);
-
+        SdkChannelPool sdkChannelPool = new HttpOrHttp2ChannelPool(channelPool,
+                                                                   bootstrap.config().group(),
+                                                                   configuration.maxConnections(),
+                                                                   configuration);
 
         // Wrap the channel pool such that we remove request-specific handlers with each request.
-        channelPool = new HandlerRemovingChannelPool(channelPool);
+        sdkChannelPool = new HandlerRemovingChannelPool(sdkChannelPool);
 
         // Wrap the channel pool such that an individual channel can only be released to the underlying pool once.
-        channelPool = new ReleaseOnceChannelPool(channelPool);
+        sdkChannelPool = new ReleaseOnceChannelPool(sdkChannelPool);
 
         // Wrap the channel pool to guarantee all channels checked out are healthy, and all unhealthy channels checked in are
         // closed.
-        channelPool = new HealthCheckedChannelPool(bootstrap.config().group(), configuration, channelPool);
+        sdkChannelPool = new HealthCheckedChannelPool(bootstrap.config().group(), configuration, sdkChannelPool);
 
         // Wrap the channel pool such that if the Promise given to acquire(Promise) is done when the channel is acquired
         // from the underlying pool, the channel is closed and released.
-        channelPool = new CancellableAcquireChannelPool(bootstrap.config().group().next(), channelPool);
+        sdkChannelPool = new CancellableAcquireChannelPool(bootstrap.config().group().next(), sdkChannelPool);
 
-        return channelPool;
+        return sdkChannelPool;
     }
 
     private SslContext sslContext(URI targetAddress) {

http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/CancellableAcquireChannelPool.java

@@ -20,7 +20,9 @@
 import io.netty.util.concurrent.EventExecutor;
 import io.netty.util.concurrent.Future;
 import io.netty.util.concurrent.Promise;
+import java.util.concurrent.CompletableFuture;
 import software.amazon.awssdk.annotations.SdkInternalApi;
+import software.amazon.awssdk.metrics.MetricCollector;
 
 /**
  * Simple decorator {@link ChannelPool} that attempts to complete the promise
@@ -29,11 +31,11 @@
  * is closed then released back to the delegate.
  */
 @SdkInternalApi
-public final class CancellableAcquireChannelPool implements ChannelPool {
+public final class CancellableAcquireChannelPool implements SdkChannelPool {
     private final EventExecutor executor;
-    private final ChannelPool delegatePool;
+    private final SdkChannelPool delegatePool;
 
-    public CancellableAcquireChannelPool(EventExecutor executor, ChannelPool delegatePool) {
+    public CancellableAcquireChannelPool(EventExecutor executor, SdkChannelPool delegatePool) {
         this.executor = executor;
         this.delegatePool = delegatePool;
     }
@@ -73,4 +75,9 @@ public Future<Void> release(Channel channel, Promise<Void> promise) {
     public void close() {
         delegatePool.close();
     }
+
+    @Override
+    public CompletableFuture<Void> collectChannelPoolMetrics(MetricCollector metrics) {
+        return delegatePool.collectChannelPoolMetrics(metrics);
+    }
 }