dongjoon-hyun commented on code in PR #918:
URL:
https://github.com/apache/spark-kubernetes-operator/pull/918#discussion_r4169285551
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptor.java:
##########
@@ -181,11 +205,12 @@ private void updateRequestMetrics(HttpRequest request) {
});
}
- private void updateResponseMetrics(HttpResponse response, long
startTimeNanos) {
+ private void updateResponseMetrics(HttpResponse response, Long
startTimeNanos) {
Objects.requireNonNull(response);
- final long latency = System.nanoTime() - startTimeNanos;
responseRateMeter.mark();
- responseLatency.update(latency);
+ if (startTimeNanos != null) {
+ responseLatency.update(System.nanoTime() - startTimeNanos);
Review Comment:
fabric8 registers `after(...)` on the response future before the
`afterFailure(...)` chain (fabric8 7.8.0 `StandardHttpClient.java` L118-119 vs
L121-124), and `CompletableFuture` runs the dependents of a future in LIFO
order. So `after(...)` of a non-2xx response runs only after the synchronous
`afterFailure(...)` work. For a `401`, `TokenRefreshInterceptor#afterFailure`
calls `config.refresh()` synchronously, which re-runs a kubeconfig exec
credential plugin (`ProcessBuilder` and `waitFor()` in `KubeConfigUtils`).
With an exec plugin which takes 1 s, a `401` sent by the server ~55 ms after
the request was recorded as ~1.1 s in my test. In-cluster, the refresh only
re-reads the service account token file, so the difference is small (50 ms ->
57 ms). Since this comes from the callback order of fabric8, could you at least
mention it in the docs, e.g. that the latency of a `401` can include the token
refresh?
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptor.java:
##########
@@ -88,6 +88,24 @@ public KubernetesMetricsInterceptor() {
}
}
+ /**
+ * Called before a request is sent, to allow wrapping the consumer of the
response body.
+ *
+ * <p>Wraps the consumer to carry the request start time. fabric8 passes the
returned consumer to
+ * {@link #after(HttpRequest, HttpResponse, AsyncBody.Consumer)}, also for a
request re-sent after
+ * {@link #afterFailure(BasicBuilder, HttpResponse, RequestTags)}, which
computes the response
+ * latency from it. A WebSocket upgrade has no consumer, so no latency is
recorded for it.
+ *
+ * @param consumer the consumer of the response body
+ * @param request the request about to be sent
+ * @return the consumer carrying the request start time
+ */
+ @Override
+ public AsyncBody.Consumer<List<ByteBuffer>> consumer(
+ AsyncBody.Consumer<List<ByteBuffer>> consumer, HttpRequest request) {
+ return new TimedConsumer(consumer, System.nanoTime());
Review Comment:
The start time is fixed once per `consumer(...)` call. But when an
interceptor's `afterFailure(...)` returns `true`,
`StandardHttpClient#consumeBytesOnce` re-sends the request with the same
`effectiveConsumer` and does not call `consumer(...)` again (fabric8 7.8.0
`StandardHttpClient.java` L128-131). So the sample of the re-sent response is
measured from the start of the **first** attempt: the first round trip, plus
the `afterFailure(...)` work, plus the re-send round trip. The first exchange
is counted twice.
For example, with the default `TokenRefreshInterceptor` and an
auto-configured service account token, a `401` after ~150 ms and a re-sent
`200` after ~54 ms were recorded as 158 ms and 212 ms. A `5xx` or `429` retry,
on the other hand, goes through `AsyncUtils` -> `consumeBytesOnce` and gets a
fresh `TimedConsumer`; `500` -> `200` was recorded as 56 ms. So the two re-send
paths record different things. This does not match `records the response
latency of each HTTP request` in the migration guide.
`testResponseLatencyOfResentRequest` checks only the counts, which 1.0 also
satisfied.
How about restarting the clock per response, e.g. keeping the start time in
an `AtomicLong` and recording `now - start.getAndSet(now)` in `after(...)`?
fabric8 runs `after(...)` of the failed response after the `afterFailure(...)`
stages, so the sample of the re-sent response becomes close to the latency of
the re-sent request itself. Alternatively, record only the first response per
consumer and document that re-sent responses are excluded, like WebSocket
upgrades. Either way, could you also check the values in the test, e.g. delay
the `403` and assert that `getSnapshot().getMin()` is below the delay?
##########
docs/migration_guide.md:
##########
@@ -81,6 +81,10 @@ affect building the project, CI, tests or examples.
documented. 1.0 capped it at about 15 seconds, so with the other defaults,
the retries now span
about 49 minutes instead of about 3 minutes. To restore the behavior before
1.1.0, set it to
`15` ([SPARK-59892](https://issues.apache.org/jira/browse/SPARK-59892)).
+- Since 1.1.0, `kubernetes.client.http.response.latency.nanos` records the
response latency of
Review Comment:
This PR conflicts with `main` here, because SPARK-59935 (94bcfa3) added an
item right after the SPARK-59892 item. Could you rebase and keep both items?
FYI, SPARK-59935 also matters for this fix. At the base of this PR and in
`branch-1.0`, `PrometheusPullModelHandler` exports the `_sum` of this histogram
as `snap.getMean() * count`. Once this PR records real latencies, it can
decrease between scrapes. In my test, 100 responses of 2 s, followed a minute
later by 100 responses of 10 ms, decreased the exported `_sum` from 200 to 117
while the true sum grew to 201. So if we backport this fix to `branch-1.0`,
SPARK-59935 is needed too.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptor.java:
##########
@@ -100,21 +118,26 @@ 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:
This is true for an HTTP request, but not for a WebSocket upgrade in fabric8
7.8.0 (`StandardHttpClient#buildWebSocketOnce`):
- An upgrade re-sent after `afterFailure(...)` returns `true` is not passed
to `after(...)` at all (L321-322), e.g. a `401` upgrade re-sent by
`TokenRefreshInterceptor` which gets `101`.
- A rejected upgrade is passed to `after(...)` once (L314-315), and again
for each interceptor whose `afterFailure(...)` returns `false` (L324-326). With
the default interceptors of the operator, a watch rejected with `403` marks
`http.response`, `http.response.403` and `4xx` five times.
The over-count is not from this PR. But how about limiting the re-send
sentence to HTTP requests, e.g. `..., and again for an HTTP request re-sent
after {@link #afterFailure(...)}`? Optionally, `docs/configuration.md` could
mention that a rejected upgrade is counted several times in
`kubernetes.client.http.response`.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptor.java:
##########
@@ -88,6 +88,24 @@ public KubernetesMetricsInterceptor() {
}
}
+ /**
+ * Called before a request is sent, to allow wrapping the consumer of the
response body.
+ *
+ * <p>Wraps the consumer to carry the request start time. fabric8 passes the
returned consumer to
Review Comment:
nit: fabric8 passes to `after(...)` the outermost consumer after all
interceptors have wrapped it (`StandardHttpClient.java` L112-119), not
necessarily the one returned here. That is why `after(...)` needs
`unwrap(...)`. It finds the start time only if every interceptor registered
after this one delegates `unwrap(...)`, and `AsyncBody.Consumer#unwrap` does
not delegate by default. This holds now, because this interceptor is the last
one which wraps the consumer. But if a subclass overrode
`SparkOperator#getClientInterceptors` and added a wrapping interceptor after
this one, e.g. one returning a lambda, the histogram would silently stay empty.
How about stating this requirement here once? The WebSocket and re-send
details are also repeated at L95-97, L126-128 and L132, so this Javadoc could
be shorter.
##########
spark-operator/src/test/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptorTest.java:
##########
@@ -122,6 +131,81 @@ void testWhenKubernetesServerNotWorking() {
}
}
+ @Test
+ @Order(3)
+ 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();
+
+ Map<String, Metric> map =
metricsInterceptor.metricRegistry().getMetrics();
+ Histogram latency = (Histogram) map.get("http.response.latency.nanos");
+ Assertions.assertEquals(1, latency.getCount());
+ Assertions.assertTrue(latency.getSnapshot().getMax() >= 50_000_000L);
+ Assertions.assertTrue(latency.getSnapshot().getMax() < 30_000_000_000L);
+ }
+ }
+
+ @Test
+ @Order(4)
+ void testResponseLatencyOfResentRequest() {
+ KubernetesMetricsInterceptor metricsInterceptor = new
KubernetesMetricsInterceptor();
+ Interceptor resendInterceptor =
+ new Interceptor() {
+ @Override
+ public CompletableFuture<Boolean> afterFailure(
+ BasicBuilder builder, HttpResponse<?> response, RequestTags
tags) {
+ return CompletableFuture.completedFuture(response.code() == 403);
+ }
+ };
+ List<Interceptor> interceptors = List.of(metricsInterceptor,
resendInterceptor);
+ try (KubernetesClient client =
+ KubernetesClientFactory.buildKubernetesClient(
+ interceptors, kubernetesClient.getConfiguration())) {
+ ConfigMap configMap = createConfigMap();
+ mockServer.expect().get().withPath(CONFIG_MAP_PATH).andReturn(403,
null).once();
+ mockServer.expect().get().withPath(CONFIG_MAP_PATH).andReturn(200,
configMap).once();
+ client.resource(configMap).get();
+
+ Map<String, Metric> map =
metricsInterceptor.metricRegistry().getMetrics();
+ await()
+ .untilAsserted(
+ () -> Assertions.assertEquals(2, ((Meter)
map.get("http.response")).getCount()));
+ Assertions.assertEquals(2, ((Histogram)
map.get("http.response.latency.nanos")).getCount());
+ }
+ }
+
+ @Test
+ @Order(5)
+ 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()) {
+ Map<String, Metric> map =
metricsInterceptor.metricRegistry().getMetrics();
+ await()
+ .untilAsserted(
+ () -> Assertions.assertEquals(2, ((Meter)
map.get("http.response")).getCount()));
+ Assertions.assertTrue(informer.isWatching());
+ Assertions.assertEquals(1, ((Meter)
map.get("http.response.101")).getCount());
Review Comment:
This can be flaky. `inform()` returns before `after(...)` of the `101` runs,
because the informer start future is completed inside the upgrade's
`CompletableFuture` chain, which runs before the `after(...)` stage registered
earlier (LIFO). `updateResponseMetrics(...)` marks `http.response` before it
lazily registers and marks `http.response.101`, and `getMetrics()` is a live
view. So if the poll sees `http.response == 2` in between,
`map.get("http.response.101")` returns `null` and this line throws an NPE, or
the count is `0`. Only the default 100 ms poll delay of Awaitility hides it
now. When I polled without the delay, 14 of 40 runs failed.
Could you assert all of them inside `untilAsserted(...)`? `assertNotNull`
keeps a missing meter an `AssertionError`, which Awaitility retries.
```java
await()
.untilAsserted(
() -> {
Assertions.assertEquals(2, ((Meter)
map.get("http.response")).getCount());
Assertions.assertNotNull(map.get("http.response.101"));
Assertions.assertEquals(1, ((Meter)
map.get("http.response.101")).getCount());
Assertions.assertEquals(
1, ((Histogram)
map.get("http.response.latency.nanos")).getCount());
});
Assertions.assertTrue(informer.isWatching());
```
##########
docs/configuration.md:
##########
@@ -234,6 +234,9 @@ 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` does not record a WebSocket
upgrade response, e.g.
Review Comment:
fabric8 calls `after(...)` when the response headers arrive, and the body is
read afterwards: `VertxHttpRequest` pauses the response, and
`StandardHttpClient.java` L117-119 calls `after(...)`. So this metric does not
include the time to read the response body. For example, a LIST of 2000 items
streamed over ~1.1 s was recorded as ~4 ms in my test. How about adding a
sentence, e.g. `It measures the time until the response headers arrive, so it
does not include the time to read the response body.`? Could you also align the
migration guide item with it?
##########
spark-operator/src/test/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptorTest.java:
##########
@@ -122,6 +131,81 @@ void testWhenKubernetesServerNotWorking() {
}
}
+ @Test
+ @Order(3)
+ 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();
+
+ Map<String, Metric> map =
metricsInterceptor.metricRegistry().getMetrics();
+ Histogram latency = (Histogram) map.get("http.response.latency.nanos");
+ Assertions.assertEquals(1, latency.getCount());
+ Assertions.assertTrue(latency.getSnapshot().getMax() >= 50_000_000L);
+ Assertions.assertTrue(latency.getSnapshot().getMax() < 30_000_000_000L);
+ }
+ }
+
+ @Test
+ @Order(4)
+ void testResponseLatencyOfResentRequest() {
+ KubernetesMetricsInterceptor metricsInterceptor = new
KubernetesMetricsInterceptor();
+ Interceptor resendInterceptor =
+ new Interceptor() {
+ @Override
+ public CompletableFuture<Boolean> afterFailure(
+ BasicBuilder builder, HttpResponse<?> response, RequestTags
tags) {
+ return CompletableFuture.completedFuture(response.code() == 403);
+ }
+ };
+ List<Interceptor> interceptors = List.of(metricsInterceptor,
resendInterceptor);
+ try (KubernetesClient client =
+ KubernetesClientFactory.buildKubernetesClient(
+ interceptors, kubernetesClient.getConfiguration())) {
+ ConfigMap configMap = createConfigMap();
+ mockServer.expect().get().withPath(CONFIG_MAP_PATH).andReturn(403,
null).once();
+ mockServer.expect().get().withPath(CONFIG_MAP_PATH).andReturn(200,
configMap).once();
+ client.resource(configMap).get();
+
+ Map<String, Metric> map =
metricsInterceptor.metricRegistry().getMetrics();
+ await()
Review Comment:
nit: This `await()` is not needed. fabric8 calls `after(...)` for both
responses on the event loop before the body of the `200` is delivered, so
`get()` returns after both calls, as `testResponseLatency` assumes without
`await()`. By default, Awaitility sleeps the poll interval (100 ms) before its
first check, so this adds ~100 ms. Could you assert directly here? The
`await()` in `testResponseLatencyOfWebSocketUpgrade` is needed, though.
##########
spark-operator/src/test/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptorTest.java:
##########
@@ -122,6 +131,81 @@ void testWhenKubernetesServerNotWorking() {
}
}
+ @Test
+ @Order(3)
+ 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();
+
+ Map<String, Metric> map =
metricsInterceptor.metricRegistry().getMetrics();
+ Histogram latency = (Histogram) map.get("http.response.latency.nanos");
Review Comment:
nit: The new tests cast `map.get(...)` six times, so a missing metric fails
with an NPE which does not name the metric. A small helper could give a clearer
failure, e.g.
```java
private static long count(Map<String, Metric> metrics, String name) {
Metric metric = metrics.get(name);
Assertions.assertNotNull(metric, name);
return metric instanceof Histogram h ? h.getCount() : ((Meter)
metric).getCount();
}
```
It would also make a missing meter an `AssertionError` inside
`untilAsserted(...)`. Keeping the literal names is good, since they also check
the documented names.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptor.java:
##########
@@ -227,4 +252,19 @@ public Optional<Pair<String, String>>
parseNamespaceScopedResource(String path)
return Optional.empty();
}
}
+
+ /** Consumer of a response body which carries the start time of its request.
*/
+ private record TimedConsumer(AsyncBody.Consumer<List<ByteBuffer>> delegate,
long startTimeNanos)
+ implements AsyncBody.Consumer<List<ByteBuffer>> {
+ @Override
+ public void consume(List<ByteBuffer> value, AsyncBody asyncBody) throws
Exception {
+ delegate.consume(value, asyncBody);
+ }
+
+ @Override
+ public <U> U unwrap(Class<U> target) {
+ U self = AsyncBody.Consumer.super.unwrap(target);
+ return self != null ? self : delegate.unwrap(target);
Review Comment:
This delegation keeps `HttpLoggingInterceptor` working at the `TRACE` level.
fabric8 installs it first, so its `DeferredLoggingConsumer` is wrapped inside
this one. But no test covers this branch. The test `log4j2.properties` filters
events below `warn`, so `HttpLoggingInterceptor` is not installed and
`TimedConsumer` is always the outermost consumer. So
`unwrap(TimedConsumer.class)` always returns `self`. If this line were dropped,
all the tests would still pass, while the `TRACE` logging of response bodies
would silently break.
Could you add a small unit test? e.g.
```java
class InnerConsumer implements AsyncBody.Consumer<List<ByteBuffer>> {
@Override
public void consume(List<ByteBuffer> value, AsyncBody asyncBody) {}
}
InnerConsumer inner = new InnerConsumer();
Assertions.assertSame(
inner, new KubernetesMetricsInterceptor().consumer(inner,
null).unwrap(InnerConsumer.class));
```
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptor.java:
##########
@@ -181,11 +205,12 @@ private void updateRequestMetrics(HttpRequest request) {
});
}
- private void updateResponseMetrics(HttpResponse response, long
startTimeNanos) {
+ private void updateResponseMetrics(HttpResponse response, Long
startTimeNanos) {
Review Comment:
nit (optional): `after(...)` is the only caller. How about updating the
histogram in `after(...)` and keeping the original signature here? It avoids
the nullable `Long`, which is boxed for every response, and the three `null`
checks across two methods. It also updates the histogram before `http.response`
is marked, which helps the tests that await `http.response`.
```java
TimedConsumer timedConsumer = consumer == null ? null :
consumer.unwrap(TimedConsumer.class);
if (timedConsumer != null) {
responseLatency.update(System.nanoTime() - timedConsumer.startTimeNanos());
}
updateResponseMetrics(response);
```
I know this form came from my previous suggestion, so either is fine.
##########
docs/migration_guide.md:
##########
@@ -81,6 +81,10 @@ affect building the project, CI, tests or examples.
documented. 1.0 capped it at about 15 seconds, so with the other defaults,
the retries now span
about 49 minutes instead of about 3 minutes. To restore the behavior before
1.1.0, set it to
`15` ([SPARK-59892](https://issues.apache.org/jira/browse/SPARK-59892)).
+- Since 1.1.0, `kubernetes.client.http.response.latency.nanos` records the
response latency of
+ each HTTP request. 1.0 recorded a few microseconds regardless of the
latency. It no longer
Review Comment:
nit: 1.0 subtracted two back-to-back `System.nanoTime()` calls with only
`Objects.requireNonNull(response)` in between. So it recorded less than a
microsecond, not a few microseconds; the median was 125-292 ns in my test. How
about `1.0 recorded nearly zero regardless of the latency.`? The PR description
has the same wording.
##########
spark-operator/src/test/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptorTest.java:
##########
@@ -122,6 +131,81 @@ void testWhenKubernetesServerNotWorking() {
}
}
+ @Test
+ @Order(3)
+ 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();
+
+ Map<String, Metric> map =
metricsInterceptor.metricRegistry().getMetrics();
+ Histogram latency = (Histogram) map.get("http.response.latency.nanos");
+ Assertions.assertEquals(1, latency.getCount());
+ Assertions.assertTrue(latency.getSnapshot().getMax() >= 50_000_000L);
+ Assertions.assertTrue(latency.getSnapshot().getMax() < 30_000_000_000L);
+ }
+ }
+
+ @Test
+ @Order(4)
+ void testResponseLatencyOfResentRequest() {
+ KubernetesMetricsInterceptor metricsInterceptor = new
KubernetesMetricsInterceptor();
+ Interceptor resendInterceptor =
Review Comment:
nit: The operator's client never re-sends a request on `403`. fabric8's
default `TokenRefreshInterceptor` re-sends on `401` through the same path. How
about testing that instead of an anonymous interceptor? The mock server config
has a static `oauthToken`, so a token provider is needed:
```java
Config config =
new ConfigBuilder(kubernetesClient.getConfiguration())
.withOauthTokenProvider(() -> "token")
.build();
```
Then return `HTTP_UNAUTHORIZED` first and `200` second. This also keeps the
production order of the interceptors. There `TokenRefreshInterceptor` comes
before this interceptor, so the `401` is not counted in `failed`. With the
anonymous interceptor registered after this one, it is.
--
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]
