dongjoon-hyun commented on code in PR #918:
URL:
https://github.com/apache/spark-kubernetes-operator/pull/918#discussion_r4437836316
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptor.java:
##########
@@ -103,21 +125,38 @@ 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} or {@code
+ * BackwardsCompatibilityInterceptor} for a legacy API version. The two
values of a request
Review Comment:
Sorry, my comment in the previous round was wrong. fabric8 installs
`BackwardsCompatibilityInterceptor` only when
`kubernetes.backwardsCompatibilityInterceptor.disable` is set to `false`,
because its default is `"true"` (fabric8 7.8.0 `HttpClientUtils.java` L75 and
L117-122, the same in 7.9.0 and 8.0.0). So with the default interceptors,
`TokenRefreshInterceptor` is the only one which re-sends a request. Could you
drop `or BackwardsCompatibilityInterceptor for a legacy API version` here, and
the same example from `What changes were proposed in this pull request?` of the
PR description, which becomes the commit message?
FYI, when it is enabled, a request can be re-sent twice, e.g. `401` -> `404`
-> `200`, and is then recorded three times. The values still add up, so
`recorded twice` in the docs is fine for the default interceptors.
nit: For an HTTP request, fabric8 calls the
`afterFailure(HttpRequest.Builder, HttpResponse, RequestTags)` overload
(`StandardHttpClient.java` L124), which delegates to the `BasicBuilder` one by
default. So linking that overload would be slightly more precise. The current
link came from my suggestion, so either is fine.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptor.java:
##########
@@ -103,21 +125,38 @@ 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} or {@code
+ * BackwardsCompatibilityInterceptor} for a legacy API version. 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());
+ Objects.requireNonNull(response);
+ TimedConsumer timedConsumer = consumer == null ? null :
consumer.unwrap(TimedConsumer.class);
Review Comment:
Now that the `owner` lookup is gone, as I suggested earlier, the corner case
of my earlier comment is back. `unwrap(TimedConsumer.class)` returns the
outermost `TimedConsumer` for every `KubernetesMetricsInterceptor` on a client,
and `restart()` resets its start time. So a second instance, e.g. of a subclass
added through `SparkOperator#getClientInterceptors`, records nearly zero for
every response, without a warning. In my test with two instances on one client,
the first recorded ~205 ms and the second 47-58 µs for the same response.
Since this can't happen with the operator's own client, mentioning it in the
Javadoc would be enough, e.g. in `consumer(...)` (L100-104):
> Only one instance per client is supported, because every instance restarts
the outermost {@code TimedConsumer}.
##########
docs/configuration.md:
##########
@@ -264,6 +264,13 @@ 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 measured from when a
request is handed to the
+HTTP client, so it also includes waiting for a connection. It does not include
the time to read
+the response body. A request re-sent by another interceptor, e.g. after a
`401`, is recorded
+twice, and the two values add up to the time from sending it until the re-sent
response arrives.
Review Comment:
This holds for a `401` which is re-sent. But when the token refresh fails or
gets no new token, e.g. with an `OAuthTokenProvider` or an exec credential
plugin, the `401` is not re-sent and has only one value. Since `after(...)`
runs after the synchronous `afterFailure(...)` work, that value includes the
whole refresh. In my test with an `OAuthTokenProvider` which blocked for 500 ms
and returned `null`, a `401` which the server returned after ~18 ms was
recorded as ~529 ms.
How about adding a sentence, e.g. `A 401 which is not re-sent, e.g. when the
token refresh fails, can include the time of the refresh.`? This does not
happen in-cluster, where the service account token file always gives a token,
so it is minor.
##########
docs/migration_guide.md:
##########
@@ -144,6 +144,12 @@ affect building the project, CI, tests or examples.
operator also redacts the configuration properties whose keys or values
contain `credential` in
its logs. To restore the behavior before 1.1.0, set it to
`(?i)secret|password|token|access[.]?key`.
+- Since 1.1.0, `kubernetes.client.http.response.latency.nanos` records the
latency of HTTP
+ responses, not including the time to read the response body. 1.0 recorded
nearly zero
+ regardless of the latency. A request re-sent by another interceptor, e.g.
after a `401`, is
+ recorded twice, and the two values add up to the time from sending it until
the re-sent response
+ arrives, e.g. including the token refresh. It no longer records a WebSocket
upgrade response,
+ e.g. of a watch, so its count can be lower than that of
`kubernetes.client.http.response`.
Review Comment:
nit (optional): The count can also stay at zero for an embedder. 1.0
recorded a value at every `after(...)`, while this records one only when
`unwrap(...)` finds the `TimedConsumer`. So if a subclass of `SparkOperator`
adds an interceptor through `getClientInterceptors(...)` which wraps the
consumer after this one without delegating `unwrap(...)`, e.g. with a lambda,
the histogram stays empty after one warning. The Javadoc and the warning
explain it, so this is minor, but a line under `### Java API` could mention it.
##########
spark-operator/src/test/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptorTest.java:
##########
@@ -171,6 +194,145 @@ 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();
+ long startTime = System.nanoTime();
+ client.resource(configMap).get();
+ long elapsed = System.nanoTime() - startTime;
+
+ Histogram latency = histogram(metricsInterceptor.metricRegistry(),
LATENCY);
+ Assertions.assertEquals(1, latency.getCount());
+ Snapshot snapshot = latency.getSnapshot();
+ Assertions.assertTrue(snapshot.getMax() >= 50_000_000L, () -> "max: " +
snapshot.getMax());
+ Assertions.assertTrue(
+ snapshot.getMax() <= elapsed,
+ () -> "max: " + snapshot.getMax() + ", elapsed: " + elapsed);
+ }
+ }
+
+ @Test
+ @Order(6)
+ void testResponseLatencyOfResentRequest() {
+ KubernetesMetricsInterceptor metricsInterceptor = new
KubernetesMetricsInterceptor();
+ MetricRegistry metrics = metricsInterceptor.metricRegistry();
+ 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();
+ long startTime = System.nanoTime();
+ client.resource(configMap).get();
+ long elapsed = System.nanoTime() - startTime;
+
+ Assertions.assertEquals(2, meterCount(metrics, "http.response"));
+ Assertions.assertEquals(1, meterCount(metrics, "http.response.401"));
+ Assertions.assertEquals(0, meterCount(metrics, "failed"));
+ Histogram latency = histogram(metrics, LATENCY);
+ 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());
+ long sum = Assertions.assertInstanceOf(SummingHistogram.class,
latency).getSum();
+ Assertions.assertTrue(sum >= 200_000_000L, () -> "sum: " + sum);
+ Assertions.assertTrue(sum <= elapsed, () -> "sum: " + sum + ", elapsed:
" + elapsed);
+ }
+ }
+
+ @Test
+ @Order(7)
+ void testResponseLatencyOfWebSocketUpgrade() {
+ KubernetesMetricsInterceptor metricsInterceptor = new
KubernetesMetricsInterceptor();
+ MetricRegistry metrics = metricsInterceptor.metricRegistry();
+ 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(metrics,
"http.response"));
+ Assertions.assertEquals(1, meterCount(metrics,
"http.response.101"));
+ Assertions.assertEquals(1, histogram(metrics,
LATENCY).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()));
+ }
+
+ @Test
+ @Order(9)
+ void testResponseLatencyWithHttpLogging() {
+ Logger logger = mock(Logger.class);
+ when(logger.isTraceEnabled()).thenReturn(true);
+ KubernetesMetricsInterceptor metricsInterceptor = new
KubernetesMetricsInterceptor();
+ // fabric8 installs HttpLoggingInterceptor first at the TRACE level, so it
is wrapped inside.
+ List<Interceptor> interceptors =
+ List.of(new HttpLoggingInterceptor(logger), metricsInterceptor);
+ try (KubernetesClient client =
+ KubernetesClientFactory.buildKubernetesClient(
+ interceptors, kubernetesClient.getConfiguration())) {
+ ConfigMap configMap = createConfigMap();
+ mockServer.expect().get().withPath(CONFIG_MAP_PATH).andReturn(200,
configMap).once();
+ client.resource(configMap).get();
+
+ Assertions.assertEquals(
+ 1, histogram(metricsInterceptor.metricRegistry(),
LATENCY).getCount());
+ // The response body is logged only if HttpLoggingInterceptor finds its
own consumer.
+ String body = "\"kind\":\"ConfigMap\"";
+ verify(logger, timeout(10_000)).trace(argThat((String m) -> m != null &&
m.contains(body)));
+ }
+ }
+
+ @Test
+ @Order(10)
+ void testResponseLatencyWithoutTimedConsumer() {
+ KubernetesMetricsInterceptor metricsInterceptor = new
KubernetesMetricsInterceptor();
+ MetricRegistry metrics = metricsInterceptor.metricRegistry();
+ AsyncBody.Consumer<List<ByteBuffer>> consumer = (value, asyncBody) -> {};
+ HttpResponse<?> response = mock(HttpResponse.class);
+ when(response.code()).thenReturn(200);
+ metricsInterceptor.after(null, response, consumer);
+ metricsInterceptor.after(null, response, consumer);
+ Assertions.assertEquals(2, meterCount(metrics, "http.response"));
+ Assertions.assertEquals(0, histogram(metrics, LATENCY).getCount());
Review Comment:
nit: The Javadoc of `consumer(...)` and the PR description say that a
warning is logged once, but this test checks only the meters. So it would still
pass if the warning were never logged, or logged for every response on the
event loop. How about also asserting the log messages, e.g. with a
`TestLogAppender` like `ConfigOptionTest#testLogEachInvalidValueOnce`? The two
`after(...)` calls here already cover the "once" part.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptor.java:
##########
@@ -233,4 +269,38 @@ 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 static final AtomicLongFieldUpdater<TimedConsumer>
START_TIME_NANOS =
+ AtomicLongFieldUpdater.newUpdater(TimedConsumer.class,
"startTimeNanos");
+
+ private final AsyncBody.Consumer<List<ByteBuffer>> delegate;
+ // Updated through START_TIME_NANOS, so that each request needs only one
allocation.
+ @SuppressWarnings({"PMD.AvoidUsingVolatile", "PMD.UnusedPrivateField"})
+ private volatile long startTimeNanos = System.nanoTime();
Review Comment:
nit (optional): I suggested this updater in the previous round, but on
second thought, it isn't worth it. It saves one small `AtomicLong` per request,
which is negligible next to the other allocations of an HTTP exchange, while it
needs a `volatile` field, two PMD suppressions, and the field name in a string,
which fails only at runtime if the field is renamed. A plain field does the
same:
```java
private final AtomicLong startTimeNanos = new AtomicLong(System.nanoTime());
long restart() {
long now = System.nanoTime();
return now - startTimeNanos.getAndSet(now);
}
```
Either is fine with me. Sorry for the churn.
##########
spark-operator/src/test/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptorTest.java:
##########
@@ -171,6 +194,145 @@ 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();
+ long startTime = System.nanoTime();
+ client.resource(configMap).get();
+ long elapsed = System.nanoTime() - startTime;
+
+ Histogram latency = histogram(metricsInterceptor.metricRegistry(),
LATENCY);
+ Assertions.assertEquals(1, latency.getCount());
+ Snapshot snapshot = latency.getSnapshot();
+ Assertions.assertTrue(snapshot.getMax() >= 50_000_000L, () -> "max: " +
snapshot.getMax());
+ Assertions.assertTrue(
+ snapshot.getMax() <= elapsed,
+ () -> "max: " + snapshot.getMax() + ", elapsed: " + elapsed);
+ }
+ }
+
+ @Test
+ @Order(6)
+ void testResponseLatencyOfResentRequest() {
+ KubernetesMetricsInterceptor metricsInterceptor = new
KubernetesMetricsInterceptor();
+ MetricRegistry metrics = metricsInterceptor.metricRegistry();
+ 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();
+ long startTime = System.nanoTime();
+ client.resource(configMap).get();
+ long elapsed = System.nanoTime() - startTime;
+
+ Assertions.assertEquals(2, meterCount(metrics, "http.response"));
+ Assertions.assertEquals(1, meterCount(metrics, "http.response.401"));
+ Assertions.assertEquals(0, meterCount(metrics, "failed"));
+ Histogram latency = histogram(metrics, LATENCY);
+ 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:
nit: Now that L262 bounds the sum by the wall-clock time, this check, which
I suggested in an earlier round, no longer adds coverage. If `restart()` did
not reset the start time, the second value would include the first one, and the
sum would exceed `elapsed`. But this check still assumes that the re-sent
request completes within 200 ms, so a long GC pause on a busy CI runner could
fail it. How about dropping it? `How was this patch tested?` in the PR
description (`one of at least 200 ms and one below 200 ms`) would need the same
update.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptor.java:
##########
@@ -103,21 +125,38 @@ 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
Review Comment:
FYI, no change is needed in this PR, since this is in your list of
follow-ups (the upstream fabric8 issue). A rejected upgrade still reaches
`after(...)` 1+N times (`StandardHttpClient.java` L314-315 and L324-326), e.g.
five times for a `403` with the default interceptors, which marks
`http.response`, its code meter and `4xx` each time. In addition, fabric8
retries an upgrade rejected with a `429` or a `5xx` with backoff
(`buildWebSocket` -> `retryWithExponentialBackoff`, L285-289), so this repeats
for each attempt. The latency histogram is no longer affected, since the
consumer is `null` there.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptor.java:
##########
@@ -103,21 +125,38 @@ 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} or {@code
+ * BackwardsCompatibilityInterceptor} for a legacy API version. 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());
+ Objects.requireNonNull(response);
+ TimedConsumer timedConsumer = consumer == null ? null :
consumer.unwrap(TimedConsumer.class);
+ if (timedConsumer != null) {
+ responseLatency.update(timedConsumer.restart());
+ } else if (consumer != null &&
missingTimedConsumerLogged.compareAndSet(false, true)) {
+ log.warn(
+ "No response latency is recorded, because an interceptor added after
{} wraps the "
+ + "response body consumer without delegating
AsyncBody.Consumer#unwrap(...).",
+ KubernetesMetricsInterceptor.class.getSimpleName());
+ }
+ updateResponseMetrics(response);
Review Comment:
nit (optional): `consumer.unwrap(...)` at L150 runs the `unwrap(...)` of the
consumers which wrap `TimedConsumer`, i.e. code of other interceptors, before
this line. If one of them throws, e.g. a wrapper whose `unwrap(...)` returns
`(U) this` for any class, which fails the cast at L150, fabric8 drops the
exception of its `thenAccept(...)` stage (`StandardHttpClient.java` L118-119).
Then `http.response` and the code meters miss the response without a log, and
the interceptors after this one don't get `after(...)` either. 1.0 marked the
meters unconditionally.
This can't happen with the default interceptors, where `TimedConsumer` is
the outermost consumer, so feel free to leave it. Otherwise, calling
`updateResponseMetrics(response)` right after
`Objects.requireNonNull(response)` would keep the meters independent of other
interceptors, and the tests don't depend on the order. I know the current order
came from my earlier suggestion.
--
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]
