peter-toth commented on code in PR #904:
URL:
https://github.com/apache/spark-kubernetes-operator/pull/904#discussion_r4130459753
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppRunningStep.java:
##########
@@ -49,6 +46,10 @@ public final class AppRunningStep extends AppReconcileStep {
@Override
public ReconcileProgress reconcile(
SparkAppContext context, SparkAppStatusRecorder statusRecorder) {
+ // The driver and executors are observed from DriverReady on, so the Kueue
Workload learns here
+ // that they are ready, before a state transition ends the reconciliation.
A failed recording is
+ // retried by a later reconciliation rather than holding back the
observation of the driver.
+ KueueWorkloadUtils.recordPodsReady(context);
Review Comment:
**Finding 1.** The return value is dropped here, so on the
`SparkApplication` path neither retry the description promises happens.
The description says:
> A conflict or a transport-level failure is retried after 5 seconds. A
persistent failure publishes the new `KueuePodsReadyUpdateFailed` warning event
and is retried with the default interval.
Both hold for `ClusterSuspendStep`, which returns
`recordPodsReady(context).orElse(proceed())`. For an application:
- a conflict or a transient failure gives up its
`completeAndRequeueAfter(STALE_WORKLOAD_REQUEUE_INTERVAL)`, and the retry lands
on whatever this reconciliation ends with instead;
- a persistent failure gives up its `completeAndDefaultRequeue()`, still
publishes the event, and again inherits the step's own progress.
What that inherits is not one interval. When the step changes nothing it
falls through to `observeDriver`, which proceeds, and
`SparkAppReconciler.reconcile` ends with `completeAndDefaultRequeue()`, i.e.
`spark.kubernetes.operator.reconciler.intervalSeconds` (120s) rather than 5s.
When the step transitions state it returns `completeAndImmediateRequeue()`.
When the driver is missing from the cache, `observeDriver` returns
`MISSING_DRIVER_REQUEUE_INTERVAL_SECONDS`. Against the 30-minute default
`waitForPodsReady` timeout none of these is dangerous, so this is about the
contract rather than about a broken resource.
The comment above the call gives a good reason not to *hold back* the driver
observation, and that reason is separable from discarding the value. The step
can keep it and apply it on the branch that would otherwise proceed:
```java
Optional<ReconcileProgress> podsReadyRetry =
KueueWorkloadUtils.recordPodsReady(context);
```
then at the bottom:
```java
if (proposedStateSummary == prevStateSummary) {
ReconcileProgress progress =
observeDriver(context, statusRecorder, List.of(new
AppDriverRunningObserver()));
// A failed recording is retried on its own schedule, but only where
the driver observation
// has nothing of its own to say.
return progress.isCompleted() ? progress :
podsReadyRetry.orElse(progress);
}
```
If you would rather keep the drop as it is, the description's retry sentence
needs the qualification, and so does the `KueuePodsReadyUpdateFailed` row in
`docs/configuration.md:97`. That row reads "Transport-level errors and
conflicts are skipped, and a persistent failure is retried with the default
interval", which drops the "retried every 5 seconds" half its
`KueueResourceFlavorReadFailed` neighbour carries — accurate for an
application, not for a cluster.
Either way this branch is the one path through `recordPodsReady` with no
test: `AppRunningStepTest.immediateRequeueWhenRunningStateChanges` verifies the
call is made, and nothing pins what happens to a failed one.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/context/BaseContext.java:
##########
@@ -104,6 +111,30 @@ public Optional<Workload> getCachedKueueWorkload() {
return josdkContext.getSecondaryResource(Workload.class);
}
+ /**
+ * Counts the ready pods of the resource per Spark role in the informer
cache, once for the
+ * lifetime of this context, since a reconciliation may check them more than
once.
+ *
+ * @return The number of the ready pods by the value of their `spark-role`
label, e.g. `executor`
+ * or `worker`.
+ */
+ public Map<String, Long> countReadyPodsByRole() {
+ synchronized (this) {
+ if (readyPodsByRole == null) {
+ readyPodsByRole =
+ josdkContext
+ .getSecondaryResourcesAsStream(Pod.class)
+ .filter(pod ->
pod.getMetadata().getLabels().containsKey(LABEL_SPARK_ROLE_NAME))
Review Comment:
**Finding 2.** This filter is not just narrowing to Spark pods, it is what
keeps the collector from throwing, and that is worth a clause of comment since
the next reader may take it for a tidy-up and inline it into the `groupingBy`
classifier.
`Collectors.groupingBy` throws `NullPointerException` when the classifier
returns null, so a pod in the stream without `spark-role` would fail the
reconciliation rather than be ignored. Such a pod is reachable on the cluster
side: `SparkClusterReconciler.prepareEventSources` selects the pod informer on
`LABEL_SPARK_CLUSTER_NAME` alone, with no operator-name term, so anything
carrying that one label is a secondary resource of the cluster. The application
informer is narrower (`commonResourceLabelsStr()` plus the app-name mapper),
but the helper is shared.
```java
// Not only a narrowing to the Spark pods: `groupingBy`
throws on a null key, and
// the cluster pod informer selects on the cluster name
label alone.
.filter(pod ->
pod.getMetadata().getLabels().containsKey(LABEL_SPARK_ROLE_NAME))
```
--
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]
