// Copyright 2025 Alibaba Group Holding Ltd. // // Licensed 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 controller import ( "context" "fmt" "time" corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/types" "k8s.io/client-go/util/retry" "k8s.io/utils/ptr" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" logf "sigs.k8s.io/controller-runtime/pkg/log" sandboxv1alpha1 "github.com/alibaba/OpenSandbox/sandbox-k8s/apis/sandbox/v1alpha1" "github.com/alibaba/OpenSandbox/sandbox-k8s/internal/controller/strategy" "github.com/alibaba/OpenSandbox/sandbox-k8s/internal/utils" controllerutils "github.com/alibaba/OpenSandbox/sandbox-k8s/internal/utils/controller" ) const internalPauseSnapshotSuffix = "-pause" const supportedPauseReplicas int32 = 1 func internalPauseSnapshotName(batchSandboxName string) string { // TODO: handle Kubernetes resource name length limits for long BatchSandbox names. return batchSandboxName + internalPauseSnapshotSuffix } func unsupportedPauseReplicasMessage(replicas *int32) string { if replicas == nil { return "pause/resume currently supports only BatchSandbox spec.replicas=1; spec.replicas is unset" } return fmt.Sprintf("pause/resume currently supports only BatchSandbox spec.replicas=1; got spec.replicas=%d", *replicas) } func ensureImagePullSecret(template *corev1.PodTemplateSpec, secretName string) { if template == nil || secretName == "" { return } for _, secret := range template.Spec.ImagePullSecrets { if secret.Name == secretName { return } } template.Spec.ImagePullSecrets = append(template.Spec.ImagePullSecrets, corev1.LocalObjectReference{Name: secretName}) } func snapshotFailureMessage(snapshot *sandboxv1alpha1.SandboxSnapshot) string { for _, cond := range snapshot.Status.Conditions { if cond.Type == sandboxv1alpha1.SandboxSnapshotConditionFailed && cond.Status == sandboxv1alpha1.ConditionTrue && cond.Message != "" { return cond.Message } } return "snapshot failed" } func copyPodTemplateMap(values map[string]string) map[string]string { if len(values) == 0 { return nil } copied := make(map[string]string, len(values)) for key, value := range values { copied[key] = value } if len(copied) == 0 { return nil } return copied } func sourcePodTemplateForPause(pod *corev1.Pod) *corev1.PodTemplateSpec { if pod == nil { return nil } labels := copyPodTemplateMap(pod.Labels) delete(labels, LabelPoolName) delete(labels, LabelPoolRevision) delete(labels, LabelBatchSandboxNameKey) delete(labels, LabelBatchSandboxPodIndexKey) spec := *pod.Spec.DeepCopy() spec.NodeName = "" return &corev1.PodTemplateSpec{ ObjectMeta: metav1.ObjectMeta{ Labels: labels, Annotations: copyPodTemplateMap(pod.Annotations), }, Spec: spec, } } func (r *BatchSandboxReconciler) deleteInternalPauseSnapshot(ctx context.Context, bs *sandboxv1alpha1.BatchSandbox) error { log := logf.FromContext(ctx) snapshot := &sandboxv1alpha1.SandboxSnapshot{} if err := r.Get(ctx, types.NamespacedName{Namespace: bs.Namespace, Name: internalPauseSnapshotName(bs.Name)}, snapshot); err != nil { if errors.IsNotFound(err) { return nil } return err } if snapshot.Status.Phase != sandboxv1alpha1.SandboxSnapshotPhaseSucceed { return nil } if err := r.Delete(ctx, snapshot); err != nil && !errors.IsNotFound(err) { return err } log.Info("Deleted SandboxSnapshot after successful resume", "snapshot", snapshot.Name) return nil } func (r *BatchSandboxReconciler) hasReadyResumePod(ctx context.Context, bs *sandboxv1alpha1.BatchSandbox) (bool, error) { poolStrategy := strategy.NewPoolStrategy(bs) if poolStrategy.IsPooledMode() { alloc, err := parseSandboxAllocation(bs) if err != nil { return false, err } for _, podName := range alloc.Pods { pod := &corev1.Pod{} if err := r.Get(ctx, types.NamespacedName{Namespace: bs.Namespace, Name: podName}, pod); err != nil { if errors.IsNotFound(err) { continue } return false, err } if utils.IsPodReady(pod) { return true, nil } } } podList := &corev1.PodList{} if err := r.List(ctx, podList, client.InNamespace(bs.Namespace), client.MatchingLabels{LabelBatchSandboxNameKey: bs.Name}, ); err != nil { return false, err } for i := range podList.Items { if utils.IsPodReady(&podList.Items[i]) { return true, nil } } pod := &corev1.Pod{} if err := r.Get(ctx, types.NamespacedName{Namespace: bs.Namespace, Name: fmt.Sprintf("%s-%d", bs.Name, batchSandboxFirstPodIndex)}, pod); err != nil { if errors.IsNotFound(err) { return false, nil } return false, err } return utils.IsPodReady(pod), nil } // dispatchPauseResume implements the 5-case dispatch table from the design doc. // Returns (result, handled, error). If handled=true, the caller should return immediately. func (r *BatchSandboxReconciler) dispatchPauseResume(ctx context.Context, bs *sandboxv1alpha1.BatchSandbox) (ctrl.Result, bool, error) { log := logf.FromContext(ctx) generation := bs.Generation pauseObservedGen := bs.Status.PauseObservedGeneration pause := bs.Spec.Pause // In-progress phases take priority over generation checks so that controller-owned // spec mutations (template solidification, image replacement, pool detachment) do not // get mistaken for new external pause/resume requests. log.Info("Dispatch: checking phase", "currentPhase", bs.Status.Phase, "generation", generation, "pauseObservedGen", pauseObservedGen, "pause", pause) if bs.Status.Phase == sandboxv1alpha1.BatchSandboxPhaseResuming { log.Info("Dispatch: phase is Resuming, continuing resume") result, err := r.continueResume(ctx, bs) if err != nil { return result, true, err } // Return handled=false to let normal flow update phase from Resuming to Succeed. return result, false, nil } if bs.Status.Phase == sandboxv1alpha1.BatchSandboxPhasePausing { log.Info("Dispatch: phase is Pausing, syncing pause state") result, err := r.syncPauseOrClear(ctx, bs) return result, true, err } if generation > pauseObservedGen { if pause != nil { if *pause { log.Info("Dispatch: handlePause", "generation", generation, "pauseObservedGeneration", pauseObservedGen) result, err := r.handlePause(ctx, bs) return result, true, err } log.Info("Dispatch: handleResume", "generation", generation, "pauseObservedGeneration", pauseObservedGen, "currentPhase", bs.Status.Phase) result, err := r.handleResume(ctx, bs) return result, true, err } // No pause intent — skip the dedicated ACK API call. The normal flow's // persistRuntimeView will update PauseObservedGeneration in its status patch. log.Info("Dispatch: no pause intent, deferring ACK to status patch", "generation", generation, "pauseObservedGeneration", pauseObservedGen) return ctrl.Result{}, false, nil } return ctrl.Result{}, false, nil } // handlePause implements the pause flow: // 1. ACK (pauseObservedGeneration + phase=Pausing) // 2. Stop task-executor tasks, if any, while keeping the source Pod allocated // 3. Create SandboxSnapshot child resource func (r *BatchSandboxReconciler) handlePause(ctx context.Context, bs *sandboxv1alpha1.BatchSandbox) (ctrl.Result, error) { log := logf.FromContext(ctx) // The pause snapshot format records one source pod's container images, so this // controller only supports rootfs pause/resume for single-replica sandboxes. if bs.Spec.Replicas == nil || *bs.Spec.Replicas != supportedPauseReplicas { msg := unsupportedPauseReplicasMessage(bs.Spec.Replicas) log.Info("Rejecting pause for unsupported replica count", "message", msg) phase := bs.Status.Phase if phase == "" { phase = sandboxv1alpha1.BatchSandboxPhaseSucceed } if err := r.ackPauseWithPhase(ctx, bs, phase, ""); err != nil { return ctrl.Result{}, err } if err := r.setCondition(ctx, bs, sandboxv1alpha1.BatchSandboxConditionPauseFailed, sandboxv1alpha1.ConditionTrue, "UnsupportedReplicas", msg); err != nil { return ctrl.Result{}, err } return ctrl.Result{}, nil } snapshot := &sandboxv1alpha1.SandboxSnapshot{} snapshotName := internalPauseSnapshotName(bs.Name) err := r.Get(ctx, types.NamespacedName{Namespace: bs.Namespace, Name: snapshotName}, snapshot) if err == nil && snapshot.DeletionTimestamp != nil { log.Info("Waiting for stale SandboxSnapshot deletion before retrying pause", "snapshot", snapshotName) return ctrl.Result{RequeueAfter: time.Second}, nil } if err == nil && (snapshot.Status.Phase == sandboxv1alpha1.SandboxSnapshotPhaseFailed || snapshot.Status.Phase == sandboxv1alpha1.SandboxSnapshotPhaseSucceed) { log.Info("Deleting terminal SandboxSnapshot for retry", "snapshot", snapshotName, "phase", snapshot.Status.Phase) if err := r.Delete(ctx, snapshot); err != nil && !errors.IsNotFound(err) { return ctrl.Result{}, err } return ctrl.Result{RequeueAfter: time.Second}, nil } if err != nil && !errors.IsNotFound(err) { return ctrl.Result{}, err } _ = r.setCondition(ctx, bs, sandboxv1alpha1.BatchSandboxConditionPauseFailed, sandboxv1alpha1.ConditionFalse, "", "") if err := r.ackPauseWithPhase(ctx, bs, sandboxv1alpha1.BatchSandboxPhasePausing, ""); err != nil { return ctrl.Result{}, err } if errors.IsNotFound(err) { if created, err := r.ensureInternalPauseSnapshot(ctx, bs, snapshotName); err != nil { return ctrl.Result{}, err } else if !created { log.Info("Waiting for task cleanup before creating SandboxSnapshot", "snapshot", snapshotName) } } return ctrl.Result{RequeueAfter: time.Second}, nil } func (r *BatchSandboxReconciler) ensureInternalPauseSnapshot(ctx context.Context, bs *sandboxv1alpha1.BatchSandbox, snapshotName string) (bool, error) { log := logf.FromContext(ctx) tasksStopped, err := r.stopTasksBeforePause(ctx, bs) if err != nil { return false, err } if !tasksStopped { return false, nil } snapshot := &sandboxv1alpha1.SandboxSnapshot{ ObjectMeta: metav1.ObjectMeta{ Name: snapshotName, Namespace: bs.Namespace, }, Spec: sandboxv1alpha1.SandboxSnapshotSpec{ SandboxName: bs.Name, }, } if err := controllerutil.SetControllerReference(bs, snapshot, r.Scheme); err != nil { return false, err } if err := r.Create(ctx, snapshot); err != nil { if errors.IsAlreadyExists(err) { return true, nil } return false, err } log.Info("Created SandboxSnapshot", "snapshot", snapshotName) return true, nil } func (r *BatchSandboxReconciler) stopTasksBeforePause(ctx context.Context, bs *sandboxv1alpha1.BatchSandbox) (bool, error) { log := logf.FromContext(ctx) taskStrategy := strategy.NewTaskSchedulingStrategy(bs) if !taskStrategy.NeedTaskScheduling() { return true, nil } poolStrategy := strategy.NewPoolStrategy(bs) pods, err := r.listPods(ctx, poolStrategy, bs) if err != nil { return false, err } sch, err := r.getTaskScheduler(ctx, bs, pods) if err != nil { return false, err } stoppingTasks := sch.StopTask() if len(stoppingTasks) > 0 { log.Info("Stopping tasks before pause", "count", len(stoppingTasks)) } if err := sch.Schedule(); err != nil { return false, fmt.Errorf("failed to stop tasks before pause: %w", err) } unfinishedTasks := r.getTasksCleanupUnfinished(bs, sch) if len(unfinishedTasks) > 0 { log.Info("Task cleanup before pause is unfinished", "unfinishedCount", len(unfinishedTasks)) return false, nil } log.Info("Task cleanup before pause is finished") return true, nil } // handleResume implements the resume flow: // 1. ACK (pauseObservedGeneration + phase=Resuming) // 2. Requeue so a subsequent pass can consume the snapshot result func (r *BatchSandboxReconciler) handleResume(ctx context.Context, bs *sandboxv1alpha1.BatchSandbox) (ctrl.Result, error) { log := logf.FromContext(ctx) _ = r.setCondition(ctx, bs, sandboxv1alpha1.BatchSandboxConditionResumeFailed, sandboxv1alpha1.ConditionFalse, "", "") log.Info("ACK Resuming phase") if err := r.ackPauseWithPhase(ctx, bs, sandboxv1alpha1.BatchSandboxPhaseResuming, ""); err != nil { return ctrl.Result{}, err } return ctrl.Result{RequeueAfter: time.Second}, nil } // syncPauseOrClear waits for the internal pause snapshot to finish and transitions the // BatchSandbox into Paused or a retryable/terminal failure state. func (r *BatchSandboxReconciler) syncPauseOrClear(ctx context.Context, bs *sandboxv1alpha1.BatchSandbox) (ctrl.Result, error) { log := logf.FromContext(ctx) snapshot := &sandboxv1alpha1.SandboxSnapshot{} err := r.Get(ctx, types.NamespacedName{Namespace: bs.Namespace, Name: internalPauseSnapshotName(bs.Name)}, snapshot) if errors.IsNotFound(err) { snapshotName := internalPauseSnapshotName(bs.Name) if created, createErr := r.ensureInternalPauseSnapshot(ctx, bs, snapshotName); createErr != nil { return ctrl.Result{}, createErr } else if !created { log.Info("SandboxSnapshot not created yet; waiting for task cleanup", "snapshot", snapshotName) } return ctrl.Result{RequeueAfter: time.Second}, nil } if err != nil { return ctrl.Result{}, err } switch snapshot.Status.Phase { case sandboxv1alpha1.SandboxSnapshotPhaseSucceed: log.Info("SandboxSnapshot Succeed, completing pause") if err := r.completePause(ctx, bs); err != nil { return ctrl.Result{}, err } return ctrl.Result{RequeueAfter: time.Second}, nil case sandboxv1alpha1.SandboxSnapshotPhaseFailed: msg := snapshotFailureMessage(snapshot) log.Info("SandboxSnapshot Failed", "message", msg) phase := sandboxv1alpha1.BatchSandboxPhaseSucceed reason := "SnapshotFailed" if _, podErr := r.findPodForSandbox(ctx, bs); podErr != nil { phase = sandboxv1alpha1.BatchSandboxPhaseFailed reason = "PodNotFound" } if err := r.ackPauseWithPhase(ctx, bs, phase, ""); err != nil { return ctrl.Result{}, err } if err := r.setCondition(ctx, bs, sandboxv1alpha1.BatchSandboxConditionPauseFailed, sandboxv1alpha1.ConditionTrue, reason, msg); err != nil { return ctrl.Result{}, err } return ctrl.Result{}, nil case sandboxv1alpha1.SandboxSnapshotPhasePending, sandboxv1alpha1.SandboxSnapshotPhaseCommitting: log.Info("SandboxSnapshot in progress", "phase", snapshot.Status.Phase) return ctrl.Result{RequeueAfter: time.Second}, nil default: return ctrl.Result{RequeueAfter: time.Second}, nil } } // completePause finalizes the pause operation: // 1. Normal mode: delete all Pods (cascade via OwnerRef) // 2. Pool mode: solidify the source Pod template and clear poolRef. Pool Controller GC // then releases and deletes the old allocated pool Pod. // 3. Set phase=Paused only after pod deletion/release is accepted. // 4. spec.pause remains unchanged; the next external request (or server retry bridge) // is responsible for changing it. func (r *BatchSandboxReconciler) completePause(ctx context.Context, bs *sandboxv1alpha1.BatchSandbox) error { log := logf.FromContext(ctx) poolStrategy := strategy.NewPoolStrategy(bs) wasPooled := poolStrategy.IsPooledMode() pods, err := r.listPods(ctx, poolStrategy, bs) if err != nil { return err } var pooledTemplate *corev1.PodTemplateSpec if wasPooled { if len(pods) == 0 { return fmt.Errorf("no allocated pods found for pooled BatchSandbox %s/%s", bs.Namespace, bs.Name) } pooledTemplate = sourcePodTemplateForPause(pods[0]) if err := retry.RetryOnConflict(retry.DefaultBackoff, func() error { latest := &sandboxv1alpha1.BatchSandbox{} if err := r.Get(ctx, types.NamespacedName{Namespace: bs.Namespace, Name: bs.Name}, latest); err != nil { return err } patch := client.MergeFrom(latest.DeepCopy()) latest.Spec.Template = pooledTemplate.DeepCopy() latest.Spec.PoolRef = "" controllerutil.RemoveFinalizer(latest, FinalizerPoolAllocation) if latest.Annotations != nil { delete(latest.Annotations, AnnoAllocReleaseKey) } return r.Patch(ctx, latest, patch) }); err != nil { return err } bs.Spec.Template = pooledTemplate.DeepCopy() bs.Spec.PoolRef = "" controllerutil.RemoveFinalizer(bs, FinalizerPoolAllocation) if bs.Annotations != nil { delete(bs.Annotations, AnnoAllocReleaseKey) } log.Info("Detached pooled BatchSandbox after pause", "sourcePod", pods[0].Name) } controllerKey := controllerutils.GetControllerKey(bs) BatchSandboxScaleExpectations.DeleteExpectations(controllerKey) log.Info("Cleared scale expectations before pod deletion", "controllerKey", controllerKey) if !wasPooled { for _, pod := range pods { if err := r.Delete(ctx, pod); err != nil && !errors.IsNotFound(err) { log.Error(err, "Failed to delete pod during pause", "pod", pod.Name) return err } log.Info("Deleted pod during pause", "pod", pod.Name) } } r.deleteTaskScheduler(ctx, bs) var latest *sandboxv1alpha1.BatchSandbox if err := retry.RetryOnConflict(retry.DefaultBackoff, func() error { latest = &sandboxv1alpha1.BatchSandbox{} if err := r.Get(ctx, types.NamespacedName{Namespace: bs.Namespace, Name: bs.Name}, latest); err != nil { return err } latest.Status.Phase = sandboxv1alpha1.BatchSandboxPhasePaused // Pool pause rewrites spec.template and clears spec.poolRef, which advances generation. // Acknowledge that controller-owned spec change only while pause is still requested, // so spec.pause=true is not replayed and a queued resume request is still observed. if latest.Spec.Pause != nil && *latest.Spec.Pause { latest.Status.PauseObservedGeneration = latest.Generation } applyBatchSandboxPhaseConditions(&latest.Status) return r.Status().Update(ctx, latest) }); err != nil { return err } r.StatusRVExpectation.Expect(latest) return nil } // continueResume continues the resume flow: // 1. Read SandboxSnapshot status for image URIs // 2. Replace template container images // 3. Pool mode: clear poolRef // 4. Leave spec.pause and spec.replicas untouched; normal reconciliation recreates // the single supported runtime replica from the rewritten template. func (r *BatchSandboxReconciler) continueResume(ctx context.Context, bs *sandboxv1alpha1.BatchSandbox) (ctrl.Result, error) { log := logf.FromContext(ctx) snapshot := &sandboxv1alpha1.SandboxSnapshot{} if err := r.Get(ctx, types.NamespacedName{Namespace: bs.Namespace, Name: internalPauseSnapshotName(bs.Name)}, snapshot); err != nil { if errors.IsNotFound(err) { readyPodExists, readyErr := r.hasReadyResumePod(ctx, bs) if readyErr != nil { return ctrl.Result{}, readyErr } if readyPodExists { log.Info("SandboxSnapshot missing for resume, but a ready pod already exists; treating resume as complete") _ = r.ackPauseWithPhase(ctx, bs, sandboxv1alpha1.BatchSandboxPhaseSucceed, "") _ = r.setCondition(ctx, bs, sandboxv1alpha1.BatchSandboxConditionResumeFailed, sandboxv1alpha1.ConditionFalse, "", "") _ = r.setCondition(ctx, bs, sandboxv1alpha1.BatchSandboxConditionPodFailed, sandboxv1alpha1.ConditionFalse, "", "") return ctrl.Result{}, nil } log.Info("SandboxSnapshot not found for resume, rolling back to Paused") _ = r.ackPauseWithPhase(ctx, bs, sandboxv1alpha1.BatchSandboxPhasePaused, "") _ = r.setCondition(ctx, bs, sandboxv1alpha1.BatchSandboxConditionResumeFailed, sandboxv1alpha1.ConditionTrue, "SnapshotNotFound", "SandboxSnapshot not found") return ctrl.Result{}, nil } return ctrl.Result{}, err } if snapshot.Status.Phase != sandboxv1alpha1.SandboxSnapshotPhaseSucceed { msg := fmt.Sprintf("snapshot not ready: phase=%s", snapshot.Status.Phase) log.Error(nil, msg) _ = r.ackPauseWithPhase(ctx, bs, sandboxv1alpha1.BatchSandboxPhasePaused, "") _ = r.setCondition(ctx, bs, sandboxv1alpha1.BatchSandboxConditionResumeFailed, sandboxv1alpha1.ConditionTrue, "SnapshotNotReady", msg) return ctrl.Result{}, nil } imageMap := make(map[string]string) for _, c := range snapshot.Status.Containers { imageMap[c.ContainerName] = c.ImageURI } var patched *sandboxv1alpha1.BatchSandbox if err := retry.RetryOnConflict(retry.DefaultBackoff, func() error { latest := &sandboxv1alpha1.BatchSandbox{} if err := r.Get(ctx, types.NamespacedName{Namespace: bs.Namespace, Name: bs.Name}, latest); err != nil { return err } patch := client.MergeFrom(latest.DeepCopy()) if latest.Spec.Template != nil { for i := range latest.Spec.Template.Spec.Containers { if img, ok := imageMap[latest.Spec.Template.Spec.Containers[i].Name]; ok { latest.Spec.Template.Spec.Containers[i].Image = img } } ensureImagePullSecret(latest.Spec.Template, r.ResumePullSecret) } if latest.Spec.PoolRef != "" { latest.Spec.PoolRef = "" controllerutil.RemoveFinalizer(latest, FinalizerPoolAllocation) } if err := r.Patch(ctx, latest, patch); err != nil { return err } patched = latest.DeepCopy() return nil }); err != nil { return ctrl.Result{}, err } if patched != nil { bs.ObjectMeta = patched.ObjectMeta bs.Spec = patched.Spec } return ctrl.Result{RequeueAfter: time.Second}, nil } func (r *BatchSandboxReconciler) ackPauseGeneration(ctx context.Context, bs *sandboxv1alpha1.BatchSandbox) error { var latest *sandboxv1alpha1.BatchSandbox if err := retry.RetryOnConflict(retry.DefaultBackoff, func() error { latest = &sandboxv1alpha1.BatchSandbox{} if err := r.Get(ctx, types.NamespacedName{Namespace: bs.Namespace, Name: bs.Name}, latest); err != nil { return err } latest.Status.PauseObservedGeneration = latest.Generation applyBatchSandboxPhaseConditions(&latest.Status) return r.Status().Update(ctx, latest) }); err != nil { return err } r.StatusRVExpectation.Expect(latest) bs.Status.PauseObservedGeneration = bs.Generation applyBatchSandboxPhaseConditions(&bs.Status) return nil } func (r *BatchSandboxReconciler) ackPauseWithPhase(ctx context.Context, bs *sandboxv1alpha1.BatchSandbox, phase sandboxv1alpha1.BatchSandboxPhase, _ string) error { var latest *sandboxv1alpha1.BatchSandbox if err := retry.RetryOnConflict(retry.DefaultBackoff, func() error { latest = &sandboxv1alpha1.BatchSandbox{} if err := r.Get(ctx, types.NamespacedName{Namespace: bs.Namespace, Name: bs.Name}, latest); err != nil { return err } latest.Status.PauseObservedGeneration = latest.Generation latest.Status.Phase = phase applyBatchSandboxPhaseConditions(&latest.Status) return r.Status().Update(ctx, latest) }); err != nil { return err } r.StatusRVExpectation.Expect(latest) bs.Status.PauseObservedGeneration = bs.Generation bs.Status.Phase = phase applyBatchSandboxPhaseConditions(&bs.Status) return nil } // findPodForSandbox finds the Pod associated with a BatchSandbox. func (r *BatchSandboxReconciler) findPodForSandbox(ctx context.Context, bs *sandboxv1alpha1.BatchSandbox) (*corev1.Pod, error) { poolStrategy := strategy.NewPoolStrategy(bs) pods, err := r.listPods(ctx, poolStrategy, bs) if err != nil { return nil, err } if len(pods) == 0 { return nil, fmt.Errorf("no pods found for BatchSandbox %s/%s", bs.Namespace, bs.Name) } return pods[0], nil } func (r *BatchSandboxReconciler) setCondition( ctx context.Context, bs *sandboxv1alpha1.BatchSandbox, conditionType sandboxv1alpha1.BatchSandboxConditionType, status string, reason string, message string, ) error { var latest *sandboxv1alpha1.BatchSandbox if err := retry.RetryOnConflict(retry.DefaultBackoff, func() error { latest = &sandboxv1alpha1.BatchSandbox{} if err := r.Get(ctx, types.NamespacedName{Namespace: bs.Namespace, Name: bs.Name}, latest); err != nil { return err } var conditions []sandboxv1alpha1.BatchSandboxCondition found := false for _, c := range latest.Status.Conditions { if c.Type == conditionType { if status == sandboxv1alpha1.ConditionFalse { continue } c.Status = status c.Reason = reason c.Message = message c.LastTransitionTime = ptr.To(metav1.Now()) found = true } conditions = append(conditions, c) } if !found && status == sandboxv1alpha1.ConditionTrue { conditions = append(conditions, sandboxv1alpha1.BatchSandboxCondition{ Type: conditionType, Status: status, Reason: reason, Message: message, LastTransitionTime: ptr.To(metav1.Now()), }) } latest.Status.Conditions = conditions return r.Status().Update(ctx, latest) }); err != nil { return err } r.StatusRVExpectation.Expect(latest) return nil }