diff --git a/httpclient5-observation/src/main/java/org/apache/hc/client5/http/observation/HttpClientObservationSupport.java b/httpclient5-observation/src/main/java/org/apache/hc/client5/http/observation/HttpClientObservationSupport.java index 6880f594b3..74f0cdcd88 100644 --- a/httpclient5-observation/src/main/java/org/apache/hc/client5/http/observation/HttpClientObservationSupport.java +++ b/httpclient5-observation/src/main/java/org/apache/hc/client5/http/observation/HttpClientObservationSupport.java @@ -43,8 +43,10 @@ import org.apache.hc.client5.http.observation.impl.ObservationClassicExecInterceptor; import org.apache.hc.client5.http.observation.impl.MeteredDnsResolver; import org.apache.hc.client5.http.observation.impl.MeteredTlsStrategy; +import org.apache.hc.client5.http.observation.interceptors.AsyncContentLengthExec; import org.apache.hc.client5.http.observation.interceptors.AsyncIoByteCounterExec; import org.apache.hc.client5.http.observation.interceptors.AsyncTimerExec; +import org.apache.hc.client5.http.observation.interceptors.ContentLengthExec; import org.apache.hc.client5.http.observation.interceptors.IoByteCounterExec; import org.apache.hc.client5.http.observation.interceptors.TimerExec; import org.apache.hc.client5.http.io.HttpClientConnectionManager; @@ -94,6 +96,8 @@ * that surrounds each execution with a start/stop span. *
  • Metric interceptors according to {@link ObservingOptions.MetricSet}: * BASIC (latency timer + response counter), IO (bytes in/out counters), + * CONTENT_LENGTH (declared request/response entity length distribution + * summaries, opt-in), * and CONN_POOL (pool gauges; classic and async variants). Pool lease * timing is available when using {@link #meteredConnectionManager} or * {@link #meteredAsyncConnectionManager}.
    • @@ -118,6 +122,10 @@ public final class HttpClientObservationSupport { * Internal ID for I/O byte counters interceptor. */ private static final String IO_ID = "metric-io"; + /** + * Internal ID for content length distribution summaries interceptor. + */ + private static final String CONTENT_LENGTH_ID = "metric-content-length"; /* ====================== Classic ====================== */ @@ -203,6 +211,9 @@ public static void enable(final HttpClientBuilder builder, if (o.metricSets.contains(ObservingOptions.MetricSet.IO)) { builder.addExecInterceptorFirst(IO_ID, new IoByteCounterExec(meterReg, o, config)); } + if (o.metricSets.contains(ObservingOptions.MetricSet.CONTENT_LENGTH)) { + builder.addExecInterceptorFirst(CONTENT_LENGTH_ID, new ContentLengthExec(meterReg, o, config)); + } if (o.metricSets.contains(ObservingOptions.MetricSet.CONN_POOL)) { ConnPoolMeters.bindTo(builder, meterReg, config); } @@ -294,6 +305,9 @@ public static void enable(final CachingHttpClientBuilder builder, if (o.metricSets.contains(ObservingOptions.MetricSet.IO)) { builder.addExecInterceptorAfter(ChainElement.CACHING.name(), IO_ID, new IoByteCounterExec(meterReg, o, config)); } + if (o.metricSets.contains(ObservingOptions.MetricSet.CONTENT_LENGTH)) { + builder.addExecInterceptorAfter(ChainElement.CACHING.name(), CONTENT_LENGTH_ID, new ContentLengthExec(meterReg, o, config)); + } if (o.metricSets.contains(ObservingOptions.MetricSet.CONN_POOL)) { ConnPoolMeters.bindTo(builder, meterReg, config); } @@ -381,6 +395,9 @@ public static void enable(final HttpAsyncClientBuilder builder, if (o.metricSets.contains(ObservingOptions.MetricSet.IO)) { builder.addExecInterceptorFirst(IO_ID, new AsyncIoByteCounterExec(meterReg, o, config)); } + if (o.metricSets.contains(ObservingOptions.MetricSet.CONTENT_LENGTH)) { + builder.addExecInterceptorFirst(CONTENT_LENGTH_ID, new AsyncContentLengthExec(meterReg, o, config)); + } if (o.metricSets.contains(ObservingOptions.MetricSet.CONN_POOL)) { ConnPoolMetersAsync.bindTo(builder, meterReg, config); } @@ -471,6 +488,9 @@ public static void enable(final CachingHttpAsyncClientBuilder builder, if (o.metricSets.contains(ObservingOptions.MetricSet.IO)) { builder.addExecInterceptorAfter(ChainElement.CACHING.name(), IO_ID, new AsyncIoByteCounterExec(meterReg, o, config)); } + if (o.metricSets.contains(ObservingOptions.MetricSet.CONTENT_LENGTH)) { + builder.addExecInterceptorAfter(ChainElement.CACHING.name(), CONTENT_LENGTH_ID, new AsyncContentLengthExec(meterReg, o, config)); + } if (o.metricSets.contains(ObservingOptions.MetricSet.CONN_POOL)) { ConnPoolMetersAsync.bindTo(builder, meterReg, config); } diff --git a/httpclient5-observation/src/main/java/org/apache/hc/client5/http/observation/ObservingOptions.java b/httpclient5-observation/src/main/java/org/apache/hc/client5/http/observation/ObservingOptions.java index d6c3d6501e..b1e1b23646 100644 --- a/httpclient5-observation/src/main/java/org/apache/hc/client5/http/observation/ObservingOptions.java +++ b/httpclient5-observation/src/main/java/org/apache/hc/client5/http/observation/ObservingOptions.java @@ -48,8 +48,25 @@ public final class ObservingOptions { * {@link org.apache.hc.client5.http.observation.HttpClientObservationSupport#meteredTlsStrategy} * and {@link org.apache.hc.client5.http.observation.HttpClientObservationSupport#meteredDnsResolver} * to wrap the underlying TLS strategy or DNS resolver. + *

    + * {@link MetricSet#CONTENT_LENGTH} records the declared request and response + * entity lengths as distribution summaries. See + * {@link org.apache.hc.client5.http.observation.interceptors.ContentLengthExec} + * for details. */ - public enum MetricSet { BASIC, IO, CONN_POOL, TLS, DNS } + public enum MetricSet { + BASIC, + IO, + CONN_POOL, + TLS, + DNS, + /** + * Distributions of the declared request and response entity lengths, in bytes. + * + * @since 5.7 + */ + CONTENT_LENGTH + } /** * How many tags each metric/trace should get. diff --git a/httpclient5-observation/src/main/java/org/apache/hc/client5/http/observation/interceptors/AsyncContentLengthExec.java b/httpclient5-observation/src/main/java/org/apache/hc/client5/http/observation/interceptors/AsyncContentLengthExec.java new file mode 100644 index 0000000000..7d66b87e78 --- /dev/null +++ b/httpclient5-observation/src/main/java/org/apache/hc/client5/http/observation/interceptors/AsyncContentLengthExec.java @@ -0,0 +1,149 @@ +/* + * ==================================================================== + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 + * + * http://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. + * ==================================================================== + * + * This software consists of voluntary contributions made by many + * individuals on behalf of the Apache Software Foundation. For more + * information on the Apache Software Foundation, please see + * . + * + */ +package org.apache.hc.client5.http.observation.interceptors; + +import java.io.IOException; + +import io.micrometer.core.instrument.DistributionSummary; +import io.micrometer.core.instrument.MeterRegistry; +import org.apache.hc.client5.http.async.AsyncExecCallback; +import org.apache.hc.client5.http.async.AsyncExecChain; +import org.apache.hc.client5.http.async.AsyncExecChainHandler; +import org.apache.hc.client5.http.observation.MetricConfig; +import org.apache.hc.client5.http.observation.ObservingOptions; +import org.apache.hc.core5.http.EntityDetails; +import org.apache.hc.core5.http.HttpException; +import org.apache.hc.core5.http.HttpRequest; +import org.apache.hc.core5.http.HttpResponse; +import org.apache.hc.core5.http.nio.AsyncDataConsumer; +import org.apache.hc.core5.http.nio.AsyncEntityProducer; +import org.apache.hc.core5.util.Args; + +/** + * Records the declared request / response entity lengths for + * asynchronous clients as {@link DistributionSummary distribution summaries}. + *

    + * Meters, tags and semantics are the same as for {@link ContentLengthExec}. The + * request length is taken from {@link AsyncEntityProducer#getContentLength()} and + * the response length from {@link EntityDetails#getContentLength()}. They are the + * declared entity lengths as the application sees them, not the number of bytes + * transferred, and they are uncompressed only if no content coding is visible to + * the application. Nothing is recorded if no response was received, for entities + * of unknown length (for example chunked) or for messages without an entity. The + * data stream is never wrapped or buffered. + *

    + * Over HTTP/2, HttpCore reports the response entity length as unknown even if + * a {@code Content-Length} header is present, so no response length is recorded + * for HTTP/2 exchanges. + *

    + * Enabled by {@link ObservingOptions.MetricSet#CONTENT_LENGTH}. + * + * @see ContentLengthExec + * @since 5.7 + */ +public final class AsyncContentLengthExec implements AsyncExecChainHandler { + + private final ObservingOptions opts; + private final ContentLengthRecorder recorder; + + /** + * Creates a new instance. + * + * @param meterRegistry meter registry to register meters with + * @param opts observation/metric options + * @param mc metric configuration + */ + public AsyncContentLengthExec(final MeterRegistry meterRegistry, + final ObservingOptions opts, + final MetricConfig mc) { + Args.notNull(meterRegistry, "meterRegistry"); + this.opts = Args.notNull(opts, "observingOptions"); + Args.notNull(mc, "metricConfig"); + this.recorder = new ContentLengthRecorder(meterRegistry, opts, mc); + } + + @Override + public void execute(final HttpRequest request, + final AsyncEntityProducer entityProducer, + final AsyncExecChain.Scope scope, + final AsyncExecChain chain, + final AsyncExecCallback callback) throws HttpException, IOException { + + if (!opts.spanSampling.test(request.getRequestUri())) { + chain.proceed(request, entityProducer, scope, callback); + return; + } + + final long requestContentLength = entityProducer != null + ? ContentLengthRecorder.normalize(entityProducer.getContentLength()) + : -1L; + + final AsyncExecCallback wrapped = new AsyncExecCallback() { + + private volatile HttpResponse response; + private volatile long responseContentLength = -1L; + + @Override + public AsyncDataConsumer handleResponse(final HttpResponse response, + final EntityDetails entityDetails) throws HttpException, IOException { + this.response = response; + if (entityDetails != null) { + responseContentLength = ContentLengthRecorder.normalize(entityDetails.getContentLength()); + } + return callback.handleResponse(response, entityDetails); + } + + @Override + public void handleInformationResponse(final HttpResponse response) throws HttpException, IOException { + callback.handleInformationResponse(response); + } + + @Override + public void completed() { + record(); + callback.completed(); + } + + @Override + public void failed(final Exception cause) { + record(); + callback.failed(cause); + } + + private void record() { + final HttpResponse rsp = response; + if (rsp != null) { + recorder.record(request.getMethod(), rsp.getCode(), scope.route.getTargetHost(), + request.getRequestUri(), requestContentLength, responseContentLength); + } + } + }; + + chain.proceed(request, entityProducer, scope, wrapped); + } + +} diff --git a/httpclient5-observation/src/main/java/org/apache/hc/client5/http/observation/interceptors/ContentLengthExec.java b/httpclient5-observation/src/main/java/org/apache/hc/client5/http/observation/interceptors/ContentLengthExec.java new file mode 100644 index 0000000000..da374021ea --- /dev/null +++ b/httpclient5-observation/src/main/java/org/apache/hc/client5/http/observation/interceptors/ContentLengthExec.java @@ -0,0 +1,128 @@ +/* + * ==================================================================== + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 + * + * http://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. + * ==================================================================== + * + * This software consists of voluntary contributions made by many + * individuals on behalf of the Apache Software Foundation. For more + * information on the Apache Software Foundation, please see + * . + * + */ +package org.apache.hc.client5.http.observation.interceptors; + +import java.io.IOException; + +import io.micrometer.core.instrument.DistributionSummary; +import io.micrometer.core.instrument.MeterRegistry; +import org.apache.hc.client5.http.classic.ExecChain; +import org.apache.hc.client5.http.classic.ExecChainHandler; +import org.apache.hc.client5.http.observation.MetricConfig; +import org.apache.hc.client5.http.observation.ObservingOptions; +import org.apache.hc.core5.http.ClassicHttpRequest; +import org.apache.hc.core5.http.ClassicHttpResponse; +import org.apache.hc.core5.http.HttpEntity; +import org.apache.hc.core5.http.HttpException; +import org.apache.hc.core5.util.Args; + +/** + * Records the declared request / response entity lengths for classic + * clients as {@link DistributionSummary distribution summaries}. + *

    + * Meters: + *

    + *

    + * Tags are the same as for the request timer: {@code method}, {@code status}, + * and when {@link ObservingOptions.TagLevel#EXTENDED} also {@code protocol} and + * {@code target}. {@link ObservingOptions#tagCustomizer} is applied and + * {@link MetricConfig#commonTags} are appended. {@link MetricConfig#percentiles} + * are published if configured. + *

    + * What is measured. The values are taken from {@link HttpEntity#getContentLength()} + * of the request and response entities as the application sees them: before the + * client applies any content coding to the request and after the client + * transparently decodes the response. They are the declared entity lengths, not + * the number of bytes transferred: they exclude headers and transfer framing, + * and they are recorded even if the body is not (fully) transmitted, for example + * when the server rejects a request before reading its body. + *

    + * The declared length is the uncompressed length only if no content coding is + * visible to the application. It is the encoded length if the application sends + * an entity that is already encoded (for example with its own + * {@code Content-Encoding: gzip}) or if content compression is disabled and the + * response is not decoded. When the client handles content coding, this metric + * is useful where limits apply to the decoded size, for example servers that + * reject requests whose decoded body exceeds a maximum content length. + *

    + * What is not measured. Nothing is recorded if no response was received + * (for example the connection could not be established), for an entity whose + * length is not known up front (for example {@code Transfer-Encoding: chunked} + * or a response the client decodes transparently), or for a message without + * an entity. Entities are never wrapped or buffered, so the instrumentation adds + * no I/O overhead. + *

    + * Enabled by {@link ObservingOptions.MetricSet#CONTENT_LENGTH}. + * + * @see AsyncContentLengthExec + * @since 5.7 + */ +public final class ContentLengthExec implements ExecChainHandler { + + private final ObservingOptions opts; + private final ContentLengthRecorder recorder; + + /** + * Creates a new instance. + * + * @param meterRegistry meter registry to register meters with + * @param opts observation/metric options + * @param mc metric configuration + */ + public ContentLengthExec(final MeterRegistry meterRegistry, + final ObservingOptions opts, + final MetricConfig mc) { + Args.notNull(meterRegistry, "meterRegistry"); + this.opts = Args.notNull(opts, "observingOptions"); + Args.notNull(mc, "metricConfig"); + this.recorder = new ContentLengthRecorder(meterRegistry, opts, mc); + } + + @Override + public ClassicHttpResponse execute(final ClassicHttpRequest request, + final ExecChain.Scope scope, + final ExecChain chain) throws IOException, HttpException { + + if (!opts.spanSampling.test(request.getRequestUri())) { + return chain.proceed(request, scope); + } + + final long requestContentLength = contentLength(request.getEntity()); + final ClassicHttpResponse response = chain.proceed(request, scope); + recorder.record(request.getMethod(), response.getCode(), scope.route.getTargetHost(), + request.getRequestUri(), requestContentLength, contentLength(response.getEntity())); + return response; + } + + private static long contentLength(final HttpEntity entity) { + return entity != null ? ContentLengthRecorder.normalize(entity.getContentLength()) : -1L; + } + +} diff --git a/httpclient5-observation/src/main/java/org/apache/hc/client5/http/observation/interceptors/ContentLengthRecorder.java b/httpclient5-observation/src/main/java/org/apache/hc/client5/http/observation/interceptors/ContentLengthRecorder.java new file mode 100644 index 0000000000..dde03bf813 --- /dev/null +++ b/httpclient5-observation/src/main/java/org/apache/hc/client5/http/observation/interceptors/ContentLengthRecorder.java @@ -0,0 +1,119 @@ +/* + * ==================================================================== + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 + * + * http://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. + * ==================================================================== + * + * This software consists of voluntary contributions made by many + * individuals on behalf of the Apache Software Foundation. For more + * information on the Apache Software Foundation, please see + * . + * + */ +package org.apache.hc.client5.http.observation.interceptors; + +import java.util.ArrayList; +import java.util.List; + +import io.micrometer.core.instrument.DistributionSummary; +import io.micrometer.core.instrument.MeterRegistry; +import io.micrometer.core.instrument.Tag; +import io.micrometer.core.instrument.binder.BaseUnits; +import org.apache.hc.client5.http.observation.MetricConfig; +import org.apache.hc.client5.http.observation.ObservingOptions; +import org.apache.hc.core5.http.HttpHost; + +/** + * Records declared request / response entity lengths, shared by + * {@link ContentLengthExec} and {@link AsyncContentLengthExec}. + */ +final class ContentLengthRecorder { + + static final String REQUEST_SUFFIX = ".request.content.length"; + static final String RESPONSE_SUFFIX = ".response.content.length"; + + private final MeterRegistry meterRegistry; + private final ObservingOptions opts; + private final MetricConfig mc; + private final String requestName; + private final String responseName; + + ContentLengthRecorder(final MeterRegistry meterRegistry, + final ObservingOptions opts, + final MetricConfig mc) { + this.meterRegistry = meterRegistry; + this.opts = opts; + this.mc = mc; + this.requestName = mc.prefix + REQUEST_SUFFIX; + this.responseName = mc.prefix + RESPONSE_SUFFIX; + } + + /** + * Records the given lengths of an exchange that received a response. + * Negative lengths (unknown length or no entity) are skipped. + */ + void record(final String method, + final int status, + final HttpHost targetHost, + final String uri, + final long requestContentLength, + final long responseContentLength) { + if (requestContentLength < 0 && responseContentLength < 0) { + return; + } + final List tags = buildTags(method, status, targetHost, uri); + if (requestContentLength >= 0) { + summary(requestName, "Declared HTTP client request entity length", tags).record(requestContentLength); + } + if (responseContentLength >= 0) { + summary(responseName, "Declared HTTP client response entity length", tags).record(responseContentLength); + } + } + + private DistributionSummary summary(final String name, final String description, final List tags) { + final DistributionSummary.Builder builder = DistributionSummary.builder(name) + .description(description) + .baseUnit(BaseUnits.BYTES) + .tags(mc.commonTags) + .tags(tags); + if (mc.percentiles != null && mc.percentiles.length > 0) { + builder.publishPercentiles(mc.percentiles); + } + return builder.register(meterRegistry); + } + + private List buildTags(final String method, + final int status, + final HttpHost targetHost, + final String uri) { + final String protocol = targetHost != null ? targetHost.getSchemeName() : null; + final String target = targetHost != null ? targetHost.getHostName() : null; + final List tags = new ArrayList<>(4); + tags.add(Tag.of("method", method)); + tags.add(Tag.of("status", Integer.toString(status))); + if (opts.tagLevel == ObservingOptions.TagLevel.EXTENDED && targetHost != null) { + tags.add(Tag.of("protocol", protocol)); + tags.add(Tag.of("target", target)); + } + return opts.tagCustomizer.apply(tags, method, status, protocol, target, uri); + } + + static long normalize(final long contentLength) { + return contentLength >= 0 ? contentLength : -1L; + } + +} diff --git a/httpclient5-observation/src/test/java/org/apache/hc/client5/http/observation/interceptors/AsyncContentLengthExecTest.java b/httpclient5-observation/src/test/java/org/apache/hc/client5/http/observation/interceptors/AsyncContentLengthExecTest.java new file mode 100644 index 0000000000..8fee976a89 --- /dev/null +++ b/httpclient5-observation/src/test/java/org/apache/hc/client5/http/observation/interceptors/AsyncContentLengthExecTest.java @@ -0,0 +1,253 @@ +/* + * ==================================================================== + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 + * + * http://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. + * ==================================================================== + * + * This software consists of voluntary contributions made by many + * individuals on behalf of the Apache Software Foundation. For more + * information on the Apache Software Foundation, please see + * . + * + */ +package org.apache.hc.client5.http.observation.interceptors; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.net.InetAddress; +import java.net.ServerSocket; +import java.net.URI; +import java.nio.charset.StandardCharsets; +import java.util.EnumSet; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; + +import io.micrometer.core.instrument.DistributionSummary; +import io.micrometer.core.instrument.MeterRegistry; +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; +import org.apache.hc.client5.http.async.methods.SimpleHttpRequest; +import org.apache.hc.client5.http.async.methods.SimpleHttpResponse; +import org.apache.hc.client5.http.async.methods.SimpleRequestBuilder; +import org.apache.hc.client5.http.async.methods.SimpleResponseConsumer; +import org.apache.hc.client5.http.impl.async.CloseableHttpAsyncClient; +import org.apache.hc.client5.http.impl.async.HttpAsyncClientBuilder; +import org.apache.hc.client5.http.impl.async.HttpAsyncClients; +import org.apache.hc.client5.http.observation.HttpClientObservationSupport; +import org.apache.hc.client5.http.observation.MetricConfig; +import org.apache.hc.client5.http.observation.ObservingOptions; +import org.apache.hc.core5.http.ContentType; +import org.apache.hc.core5.http.HttpStatus; +import org.apache.hc.core5.http.impl.bootstrap.HttpServer; +import org.apache.hc.core5.http.impl.bootstrap.ServerBootstrap; +import org.apache.hc.core5.http.io.entity.EntityUtils; +import org.apache.hc.core5.http.io.entity.InputStreamEntity; +import org.apache.hc.core5.http.io.entity.StringEntity; +import org.apache.hc.core5.http.message.BasicHttpRequest; +import org.apache.hc.core5.http.nio.AsyncRequestProducer; +import org.apache.hc.core5.http.nio.entity.StringAsyncEntityProducer; +import org.apache.hc.core5.http.nio.support.BasicRequestProducer; +import org.apache.hc.core5.io.CloseMode; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +class AsyncContentLengthExecTest { + + private static final String REQUEST = "test.request.content.length"; + private static final String RESPONSE = "test.response.content.length"; + + private HttpServer server; + private String baseUrl; + private MeterRegistry meters; + + @BeforeEach + void setUp() throws IOException { + server = ServerBootstrap.bootstrap() + .setLocalAddress(InetAddress.getLoopbackAddress()) + .setListenerPort(0) + .register("localhost", "/known", (request, response, context) -> { + EntityUtils.consume(request.getEntity()); + response.setCode(HttpStatus.SC_OK); + response.setEntity(new StringEntity("OK!", ContentType.TEXT_PLAIN)); + }) + .register("localhost", "/chunked", (request, response, context) -> { + EntityUtils.consume(request.getEntity()); + response.setCode(HttpStatus.SC_OK); + response.setEntity(new InputStreamEntity( + new ByteArrayInputStream("chunky".getBytes(StandardCharsets.US_ASCII)), -1, + ContentType.TEXT_PLAIN)); + }) + .register("localhost", "/empty", (request, response, context) -> + response.setCode(HttpStatus.SC_NO_CONTENT)) + .register("localhost", "/error", (request, response, context) -> { + EntityUtils.consume(request.getEntity()); + response.setCode(HttpStatus.SC_INTERNAL_SERVER_ERROR); + response.setEntity(new StringEntity("Boom!!", ContentType.TEXT_PLAIN)); + }) + .create(); + server.start(); + baseUrl = "http://localhost:" + server.getLocalPort(); + meters = new SimpleMeterRegistry(); + } + + @AfterEach + void tearDown() { + if (server != null) { + server.close(CloseMode.GRACEFUL); + } + } + + private static ObservingOptions options() { + return ObservingOptions.builder() + .metrics(EnumSet.of(ObservingOptions.MetricSet.CONTENT_LENGTH)) + .build(); + } + + private static MetricConfig config() { + return MetricConfig.builder().prefix("test").percentiles().build(); + } + + private SimpleHttpResponse execute(final AsyncRequestProducer producer) throws Exception { + final HttpAsyncClientBuilder builder = HttpAsyncClients.custom().disableAutomaticRetries(); + builder.addExecInterceptorFirst("content-length", new AsyncContentLengthExec(meters, options(), config())); + try (final CloseableHttpAsyncClient client = builder.build()) { + client.start(); + return client.execute(producer, SimpleResponseConsumer.create(), null).get(20, TimeUnit.SECONDS); + } + } + + private SimpleHttpResponse execute(final SimpleHttpRequest request) throws Exception { + final HttpAsyncClientBuilder builder = HttpAsyncClients.custom().disableAutomaticRetries(); + builder.addExecInterceptorFirst("content-length", new AsyncContentLengthExec(meters, options(), config())); + try (final CloseableHttpAsyncClient client = builder.build()) { + client.start(); + return client.execute(request, null).get(20, TimeUnit.SECONDS); + } + } + + @Test + void recordsKnownRequestAndResponseLengths() throws Exception { + final SimpleHttpResponse response = execute(SimpleRequestBuilder.post(baseUrl + "/known") + .setBody("HELLO", ContentType.TEXT_PLAIN) + .build()); + assertEquals(200, response.getCode()); + + final DistributionSummary req = meters.find(REQUEST).tag("method", "POST").tag("status", "200").summary(); + assertNotNull(req); + assertEquals(1, req.count()); + assertEquals(5.0, req.totalAmount()); + assertEquals("bytes", req.getId().getBaseUnit()); + + final DistributionSummary rsp = meters.find(RESPONSE).tag("method", "POST").tag("status", "200").summary(); + assertNotNull(rsp); + assertEquals(1, rsp.count()); + assertEquals(3.0, rsp.totalAmount()); + } + + @Test + void recordsNothingForChunkedBodies() throws Exception { + final SimpleHttpResponse response = execute(new BasicRequestProducer( + new BasicHttpRequest("POST", URI.create(baseUrl + "/chunked")), + new StringAsyncEntityProducer("HELLO", ContentType.TEXT_PLAIN))); + assertEquals(200, response.getCode()); + assertEquals("chunky", response.getBodyText()); + + assertNull(meters.find(REQUEST).summary()); + assertNull(meters.find(RESPONSE).summary()); + } + + @Test + void recordsNothingWithoutEntity() throws Exception { + final SimpleHttpResponse response = execute(SimpleRequestBuilder.get(baseUrl + "/empty").build()); + assertEquals(204, response.getCode()); + + assertNull(meters.find(REQUEST).summary()); + assertNull(meters.find(RESPONSE).summary()); + } + + @Test + void recordsResponseOnlyForGet() throws Exception { + execute(SimpleRequestBuilder.get(baseUrl + "/known").build()); + + assertNull(meters.find(REQUEST).summary()); + final DistributionSummary rsp = meters.find(RESPONSE).tag("method", "GET").summary(); + assertNotNull(rsp); + assertEquals(3.0, rsp.totalAmount()); + } + + @Test + void recordsErrorResponsesWithStatusTag() throws Exception { + final SimpleHttpResponse response = execute(SimpleRequestBuilder.post(baseUrl + "/error") + .setBody("HELLO", ContentType.TEXT_PLAIN) + .build()); + assertEquals(500, response.getCode()); + + final DistributionSummary req = meters.find(REQUEST).tag("status", "500").summary(); + assertNotNull(req); + assertEquals(5.0, req.totalAmount()); + final DistributionSummary rsp = meters.find(RESPONSE).tag("status", "500").summary(); + assertNotNull(rsp); + assertEquals(6.0, rsp.totalAmount()); + } + + @Test + void recordsNothingWhenNoResponseIsReceived() throws Exception { + final int closedPort; + try (final ServerSocket socket = new ServerSocket(0, 1, InetAddress.getLoopbackAddress())) { + closedPort = socket.getLocalPort(); + } + + assertThrows(ExecutionException.class, () -> execute( + SimpleRequestBuilder.post("http://localhost:" + closedPort + "/known") + .setBody("HELLO", ContentType.TEXT_PLAIN) + .build())); + + assertNull(meters.find(REQUEST).summary()); + assertNull(meters.find(RESPONSE).summary()); + } + + @Test + void enabledOnlyViaContentLengthMetricSet() throws Exception { + final MeterRegistry defaultMeters = new SimpleMeterRegistry(); + final HttpAsyncClientBuilder defaults = HttpAsyncClients.custom(); + HttpClientObservationSupport.enable(defaults, null, defaultMeters, ObservingOptions.DEFAULT); + try (final CloseableHttpAsyncClient client = defaults.build()) { + client.start(); + client.execute(SimpleRequestBuilder.post(baseUrl + "/known") + .setBody("HELLO", ContentType.TEXT_PLAIN) + .build(), null).get(20, TimeUnit.SECONDS); + } + assertNull(defaultMeters.find("http.client.request.content.length").summary()); + + final HttpAsyncClientBuilder optedIn = HttpAsyncClients.custom(); + HttpClientObservationSupport.enable(optedIn, null, meters, options()); + try (final CloseableHttpAsyncClient client = optedIn.build()) { + client.start(); + client.execute(SimpleRequestBuilder.post(baseUrl + "/known") + .setBody("HELLO", ContentType.TEXT_PLAIN) + .build(), null).get(20, TimeUnit.SECONDS); + } + assertNotNull(meters.find("http.client.request.content.length").summary()); + assertNotNull(meters.find("http.client.response.content.length").summary()); + } + +} diff --git a/httpclient5-observation/src/test/java/org/apache/hc/client5/http/observation/interceptors/ContentLengthExecTest.java b/httpclient5-observation/src/test/java/org/apache/hc/client5/http/observation/interceptors/ContentLengthExecTest.java new file mode 100644 index 0000000000..8a71873f86 --- /dev/null +++ b/httpclient5-observation/src/test/java/org/apache/hc/client5/http/observation/interceptors/ContentLengthExecTest.java @@ -0,0 +1,277 @@ +/* + * ==================================================================== + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 + * + * http://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. + * ==================================================================== + * + * This software consists of voluntary contributions made by many + * individuals on behalf of the Apache Software Foundation. For more + * information on the Apache Software Foundation, please see + * . + * + */ +package org.apache.hc.client5.http.observation.interceptors; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.net.InetAddress; +import java.net.ServerSocket; +import java.nio.charset.StandardCharsets; +import java.util.EnumSet; + +import io.micrometer.core.instrument.DistributionSummary; +import io.micrometer.core.instrument.MeterRegistry; +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; +import org.apache.hc.client5.http.impl.classic.CloseableHttpClient; +import org.apache.hc.client5.http.impl.classic.HttpClientBuilder; +import org.apache.hc.client5.http.impl.classic.HttpClients; +import org.apache.hc.client5.http.observation.HttpClientObservationSupport; +import org.apache.hc.client5.http.observation.MetricConfig; +import org.apache.hc.client5.http.observation.ObservingOptions; +import org.apache.hc.core5.http.ClassicHttpRequest; +import org.apache.hc.core5.http.ContentType; +import org.apache.hc.core5.http.HttpHost; +import org.apache.hc.core5.http.HttpStatus; +import org.apache.hc.core5.http.impl.bootstrap.HttpServer; +import org.apache.hc.core5.http.impl.bootstrap.ServerBootstrap; +import org.apache.hc.core5.http.io.entity.EntityUtils; +import org.apache.hc.core5.http.io.entity.InputStreamEntity; +import org.apache.hc.core5.http.io.entity.StringEntity; +import org.apache.hc.core5.http.io.support.ClassicRequestBuilder; +import org.apache.hc.core5.io.CloseMode; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +class ContentLengthExecTest { + + private static final String REQUEST = "test.request.content.length"; + private static final String RESPONSE = "test.response.content.length"; + + private HttpServer server; + private HttpHost target; + private MeterRegistry meters; + + @BeforeEach + void setUp() throws IOException { + server = ServerBootstrap.bootstrap() + .setLocalAddress(InetAddress.getLoopbackAddress()) + .setListenerPort(0) + .register("localhost", "/known", (request, response, context) -> { + EntityUtils.consume(request.getEntity()); + response.setCode(HttpStatus.SC_OK); + response.setEntity(new StringEntity("OK!", ContentType.TEXT_PLAIN)); + }) + .register("localhost", "/chunked", (request, response, context) -> { + EntityUtils.consume(request.getEntity()); + response.setCode(HttpStatus.SC_OK); + response.setEntity(new InputStreamEntity( + new ByteArrayInputStream("chunky".getBytes(StandardCharsets.US_ASCII)), -1, + ContentType.TEXT_PLAIN)); + }) + .register("localhost", "/empty", (request, response, context) -> + response.setCode(HttpStatus.SC_NO_CONTENT)) + .register("localhost", "/error", (request, response, context) -> { + EntityUtils.consume(request.getEntity()); + response.setCode(HttpStatus.SC_INTERNAL_SERVER_ERROR); + response.setEntity(new StringEntity("Boom!!", ContentType.TEXT_PLAIN)); + }) + .create(); + server.start(); + target = new HttpHost("http", "localhost", server.getLocalPort()); + meters = new SimpleMeterRegistry(); + } + + @AfterEach + void tearDown() { + if (server != null) { + server.close(CloseMode.GRACEFUL); + } + } + + private static ObservingOptions options(final ObservingOptions.TagLevel tagLevel) { + return ObservingOptions.builder() + .metrics(EnumSet.of(ObservingOptions.MetricSet.CONTENT_LENGTH)) + .tagLevel(tagLevel) + .build(); + } + + private static MetricConfig config() { + return MetricConfig.builder().prefix("test").percentiles().build(); + } + + private void execute(final ObservingOptions opts, final MetricConfig mc, + final HttpHost host, final ClassicHttpRequest request) throws IOException { + final HttpClientBuilder builder = HttpClients.custom().disableAutomaticRetries(); + builder.addExecInterceptorFirst("content-length", new ContentLengthExec(meters, opts, mc)); + try (final CloseableHttpClient client = builder.build()) { + client.execute(host, request, response -> { + EntityUtils.consume(response.getEntity()); + return null; + }); + } + } + + @Test + void recordsKnownRequestAndResponseLengths() throws Exception { + execute(options(ObservingOptions.TagLevel.LOW), config(), target, + ClassicRequestBuilder.post("/known") + .setEntity(new StringEntity("HELLO", ContentType.TEXT_PLAIN)) + .build()); + + final DistributionSummary req = meters.find(REQUEST).tag("method", "POST").tag("status", "200").summary(); + assertNotNull(req); + assertEquals(1, req.count()); + assertEquals(5.0, req.totalAmount()); + assertEquals("bytes", req.getId().getBaseUnit()); + + final DistributionSummary rsp = meters.find(RESPONSE).tag("method", "POST").tag("status", "200").summary(); + assertNotNull(rsp); + assertEquals(1, rsp.count()); + assertEquals(3.0, rsp.totalAmount()); + + assertNull(meters.find(REQUEST).tagKeys("protocol").summary()); + assertNull(meters.find(REQUEST).tagKeys("uri").summary()); + } + + @Test + void recordsNothingForChunkedBodies() throws Exception { + execute(options(ObservingOptions.TagLevel.LOW), config(), target, + ClassicRequestBuilder.post("/chunked") + .setEntity(new InputStreamEntity( + new ByteArrayInputStream("HELLO".getBytes(StandardCharsets.US_ASCII)), -1, + ContentType.TEXT_PLAIN)) + .build()); + + assertNull(meters.find(REQUEST).summary()); + assertNull(meters.find(RESPONSE).summary()); + } + + @Test + void recordsNothingWithoutEntity() throws Exception { + execute(options(ObservingOptions.TagLevel.LOW), config(), target, + ClassicRequestBuilder.get("/empty").build()); + + assertNull(meters.find(REQUEST).summary()); + assertNull(meters.find(RESPONSE).summary()); + } + + @Test + void recordsResponseOnlyForGet() throws Exception { + execute(options(ObservingOptions.TagLevel.LOW), config(), target, + ClassicRequestBuilder.get("/known").build()); + + assertNull(meters.find(REQUEST).summary()); + final DistributionSummary rsp = meters.find(RESPONSE).tag("method", "GET").summary(); + assertNotNull(rsp); + assertEquals(3.0, rsp.totalAmount()); + } + + @Test + void recordsErrorResponsesWithStatusTag() throws Exception { + execute(options(ObservingOptions.TagLevel.LOW), config(), target, + ClassicRequestBuilder.post("/error") + .setEntity(new StringEntity("HELLO", ContentType.TEXT_PLAIN)) + .build()); + + final DistributionSummary req = meters.find(REQUEST).tag("status", "500").summary(); + assertNotNull(req); + assertEquals(5.0, req.totalAmount()); + final DistributionSummary rsp = meters.find(RESPONSE).tag("status", "500").summary(); + assertNotNull(rsp); + assertEquals(6.0, rsp.totalAmount()); + } + + @Test + void recordsNothingWhenNoResponseIsReceived() throws Exception { + final int closedPort; + try (final ServerSocket socket = new ServerSocket(0, 1, InetAddress.getLoopbackAddress())) { + closedPort = socket.getLocalPort(); + } + final HttpHost unreachable = new HttpHost("http", "localhost", closedPort); + + assertThrows(IOException.class, () -> execute(options(ObservingOptions.TagLevel.LOW), config(), unreachable, + ClassicRequestBuilder.post("/known") + .setEntity(new StringEntity("HELLO", ContentType.TEXT_PLAIN)) + .build())); + + assertNull(meters.find(REQUEST).summary()); + assertNull(meters.find(RESPONSE).summary()); + } + + @Test + void extendedTagsAndCommonTags() throws Exception { + final MetricConfig mc = MetricConfig.builder().prefix("test").addCommonTag("app", "demo").build(); + execute(options(ObservingOptions.TagLevel.EXTENDED), mc, target, + ClassicRequestBuilder.post("/known") + .setEntity(new StringEntity("HELLO", ContentType.TEXT_PLAIN)) + .build()); + + final DistributionSummary req = meters.find(REQUEST) + .tag("protocol", "http") + .tag("target", "localhost") + .tag("app", "demo") + .summary(); + assertNotNull(req); + assertEquals(5.0, req.totalAmount()); + } + + @Test + void skipsUnsampledRequests() throws Exception { + final ObservingOptions opts = ObservingOptions.builder() + .metrics(EnumSet.of(ObservingOptions.MetricSet.CONTENT_LENGTH)) + .spanSampling(uri -> false) + .build(); + execute(opts, config(), target, + ClassicRequestBuilder.post("/known") + .setEntity(new StringEntity("HELLO", ContentType.TEXT_PLAIN)) + .build()); + + assertNull(meters.find(REQUEST).summary()); + assertNull(meters.find(RESPONSE).summary()); + } + + @Test + void enabledOnlyViaContentLengthMetricSet() throws Exception { + final MeterRegistry defaultMeters = new SimpleMeterRegistry(); + final HttpClientBuilder defaults = HttpClients.custom(); + HttpClientObservationSupport.enable(defaults, null, defaultMeters, ObservingOptions.DEFAULT); + try (final CloseableHttpClient client = defaults.build()) { + client.execute(target, ClassicRequestBuilder.post("/known") + .setEntity(new StringEntity("HELLO", ContentType.TEXT_PLAIN)) + .build(), response -> null); + } + assertNull(defaultMeters.find("http.client.request.content.length").summary()); + + final HttpClientBuilder optedIn = HttpClients.custom(); + HttpClientObservationSupport.enable(optedIn, null, meters, options(ObservingOptions.TagLevel.LOW)); + try (final CloseableHttpClient client = optedIn.build()) { + client.execute(target, ClassicRequestBuilder.post("/known") + .setEntity(new StringEntity("HELLO", ContentType.TEXT_PLAIN)) + .build(), response -> null); + } + assertNotNull(meters.find("http.client.request.content.length").summary()); + assertNotNull(meters.find("http.client.response.content.length").summary()); + assertNull(meters.find("http.client.request.bytes").counter()); + } + +}