dongjoon-hyun commented on code in PR #909:
URL:
https://github.com/apache/spark-kubernetes-operator/pull/909#discussion_r4235489166
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/ConfigurableEventRecorder.java:
##########
@@ -207,14 +217,19 @@ private boolean withinMinInterval(EventRecord event,
String uid) {
AtomicBoolean admitted = new AtomicBoolean();
// compute is atomic per key, so concurrent reconcile threads cannot both
admit the same event.
lastRecorded.compute(
- new RecordedEvent(uid, event.reason(), event.message()),
+ new RecordedEvent(uid, event.type(), event.reason(), event.message()),
(key, previous) -> {
if (previous == null || now - previous >= minInterval) {
admitted.set(true);
return now;
}
return previous;
});
+ if (admitted.get() && event.type() == EventType.WARNING) {
Review Comment:
This works in one direction only. A published `Warning` clears the `Normal`
entries of the resource, but a published `Normal` does not clear its `Warning`
entries. For example:
1. t0: the lookup fails with 403 and `SuspendCheckFailed` is published.
2. t1: the lookup succeeds and `SuspendHeld` is published, since the warning
cleared its entry.
3. t2 < t0 + `minIntervalSeconds`: the lookup fails again with the same 403,
so the same message is dropped as a repeat.
Then `kubectl describe` shows `SuspendHeld` as the newest event, while the
check is failing and the `Workload` is kept. Should an admitted `Normal` also
clear the `Warning` entries of the same resource, or should the check use the
order of the last two events of the resource?
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/SuspendUtils.java:
##########
@@ -89,4 +94,39 @@ static ReconcileProgress holdForSuspend(final BaseContext<?>
context, final Stri
return completeAndRequeueAfter(
Duration.ofSeconds(SUSPEND_HOLD_REQUEUE_INTERVAL_SECONDS.getValue()));
}
+
+ /**
+ * Reports a failed check of whether the resources of a suspended resource
were requested, and
+ * returns the progress to retry it with. Whether they are live is unknown
then, so the resource
+ * is neither held, which would claim in an event that none was requested
and release the Kueue
+ * quota of a running driver or master, nor started. Like a failed Kueue
admission request, a
+ * transport level failure is not published, since writing an event would
only add load to an API
+ * server that is often the cause of it. A throttled read is not published
for the same reason,
+ * since the API server is overloaded then by definition. Anything else is,
since a suspended
+ * resource may have no persisted status to show it and would otherwise be
retried without any
+ * signal.
+ *
+ * @param context The context of the suspended resource.
+ * @param e The failure to report.
+ * @param requested The resources whose request could not be checked, as
named in the event and
+ * the log, e.g. {@code "driver"}.
+ * @return The progress to return, requeued after the steady-state interval.
+ */
+ static ReconcileProgress retryAfterCheckFailure(
+ final BaseContext<?> context, final KubernetesClientException e, final
String requested) {
+ String what =
+ "Failed to check whether the "
+ + requested
+ + " of the suspended "
+ + context.getResource().getKind()
+ + " was requested";
+ log.warn("{}, will retry.", what, e);
+ if (!ReconcilerUtils.isTransientError(e) && e.getCode() !=
HTTP_TOO_MANY_REQUESTS) {
Review Comment:
The PR description says that 429 is published ("Any other failure, e.g. 403,
429 or 500, is published as a `SuspendCheckFailed` Warning event"). But this
line skips 429, and so do the docs and the tests (`@ValueSource(ints = {503,
429})` expects no event). Could you update the PR description to match the code?
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterInitStep.java:
##########
@@ -273,6 +293,18 @@ private boolean isMasterRequested(SparkClusterContext
context) {
// read as absent, and getResourceStrictly does the same for a transient
failure, a 500 or a
// 429, since the create path it serves re-reads anyway. Either would
release the Kueue quota
// of a running master. Only a 404 may mean absent, which the client
reports as null.
- return
context.getClient().resource(context.getMasterStatefulSetSpec()).get() != null;
+ StatefulSet master =
context.getClient().resource(context.getMasterStatefulSetSpec()).get();
+ // The name alone is not enough: a StatefulSet left by a deleted
SparkCluster of the same
+ // name, which is still being garbage collected, was not requested by this
one, and would
+ // otherwise move a recreated cluster which never ran to Suspended.
+ return master != null && isOwnedBy(master, context.getResource());
Review Comment:
`isMasterRequested` is also called by the non-suspended Kueue path:
`holdForKueueAdmission` and `handleDequeuedWorkload`. So this owner check
changes that path too. For example, a non-suspended queued `SparkCluster`
reuses the name of a deleted one whose master `StatefulSet` is still being
garbage collected. Before this PR, the leftover master counted as requested, so
only `applyAdmittedPodSetInfos` ran. Now it counts as absent, so a new
admission is requested, or `handleDequeuedWorkload` releases the pending
`Workload`. This is probably the more correct behavior, but the PR description
does not mention it and no test covers it without `suspend`. Could you add a
test, and mention it in the PR description?
##########
spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterInitStepTest.java:
##########
@@ -843,19 +867,21 @@ void kueueTransientApiFailurePublishesNoEvent() {
verifyNoInteractions(eventRecorder);
}
- @Test
- void suspendedClusterWithUnverifiableMasterIsNotHeld() {
+ @ParameterizedTest
+ @ValueSource(ints = {503, 429})
Review Comment:
nit: The parameterized tests cover 503 and 429 (no event) and 403 and 500
(event). They do not cover a failure with no response code, e.g. a broken
connection, which `isTransientError` classifies by its cause. A regression
there would decide whether `SuspendCheckFailed` is written to an API server
which is already failing. How about adding a `NO_RESPONSE_CODE` case with an
`IOException` cause, both here and in `AppInitStepTest`?
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/ConfigurableEventRecorder.java:
##########
@@ -207,14 +217,19 @@ private boolean withinMinInterval(EventRecord event,
String uid) {
AtomicBoolean admitted = new AtomicBoolean();
// compute is atomic per key, so concurrent reconcile threads cannot both
admit the same event.
lastRecorded.compute(
- new RecordedEvent(uid, event.reason(), event.message()),
+ new RecordedEvent(uid, event.type(), event.reason(), event.message()),
(key, previous) -> {
if (previous == null || now - previous >= minInterval) {
admitted.set(true);
return now;
}
return previous;
});
+ if (admitted.get() && event.type() == EventType.WARNING) {
+ lastRecorded
+ .keySet()
+ .removeIf(key -> key.type() == EventType.NORMAL &&
key.uid().equals(uid));
Review Comment:
Each admitted `Warning` scans the whole `lastRecorded` map, which holds the
events of all resources in the last interval. During an API incident, many
resources can publish warnings which are not transient, e.g. 403 or 500. Then
the total cost grows quadratically, and it runs on the reconcile threads.
Keying the state by uid, e.g. `Map<String, Map<RecordedEvent, Long>>`, or
keeping a per-uid "last warning" timestamp which `compute` compares, would make
this O(1).
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterInitStep.java:
##########
@@ -260,11 +264,27 @@ private Optional<ReconcileProgress> holdForKueueAdmission(
}
/**
- * Checks whether the master StatefulSet has already been requested, e.g.
when the status update
- * to RunningHealthy failed after the resources were created.
+ * Checks whether the cluster has had no state but its initial Submitted
state, which is never
+ * persisted while it is suspended. A cluster resumed from Suspended or
queued again after a Kueue
+ * eviction has persisted another Submitted state instead, at a later key,
since the keys of the
+ * state transition history keep increasing even when the history is
trimmed. Only a cluster in
+ * its initial submission may have a master which was never requested, so
only it looks the
+ * master up while suspended. A change which persists another state before
the master is
+ * requested must keep this check in step with it.
+ *
+ * @param cluster The SparkCluster.
+ * @return True if the cluster is in its initial submission, false otherwise.
+ */
+ private static boolean isInitialSubmission(SparkCluster cluster) {
+ return cluster.getStatus().getStateTransitionHistory().lastKey() == 0L;
Review Comment:
This check assumes that nothing persists a state before the master is
requested. The Javadoc says so itself: "A change which persists another state
before the master is requested must keep this check in step with it." Suppose a
later change persists an intermediate state first, e.g. a Kueue pending status.
Then a suspended first attempt would skip the master lookup and go to
`Suspended` directly. `ClusterSuspendStep` would delete the StatefulSets by
name, and a cluster which never started would be shown as released. Could we
use a more explicit signal instead, e.g. whether the status was ever persisted?
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterInitStep.java:
##########
@@ -273,6 +293,18 @@ private boolean isMasterRequested(SparkClusterContext
context) {
// read as absent, and getResourceStrictly does the same for a transient
failure, a 500 or a
// 429, since the create path it serves re-reads anyway. Either would
release the Kueue quota
// of a running master. Only a 404 may mean absent, which the client
reports as null.
- return
context.getClient().resource(context.getMasterStatefulSetSpec()).get() != null;
+ StatefulSet master =
context.getClient().resource(context.getMasterStatefulSetSpec()).get();
+ // The name alone is not enough: a StatefulSet left by a deleted
SparkCluster of the same
+ // name, which is still being garbage collected, was not requested by this
one, and would
+ // otherwise move a recreated cluster which never ran to Suspended.
+ return master != null && isOwnedBy(master, context.getResource());
+ }
+
+ private static boolean isOwnedBy(StatefulSet statefulSet, SparkCluster
cluster) {
Review Comment:
nit: This repeats the owner reference scan which already exists in
`KueueWorkloadUtils.controllerUid`, and also in `DriverResourceDecorator`. How
about a shared helper, e.g. in `ModelUtils` next to `buildOwnerReferenceTo`?
Then a future fix, e.g. also requiring `controller=true` as `controllerUid`
does, would be needed in one place only.
--
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]
