dongjoon-hyun commented on code in PR #973:
URL:
https://github.com/apache/spark-kubernetes-operator/pull/973#discussion_r4437164931
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppInitStep.java:
##########
@@ -127,33 +128,13 @@ public ReconcileProgress reconcile(
if (kueueHold.isPresent()) {
return kueueHold.get();
}
- List<HasMetadata> preResourcesSpec = context.getDriverPreResourcesSpec();
- for (HasMetadata resource : preResourcesSpec) {
- Optional<HasMetadata> createdResource =
- ReconcilerUtils.getOrCreateSecondaryResource(context.getClient(),
resource);
- if (createdResource.isEmpty()) {
- return appendStateAndImmediateRequeue(
- context, statusRecorder, creationFailureState(resource));
- }
- }
- Optional<Pod> driverPod =
- ReconcilerUtils.getOrCreateSecondaryResource(
- context.getClient(), context.getDriverPodSpec());
- if (driverPod.isPresent()) {
- DriverResourceDecorator decorator = new
DriverResourceDecorator(driverPod.get());
- preResourcesSpec.forEach(decorator::decorate);
-
context.getClient().resourceList(preResourcesSpec).forceConflicts().serverSideApply();
- List<HasMetadata> driverResources = context.getDriverResourcesSpec();
- driverResources.forEach(decorator::decorate);
- for (HasMetadata resource : driverResources) {
- Optional<HasMetadata> createdResource =
-
ReconcilerUtils.getOrCreateSecondaryResource(context.getClient(), resource);
- if (createdResource.isEmpty()) {
- return appendStateAndImmediateRequeue(
- context, statusRecorder, creationFailureState(resource));
- }
- }
+ Optional<ReconcileProgress> creationFailure =
+ requestDriverAndResources(context, statusRecorder);
+ if (creationFailure.isPresent()) {
+ return creationFailure.get();
}
+ } catch (SecondaryResourceBackoffException e) {
Review Comment:
This exception is caught only here, but `ClusterInitStep` reaches
`getOrCreateSecondaryResource` too, via `KueueWorkloadUtils.holdForAdmission`
-> `requestAdmission` (I'm leaving this here since `ClusterInitStep` is not
changed by this PR). Since `holdForAdmission` catches only
`KubernetesClientException` and `IllegalStateException`, a parked Workload
creation of a Kueue-queued `SparkCluster`, e.g. on a 429 with `Retry-After` of
3s or more, falls into the `catch (Exception e)` of `ClusterInitStep#reconcile`
and fails the cluster with the terminal `SchedulingFailure`.
Before this PR, the same case slept in place, and even when the attempts ran
out, the `KubernetesClientException` was retried via `retryAfterRequestFailure`
instead of failing the cluster. Could you handle this exception for the cluster
as well, e.g. in `holdForAdmission`?
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/SecondaryResourceRetryTracker.java:
##########
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.k8s.operator.utils;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+
+/**
+ * Tracks, in memory only, how many consecutive attempts a secondary resource
creation has
+ * already made across reconciles, so that a backoff parked via {@link
+ * SecondaryResourceBackoffException} resumes where it left off instead of
restarting from the
+ * first attempt. There is no delete hook to clear an entry whose owner is
removed mid-backoff, so
+ * a stale entry is instead dropped lazily the next time its key is looked up.
+ */
+final class SecondaryResourceRetryTracker {
+
+ private static final long STALE_AFTER_NANOS = TimeUnit.MINUTES.toNanos(30);
+
+ private static final ConcurrentHashMap<String, Entry> ATTEMPT_COUNTS = new
ConcurrentHashMap<>();
+
+ private SecondaryResourceRetryTracker() {}
+
+ static long startAttemptCount(HasMetadata resource) {
+ Entry entry = ATTEMPT_COUNTS.get(key(resource));
+ if (entry == null || System.nanoTime() - entry.recordedAtNanos() >
STALE_AFTER_NANOS) {
+ return 1L;
+ }
+ return entry.attemptCount();
+ }
+
+ static void recordAttemptCount(HasMetadata resource, long attemptCount) {
+ ATTEMPT_COUNTS.put(key(resource), new Entry(attemptCount,
System.nanoTime()));
+ }
+
+ static void clear(HasMetadata resource) {
+ ATTEMPT_COUNTS.remove(key(resource));
+ }
+
+ private static String key(HasMetadata resource) {
Review Comment:
The key has no owner identity, so a parked count can outlive its owner.
E.g., when a `SparkApplication` whose driver pod creation was parked twice
(count 3) is deleted and submitted again with the same name within 30 minutes,
the new application resumes at 3, because its driver pod gets the same name,
which is derived from the application name and the attempt id. Then its first
429 fails it with `SchedulingFailure` without any retry. The same holds for
other deterministic names, e.g. the Kueue Workload (`<kind>-<name>`). Could you
include the owner UID (and the attempt id) in the key, or drop the owner's
entries on cleanup?
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppInitStep.java:
##########
@@ -311,6 +338,22 @@ private static boolean hasLiveDriverPod(SparkAppContext
context, SparkApplicatio
.anyMatch(pod -> SparkAppContext.isLiveDriverPod(app, pod));
}
+ /**
+ * Requeues after a backoff too long to block the reconciler thread for, see
{@link
+ * SecondaryResourceBackoffException}. This is not a failure, so no state is
appended and the
+ * application stays Initializing.
+ *
+ * @param e The exception carrying the requeue delay.
+ * @return The progress to requeue the reconcile after the delay.
+ */
+ private ReconcileProgress
requeueAfterBackoff(SecondaryResourceBackoffException e) {
+ log.info(
Review Comment:
nit: Since `BackoffUtils.backoffSleep` is skipped on this path, this is the
only log of the backoff, but it has only the delay. How about including
`e.getMessage()`, which already has the kind, the name and the attempt count,
so that a parked retry can be traced to its resource?
##########
spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppInitStepTest.java:
##########
@@ -366,6 +367,48 @@ void
nonTrimModeRestartBackoffElapsedProceedsToDriverCreation() {
application.getStatus().getCurrentState().getCurrentStateSummary());
}
+ @Test
+ void driverCreationBackoffAboveThresholdIsParkedNotFailed() {
+ // A Retry-After this long exceeds the default blocking threshold (see
+ //
SparkOperatorConf.API_SECONDARY_RESOURCE_CREATE_MAX_BLOCKING_DELAY_MILLIS), so
the
+ // reconciler must requeue instead of blocking the thread and instead of
failing the
+ // application with the terminal SchedulingFailure.
+ AppInitStep appInitStep = new AppInitStep();
+ SparkAppContext mockContext = mock(SparkAppContext.class);
+ SparkAppStatusRecorder recorder = mock(SparkAppStatusRecorder.class);
+ SparkApplication application = new SparkApplication();
+ application.setMetadata(applicationMetadata);
+ KubernetesClient failingClient = mock(KubernetesClient.class,
RETURNS_DEEP_STUBS);
+ KubernetesClientException tooManyRequests =
+ new KubernetesClientException(
+ "Too Many Requests",
+ 429,
+ new StatusBuilder()
+ .withCode(429)
+ .withNewDetails()
+ .withRetryAfterSeconds(5)
+ .endDetails()
+ .build());
+ when(failingClient.resource(any(Pod.class)).get()).thenReturn(null);
+
when(failingClient.resource(any(Pod.class)).create()).thenThrow(tooManyRequests);
+ when(mockContext.getResource()).thenReturn(application);
+ when(mockContext.getClient()).thenReturn(failingClient);
+ when(mockContext.getEventRecorder()).thenReturn(eventRecorder);
+ when(mockContext.getDriverPreResourcesSpec()).thenReturn(List.of());
+ when(mockContext.getDriverPodSpec()).thenReturn(driverPodSpec);
Review Comment:
This uses the shared `driverPodSpec` (`default/driver-pod`, used by many
tests of this class), so it leaves `Pod/default/driver-pod` with count 2 in the
static tracker. It may not break anything today, but a later test in the same
JVM whose creation of this pod fails would start at attempt 2 depending on the
run order, which the PR description says the new tests avoid with unique names.
Could you use a uniquely named pod here as well?
##########
spark-operator/src/test/java/org/apache/spark/k8s/operator/utils/ReconcilerUtilsTest.java:
##########
@@ -342,6 +347,109 @@ void
retriesCreateOnTransient5xxHonoringRetryAfterWhenPresent() {
assertTrue(elapsedMillis >= 1000L, "should have slept for the
server-requested 1s");
}
+ @Test
+ void throwsBackoffExceptionInsteadOfSleepingAboveTheBlockingThreshold() {
+ Pod pod = buildPod("backoff-threshold-pod");
+ KubernetesClient mockClient = mock(KubernetesClient.class);
+ NamespaceableResource<Pod> mockResource = mockClientReturning(mockClient,
pod);
+ when(mockResource.get()).thenReturn(null);
+ // 5s Retry-After exceeds the default 3s blocking threshold, so this must
not sleep.
+ KubernetesClientException tooManyRequests =
Review Comment:
nit: The same 429 with `retryAfterSeconds(5)` is built in the three new
tests here and in `AppInitStepTest`. How about a small helper, e.g.
`tooManyRequests(int retryAfterSeconds)`?
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/ReconcilerUtils.java:
##########
@@ -169,20 +176,34 @@ public static <T extends HasMetadata> Optional<T>
getOrCreateSecondaryResource(
// GET to avoid a duplicate create for transport failures,
timeouts and transient 5xx
current = getResource(client, resource);
if (current.isPresent()) {
+ SecondaryResourceRetryTracker.clear(resource);
return current;
}
} else {
+ SecondaryResourceRetryTracker.clear(resource);
throw e;
}
if (++attemptCount > maxAttempts) {
+ SecondaryResourceRetryTracker.clear(resource);
log.error("Max Retries exceeded while trying to create resource");
throw e;
}
if (shouldBackoffBeforeRetry(e)) {
- BackoffUtils.backoffSleep(e, attemptCount, maxAttempts);
+ long delayMillis =
+
BackoffUtils.computeDelayMillis(BackoffUtils.getRetryAfterMillis(e),
attemptCount);
+ long maxBlockingDelayMillis =
+
API_SECONDARY_RESOURCE_CREATE_MAX_BLOCKING_DELAY_MILLIS.getValue();
+ if (delayMillis > maxBlockingDelayMillis) {
+ SecondaryResourceRetryTracker.recordAttemptCount(resource,
attemptCount);
Review Comment:
The tracker keeps only the attempt count, not when the parked retry is due.
So a reconcile which comes before `requeueAfter`, e.g. one triggered by the
events of the driver pod created earlier in the same reconcile, sends the
create again right away. Under a 429 storm, the remaining attempts can be used
up within a few seconds, failing the application with `SchedulingFailure`
without honoring the server-requested `Retry-After`, while `Thread.sleep`
guaranteed the delay between attempts before this PR.
How about recording the due time too, and requeueing for the remaining time
without sending the create when a reconcile comes early?
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/config/SparkOperatorConf.java:
##########
@@ -928,6 +928,23 @@ public final class SparkOperatorConf {
.defaultValue(500L)
.build();
+ /**
+ * Maximum backoff delay (in milliseconds) the reconciler still blocks a
thread for with
+ * Thread.sleep when creating secondary resources. A delay above this is
requeued instead,
+ * since parking it would cost more in reconcile-queue latency than the wait
it would save.
Review Comment:
nit: The `since` clause is the reason to block on a short delay, but it is
attached to the requeue of a long one, so it reads reversed. How about e.g.:
> Maximum backoff delay (in milliseconds) the reconciler still blocks a
thread for with Thread.sleep when creating secondary resources, since
requeueing a delay this short would cost more in reconcile-queue latency than
the wait it would save. A longer delay is requeued instead.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppInitStep.java:
##########
@@ -186,6 +167,52 @@ public ReconcileProgress reconcile(
return attemptStatusUpdate(context, statusRecorder, updatedStatus,
completeAndDefaultRequeue());
}
+ /**
+ * Requests the driver pre-resources, the driver pod, and the driver
resources, in order,
+ * decorating the driver resources with the driver pod's identity once it is
requested. A
+ * resource whose creation fails after exhausting the retries of {@link
+ * ReconcilerUtils#getOrCreateSecondaryResource} moves the application to
SchedulingFailure.
+ *
+ * @param context The SparkAppContext for the application.
+ * @param statusRecorder The SparkAppStatusRecorder for recording status
updates.
+ * @return The progress to return on a creation failure, or empty to proceed.
+ * @throws SecondaryResourceBackoffException if a retry needs a longer wait
than the reconciler
+ * should block its thread for.
+ */
+ private Optional<ReconcileProgress> requestDriverAndResources(
+ SparkAppContext context, SparkAppStatusRecorder statusRecorder) {
+ List<HasMetadata> preResourcesSpec = context.getDriverPreResourcesSpec();
+ for (HasMetadata resource : preResourcesSpec) {
+ Optional<HasMetadata> createdResource =
+ ReconcilerUtils.getOrCreateSecondaryResource(context.getClient(),
resource);
+ if (createdResource.isEmpty()) {
+ return Optional.of(
+ appendStateAndImmediateRequeue(
+ context, statusRecorder, creationFailureState(resource)));
+ }
+ }
+ Optional<Pod> driverPod =
+ ReconcilerUtils.getOrCreateSecondaryResource(
+ context.getClient(), context.getDriverPodSpec());
+ if (driverPod.isPresent()) {
+ DriverResourceDecorator decorator = new
DriverResourceDecorator(driverPod.get());
+ preResourcesSpec.forEach(decorator::decorate);
+
context.getClient().resourceList(preResourcesSpec).forceConflicts().serverSideApply();
Review Comment:
Since a parked retry runs the whole init path again on the next reconcile,
each retry also re-sends a GET per pre-resource, the GET of the driver pod and,
once the driver pod exists, this server-side apply of all pre-resources (plus
the Workload GET of a Kueue-queued application), while the in-place retry
before this PR re-sent only the failed create. This adds API calls exactly
while the API server is throttling the operator. Skipping the requests while
the parked retry is not due yet (see the comment on `ReconcilerUtils`) would
avoid most of them.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/ReconcilerUtils.java:
##########
@@ -143,10 +148,11 @@ public static <T extends HasMetadata> Optional<T>
getOrCreateSecondaryResource(
// Adding retry logic to overcome known k8s issue:
// https://github.com/kubernetes/kubernetes/issues/67761
long maxAttempts = API_SECONDARY_RESOURCE_CREATE_MAX_ATTEMPTS.getValue();
- long attemptCount = 1;
+ long attemptCount =
SecondaryResourceRetryTracker.startAttemptCount(resource);
Review Comment:
Some exits leave the entry behind, e.g. when the `getResourceStrictly` call
above throws a non-retryable error, like a 403 after an RBAC change, on the
call after a park, or when `create()` throws something other than
`KubernetesClientException`. A later creation with the same key then resumes
from that count (see the comment on the key).
Instead of the six `clear()` calls, how about removing the count when
reading it at the beginning of this method, and recording it only right before
throwing `SecondaryResourceBackoffException`? Then every other exit drops the
entry by construction.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/SecondaryResourceRetryTracker.java:
##########
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.k8s.operator.utils;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+
+/**
+ * Tracks, in memory only, how many consecutive attempts a secondary resource
creation has
+ * already made across reconciles, so that a backoff parked via {@link
+ * SecondaryResourceBackoffException} resumes where it left off instead of
restarting from the
+ * first attempt. There is no delete hook to clear an entry whose owner is
removed mid-backoff, so
+ * a stale entry is instead dropped lazily the next time its key is looked up.
+ */
+final class SecondaryResourceRetryTracker {
+
+ private static final long STALE_AFTER_NANOS = TimeUnit.MINUTES.toNanos(30);
+
+ private static final ConcurrentHashMap<String, Entry> ATTEMPT_COUNTS = new
ConcurrentHashMap<>();
+
+ private SecondaryResourceRetryTracker() {}
+
+ static long startAttemptCount(HasMetadata resource) {
+ Entry entry = ATTEMPT_COUNTS.get(key(resource));
+ if (entry == null || System.nanoTime() - entry.recordedAtNanos() >
STALE_AFTER_NANOS) {
Review Comment:
A stale entry is ignored here, but not removed, and nothing removes the
entry of a key which is not looked up again, e.g. of an application deleted
while parked. So, on a platform which submits uniquely named applications under
recurring throttling, the static map keeps growing for the lifetime of the
operator. Could you evict stale entries, e.g. by sweeping them in
`recordAttemptCount`?
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/config/SparkOperatorConf.java:
##########
@@ -928,6 +928,23 @@ public final class SparkOperatorConf {
.defaultValue(500L)
.build();
+ /**
+ * Maximum backoff delay (in milliseconds) the reconciler still blocks a
thread for with
+ * Thread.sleep when creating secondary resources. A delay above this is
requeued instead,
+ * since parking it would cost more in reconcile-queue latency than the wait
it would save.
+ */
+ public static final ConfigOption<Long>
API_SECONDARY_RESOURCE_CREATE_MAX_BLOCKING_DELAY_MILLIS =
+ ConfigOption.<Long>builder()
+
.key("spark.kubernetes.operator.api.secondaryResourceCreateMaxBlockingDelayMillis")
+ .enableDynamicOverride(false)
+ .description(
+ "Maximum backoff delay (in milliseconds) the reconciler still
blocks a thread "
+ + "for with Thread.sleep when creating secondary resources
for Spark "
+ + "application. A delay above this is requeued instead of
blocked on.")
+ .typeParameterClass(Long.class)
+ .defaultValue(3000L)
Review Comment:
Since this applies by default, a 1.0 deployment which tuned
`spark.kubernetes.operator.api.secondaryResourceCreateInitialBackoffMillis`,
`...BackoffMultiplier` or `...MaxAttempts` so that a backoff exceeds 3s now
gets a requeue instead of the in-place sleep of 1.0.0 (`Retry-After` is not
honored in 1.0.0, so only such tuned settings are affected). Should
`docs/migration_guide.md` have an item for it, with how to restore the previous
behavior, i.e. setting this to `secondaryResourceCreateMaxBackoffMillis` +
`secondaryResourceCreateBackoffJitterMillis` or more?
##########
spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/SparkAppReconcilerTest.java:
##########
@@ -389,16 +391,16 @@ void
secondaryResourceCreationRetriesWithBackoffOnConflict() {
when(mockResource.get()).thenReturn(null);
when(mockResource.create()).thenThrow(conflict).thenReturn(pod);
- long start = System.currentTimeMillis();
- Optional<Pod> result =
ReconcilerUtils.getOrCreateSecondaryResource(mockClient, pod);
- long elapsed = System.currentTimeMillis() - start;
-
- assertTrue(result.isPresent());
- verify(mockResource, times(2)).create();
- assertTrue(elapsed >= 5000L,
- "Expected backoff delay of at least "
- + "5000ms, but was: "
- + elapsed + "ms");
+ // The 5000ms initial backoff exceeds the default blocking
threshold, so the retry is
+ // parked via a requeue instead of blocking the reconciler thread
with Thread.sleep.
+ SecondaryResourceBackoffException ex =
Review Comment:
This was the only test of the in-place backoff on a 409 Conflict, i.e.
sleeping and then retrying successfully, and now it covers only the park. So a
regression of that path, e.g. of the 409 branch of `shouldBackoffBeforeRetry`,
would go unnoticed. Could you keep the original scenario with an initial
backoff below the threshold, and test the park separately? The test name no
longer matches what it verifies, either.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/SecondaryResourceBackoffException.java:
##########
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.k8s.operator.utils;
+
+import java.time.Duration;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.fabric8.kubernetes.client.KubernetesClientException;
+import lombok.Getter;
+
+/**
+ * Signals that {@link ReconcilerUtils#getOrCreateSecondaryResource} needs to
wait longer than the
+ * reconciler should block a thread for before retrying resource creation.
Callers must catch this
+ * ahead of any broader {@code catch (Exception e)} and requeue the reconcile
after the delay
+ * carried by this exception instead of treating it as a failure.
+ */
+@Getter
+public final class SecondaryResourceBackoffException extends RuntimeException {
+
+ private final HasMetadata resource;
Review Comment:
nit: Outside of the tests, only `requeueAfter` is read, while the message
already has the resource, the attempt count and the max attempts. How about
keeping only `requeueAfter` as a field? It also avoids keeping the whole
desired resource, e.g. a driver pod spec, in the exception.
--
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]
