diff --git a/cocoonset/agents.go b/cocoonset/agents.go index dd30288..79a9dcd 100644 --- a/cocoonset/agents.go +++ b/cocoonset/agents.go @@ -169,7 +169,7 @@ func (r *Reconciler) rebuildSubAgent(ctx context.Context, logger *log.Fields, po history := readRebuildHistory(cs) entry := history[slot] if entry.Count >= maxRebuildAttempts { - if err := r.patchPodAnnotation(ctx, pod, annotationDeadLetter, "true"); err != nil { + if err := r.patchAnnotation(ctx, pod, annotationDeadLetter, "true"); err != nil { return false, 0, err } metrics.SubAgentDeadLetterTotal.WithLabelValues(cs.Namespace, cs.Name).Inc() @@ -203,13 +203,18 @@ func (r *Reconciler) rebuildSubAgent(ctx context.Context, logger *log.Fields, po return true, 0, nil } -func (r *Reconciler) patchPodAnnotation(ctx context.Context, pod *corev1.Pod, key, value string) error { - patch, err := commonk8s.AnnotationsMergePatch(map[string]any{key: value}) +// patchAnnotation merge-patches one annotation on obj; an empty value deletes the key. +func (r *Reconciler) patchAnnotation(ctx context.Context, obj client.Object, key, value string) error { + var v any = value + if value == "" { + v = nil + } + patch, err := commonk8s.AnnotationsMergePatch(map[string]any{key: v}) if err != nil { - return fmt.Errorf("build patch for pod %s/%s annotation %s: %w", pod.Namespace, pod.Name, key, err) + return fmt.Errorf("build patch for %T %s/%s annotation %s: %w", obj, obj.GetNamespace(), obj.GetName(), key, err) } - if err := r.Patch(ctx, pod, client.RawPatch(types.MergePatchType, patch)); err != nil { - return fmt.Errorf("patch pod %s/%s annotation %s: %w", pod.Namespace, pod.Name, key, err) + if err := r.Patch(ctx, obj, client.RawPatch(types.MergePatchType, patch)); err != nil { + return fmt.Errorf("patch %T %s/%s annotation %s: %w", obj, obj.GetNamespace(), obj.GetName(), key, err) } return nil } diff --git a/cocoonset/pods.go b/cocoonset/pods.go index eec3c72..74829bb 100644 --- a/cocoonset/pods.go +++ b/cocoonset/pods.go @@ -116,18 +116,33 @@ func hostnameAffinity(nodeName string) *corev1.Affinity { return &corev1.Affinity{ NodeAffinity: &corev1.NodeAffinity{ RequiredDuringSchedulingIgnoredDuringExecution: &corev1.NodeSelector{ - NodeSelectorTerms: []corev1.NodeSelectorTerm{{ - MatchExpressions: []corev1.NodeSelectorRequirement{{ - Key: corev1.LabelHostname, - Operator: corev1.NodeSelectorOpIn, - Values: []string{nodeName}, - }}, - }}, + NodeSelectorTerms: []corev1.NodeSelectorTerm{hostnameSelectorTerm(nodeName)}, }, }, } } +func preferredHostnameAffinity(nodeName string) *corev1.Affinity { + return &corev1.Affinity{ + NodeAffinity: &corev1.NodeAffinity{ + PreferredDuringSchedulingIgnoredDuringExecution: []corev1.PreferredSchedulingTerm{{ + Weight: 100, + Preference: hostnameSelectorTerm(nodeName), + }}, + }, + } +} + +func hostnameSelectorTerm(nodeName string) corev1.NodeSelectorTerm { + return corev1.NodeSelectorTerm{ + MatchExpressions: []corev1.NodeSelectorRequirement{{ + Key: corev1.LabelHostname, + Operator: corev1.NodeSelectorOpIn, + Values: []string{nodeName}, + }}, + } +} + func buildToolboxPod(cs *cocoonv1.CocoonSet, tb cocoonv1.ToolboxSpec, scheme *runtime.Scheme) (*corev1.Pod, error) { podName := toolboxPodName(cs.Name, tb.Name) vmName := meta.VMNameForPod(cs.Namespace, podName) diff --git a/cocoonset/reconciler.go b/cocoonset/reconciler.go index 11a6fb4..28e584d 100644 --- a/cocoonset/reconciler.go +++ b/cocoonset/reconciler.go @@ -35,9 +35,10 @@ const ( // and toolbox pods to match the declared spec. type Reconciler struct { client.Client - Scheme *runtime.Scheme - Registry snapshot.Registry - Recorder record.EventRecorder + APIReader client.Reader + Scheme *runtime.Scheme + Registry snapshot.Registry + Recorder record.EventRecorder // Concurrency caps in-flight reconciles; at 1 one slow registry probe // stalls every other CocoonSet. Concurrency int @@ -105,6 +106,9 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu } if cs.Spec.Suspend { + if cs.Spec.HibernatePolicy.Default() == cocoonv1.HibernatePolicyRelease { + return r.reconcileSuspendRelease(ctx, &cs, classified) + } return r.reconcileSuspend(ctx, &cs, classified) } @@ -114,6 +118,12 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu return res, err } + // Released-seat wake runs before createMainAgent, whose restore intent + // only covers CocoonHibernation CRs and would fresh-boot over the snapshot. + if handled, res, err := r.reconcileWake(ctx, &cs, classified); handled { + return res, err + } + if err := r.applyUnsuspend(ctx, cs.Namespace, classified); err != nil { return ctrl.Result{}, err } diff --git a/cocoonset/slotrelease.go b/cocoonset/slotrelease.go new file mode 100644 index 0000000..80a7a25 --- /dev/null +++ b/cocoonset/slotrelease.go @@ -0,0 +1,214 @@ +package cocoonset + +import ( + "context" + "fmt" + "maps" + "slices" + + "github.com/projecteru2/core/log" + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + + cocoonv1 "github.com/cocoonstack/cocoon-common/apis/v1" + "github.com/cocoonstack/cocoon-common/meta" + "github.com/cocoonstack/cocoon-operator/metrics" + "github.com/cocoonstack/cocoon-operator/snapshot" +) + +// reconcileSuspendRelease drives a suspended release-policy CocoonSet to zero +// pods so its scheduling seat frees; pods are deleted only after every managed +// VM's :hibernate snapshot is verified in the registry. +func (r *Reconciler) reconcileSuspendRelease(ctx context.Context, cs *cocoonv1.CocoonSet, classified classifiedPods) (ctrl.Result, error) { + logger := log.WithFunc("cocoonset.Reconciler.reconcileSuspendRelease") + + if r.Registry == nil { + return ctrl.Result{}, fmt.Errorf("hibernatePolicy=release on %s/%s requires a configured registry", cs.Namespace, cs.Name) + } + + if len(classified.allByName) == 0 { + // Seat already released, or suspended before first boot — nothing to + // snapshot; settle Suspended. + return ctrl.Result{}, r.patchStatus(ctx, cs, buildStatus(cs, classified, cocoonv1.CocoonSetPhaseSuspended)) + } + + if err := r.applySuspend(ctx, classified); err != nil { + return ctrl.Result{}, err + } + allHibernated, err := r.allOwnedPodsHibernated(ctx, cs, classified) + if err != nil { + return ctrl.Result{}, err + } + if !allHibernated { + return ctrl.Result{RequeueAfter: requeueSuspendPoll}, + r.patchStatus(ctx, cs, buildStatus(cs, classified, cocoonv1.CocoonSetPhaseSuspending)) + } + + // Stash before the first delete: delete-time GC needs the vm names (status + // rebuilds empty once the pods are gone) and wake needs the node hint. + if err := r.stashDeleteVMNames(ctx, cs, podsSlice(classified)); err != nil { + return ctrl.Result{}, fmt.Errorf("stash vm names before slot release: %w", err) + } + if main := classified.main; main != nil && main.Spec.NodeName != "" { + if err := r.patchAnnotation(ctx, cs, meta.AnnotationHibernatedOnNode, main.Spec.NodeName); err != nil { + return ctrl.Result{}, err + } + } + + deleteErr := classified.forEachSorted(ctx, func(pod *corev1.Pod) error { + if podIsTerminal(pod) { + // Terminal pods carry no VM state; keep them for post-unsuspend triage. + return nil + } + logger.Infof(ctx, "slot release: deleting hibernated pod %s/%s (node=%s)", pod.Namespace, pod.Name, pod.Spec.NodeName) + if err := r.Delete(ctx, pod); err != nil && !apierrors.IsNotFound(err) { + return fmt.Errorf("slot release: delete pod %s/%s: %w", pod.Namespace, pod.Name, err) + } + metrics.SlotReleasePodsDeletedTotal.WithLabelValues(pod.Namespace, cs.Name).Inc() + return nil + }) + if deleteErr != nil { + return ctrl.Result{}, deleteErr + } + // Suspended lands with the deletes; wake's stale-cache arm keys on this receipt. + return ctrl.Result{RequeueAfter: requeueSuspendPoll}, + r.patchStatus(ctx, cs, buildStatus(cs, classified, cocoonv1.CocoonSetPhaseSuspended)) +} + +// reconcileWake owns the reconcile of a set coming back from a pod-less +// Suspended state: durable Waking phase first, restore pod second, tag drop +// only once the VM is live. Gated on phase, not current policy — a policy +// edit made while suspended must not fresh-boot over the snapshot. +func (r *Reconciler) reconcileWake(ctx context.Context, cs *cocoonv1.CocoonSet, classified classifiedPods) (bool, ctrl.Result, error) { + logger := log.WithFunc("cocoonset.Reconciler.reconcileWake") + main := classified.main + waking := cs.Status.Phase == cocoonv1.CocoonSetPhaseWaking + suspended := cs.Status.Phase == cocoonv1.CocoonSetPhaseSuspended + // A fast unsuspend can land before the deletes settle Suspended; the node + // hint keeps the wake engaged (phase-scoped so a stale hint never fires). + suspending := cs.Status.Phase == cocoonv1.CocoonSetPhaseSuspending && + cs.Annotations[meta.AnnotationHibernatedOnNode] != "" + // A stale-phase read can overwrite Waking; a restore-marked, non-hibernating + // main with the hint set is an unfinished wake (completion clears the hint). + cleanupPending := main != nil && meta.ReadRestoreFromHibernate(main) && + !bool(meta.ReadHibernateState(main)) && cs.Annotations[meta.AnnotationHibernatedOnNode] != "" + waking = waking || cleanupPending + if (!waking && !suspended && !suspending) || r.Registry == nil { + return false, ctrl.Result{}, nil + } + + switch { + case main == nil: + return r.startReleasedWake(ctx, cs, classified) + + case waking && !vmLive(main): + // Restore in flight. Unschedulable is the out-of-stock signal; + // surface it but keep waiting — a seat may free up. + if msg := podUnschedulable(main); msg != "" { + metrics.SlotReleaseWakeUnschedulableTotal.WithLabelValues(cs.Namespace, cs.Name).Inc() + if r.Recorder != nil { + r.Recorder.Eventf(cs, corev1.EventTypeWarning, "WakeNoCapacity", "main pod %s unschedulable: %s", main.Name, msg) + } + } + return true, ctrl.Result{RequeueAfter: requeueSuspendPoll}, + r.patchStatus(ctx, cs, buildStatus(cs, classified, cocoonv1.CocoonSetPhaseWaking)) + + case waking && vmLive(main): + vmName := meta.ParseVMSpec(main).VMName + logger.Infof(ctx, "wake %s/%s: restored on %s, dropping hibernate snapshot", cs.Namespace, cs.Name, main.Spec.NodeName) + if err := r.Registry.DeleteManifest(ctx, vmName, meta.HibernateSnapshotTag); err != nil { + return true, ctrl.Result{}, fmt.Errorf("wake: drop hibernate snapshot %s: %w", vmName, err) + } + placement := "pool" + if hint := cs.Annotations[meta.AnnotationHibernatedOnNode]; hint != "" && hint == main.Spec.NodeName { + placement = "hint-node" + } + metrics.SlotReleaseWakeTotal.WithLabelValues(cs.Namespace, cs.Name, placement).Inc() + if err := r.patchAnnotation(ctx, cs, meta.AnnotationHibernatedOnNode, ""); err != nil { + return true, ctrl.Result{}, err + } + // Auto-derived phase; the requeued pass settles Running/Scaling. + return true, ctrl.Result{Requeue: true}, r.patchStatus(ctx, cs, buildStatus(cs, classified, "")) + + case (suspended || suspending) && cs.Annotations[meta.AnnotationHibernatedOnNode] != "" && !podIsTerminal(main): + // Either a stale view of the deleted main or a delete that never ran; + // only an uncached read can tell them apart. + return r.confirmReleasedDelete(ctx, main) + + default: + // Retain placeholder or kept terminal pod: the normal flow owns these. + return false, ctrl.Result{}, nil + } +} + +// startReleasedWake probes the registry and recreates the main pod with +// restore intent; handled=false only when there is no snapshot to restore. +func (r *Reconciler) startReleasedWake(ctx context.Context, cs *cocoonv1.CocoonSet, classified classifiedPods) (bool, ctrl.Result, error) { + logger := log.WithFunc("cocoonset.Reconciler.startReleasedWake") + vmName := meta.VMNameForDeployment(cs.Namespace, cs.Name, 0) + present, probeErr := snapshot.HasHibernateSnapshot(ctx, r.Registry, vmName) + if probeErr != nil { + // Fail closed: falling through would fresh-boot over a real snapshot. + return true, ctrl.Result{}, fmt.Errorf("wake: %w", probeErr) + } + if !present { + // No snapshot: suspended before first boot; the normal flow fresh-boots. + return false, ctrl.Result{}, nil + } + // Persist Waking before the create so a crash between the two steps + // resumes here instead of fresh-booting. + if err := r.patchStatus(ctx, cs, buildStatus(cs, classified, cocoonv1.CocoonSetPhaseWaking)); err != nil { + return true, ctrl.Result{}, err + } + pod, err := buildAgentPod(cs, 0, "", "", r.Scheme) + if err != nil { + return true, ctrl.Result{}, fmt.Errorf("wake: build main: %w", err) + } + meta.MarkRestoreFromHibernate(pod) + // Soft-prefer the hibernated-on seat; a spec.nodeName pin already set + // a required affinity in buildAgentPod and wins. + if node := cs.Annotations[meta.AnnotationHibernatedOnNode]; node != "" && pod.Spec.Affinity == nil { + pod.Spec.Affinity = preferredHostnameAffinity(node) + } + if err := r.Create(ctx, pod); err != nil { + if apierrors.IsAlreadyExists(err) { + // Old pod still Terminating; requeue and wait. + return true, ctrl.Result{RequeueAfter: requeueWaitForMain}, nil + } + return true, ctrl.Result{}, fmt.Errorf("wake: create main: %w", err) + } + logger.Infof(ctx, "wake %s/%s: recreated main unpinned, preferred_node=%s", cs.Namespace, cs.Name, cs.Annotations[meta.AnnotationHibernatedOnNode]) + return true, ctrl.Result{RequeueAfter: requeueWaitForMain}, nil +} + +func (r *Reconciler) confirmReleasedDelete(ctx context.Context, main *corev1.Pod) (bool, ctrl.Result, error) { + var live corev1.Pod + err := r.APIReader.Get(ctx, client.ObjectKeyFromObject(main), &live) + switch { + case apierrors.IsNotFound(err) || (err == nil && live.DeletionTimestamp != nil): + return true, ctrl.Result{RequeueAfter: requeueWaitForMain}, nil + case err != nil: + return true, ctrl.Result{}, fmt.Errorf("wake: confirm delete of %s/%s: %w", main.Namespace, main.Name, err) + default: + return false, ctrl.Result{}, nil + } +} + +func podUnschedulable(pod *corev1.Pod) string { + for _, c := range pod.Status.Conditions { + if c.Type == corev1.PodScheduled && c.Status == corev1.ConditionFalse && c.Reason == corev1.PodReasonUnschedulable { + return c.Message + } + } + return "" +} + +func podsSlice(c classifiedPods) []corev1.Pod { + out := make([]corev1.Pod, 0, len(c.allByName)) + for _, name := range slices.Sorted(maps.Keys(c.allByName)) { + out = append(out, *c.allByName[name]) + } + return out +} diff --git a/cocoonset/slotrelease_test.go b/cocoonset/slotrelease_test.go new file mode 100644 index 0000000..e9b1549 --- /dev/null +++ b/cocoonset/slotrelease_test.go @@ -0,0 +1,503 @@ +package cocoonset + +import ( + "errors" + "slices" + "strings" + "testing" + + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/client" + ctrlfake "sigs.k8s.io/controller-runtime/pkg/client/fake" + + cocoonv1 "github.com/cocoonstack/cocoon-common/apis/v1" + "github.com/cocoonstack/cocoon-common/meta" +) + +var ( + relVMName = meta.VMNameForDeployment("ns", "demo", 0) + relHibernateTagKey = relVMName + ":" + meta.HibernateSnapshotTag +) + +func TestSuspendReleaseRequiresRegistry(t *testing.T) { + cs := relCocoonSet() + r := &Reconciler{Scheme: testScheme(t)} + if _, err := r.reconcileSuspendRelease(t.Context(), cs, classifiedPods{}); err == nil { + t.Error("release without a registry must fail closed, not release the seat") + } +} + +func TestSuspendReleaseNoPodsSettlesSuspended(t *testing.T) { + cs := relCocoonSet() + cli := relClient(t, cs) + r := &Reconciler{Client: cli, Scheme: testScheme(t), Registry: &fakeRegistry{}} + + if _, err := r.reconcileSuspendRelease(t.Context(), cs, classifiedPods{}); err != nil { + t.Fatalf("reconcileSuspendRelease: %v", err) + } + got := mustGetCS(t, cli) + if got.Status.Phase != cocoonv1.CocoonSetPhaseSuspended { + t.Errorf("pod-less suspended set must settle Suspended, got %q", got.Status.Phase) + } +} + +func TestSuspendReleaseWaitsForSnapshotVerification(t *testing.T) { + cs := relCocoonSet() + pod := relHibernatedPod(t, cs, "node-a") + cli := relClient(t, cs, pod) + r := &Reconciler{Client: cli, Scheme: testScheme(t), Registry: &fakeRegistry{}} // snapshot absent + + if _, err := r.reconcileSuspendRelease(t.Context(), cs, singlePod(pod)); err != nil { + t.Fatalf("reconcileSuspendRelease: %v", err) + } + var got corev1.Pod + if err := cli.Get(t.Context(), types.NamespacedName{Namespace: "ns", Name: "demo-0"}, &got); err != nil { + t.Errorf("pod must survive until the snapshot is registry-verified: %v", err) + } +} + +func TestSuspendReleaseDeletesVerifiedPodAndStashes(t *testing.T) { + cs := relCocoonSet() + pod := relHibernatedPod(t, cs, "node-a") + reg := &fakeRegistry{present: map[string]bool{relHibernateTagKey: true}} + cli := relClient(t, cs, pod) + r := &Reconciler{Client: cli, Scheme: testScheme(t), Registry: reg} + + if _, err := r.reconcileSuspendRelease(t.Context(), cs, singlePod(pod)); err != nil { + t.Fatalf("reconcileSuspendRelease: %v", err) + } + var gotPod corev1.Pod + if err := cli.Get(t.Context(), types.NamespacedName{Namespace: "ns", Name: "demo-0"}, &gotPod); !apierrors.IsNotFound(err) { + t.Errorf("verified hibernated pod must be deleted to release the seat, err=%v", err) + } + gotCS := mustGetCS(t, cli) + if gotCS.Annotations[meta.AnnotationHibernatedOnNode] != "node-a" { + t.Errorf("must stash the hibernated-on node hint, got %q", gotCS.Annotations[meta.AnnotationHibernatedOnNode]) + } + if names := gotCS.Annotations[annotationDeleteVMNames]; !strings.Contains(names, relVMName) { + t.Errorf("must stash vm names for delete-time GC before releasing pods, got %q", names) + } + if gotCS.Status.Phase != cocoonv1.CocoonSetPhaseSuspended { + t.Errorf("Suspended must land with the deletes, got %q", gotCS.Status.Phase) + } + if len(reg.deleted) != 0 { + t.Errorf("suspend must never drop the hibernate snapshot: %v", reg.deleted) + } +} + +func TestSuspendReleaseKeepsTerminalPod(t *testing.T) { + cs := relCocoonSet() + pod := mustBuildAgentPod(t, cs, 0, "", "", testScheme(t)) + pod.Spec.NodeName = "node-a" + meta.LifecycleStatus{State: meta.LifecycleStateFailed, ObservedGeneration: cs.Generation}.Apply(pod) + cli := relClient(t, cs, pod) + r := &Reconciler{Client: cli, Scheme: testScheme(t), Registry: &fakeRegistry{}} + + if _, err := r.reconcileSuspendRelease(t.Context(), cs, singlePod(pod)); err != nil { + t.Fatalf("reconcileSuspendRelease: %v", err) + } + var got corev1.Pod + if err := cli.Get(t.Context(), types.NamespacedName{Namespace: "ns", Name: "demo-0"}, &got); err != nil { + t.Errorf("terminal pod must be kept for triage, not released: %v", err) + } + gotCS := mustGetCS(t, cli) + if gotCS.Status.Phase != cocoonv1.CocoonSetPhaseSuspended { + t.Errorf("only-terminal set must settle Suspended like retain does, got %q", gotCS.Status.Phase) + } +} + +func TestWakeRecreatesWithRestoreAndPreferredAffinity(t *testing.T) { + cs := relCocoonSet(func(cs *cocoonv1.CocoonSet) { + cs.Spec.Suspend = false + cs.Status.Phase = cocoonv1.CocoonSetPhaseSuspended + cs.Annotations = map[string]string{meta.AnnotationHibernatedOnNode: "node-a"} + }) + reg := &fakeRegistry{present: map[string]bool{relHibernateTagKey: true}} + cli := relClient(t, cs) + r := &Reconciler{Client: cli, Scheme: testScheme(t), Registry: reg} + + handled, _, err := r.reconcileWake(t.Context(), cs, classifiedPods{}) + if err != nil || !handled { + t.Fatalf("handled=%v err=%v", handled, err) + } + var got corev1.Pod + if err := cli.Get(t.Context(), types.NamespacedName{Namespace: "ns", Name: "demo-0"}, &got); err != nil { + t.Fatalf("recreated pod not found: %v", err) + } + if !meta.ReadRestoreFromHibernate(&got) { + t.Error("recreated pod must carry restore-from-hibernate") + } + if got.Spec.NodeName != "" { + t.Errorf("wake must not hard-pin a node, got %q", got.Spec.NodeName) + } + na := got.Spec.Affinity + if na == nil || na.NodeAffinity == nil || len(na.NodeAffinity.PreferredDuringSchedulingIgnoredDuringExecution) != 1 || + na.NodeAffinity.PreferredDuringSchedulingIgnoredDuringExecution[0].Preference.MatchExpressions[0].Values[0] != "node-a" { + t.Errorf("wake must soft-prefer the hibernated-on node, got %+v", na) + } + if na.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution != nil { + t.Error("the node hint must be preferred, never required") + } + gotCS := mustGetCS(t, cli) + if gotCS.Status.Phase != cocoonv1.CocoonSetPhaseWaking { + t.Errorf("wake must persist the Waking phase before creating the pod, got %q", gotCS.Status.Phase) + } + if len(reg.deleted) != 0 { + t.Errorf("tag must survive until the restored VM is live: %v", reg.deleted) + } +} + +func TestWakeRespectsNodeNamePin(t *testing.T) { + cs := relCocoonSet(func(cs *cocoonv1.CocoonSet) { + cs.Spec.Suspend = false + cs.Spec.NodeName = "node-b" + cs.Status.Phase = cocoonv1.CocoonSetPhaseSuspended + cs.Annotations = map[string]string{meta.AnnotationHibernatedOnNode: "node-a"} + }) + reg := &fakeRegistry{present: map[string]bool{relHibernateTagKey: true}} + cli := relClient(t, cs) + r := &Reconciler{Client: cli, Scheme: testScheme(t), Registry: reg} + + handled, _, err := r.reconcileWake(t.Context(), cs, classifiedPods{}) + if err != nil || !handled { + t.Fatalf("handled=%v err=%v", handled, err) + } + var got corev1.Pod + if err := cli.Get(t.Context(), types.NamespacedName{Namespace: "ns", Name: "demo-0"}, &got); err != nil { + t.Fatalf("recreated pod not found: %v", err) + } + req := got.Spec.Affinity.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution + if req == nil || req.NodeSelectorTerms[0].MatchExpressions[0].Values[0] != "node-b" { + t.Errorf("spec.nodeName pin must win over the wake hint, got %+v", got.Spec.Affinity) + } +} + +func TestWakeRestoresOnFastUnsuspendFromSuspending(t *testing.T) { + // Deletes ran but the phase still reads Suspending: a fast unsuspend here + // must restore — falling through would fresh-boot over the snapshot. + cs := relCocoonSet(func(cs *cocoonv1.CocoonSet) { + cs.Spec.Suspend = false + cs.Status.Phase = cocoonv1.CocoonSetPhaseSuspending + cs.Annotations = map[string]string{meta.AnnotationHibernatedOnNode: "node-a"} + }) + reg := &fakeRegistry{present: map[string]bool{relHibernateTagKey: true}} + cli := relClient(t, cs) + r := &Reconciler{Client: cli, Scheme: testScheme(t), Registry: reg} + + handled, _, err := r.reconcileWake(t.Context(), cs, classifiedPods{}) + if err != nil || !handled { + t.Fatalf("handled=%v err=%v", handled, err) + } + var got corev1.Pod + if err := cli.Get(t.Context(), types.NamespacedName{Namespace: "ns", Name: "demo-0"}, &got); err != nil { + t.Fatalf("recreated pod not found: %v", err) + } + if !meta.ReadRestoreFromHibernate(&got) { + t.Error("fast unsuspend from Suspending must restore, not fresh-boot") + } + if mustGetCS(t, cli).Status.Phase != cocoonv1.CocoonSetPhaseWaking { + t.Error("wake must persist Waking from the Suspending window too") + } +} + +func TestWakeLeavesSuspendingWithoutHintToNormalFlow(t *testing.T) { + // Suspending without the hint is not a released seat; the normal flow owns it. + cs := relCocoonSet(func(cs *cocoonv1.CocoonSet) { + cs.Spec.Suspend = false + cs.Status.Phase = cocoonv1.CocoonSetPhaseSuspending + }) + r := &Reconciler{Scheme: testScheme(t), Registry: &fakeRegistry{probeErr: errors.New("boom")}} + handled, _, err := r.reconcileWake(t.Context(), cs, classifiedPods{}) + if handled || err != nil { + t.Errorf("hint-less Suspending must not engage the wake, handled=%v err=%v", handled, err) + } +} + +func TestWakeWaitsOutStalePodCacheThenRestores(t *testing.T) { + // Fresh CS view (suspend=false, Suspended) with a stale pod cache still + // showing the deleted hibernated main: wait, then restore once it clears. + cs := relCocoonSet(func(cs *cocoonv1.CocoonSet) { + cs.Spec.Suspend = false + cs.Status.Phase = cocoonv1.CocoonSetPhaseSuspended + cs.Annotations = map[string]string{meta.AnnotationHibernatedOnNode: "node-a"} + }) + stale := relHibernatedPod(t, cs, "node-a") + meta.HibernateState(true).Apply(stale) + reg := &fakeRegistry{present: map[string]bool{relHibernateTagKey: true}} + cli := relClient(t, cs) // pod absent: the live read sees the delete + r := &Reconciler{Client: cli, APIReader: cli, Scheme: testScheme(t), Registry: reg} + + handled, _, err := r.reconcileWake(t.Context(), cs, singlePod(stale)) + if err != nil || !handled { + t.Fatalf("stale-cache pass: handled=%v err=%v", handled, err) + } + if len(reg.deleted) != 0 { + t.Errorf("waiting must not drop the tag: %v", reg.deleted) + } + + handled, _, err = r.reconcileWake(t.Context(), cs, classifiedPods{}) + if err != nil || !handled { + t.Fatalf("post-clear pass: handled=%v err=%v", handled, err) + } + var got corev1.Pod + if err := cli.Get(t.Context(), types.NamespacedName{Namespace: "ns", Name: "demo-0"}, &got); err != nil { + t.Fatalf("restore pod not found: %v", err) + } + if !meta.ReadRestoreFromHibernate(&got) { + t.Error("post-clear pass must restore, not fresh-boot") + } +} + +func TestWakeWaitsOutStaleSuspendingViewThenRestores(t *testing.T) { + // Delete ran but the Suspended receipt never landed: a stale main view + // under Suspending must wait, not fresh-boot. + cs := relCocoonSet(func(cs *cocoonv1.CocoonSet) { + cs.Spec.Suspend = false + cs.Status.Phase = cocoonv1.CocoonSetPhaseSuspending + cs.Annotations = map[string]string{meta.AnnotationHibernatedOnNode: "node-a"} + }) + stale := relHibernatedPod(t, cs, "node-a") + meta.HibernateState(true).Apply(stale) + reg := &fakeRegistry{present: map[string]bool{relHibernateTagKey: true}} + cli := relClient(t, cs) // pod absent: the live read sees the delete + r := &Reconciler{Client: cli, APIReader: cli, Scheme: testScheme(t), Registry: reg} + + handled, _, err := r.reconcileWake(t.Context(), cs, singlePod(stale)) + if err != nil || !handled { + t.Fatalf("stale-view pass: handled=%v err=%v", handled, err) + } + handled, _, err = r.reconcileWake(t.Context(), cs, classifiedPods{}) + if err != nil || !handled { + t.Fatalf("post-clear pass: handled=%v err=%v", handled, err) + } + var got corev1.Pod + if err := cli.Get(t.Context(), types.NamespacedName{Namespace: "ns", Name: "demo-0"}, &got); err != nil { + t.Fatalf("restore pod not found: %v", err) + } + if !meta.ReadRestoreFromHibernate(&got) { + t.Error("receipt-less Suspending must still restore, not fresh-boot") + } +} + +func TestWakeLeavesUndeletedMainToInPlaceWake(t *testing.T) { + // The delete never finished (pod truly alive): in-place unsuspend owns it. + cs := relCocoonSet(func(cs *cocoonv1.CocoonSet) { + cs.Spec.Suspend = false + cs.Status.Phase = cocoonv1.CocoonSetPhaseSuspending + cs.Annotations = map[string]string{meta.AnnotationHibernatedOnNode: "node-a"} + }) + main := relHibernatedPod(t, cs, "node-a") + meta.HibernateState(true).Apply(main) + cli := relClient(t, cs, main) + r := &Reconciler{Client: cli, APIReader: cli, Scheme: testScheme(t), Registry: &fakeRegistry{}} + + handled, _, err := r.reconcileWake(t.Context(), cs, singlePod(main)) + if handled || err != nil { + t.Errorf("alive hibernated main must wake in place via the normal flow, handled=%v err=%v", handled, err) + } +} + +func TestWakeFreshBootsWhenNoSnapshot(t *testing.T) { + cs := relCocoonSet(func(cs *cocoonv1.CocoonSet) { + cs.Spec.Suspend = false + cs.Status.Phase = cocoonv1.CocoonSetPhaseSuspended + }) + r := &Reconciler{Scheme: testScheme(t), Registry: &fakeRegistry{}} + handled, _, err := r.reconcileWake(t.Context(), cs, classifiedPods{}) + if err != nil { + t.Fatalf("reconcileWake: %v", err) + } + if handled { + t.Error("suspended-before-first-boot has nothing to restore; the normal flow owns it") + } +} + +func TestWakeProbeErrorFailsClosed(t *testing.T) { + cs := relCocoonSet(func(cs *cocoonv1.CocoonSet) { + cs.Spec.Suspend = false + cs.Status.Phase = cocoonv1.CocoonSetPhaseSuspended + }) + r := &Reconciler{Scheme: testScheme(t), Registry: &fakeRegistry{probeErr: errors.New("boom")}} + handled, _, err := r.reconcileWake(t.Context(), cs, classifiedPods{}) + if !handled || err == nil { + t.Errorf("probe failure must own the reconcile (handled=%v err=%v): falling through would fresh-boot over the snapshot", handled, err) + } +} + +func TestWakeWaitsWhileRestoring(t *testing.T) { + cs := relCocoonSet(func(cs *cocoonv1.CocoonSet) { + cs.Spec.Suspend = false + cs.Status.Phase = cocoonv1.CocoonSetPhaseWaking + }) + main := migMainPod(t, cs, "", "", false) // unscheduled, no VMID + reg := &fakeRegistry{present: map[string]bool{relHibernateTagKey: true}} + cli := relClient(t, cs, main) + r := &Reconciler{Client: cli, Scheme: testScheme(t), Registry: reg} + + handled, _, err := r.reconcileWake(t.Context(), cs, singlePod(main)) + if err != nil || !handled { + t.Fatalf("handled=%v err=%v", handled, err) + } + if len(reg.deleted) != 0 { + t.Errorf("must not drop the snapshot before the restored VM is live: %v", reg.deleted) + } +} + +func TestWakeDropsTagAndClearsHintWhenLive(t *testing.T) { + cs := relCocoonSet(func(cs *cocoonv1.CocoonSet) { + cs.Spec.Suspend = false + cs.Status.Phase = cocoonv1.CocoonSetPhaseWaking + cs.Annotations = map[string]string{meta.AnnotationHibernatedOnNode: "node-a"} + }) + main := migMainPod(t, cs, "node-c", "vmid-new", true) + reg := &fakeRegistry{present: map[string]bool{relHibernateTagKey: true}} + cli := relClient(t, cs, main) + r := &Reconciler{Client: cli, Scheme: testScheme(t), Registry: reg} + + handled, _, err := r.reconcileWake(t.Context(), cs, singlePod(main)) + if err != nil || !handled { + t.Fatalf("handled=%v err=%v", handled, err) + } + if !slices.Contains(reg.deleted, relHibernateTagKey) { + t.Errorf("live restore must drop the hibernate snapshot, deleted=%v", reg.deleted) + } + gotCS := mustGetCS(t, cli) + if _, ok := gotCS.Annotations[meta.AnnotationHibernatedOnNode]; ok { + t.Error("node hint must be cleared once the wake completes") + } +} + +func TestWakeIgnoresRunningPhase(t *testing.T) { + cs := relCocoonSet(func(cs *cocoonv1.CocoonSet) { + cs.Spec.Suspend = false + cs.Status.Phase = cocoonv1.CocoonSetPhaseRunning + }) + // probeErr proves the registry is never consulted: a stale tag on a + // running set must not be trusted (it would roll the VM back). + r := &Reconciler{Scheme: testScheme(t), Registry: &fakeRegistry{probeErr: errors.New("boom")}} + handled, _, err := r.reconcileWake(t.Context(), cs, classifiedPods{}) + if handled || err != nil { + t.Errorf("wake must only engage from Suspended/Waking, handled=%v err=%v", handled, err) + } +} + +func TestWakeCompletionSurvivesStalePhase(t *testing.T) { + // A stale-phase reconcile overwrote Waking mid-wake (e.g. to Pending); the + // restore-marked live pod + present node hint must keep completion reachable. + cs := relCocoonSet(func(cs *cocoonv1.CocoonSet) { + cs.Spec.Suspend = false + cs.Status.Phase = cocoonv1.CocoonSetPhasePending + cs.Annotations = map[string]string{meta.AnnotationHibernatedOnNode: "node-a"} + }) + main := migMainPod(t, cs, "node-a", "vmid-new", true) + meta.MarkRestoreFromHibernate(main) + reg := &fakeRegistry{present: map[string]bool{relHibernateTagKey: true}} + cli := relClient(t, cs, main) + r := &Reconciler{Client: cli, Scheme: testScheme(t), Registry: reg} + + handled, _, err := r.reconcileWake(t.Context(), cs, singlePod(main)) + if err != nil || !handled { + t.Fatalf("handled=%v err=%v", handled, err) + } + if !slices.Contains(reg.deleted, relHibernateTagKey) { + t.Errorf("completion must survive a stale/overwritten phase, deleted=%v", reg.deleted) + } + gotCS := mustGetCS(t, cli) + if _, ok := gotCS.Annotations[meta.AnnotationHibernatedOnNode]; ok { + t.Error("hint must be cleared so the fallback retires") + } +} + +func TestWakeWaitsOnStalePhaseWhileRestoring(t *testing.T) { + // Same staleness race but the VM is not live yet: re-enter the waiting + // branch (and repaint Waking), do not drop the tag. + cs := relCocoonSet(func(cs *cocoonv1.CocoonSet) { + cs.Spec.Suspend = false + cs.Status.Phase = cocoonv1.CocoonSetPhasePending + cs.Annotations = map[string]string{meta.AnnotationHibernatedOnNode: "node-a"} + }) + main := migMainPod(t, cs, "node-a", "", false) + meta.MarkRestoreFromHibernate(main) + reg := &fakeRegistry{present: map[string]bool{relHibernateTagKey: true}} + cli := relClient(t, cs, main) + r := &Reconciler{Client: cli, Scheme: testScheme(t), Registry: reg} + + handled, _, err := r.reconcileWake(t.Context(), cs, singlePod(main)) + if err != nil || !handled { + t.Fatalf("handled=%v err=%v", handled, err) + } + if len(reg.deleted) != 0 { + t.Errorf("must not drop the snapshot before the restored VM is live: %v", reg.deleted) + } +} + +func TestWakeDisengagesAfterCleanup(t *testing.T) { + // Steady state: the pod keeps its restore annotation for life but the hint + // is gone — no re-engage (probeErr proves the registry is never consulted). + cs := relCocoonSet(func(cs *cocoonv1.CocoonSet) { + cs.Spec.Suspend = false + cs.Status.Phase = cocoonv1.CocoonSetPhaseRunning + }) + main := migMainPod(t, cs, "node-a", "vmid-new", true) + meta.MarkRestoreFromHibernate(main) + r := &Reconciler{Scheme: testScheme(t), Registry: &fakeRegistry{probeErr: errors.New("boom")}} + + handled, _, err := r.reconcileWake(t.Context(), cs, singlePod(main)) + if handled || err != nil { + t.Errorf("completed wake must hand back to the normal flow, handled=%v err=%v", handled, err) + } +} + +func TestWakeLeavesSuspendedPlaceholderToNormalFlow(t *testing.T) { + cs := relCocoonSet(func(cs *cocoonv1.CocoonSet) { + cs.Spec.Suspend = false + cs.Status.Phase = cocoonv1.CocoonSetPhaseSuspended + }) + main := relHibernatedPod(t, cs, "node-a") + meta.HibernateState(true).Apply(main) + r := &Reconciler{Scheme: testScheme(t), Registry: &fakeRegistry{}} + + handled, _, err := r.reconcileWake(t.Context(), cs, singlePod(main)) + if handled || err != nil { + t.Errorf("a suspended set with a placeholder pod wakes in place via applyUnsuspend, handled=%v err=%v", handled, err) + } +} + +func relCocoonSet(mods ...func(*cocoonv1.CocoonSet)) *cocoonv1.CocoonSet { + return newCocoonSet("demo", append([]func(*cocoonv1.CocoonSet){func(cs *cocoonv1.CocoonSet) { + cs.Generation = 1 + cs.Spec.Suspend = true + cs.Spec.HibernatePolicy = cocoonv1.HibernatePolicyRelease + }}, mods...)...) +} + +// relHibernatedPod builds the main pod in the state vk publishes after a +// completed hibernate: lifecycle=hibernated at the CR's generation. +func relHibernatedPod(t *testing.T, cs *cocoonv1.CocoonSet, node string) *corev1.Pod { + t.Helper() + pod := mustBuildAgentPod(t, cs, 0, "", "", testScheme(t)) + pod.Spec.NodeName = node + meta.LifecycleStatus{State: meta.LifecycleStateHibernated, ObservedGeneration: cs.Generation}.Apply(pod) + return pod +} + +func singlePod(pod *corev1.Pod) classifiedPods { + return classifiedPods{main: pod, allByName: map[string]*corev1.Pod{pod.Name: pod}} +} + +func relClient(t *testing.T, objs ...client.Object) client.WithWatch { + t.Helper() + return ctrlfake.NewClientBuilder().WithScheme(testScheme(t)). + WithObjects(objs...).WithStatusSubresource(&cocoonv1.CocoonSet{}).Build() +} + +func mustGetCS(t *testing.T, cli client.Client) cocoonv1.CocoonSet { + t.Helper() + var cs cocoonv1.CocoonSet + if err := cli.Get(t.Context(), types.NamespacedName{Namespace: "ns", Name: "demo"}, &cs); err != nil { + t.Fatalf("get cocoonset: %v", err) + } + return cs +} diff --git a/config/crd/bases/cocoonset.cocoonstack.io_cocoonsets.yaml b/config/crd/bases/cocoonset.cocoonstack.io_cocoonsets.yaml index 2c1f6a5..68c738d 100644 --- a/config/crd/bases/cocoonset.cocoonstack.io_cocoonsets.yaml +++ b/config/crd/bases/cocoonset.cocoonstack.io_cocoonsets.yaml @@ -248,6 +248,16 @@ spec: required: - image type: object + hibernatePolicy: + default: retain + description: |- + HibernatePolicy keeps (retain) or frees (release) the VM's scheduling + seat while suspended; release requires a main-only set until sub-agent/ + toolbox restore intent is extended. + enum: + - retain + - release + type: string nodeName: description: |- NodeName pins the VM to a node (cross-node migrate). Empty = let the @@ -421,6 +431,11 @@ spec: required: - agent type: object + x-kubernetes-validations: + - message: hibernatePolicy=release requires agent.replicas=0 and no toolboxes + rule: '!has(self.hibernatePolicy) || self.hibernatePolicy != ''release'' + || ((!has(self.agent.replicas) || self.agent.replicas == 0) && (!has(self.toolboxes) + || size(self.toolboxes) == 0))' status: description: CocoonSetStatus represents the observed state of a CocoonSet. properties: @@ -530,6 +545,7 @@ spec: - Scaling - Suspending - Suspended + - Waking - Migrating - Failed type: string diff --git a/go.mod b/go.mod index be5b701..7d988f4 100644 --- a/go.mod +++ b/go.mod @@ -3,7 +3,7 @@ module github.com/cocoonstack/cocoon-operator go 1.26.5 require ( - github.com/cocoonstack/cocoon-common v0.2.7 + github.com/cocoonstack/cocoon-common v0.2.8-0.20260724145826-4363e4ff03ee github.com/go-logr/logr v1.4.3 github.com/google/go-containerregistry v0.21.7 github.com/projecteru2/core v0.0.0-20241016125006-ff909eefe04c diff --git a/go.sum b/go.sum index 562006a..0f06d53 100644 --- a/go.sum +++ b/go.sum @@ -29,8 +29,8 @@ github.com/cockroachdb/logtags v0.0.0-20230118201751-21c54148d20b h1:r6VH0faHjZe github.com/cockroachdb/logtags v0.0.0-20230118201751-21c54148d20b/go.mod h1:Vz9DsVWQQhf3vs21MhPMZpMGSht7O/2vFW2xusFUVOs= github.com/cockroachdb/redact v1.1.3 h1:AKZds10rFSIj7qADf0g46UixK8NNLwWTNdCIGS5wfSQ= github.com/cockroachdb/redact v1.1.3/go.mod h1:BVNblN9mBWFyMyqK1k3AAiSxhvhfK2oOZZ2lK+dpvRg= -github.com/cocoonstack/cocoon-common v0.2.7 h1:0c3Sc3aqtaHxpOHuc45jCaeR4XnMcaYbeic9d3jPP+Y= -github.com/cocoonstack/cocoon-common v0.2.7/go.mod h1:2TySEuBWhZIhxJkHbjcoQghwcDqAJ0y48c+/wag7Cws= +github.com/cocoonstack/cocoon-common v0.2.8-0.20260724145826-4363e4ff03ee h1:KshBIXkOlQMNdL7DPoYXitxq+qeWyQSqQsvIpDV0gv4= +github.com/cocoonstack/cocoon-common v0.2.8-0.20260724145826-4363e4ff03ee/go.mod h1:VSfgYiWxoHRnWybzQNaxmK4kVoLT9ffiMKll5/i92CM= github.com/codegangsta/inject v0.0.0-20150114235600-33e0aa1cb7c0/go.mod h1:4Zcjuz89kmFXt9morQgcfYZAYZ5n8WHjt81YYWIwtTM= github.com/coreos/etcd v3.3.10+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE= github.com/coreos/go-etcd v2.0.0+incompatible/go.mod h1:Jez6KQU2B/sWsbdaef3ED8NzMklzPG4d5KIOhIy30Tk= diff --git a/main.go b/main.go index fab8cc5..8723610 100644 --- a/main.go +++ b/main.go @@ -116,6 +116,7 @@ func main() { if err = (&cocoonset.Reconciler{ Client: mgr.GetClient(), + APIReader: mgr.GetAPIReader(), Scheme: mgr.GetScheme(), Registry: registry, Recorder: recorder, diff --git a/metrics/metrics.go b/metrics/metrics.go index fe96b40..ba166b4 100644 --- a/metrics/metrics.go +++ b/metrics/metrics.go @@ -6,6 +6,9 @@ import "github.com/prometheus/client_golang/prometheus" const ( metricNamespace = "cocoon" metricSubsystem = "operator" + + labelNamespace = "namespace" + labelCocoonSet = "cocoonset" ) var ( @@ -16,7 +19,7 @@ var ( Name: "subagent_rebuild_total", Help: "Number of sub-agent rebuilds triggered by triageSubAgent.", }, - []string{"namespace", "cocoonset"}, + []string{labelNamespace, labelCocoonSet}, ) SubAgentDeadLetterTotal = prometheus.NewCounterVec( @@ -26,7 +29,7 @@ var ( Name: "subagent_dead_letter_total", Help: "Number of sub-agents marked dead-letter after exhausting rebuild attempts.", }, - []string{"namespace", "cocoonset"}, + []string{labelNamespace, labelCocoonSet}, ) HibernatePhaseDurationSeconds = prometheus.NewHistogramVec( @@ -60,6 +63,36 @@ var ( }, []string{"phase"}, ) + + SlotReleasePodsDeletedTotal = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: metricNamespace, + Subsystem: metricSubsystem, + Name: "slot_release_pods_deleted_total", + Help: "Number of hibernated pods deleted by release-policy suspend to free their scheduling seat.", + }, + []string{labelNamespace, labelCocoonSet}, + ) + + SlotReleaseWakeTotal = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: metricNamespace, + Subsystem: metricSubsystem, + Name: "slot_release_wake_total", + Help: "Number of released-seat wakes that reached a live VM, by placement (hint-node=landed back on the hibernated-on node, pool=rescheduled elsewhere). Every release wake restores via registry pull.", + }, + []string{labelNamespace, labelCocoonSet, "placement"}, + ) + + SlotReleaseWakeUnschedulableTotal = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: metricNamespace, + Subsystem: metricSubsystem, + Name: "slot_release_wake_unschedulable_total", + Help: "Number of reconcile passes that observed a released-seat wake pod Unschedulable (out of capacity).", + }, + []string{labelNamespace, labelCocoonSet}, + ) ) // Register installs all operator collectors into reg so they surface on the /metrics endpoint. @@ -70,5 +103,8 @@ func Register(reg prometheus.Registerer) { HibernatePhaseDurationSeconds, WakePhaseDurationSeconds, LifecycleStateFailedObservedTotal, + SlotReleasePodsDeletedTotal, + SlotReleaseWakeTotal, + SlotReleaseWakeUnschedulableTotal, ) }