dongjoon-hyun commented on code in PR #918:
URL:
https://github.com/apache/spark-kubernetes-operator/pull/918#discussion_r4160508564
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptor.java:
##########
@@ -88,6 +90,24 @@ public KubernetesMetricsInterceptor() {
}
}
+ /**
+ * Called before a request is sent, to allow wrapping the consumer of the
response body.
+ *
+ * <p>Records the request start time, which {@link #after(HttpRequest,
HttpResponse,
+ * AsyncBody.Consumer)} uses to compute the response latency. Both methods
receive the same
+ * request instance.
Review Comment:
This is true for a request passed to `consumer(...)`, but fabric8 also calls
`after(...)` without a preceding `consumer(...)`, e.g. for a WebSocket upgrade,
and calls it twice for a request re-sent after `afterFailure(...)`. This is why
the `null` check in `updateResponseMetrics` is needed. How about describing it
like the following?
> For an HTTP exchange, fabric8 passes the same request instance to both
methods. `after(...)` may also be called without a recorded start time, e.g.
for a WebSocket upgrade or a re-sent request, and then no latency is recorded.
In addition, the existing Javadoc of `after(...)` ("non-WebSocket HTTP
response") and `afterConnectionFailure(...)` ("each failed connection attempt")
does not match fabric8 7.8.0.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptor.java:
##########
@@ -60,6 +60,8 @@ public class KubernetesMetricsInterceptor implements
Interceptor, Source {
private final Meter requestRateMeter;
private final Meter responseRateMeter;
private final Map<String, Meter> namespacedResourceMethodMeters = new
ConcurrentHashMap<>();
+ private final Map<HttpRequest, Long> requestStartTimeNanos =
Review Comment:
nit (optional): fabric8 already hands a per-exchange object from
`consumer(...)` to `after(...)`. `StandardHttpClient#consumeBytesOnce` passes
the consumer returned by the `consumer(...)` chain as the third argument of
`after(...)`, and fabric8's `HttpLoggingInterceptor` uses it via
`DeferredLoggingConsumer` and `AsyncBody.Consumer#unwrap`. A delegating
consumer which carries the start time would remove this global map, its lock,
the request identity assumption and the cleanup in
`afterConnectionFailure(...)`. It would also time a re-sent request.
```java
@Override
public AsyncBody.Consumer<List<ByteBuffer>> consumer(
AsyncBody.Consumer<List<ByteBuffer>> consumer, HttpRequest request) {
return new TimedConsumer(consumer, System.nanoTime());
}
@Override
public void after(
HttpRequest request,
HttpResponse<?> response,
AsyncBody.Consumer<List<ByteBuffer>> consumer) {
TimedConsumer timed = consumer == null ? null :
consumer.unwrap(TimedConsumer.class);
updateResponseMetrics(response, timed == null ? null :
timed.startTimeNanos());
}
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);
}
}
```
`unwrap` should delegate to the wrapped consumer. Otherwise,
`HttpLoggingInterceptor`, which is installed first at the `TRACE` level, cannot
find its own consumer. A WebSocket upgrade still has no start time because
fabric8 passes `null` as the consumer.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptor.java:
##########
@@ -21,21 +21,21 @@
import static
org.apache.spark.k8s.operator.config.SparkOperatorConf.KUBERNETES_CLIENT_METRICS_GROUP_BY_RESPONSE_CODE_GROUP_ENABLED;
+import com.codahale.metrics.Histogram;
Review Comment:
This reorders the whole import block and violates the Spotless rule,
`importOrder('java', 'javax', 'scala', '', 'org.apache.spark')`, in
`build.gradle`. `./gradlew spotlessCheck` fails on this file, so `./gradlew
build` in CI fails too. Could you run `./gradlew spotlessApply`? Only
`java.util.Collections` and `java.util.WeakHashMap` need to be added to the
`java` group.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptor.java:
##########
@@ -114,7 +134,7 @@ public void after(
HttpRequest request,
HttpResponse<?> response,
AsyncBody.Consumer<List<ByteBuffer>> consumer) {
- updateResponseMetrics(response, System.nanoTime());
+ updateResponseMetrics(response, requestStartTimeNanos.remove(request));
Review Comment:
When `afterFailure(...)` of an interceptor returns `true`, e.g. fabric8's
default `TokenRefreshInterceptor` on `401`,
`StandardHttpClient#consumeBytesOnce` re-sends the request without calling
`consumer(...)` and calls `after(...)` again with the same request instance.
Since the first `after(...)` already removed the entry, the re-sent response,
which is what the caller gets, has no latency sample. With a mock server
returning `401` and then `200`, I got two `http.response` marks and only one
latency sample. This is rare, but the delegating consumer in the comment on the
field above would cover it.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptor.java:
##########
@@ -144,6 +164,7 @@ public CompletableFuture<Boolean> afterFailure(
*/
@Override
public void afterConnectionFailure(HttpRequest request, Throwable failure) {
+ requestStartTimeNanos.remove(request);
Review Comment:
This `remove(request)` never matches. fabric8 calls
`afterConnectionFailure(...)` from `StandardHttpClient#shouldRetry` with the
original request of `consumeBytes(...)`, while `consumer(...)` gets the request
rebuilt by `copy.build()` in `consumeBytesOnce(...)` for each attempt, and
`StandardHttpRequest` does not override `equals`/`hashCode`. Moreover, fabric8
does not call this method for the last attempt or for
`KubernetesClientFactory.withoutRetries` clients. With a retry limit of 2
against a stopped mock server, 3 entries were put and all 3 remained after 2
calls of this method. Since the weak keys prevent a leak, could you remove this
line and mention in a comment that GC reclaims the entries of requests without
a response?
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptor.java:
##########
@@ -181,11 +202,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) {
Review Comment:
fabric8 7.8.0 calls `after(...)` for a WebSocket upgrade response without
calling `consumer(...)` (`StandardHttpClient#buildWebSocketOnce` passes
`webSocketUpgradeResponse.request()`), so `startTimeNanos` is always `null`
here for WebSocket. Since informer watches use WebSocket by default, each watch
connection marks `http.response` and `http.response.101` without a latency
sample. With a watch on the mock server, I got two `http.response` marks (`101`
and `201`) and one latency sample. Previously, the histogram count was equal to
the `http.response` count. Since the start time cannot be correlated on this
path, could you document that WebSocket upgrades are excluded, e.g. in
`docs/configuration.md`?
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptor.java:
##########
@@ -181,11 +202,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:
This changes a user-visible behavior of the latest release, because this
metric is enabled by default and documented in 1.0.0. The values jump from
microseconds to real latencies, and the count no longer includes WebSocket
upgrades and re-sent requests. As `AGENTS.md` requires, could you add an item
to `docs/migration_guide.md` under `## Upgrading from 1.0 to 1.1` > `###
Operator`?
##########
spark-operator/src/test/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptorTest.java:
##########
@@ -122,6 +123,31 @@ void testWhenKubernetesServerNotWorking() {
}
}
+ @Test
+ @Order(3)
+ void testResponseLatency() {
Review Comment:
This test covers a single delayed GET only. Could you add test coverage for
the other paths changed by this PR? For example:
- After `testWhenKubernetesServerNotWorking`, the start time map should be
empty. It fails now because of the no-op cleanup in
`afterConnectionFailure(...)`.
- For a watch (WebSocket upgrade) and for a `401` → `200` re-send, the
latency count compared with the `http.response` count.
##########
spark-operator/src/test/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptorTest.java:
##########
@@ -122,6 +123,31 @@ 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(500)
+
.withPath("/api/v1/namespaces/spark-system/configmaps/spark-job-operator-configuration")
+ .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() >= 500_000_000L);
Review Comment:
nit: Could you add an upper bound too, e.g. `< 30_000_000_000L`?
`System.nanoTime()` is usually the time since boot, so a regression which
records an absolute timestamp instead of a difference would still pass this
check.
##########
spark-operator/src/test/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptorTest.java:
##########
@@ -122,6 +123,31 @@ 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(500)
Review Comment:
nit: Since the previous code recorded less than a microsecond, a smaller
delay, e.g. `.delay(50)` with `>= 50_000_000L`, would catch the regression
equally deterministically and save about 0.45s per run.
--
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]
