dongjoon-hyun commented on code in PR #918:
URL:
https://github.com/apache/spark-kubernetes-operator/pull/918#discussion_r4203189076
##########
docs/configuration.md:
##########
@@ -265,6 +265,12 @@ 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` ends when the response headers
arrive, so it does
+not include reading the body. A request re-sent after a `401` is recorded
twice, and the two
+values add up to the time from sending it until the re-sent response arrives,
including the token
+refresh. WebSocket upgrades, e.g. of a watch, are not recorded, and a rejected
one can be counted
Review Comment:
nit: `can be counted more than once` understates this.
`StandardHttpClient#buildWebSocketOnce` calls `after(...)` of all interceptors
once (L314-315), and again for each interceptor whose `afterFailure(...)`
returns `false` (L324-326). So with the default interceptors of the operator
(`HEADER`, `IMPERSONATOR`, `TOKEN` and this one), a watch rejected with `403`,
e.g. without the RBAC `watch` permission, always marks `http.response`,
`http.response.403` and `4xx` five times (six at the `TRACE` level), while
`http.request` and `failed` are marked once. An upgrade re-sent after a `401`
marks the `401` three times, and its `101` is not counted. How about `..., and
a rejected one is counted several times in kubernetes.client.http.response and
the response code meters.`?
##########
docs/configuration.md:
##########
@@ -265,6 +265,12 @@ 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.
Review Comment:
FYI, this is not from this PR. The table above names this meter
`kubernetes.client.http.response.failed` (L251) and the group meters
`kubernetes.client.http.response.1xx` to `5xx` (L255-259). But they are
registered at the root of the source, so their names are
`kubernetes.client.failed` and `kubernetes.client.1xx` to `5xx`, as this
paragraph and the tests use. L251 also says that it tracks the requests which
have no response, while `afterFailure(...)` marks it for every non-2xx response
which reaches this interceptor, e.g. a `404` of a `GET`. This can be fixed
separately.
##########
docs/migration_guide.md:
##########
@@ -132,6 +132,12 @@ affect building the project, CI, tests or examples.
services outside the cluster. To restore the behavior before 1.1.0, set
`spark.kubernetes.operator.events.enabled` to `false`. See
[Kubernetes Events](configuration.md#kubernetes-events) for details.
+- Since 1.1.0, `kubernetes.client.http.response.latency.nanos` records the
latency of HTTP
Review Comment:
FYI, this is not from this PR. With
`spark.kubernetes.operator.metrics.sanitizePrometheusMetricsNameEnabled=false`,
`PrometheusPullModelHandler` keeps the name
`kubernetes.client.http.response.latency.nanos`, because it replaces only
`_nanos` with `_seconds`, but still converts the values to seconds, e.g.
`{quantile="0.5"} 0.05` for 50 ms. It was invisible while the values were
nearly zero. Since Prometheus cannot parse the dotted names of that mode
anyway, this is minor.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptor.java:
##########
@@ -102,21 +122,35 @@ 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 {@link
#afterFailure(BasicBuilder, HttpResponse,
Review Comment:
nit: `{@link #afterFailure(...)}` links to this class's `afterFailure(...)`,
which always returns `false` (L170). The re-send comes from another
interceptor, e.g. fabric8's `TokenRefreshInterceptor` on a `401`. Since it
comes before this one, this class's `afterFailure(...)` is not even called
then. How about `{@link Interceptor#afterFailure(BasicBuilder, HttpResponse,
RequestTags)}`, or `..., and again for an HTTP request re-sent after another
interceptor's afterFailure(...) returns true, e.g. TokenRefreshInterceptor on a
401`? I know this wording came from my previous suggestion.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptor.java:
##########
@@ -183,11 +218,9 @@ private void updateRequestMetrics(HttpRequest request) {
});
}
- private void updateResponseMetrics(HttpResponse response, long
startTimeNanos) {
+ private void updateResponseMetrics(HttpResponse response) {
Objects.requireNonNull(response);
- final long latency = System.nanoTime() - startTimeNanos;
responseRateMeter.mark();
- responseLatency.update(latency);
getMeterByResponseCode(response.code()).mark();
if
(KUBERNETES_CLIENT_METRICS_GROUP_BY_RESPONSE_CODE_GROUP_ENABLED.getValue()) {
responseCodeGroupMeters.get(response.code() / 100 - 1).mark();
Review Comment:
FYI, this is not from this PR either. For a status code outside 100-599,
e.g. `HTTP/1.1 999` from a proxy, which Netty accepts, this line throws an
`IndexOutOfBoundsException` on the event loop. fabric8 drops the future of the
`after(...)` stage (`StandardHttpClient.java` L118-119) without a log, so the
interceptors after this one, e.g. a second metrics interceptor, miss the
response. The base revision behaves the same, so this can be fixed separately
with a bounds check.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptor.java:
##########
@@ -229,4 +262,26 @@ 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
Review Comment:
nit (optional): The restart at each response works around fabric8's re-send
path, which calls `after(...)` again with the same consumer without
`before(...)` and `consumer(...)`, although the `Interceptor` Javadoc says each
`before(...)` is matched with a call to `after(...)` or
`afterConnectionFailure(...)` (L24-26). Likewise, a re-sent WebSocket upgrade
gets no `after(...)`, and a rejected one gets 1+N. This is unchanged in fabric8
7.9.0, 8.0.0 and `main`, and I couldn't find an upstream issue. If you file
one, it would be nice to reference it here, so that we can revisit this when
fabric8 changes.
##########
docs/configuration.md:
##########
@@ -265,6 +265,12 @@ 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` ends when the response headers
arrive, so it does
Review Comment:
nit: My previous suggestion here does not hold for the `401` of a re-sent
request when the token is refreshed synchronously. fabric8 registers
`after(...)` on the response future before the `afterFailure(...)` chain
(fabric8 7.8.0 `StandardHttpClient.java` L118-119 vs L121-139), and
`CompletableFuture` runs the later dependent first. So
`TokenRefreshInterceptor` refreshes the token, e.g. with an
`OAuthTokenProvider`, a service account token file or an exec credential
plugin, and dispatches the re-send before `after(...)` of the `401` runs. With
a 1 s synchronous refresh, a `401` whose headers arrived after ~225 ms was
recorded as ~1234 ms, and the re-sent `200` as ~62 ms. Only the sum in the next
sentence holds.
How about limiting this sentence to the body, e.g. `It does not include the
time to read the response body.`? The same goes for `until their headers
arrive` in the migration guide.
##########
spark-operator/src/test/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptorTest.java:
##########
@@ -125,6 +138,137 @@ 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();
+
+ 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(4)
+ 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(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()) {
+ await()
Review Comment:
nit: Now that all the assertions are inside `untilAsserted(...)`, this
`await()` can poll from the start, e.g. `await().pollDelay(Duration.ZERO)`.
`inform()` returns only after the `101` upgrade completes, and `after(...)` of
the upgrade runs right after it. So Awaitility's default poll delay (100 ms) is
idle in every run. With `pollDelay(Duration.ZERO)`, the first poll passed in 60
of 60 runs (p50 0.25 ms vs ~104 ms).
##########
spark-operator/src/test/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptorTest.java:
##########
@@ -125,6 +138,137 @@ 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();
+
+ 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(4)
+ 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(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()) {
+ await()
+ .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(6)
+ void testResponseLatencyOfTwoInterceptors() {
+ KubernetesMetricsInterceptor first = new KubernetesMetricsInterceptor();
+ KubernetesMetricsInterceptor second = new KubernetesMetricsInterceptor()
{};
+ List<Interceptor> interceptors = List.of(first, second);
+ 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();
+
+ for (KubernetesMetricsInterceptor interceptor : List.of(first, second)) {
+ Histogram latency = latency(interceptor);
+ Assertions.assertEquals(1, latency.getCount());
+ Snapshot snapshot = latency.getSnapshot();
+ Assertions.assertTrue(snapshot.getMin() >= 50_000_000L, () -> "min: "
+ snapshot.getMin());
+ }
+ }
+ }
+
+ @Test
+ @Order(7)
+ 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 long meterCount(KubernetesMetricsInterceptor interceptor,
String name) {
Review Comment:
nit (optional): As mentioned in the previous round,
`KubernetesClientFactoryTest#meterCount` has the same name and signature but
returns `0` for a missing meter. A shared helper, e.g. in `TestUtils`, would
keep the two from diverging.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptor.java:
##########
@@ -139,7 +173,8 @@ public CompletableFuture<Boolean> afterFailure(
/**
* Called after a connection attempt fails.
*
- * <p>This method will be invoked on each failed connection attempt.
+ * <p>fabric8 invokes this method only for a failed connection attempt which
has a retry left, not
Review Comment:
This is not true for a failure which fabric8 does not retry. `AsyncUtils`
calls `shouldRetry` while the request has a retry left (L76-78), and
`StandardHttpClient#shouldRetry` calls `afterConnectionFailure(...)` for every
failure (L192-193) before it returns `-1` for a non-`IOException`, e.g. a
request timeout (Vert.x `NoStackTraceTimeoutException` is a
`TimeoutException`), or a terminal TLS trust failure. With 3 retries left, a
timed-out request got one `before(...)` and one `afterConnectionFailure(...)`
call and no retry, so this method ran for its last attempt. My earlier comment
was about a refused connection, which fabric8 retries. How about reusing the
wording of `docs/configuration.md`, e.g. `fabric8 invokes this method only
while the request has a retry left, even if it does not retry it, e.g. on a
request timeout.`?
##########
spark-operator/src/test/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptorTest.java:
##########
@@ -125,6 +138,137 @@ 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();
+
+ 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(4)
+ 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(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()) {
+ await()
+ .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(6)
+ void testResponseLatencyOfTwoInterceptors() {
Review Comment:
The PR description lists the other four new tests, but not this one. Could
you update it? In addition:
- `What changes were proposed in this pull request?` does not mention that
each interceptor finds its own `TimedConsumer` via `owner`, or the docs and
Javadoc changes.
- `measured from the previous response rather than from the first attempt`
holds only for an asynchronous token refresh. With a synchronous one, the `401`
value includes the refresh. How about describing the sum, like the docs?
Since `dev/merge_spark_pr.py` uses the PR description as the commit message,
it would be good to keep it in sync.
--
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]
