dongjoon-hyun commented on code in PR #918:
URL:
https://github.com/apache/spark-kubernetes-operator/pull/918#discussion_r4268244084
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptor.java:
##########
@@ -103,21 +123,36 @@ 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
+ @SuppressWarnings("PMD.CompareObjectsWithEquals")
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);
+ while (timedConsumer != null && timedConsumer.owner() != this) {
Review Comment:
Do we need the `owner` check and this loop? `KubernetesClientFactory`
registers interceptors with
`addOrReplaceInterceptor(interceptor.getClass().getName(), ...)`, so two
instances of the same class replace each other, and `SparkOperator` creates
only one `KubernetesMetricsInterceptor`. The only way to get two
`TimedConsumer`s into one chain is an anonymous subclass, as in
`testResponseLatencyOfTwoInterceptors`. Without that case, a plain
`consumer.unwrap(TimedConsumer.class)` would be enough, and we could drop the
`owner` component, the `PMD.CompareObjectsWithEquals` suppression, and that
test.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptor.java:
##########
@@ -140,7 +175,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 while the request has a retry left,
even if it does not
Review Comment:
This Javadoc correction looks right to me, but it isn't related to the
latency fix. The same applies to the new doc sentence that a rejected WebSocket
upgrade is counted several times in `kubernetes.client.http.response` and the
response code meters. Could we move these to a separate PR, to keep this one
scoped to SPARK-59928?
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptor.java:
##########
@@ -233,4 +267,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
+ * response so that the values of a re-sent request add up to its latency.
It also carries the
+ * interceptor which created it, so that each interceptor on a client uses
its own start time.
+ */
+ private record TimedConsumer(
Review Comment:
nit. A `record` is meant to be a shallow-immutable data carrier, but
`startTimeNanos()` returns an `AtomicLong` that `after(...)` mutates. The
record also generates `equals`/`hashCode`/`toString` over the whole delegate
chain and the owner interceptor, and this takes two allocations per request. A
`private static final class` with an explicit `delegate` and one `volatile
long` (or `AtomicLong`) field would do the same job in one allocation.
##########
docs/configuration.md:
##########
@@ -264,6 +264,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` does not include the time to
read the response
Review Comment:
The start time is taken in `consumer(...)`, which runs before
`consumeBytesDirect(...)`. So the value also includes client-side time, such as
waiting for a Vert.x pool connection and connection/TLS setup. For example,
under many concurrent reconciles, this metric can rise while the API server
itself is fast. The metrics table describes it as "HTTP response latency from
the Kubernetes API Server". Could we mention that it is measured from when the
request is handed to the HTTP client?
--
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]
