dongjoon-hyun commented on code in PR #918:
URL:
https://github.com/apache/spark-kubernetes-operator/pull/918#discussion_r4295960808
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptor.java:
##########
@@ -103,21 +123,31 @@ public void before(BasicBuilder builder, HttpRequest
request, RequestTags tags)
}
/**
- * Called after a non-WebSocket HTTP response is received. The body might or
might not be already
- * consumed.
+ * Called after an HTTP response is received. The body might or might not be
already consumed.
*
* <p>Should be used to analyze response codes and headers, original
response shouldn't be
* altered.
*
+ * <p>fabric8 also calls this for a WebSocket upgrade response with a {@code
null} consumer, and
+ * again for an HTTP request re-sent after another interceptor's {@link
+ * Interceptor#afterFailure(BasicBuilder, HttpResponse, RequestTags)}
returns {@code true}, e.g.
+ * fabric8's {@code TokenRefreshInterceptor} on a {@code 401}. The two
values of a request re-sent
+ * this way add up to the time from sending it until the re-sent response
arrives.
+ *
* @param request the original request sent to the server.
* @param response the response received from the server.
+ * @param consumer the consumer of the response body, or {@code null} for a
WebSocket upgrade.
*/
@Override
public void after(
HttpRequest request,
HttpResponse<?> response,
AsyncBody.Consumer<List<ByteBuffer>> consumer) {
- updateResponseMetrics(response, System.nanoTime());
+ TimedConsumer timedConsumer = consumer == null ? null :
consumer.unwrap(TimedConsumer.class);
+ if (timedConsumer != null) {
+ responseLatency.update(timedConsumer.restart());
Review Comment:
nit: The latency is now recorded before `updateResponseMetrics` runs
`Objects.requireNonNull(response)`. With a null response, a sample would be
added and then the NPE would skip `responseRateMeter.mark()`. Previously the
check came before any metric update. Passing the latency, or the
`TimedConsumer`, into `updateResponseMetrics` and updating it after the check
would keep the counts consistent.
##########
docs/configuration.md:
##########
@@ -264,6 +264,11 @@ has a retry left for it, even if it does not retry it,
e.g. on a request timeout
a client which never retries, i.e. a Kubernetes event write or the recording
of the Kueue
`PodsReady` condition, is not counted there when it gets no response.
+`kubernetes.client.http.response.latency.nanos` is the time from when a
request is handed to the
Review Comment:
It may be worth adding that a request retried with backoff (429 / 5xx) is
recorded once per attempt, and the backoff is not included.
`StandardHttpClient.consumeBytes` calls `consumeBytesOnce` for each retry, and
that wraps the original consumer again, so each attempt gets a new start time.
A `401` re-send, in contrast, reuses the consumer and its values add up to the
total. Without this note, a reader may expect the attempts of a retried request
to add up as well.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptor.java:
##########
@@ -103,21 +123,31 @@ public void before(BasicBuilder builder, HttpRequest
request, RequestTags tags)
}
/**
- * Called after a non-WebSocket HTTP response is received. The body might or
might not be already
- * consumed.
+ * Called after an HTTP response is received. The body might or might not be
already consumed.
*
* <p>Should be used to analyze response codes and headers, original
response shouldn't be
* altered.
*
+ * <p>fabric8 also calls this for a WebSocket upgrade response with a {@code
null} consumer, and
+ * again for an HTTP request re-sent after another interceptor's {@link
+ * Interceptor#afterFailure(BasicBuilder, HttpResponse, RequestTags)}
returns {@code true}, e.g.
+ * fabric8's {@code TokenRefreshInterceptor} on a {@code 401}. The two
values of a request re-sent
+ * this way add up to the time from sending it until the re-sent response
arrives.
+ *
* @param request the original request sent to the server.
* @param response the response received from the server.
+ * @param consumer the consumer of the response body, or {@code null} for a
WebSocket upgrade.
*/
@Override
public void after(
HttpRequest request,
HttpResponse<?> response,
AsyncBody.Consumer<List<ByteBuffer>> consumer) {
- updateResponseMetrics(response, System.nanoTime());
+ TimedConsumer timedConsumer = consumer == null ? null :
consumer.unwrap(TimedConsumer.class);
Review Comment:
If an interceptor added after this one wraps the consumer without delegating
`unwrap(...)` (the default only matches its own class), this returns `null`.
The histogram then silently stops getting samples while `http.response` keeps
counting, which is the same silent failure this PR fixes.
`SparkOperator#getClientInterceptors` is `protected`, so a subclass can add
one. Could we log once (debug or warn) when `consumer` is non-null but has no
`TimedConsumer`?
##########
spark-operator/src/test/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptorTest.java:
##########
@@ -171,6 +186,104 @@ void testResponseCodeGroups() {
Assertions.assertEquals(1, meters.get("5xx").getCount());
}
+ @Test
+ @Order(5)
+ void testResponseLatency() {
+ KubernetesMetricsInterceptor metricsInterceptor = new
KubernetesMetricsInterceptor();
+ List<Interceptor> interceptors = List.of(metricsInterceptor);
+ try (KubernetesClient client =
+ KubernetesClientFactory.buildKubernetesClient(
+ interceptors, kubernetesClient.getConfiguration())) {
+ ConfigMap configMap = createConfigMap();
+ mockServer
+ .expect()
+ .get()
+ .delay(50)
+ .withPath(CONFIG_MAP_PATH)
+ .andReturn(200, configMap)
+ .once();
+ client.resource(configMap).get();
+
+ Histogram latency = latency(metricsInterceptor);
+ Assertions.assertEquals(1, latency.getCount());
+ Snapshot snapshot = latency.getSnapshot();
+ Assertions.assertTrue(snapshot.getMax() >= 50_000_000L, () -> "max: " +
snapshot.getMax());
+ Assertions.assertTrue(
+ snapshot.getMax() < 30_000_000_000L, () -> "max: " +
snapshot.getMax());
Review Comment:
nit: A 30 s upper bound would only catch a garbage value. Bounding it by the
wall-clock time measured around `client.resource(configMap).get()` would also
catch an over-measurement, e.g. a start time taken too early.
##########
docs/configuration.md:
##########
@@ -264,6 +264,11 @@ has a retry left for it, even if it does not retry it,
e.g. on a request timeout
a client which never retries, i.e. a Kubernetes event write or the recording
of the Kueue
`PodsReady` condition, is not counted there when it gets no response.
+`kubernetes.client.http.response.latency.nanos` is the time from when a
request is handed to the
+HTTP client until its response headers arrive, so it also includes waiting for
a connection. A
+request re-sent after a `401` is recorded twice. WebSocket upgrades, e.g. of a
watch, are not
Review Comment:
A `401` is not the only case. fabric8 7.8.0 always installs
`BackwardsCompatibilityInterceptor` in
`HttpClientUtils.createApplicableInterceptors` (unless
`kubernetes.backwardsCompatibilityInterceptor.disable` is set). For a legacy
group/version, its `afterFailure(...)` returns `true` on a 404/500 and re-sends
to the newer version through the same `effectiveConsumer`, so that request is
also recorded twice. The same wording appears in `docs/migration_guide.md` and
in the Javadoc of `after(...)`. How about "a request re-sent by another
interceptor, e.g. after a `401`"?
##########
spark-operator/src/test/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptorTest.java:
##########
@@ -171,6 +186,104 @@ void testResponseCodeGroups() {
Assertions.assertEquals(1, meters.get("5xx").getCount());
}
+ @Test
+ @Order(5)
+ void testResponseLatency() {
+ KubernetesMetricsInterceptor metricsInterceptor = new
KubernetesMetricsInterceptor();
+ List<Interceptor> interceptors = List.of(metricsInterceptor);
+ try (KubernetesClient client =
+ KubernetesClientFactory.buildKubernetesClient(
+ interceptors, kubernetesClient.getConfiguration())) {
+ ConfigMap configMap = createConfigMap();
+ mockServer
+ .expect()
+ .get()
+ .delay(50)
+ .withPath(CONFIG_MAP_PATH)
+ .andReturn(200, configMap)
+ .once();
+ client.resource(configMap).get();
+
+ Histogram latency = latency(metricsInterceptor);
+ Assertions.assertEquals(1, latency.getCount());
+ Snapshot snapshot = latency.getSnapshot();
+ Assertions.assertTrue(snapshot.getMax() >= 50_000_000L, () -> "max: " +
snapshot.getMax());
+ Assertions.assertTrue(
+ snapshot.getMax() < 30_000_000_000L, () -> "max: " +
snapshot.getMax());
+ }
+ }
+
+ @Test
+ @Order(6)
+ void testResponseLatencyOfResentRequest() {
+ KubernetesMetricsInterceptor metricsInterceptor = new
KubernetesMetricsInterceptor();
+ List<Interceptor> interceptors = List.of(metricsInterceptor);
+ Config config =
+ new ConfigBuilder(kubernetesClient.getConfiguration())
+ .withOauthTokenProvider(() -> "token")
+ .build();
+ try (KubernetesClient client =
+ KubernetesClientFactory.buildKubernetesClient(interceptors, config)) {
+ ConfigMap configMap = createConfigMap();
+ mockServer
+ .expect()
+ .get()
+ .delay(200)
+ .withPath(CONFIG_MAP_PATH)
+ .andReturn(HTTP_UNAUTHORIZED, new
StatusBuilder().withCode(HTTP_UNAUTHORIZED).build())
+ .once();
+ mockServer.expect().get().withPath(CONFIG_MAP_PATH).andReturn(200,
configMap).once();
+ client.resource(configMap).get();
+
+ Assertions.assertEquals(2, meterCount(metricsInterceptor,
"http.response"));
+ Assertions.assertEquals(1, meterCount(metricsInterceptor,
"http.response.401"));
+ Assertions.assertEquals(0, meterCount(metricsInterceptor, "failed"));
+ Histogram latency = latency(metricsInterceptor);
+ Assertions.assertEquals(2, latency.getCount());
+ Snapshot snapshot = latency.getSnapshot();
+ Assertions.assertTrue(snapshot.getMax() >= 200_000_000L, () -> "max: " +
snapshot.getMax());
+ Assertions.assertTrue(snapshot.getMin() < 200_000_000L, () -> "min: " +
snapshot.getMin());
+ }
+ }
+
+ @Test
+ @Order(7)
+ void testResponseLatencyOfWebSocketUpgrade() {
+ KubernetesMetricsInterceptor metricsInterceptor = new
KubernetesMetricsInterceptor();
+ List<Interceptor> interceptors = List.of(metricsInterceptor);
+ try (KubernetesClient client =
+ KubernetesClientFactory.buildKubernetesClient(
+ interceptors, kubernetesClient.getConfiguration());
+ SharedIndexInformer<ConfigMap> informer =
+ client.configMaps().inNamespace("spark-system").inform()) {
+ await()
+ .pollDelay(Duration.ZERO)
+ .untilAsserted(
+ () -> {
+ Assertions.assertEquals(2, meterCount(metricsInterceptor,
"http.response"));
+ Assertions.assertEquals(1, meterCount(metricsInterceptor,
"http.response.101"));
+ Assertions.assertEquals(1,
latency(metricsInterceptor).getCount());
+ });
+ Assertions.assertTrue(informer.isWatching());
+ }
+ }
+
+ @Test
+ @Order(8)
+ void testConsumerUnwrapsDelegate() {
+ AsyncBody.Consumer<List<ByteBuffer>> delegate = (value, asyncBody) -> {};
+ AsyncBody.Consumer<List<ByteBuffer>> consumer =
+ new KubernetesMetricsInterceptor().consumer(delegate, null);
+ Assertions.assertSame(delegate, consumer.unwrap(delegate.getClass()));
+ }
+
+ private static Histogram latency(KubernetesMetricsInterceptor interceptor) {
Review Comment:
nit: This repeats the lookup-and-assert-non-null pattern of
`TestUtils.meterCount`, for a histogram. How about moving it to `TestUtils`
next to `meterCount`?
##########
spark-operator/src/test/java/org/apache/spark/k8s/operator/utils/TestUtils.java:
##########
@@ -66,6 +69,12 @@ public static long calculateElapsedTimeInMills(long
startTime) {
return System.currentTimeMillis() - startTime;
}
+ public static long meterCount(KubernetesMetricsInterceptor interceptor,
String name) {
Review Comment:
nit: The move also changes the semantics. The original used `meter(name)`,
which creates a missing meter and returns 0, and this version fails on a
missing meter. That is fine for the current callers, whose meters are all
registered in the constructor, but it makes the generic `TestUtils` depend on
`KubernetesMetricsInterceptor`. Taking a `MetricRegistry` argument would avoid
that dependency.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptor.java:
##########
@@ -233,4 +261,33 @@ public Optional<Pair<String, String>>
parseNamespaceScopedResource(String path)
return Optional.empty();
}
}
+
+ /**
+ * Consumer of a response body which carries the start time of its request,
restarted at each
+ * response so that the values of a re-sent request add up to its latency.
+ */
+ private static final class TimedConsumer implements
AsyncBody.Consumer<List<ByteBuffer>> {
+ private final AsyncBody.Consumer<List<ByteBuffer>> delegate;
+ private final AtomicLong startTimeNanos = new
AtomicLong(System.nanoTime());
Review Comment:
nit: This allocates two objects per HTTP request, the `TimedConsumer` and an
`AtomicLong`. A `volatile long startTimeNanos` updated through a static
`AtomicLongFieldUpdater<TimedConsumer>` keeps the atomic `getAndSet` in
`restart()` and needs only one allocation.
##########
spark-operator/src/test/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptorTest.java:
##########
@@ -171,6 +186,104 @@ void testResponseCodeGroups() {
Assertions.assertEquals(1, meters.get("5xx").getCount());
}
+ @Test
+ @Order(5)
+ void testResponseLatency() {
+ KubernetesMetricsInterceptor metricsInterceptor = new
KubernetesMetricsInterceptor();
+ List<Interceptor> interceptors = List.of(metricsInterceptor);
+ try (KubernetesClient client =
+ KubernetesClientFactory.buildKubernetesClient(
+ interceptors, kubernetesClient.getConfiguration())) {
+ ConfigMap configMap = createConfigMap();
+ mockServer
+ .expect()
+ .get()
+ .delay(50)
+ .withPath(CONFIG_MAP_PATH)
+ .andReturn(200, configMap)
+ .once();
+ client.resource(configMap).get();
+
+ Histogram latency = latency(metricsInterceptor);
+ Assertions.assertEquals(1, latency.getCount());
+ Snapshot snapshot = latency.getSnapshot();
+ Assertions.assertTrue(snapshot.getMax() >= 50_000_000L, () -> "max: " +
snapshot.getMax());
+ Assertions.assertTrue(
+ snapshot.getMax() < 30_000_000_000L, () -> "max: " +
snapshot.getMax());
+ }
+ }
+
+ @Test
+ @Order(6)
+ void testResponseLatencyOfResentRequest() {
+ KubernetesMetricsInterceptor metricsInterceptor = new
KubernetesMetricsInterceptor();
+ List<Interceptor> interceptors = List.of(metricsInterceptor);
+ Config config =
+ new ConfigBuilder(kubernetesClient.getConfiguration())
+ .withOauthTokenProvider(() -> "token")
+ .build();
+ try (KubernetesClient client =
+ KubernetesClientFactory.buildKubernetesClient(interceptors, config)) {
+ ConfigMap configMap = createConfigMap();
+ mockServer
+ .expect()
+ .get()
+ .delay(200)
+ .withPath(CONFIG_MAP_PATH)
+ .andReturn(HTTP_UNAUTHORIZED, new
StatusBuilder().withCode(HTTP_UNAUTHORIZED).build())
+ .once();
+ mockServer.expect().get().withPath(CONFIG_MAP_PATH).andReturn(200,
configMap).once();
+ client.resource(configMap).get();
+
+ Assertions.assertEquals(2, meterCount(metricsInterceptor,
"http.response"));
+ Assertions.assertEquals(1, meterCount(metricsInterceptor,
"http.response.401"));
+ Assertions.assertEquals(0, meterCount(metricsInterceptor, "failed"));
+ Histogram latency = latency(metricsInterceptor);
+ Assertions.assertEquals(2, latency.getCount());
+ Snapshot snapshot = latency.getSnapshot();
+ Assertions.assertTrue(snapshot.getMax() >= 200_000_000L, () -> "max: " +
snapshot.getMax());
+ Assertions.assertTrue(snapshot.getMin() < 200_000_000L, () -> "min: " +
snapshot.getMin());
Review Comment:
This checks that one value is >= 200 ms and the other < 200 ms, but not the
claim in the docs that the two values add up to the time until the re-sent
response arrives. Measuring the wall-clock time around
`client.resource(configMap).get()` and asserting that the sum of the two
samples is at least 200 ms and at most that time would cover it. Also, the
`HttpLoggingInterceptor` compatibility is only covered with a lambda delegate
in `testConsumerUnwrapsDelegate`. An end-to-end case with TRACE logging enabled
would cover the real `DeferredLoggingConsumer` wrapping.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
