TQJADE commented on code in PR #909:
URL:
https://github.com/apache/spark-kubernetes-operator/pull/909#discussion_r4160683728
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterInitStep.java:
##########
@@ -221,19 +225,22 @@ private ReconcileProgress failScheduling(
*
* @param context The SparkClusterContext for the cluster.
* @param cluster The SparkCluster.
+ * @param masterRequested Whether the master was found requested already in
this reconcile, e.g.
+ * by the suspend check, so that it is not looked up again.
* @return The progress to return while the admission is not granted, or
empty to proceed.
*/
private Optional<ReconcileProgress> holdForKueueAdmission(
- SparkClusterContext context, SparkCluster cluster) {
+ SparkClusterContext context, SparkCluster cluster, boolean
masterRequested) {
try {
if (!KueueWorkloadFactory.hasQueueName(cluster)) {
- return KueueWorkloadUtils.handleDequeuedWorkload(context, () ->
isMasterRequested(context));
+ return KueueWorkloadUtils.handleDequeuedWorkload(
+ context, () -> masterRequested || isMasterRequested(context));
Review Comment:
moot after P1
##########
spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterInitStepTest.java:
##########
@@ -405,6 +408,61 @@ void suspendAfterMasterRequestedCompletesInitialization() {
verify(statefulSetApplicable, times(2)).serverSideApply();
}
+ @Test
+ @SuppressWarnings("unchecked")
+ void suspendedQueuedClusterWithRequestedMasterIsLookedUpOnce() {
+ // The suspend check already found the master, so the Kueue admission must
not read it again:
+ // a failure of that second read would be reported as
KueueAdmissionRequestFailed rather than
+ // SuspendCheckFailed, and a null would request the admission of a running
master
+ ClusterInitStep clusterInitStep = new ClusterInitStep();
+ SparkClusterContext mockContext = mock(SparkClusterContext.class);
+ SparkClusterStatusRecorder recorder =
mock(SparkClusterStatusRecorder.class);
+ SparkCluster cluster = buildKueueCluster();
+ cluster.getSpec().setSuspend(true);
+ KubernetesClient mockClient = mock(KubernetesClient.class,
RETURNS_DEEP_STUBS);
+ when(mockClient.resource(masterStatefulSetSpec).get())
+ .thenReturn(masterStatefulSetSpec)
+ .thenThrow(new KubernetesClientException("forbidden", 403, null));
+ stubWorkloadRead(mockClient, admittedWorkload(cluster, Map.of()));
+ ServerSideApplicable<Service> serviceApplicable =
mock(ServerSideApplicable.class);
+ ServiceResource<Service> serviceResource = mock(ServiceResource.class);
+ when(serviceResource.forceConflicts()).thenReturn(serviceApplicable);
+
when(mockClient.services().resource(any(Service.class))).thenReturn(serviceResource);
+ ServerSideApplicable<StatefulSet> statefulSetApplicable =
mock(ServerSideApplicable.class);
+ RollableScalableResource<StatefulSet> statefulSetResource =
+ mock(RollableScalableResource.class);
+
when(statefulSetResource.forceConflicts()).thenReturn(statefulSetApplicable);
+ when(mockClient.apps().statefulSets().resource(any(StatefulSet.class)))
+ .thenReturn(statefulSetResource);
+ ServerSideApplicable<NetworkPolicy> networkPolicyApplicable =
mock(ServerSideApplicable.class);
+ Resource<NetworkPolicy> networkPolicyResource = mock(Resource.class);
+
when(networkPolicyResource.forceConflicts()).thenReturn(networkPolicyApplicable);
+
when(mockClient.network().networkPolicies().resource(any(NetworkPolicy.class)))
+ .thenReturn(networkPolicyResource);
+ when(mockContext.getResource()).thenReturn(cluster);
+ when(mockContext.getClient()).thenReturn(mockClient);
+
when(mockContext.getMasterServiceSpec()).thenReturn(service("cluster1-master-svc"));
+
when(mockContext.getWorkerServiceSpec()).thenReturn(service("cluster1-worker-svc"));
+
when(mockContext.getMasterStatefulSetSpec()).thenReturn(masterStatefulSetSpec);
+
when(mockContext.getWorkerStatefulSetSpec()).thenReturn(workerStatefulSetSpec);
+
when(mockContext.getWorkerNetworkPolicySpec()).thenReturn(networkPolicy("cluster1-worker"));
+
when(mockContext.getHorizontalPodAutoscalerSpec()).thenReturn(Optional.empty());
+
when(mockContext.getPodDisruptionBudgetSpec()).thenReturn(Optional.empty());
+ when(mockContext.getEventRecorder()).thenReturn(eventRecorder);
+
+ ReconcileProgress progress = clusterInitStep.reconcile(mockContext,
recorder);
+
+ Assertions.assertEquals(ReconcileProgress.completeAndDefaultRequeue(),
progress);
+ verify(mockClient.resource(masterStatefulSetSpec)).get();
+ verify(mockClient, never()).resource(any(Workload.class));
Review Comment:
moot after P1
--
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]
