From 5265d32b7caa5f44e79ddd3c2134b4a0a22b4a00 Mon Sep 17 00:00:00 2001 From: tonic Date: Thu, 2 Jul 2026 15:51:38 +0800 Subject: [PATCH 1/2] feat(cocoonset): cross-node migration (nodeName affinity + migration state machine) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Rewritten on current main atop the merged restore-from-hibernate producer (#14): the control plane patches CocoonSet.spec.nodeName and the operator hibernates the main agent, waits for the :hibernate snapshot in the OCI registry, deletes the old pod, recreates it with hostname nodeAffinity + restore-from-hibernate, and drops the snapshot once the restored VM runs with a fresh VMID. Decisions are pure functions of durable state, so every step is idempotent and crash-recoverable. Hardening over the original branch: - a registry probe error owns the reconcile — falling through would let applyUnsuspend unwind the migration or fresh-boot over the only snapshot - a :hibernate tag on a never-quiesced pod is a leftover and is dropped, not restored - re-targeting nodeName back mid-migration wakes the pod in place instead of deadlocking; CR-owned hibernation short-circuits before the registry probe - clearing nodeName in the deleted-pod window finishes the restore instead of stranding the snapshot - steady-state pinned sets skip the registry probe; a CR wake mid-flight is not repainted as a migration Scoped to the main agent (slot 0); sub-agents follow via their hard bind. --- cocoonset/migrate.go | 169 +++++++++ cocoonset/migrate_test.go | 333 ++++++++++++++++++ cocoonset/pods.go | 20 ++ cocoonset/pods_test.go | 58 +++ cocoonset/reconciler.go | 7 + .../cocoonset.cocoonstack.io_cocoonsets.yaml | 9 + 6 files changed, 596 insertions(+) create mode 100644 cocoonset/migrate.go create mode 100644 cocoonset/migrate_test.go diff --git a/cocoonset/migrate.go b/cocoonset/migrate.go new file mode 100644 index 0000000..e0815ef --- /dev/null +++ b/cocoonset/migrate.go @@ -0,0 +1,169 @@ +package cocoonset + +import ( + "cmp" + "context" + "fmt" + + "github.com/projecteru2/core/log" + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + ctrl "sigs.k8s.io/controller-runtime" + + cocoonv1 "github.com/cocoonstack/cocoon-common/apis/v1" + commonk8s "github.com/cocoonstack/cocoon-common/k8s" + "github.com/cocoonstack/cocoon-common/meta" +) + +// reconcileMigration drives cross-node migration of the main agent (slot 0): +// quiesce -> snapshot -> recreate on the target with restore-from-hibernate -> +// drop the snapshot; idempotent over durable state, handled=false hands back. +// Never lose live state: the old pod dies only after the snapshot exists AND +// this controller quiesced it; the snapshot drops only once the new VM runs. +func (r *Reconciler) reconcileMigration(ctx context.Context, cs *cocoonv1.CocoonSet, classified classifiedPods) (bool, ctrl.Result, error) { + desired := cs.Spec.NodeName + migrating := cs.Status.Phase == cocoonv1.CocoonSetPhaseMigrating + if r.Registry == nil || (desired == "" && !migrating) { + return false, ctrl.Result{}, nil + } + main := classified.main + // Steady state of a pinned set: skip the registry probe — safe because the + // Migrating phase is persisted before the first side effect. + if !migrating && podSettledOn(main, desired) { + return false, ctrl.Result{}, nil + } + // A CR-owned hibernation quiesced on the target is no migration either; + // short-circuit before the probe — CR hibernation is the long-lived idle state. + if main != nil && bool(meta.ReadHibernateState(main)) && (desired == "" || main.Spec.NodeName == desired) { + hibByCR, err := r.podsHibernatedByCR(ctx, cs.Namespace) + if err != nil { + return true, ctrl.Result{}, fmt.Errorf("migrate: %w", err) + } + if _, owned := hibByCR[main.Name]; owned { + return false, ctrl.Result{}, nil + } + } + vmName := meta.VMNameForDeployment(cs.Namespace, cs.Name, 0) + snap, err := r.hasHibernateSnapshot(ctx, vmName) + if err != nil { + // handled=true: falling through to the normal flow would clear the + // hibernate annotation mid-migration or fresh-boot over the snapshot. + return true, ctrl.Result{}, fmt.Errorf("migrate: %w", err) + } + + if !snap { + if desired == "" || main == nil || main.Spec.NodeName == "" || main.Spec.NodeName == desired { + // Settled / aborted / fresh create: the normal flow takes it from here. + return false, ctrl.Result{}, nil + } + return r.startMigration(ctx, cs, classified, desired) + } + return r.advanceMigration(ctx, cs, classified, vmName, desired) +} + +// startMigration quiesces the wrong-node main pod so vk pushes the :hibernate +// snapshot; Migrating persists first so the fast-path can trust the phase. +func (r *Reconciler) startMigration(ctx context.Context, cs *cocoonv1.CocoonSet, classified classifiedPods, desired string) (bool, ctrl.Result, error) { + logger := log.WithFunc("cocoonset.Reconciler.startMigration") + main := classified.main + if !meta.ReadHibernateState(main) { + logger.Infof(ctx, "migrate %s/%s: %s -> %s, hibernating", cs.Namespace, cs.Name, main.Spec.NodeName, desired) + } + if err := r.patchStatus(ctx, cs, buildStatus(cs, classified, cocoonv1.CocoonSetPhaseMigrating)); err != nil { + return true, ctrl.Result{}, fmt.Errorf("migrate: patch migrating status %s/%s: %w", cs.Namespace, cs.Name, err) + } + if err := commonk8s.PatchHibernateState(ctx, r.Client, main, true); err != nil { + return true, ctrl.Result{}, fmt.Errorf("migrate: patch hibernate on %s/%s: %w", main.Namespace, main.Name, err) + } + return true, ctrl.Result{RequeueAfter: requeueMigratePoll}, nil +} + +// advanceMigration steps a migration whose :hibernate snapshot exists through +// teardown -> recreate -> restore-wait -> snapshot drop. +func (r *Reconciler) advanceMigration(ctx context.Context, cs *cocoonv1.CocoonSet, classified classifiedPods, vmName, desired string) (bool, ctrl.Result, error) { + logger := log.WithFunc("cocoonset.Reconciler.advanceMigration") + main := classified.main + + switch { + case main != nil && desired != "" && main.Spec.NodeName != "" && main.Spec.NodeName != desired: + // A tag this controller never quiesced is a leftover (suspend/unsuspend + // never deletes it); restoring it would roll the VM back. Drop it. + if !meta.ReadHibernateState(main) { + logger.Warnf(ctx, "migrate %s/%s: stale hibernate snapshot for %s, dropping it first", cs.Namespace, cs.Name, vmName) + if err := r.Registry.DeleteManifest(ctx, vmName, meta.HibernateSnapshotTag); err != nil { + return true, ctrl.Result{}, fmt.Errorf("migrate: drop stale hibernate snapshot %s: %w", vmName, err) + } + return r.markMigrating(ctx, cs, classified) + } + // Snapshot landed; tear down the old-node pod. NodeName != "" spares the + // just-recreated restore pod (still unscheduled), else it loops delete/recreate. + logger.Infof(ctx, "migrate %s/%s: snapshot in registry, deleting old pod on %s", cs.Namespace, cs.Name, main.Spec.NodeName) + if err := r.Delete(ctx, main); err != nil && !apierrors.IsNotFound(err) { + return true, ctrl.Result{}, fmt.Errorf("migrate: delete old main %s/%s: %w", main.Namespace, main.Name, err) + } + return r.markMigrating(ctx, cs, classified) + + case main == nil: + // Recreating also finishes an aborted migration — never strand the snapshot. + pod, err := buildAgentPod(cs, 0, "", "", r.Scheme) + if err != nil { + return true, ctrl.Result{}, fmt.Errorf("migrate: build main: %w", err) + } + meta.MarkRestoreFromHibernate(pod) + if err := r.Create(ctx, pod); err != nil { + if apierrors.IsAlreadyExists(err) { + // Old pod still Terminating; wait. + return true, ctrl.Result{RequeueAfter: requeueWaitForMain}, nil + } + return true, ctrl.Result{}, fmt.Errorf("migrate: recreate main on %s: %w", cmp.Or(desired, "any node"), err) + } + logger.Infof(ctx, "migrate %s/%s: recreated main on %s (restore-from-hibernate)", cs.Namespace, cs.Name, cmp.Or(desired, "any node")) + return r.markMigrating(ctx, cs, classified) + + case bool(meta.ReadHibernateState(main)) && (desired == "" || main.Spec.NodeName == desired): + // Quiesced on the target: a re-target back mid-migration or an unsuspend + // racing the tag (CR-owned was excluded pre-probe). Wake it in place. + logger.Infof(ctx, "migrate %s/%s: waking %s in place", cs.Namespace, cs.Name, main.Name) + if err := commonk8s.PatchHibernateState(ctx, r.Client, main, false); err != nil { + return true, ctrl.Result{}, fmt.Errorf("migrate: clear hibernate on %s/%s: %w", main.Namespace, main.Name, err) + } + return r.markMigrating(ctx, cs, classified) + + case !vmLive(main): + // Restoring — wait for vk. Without the durable Migrating phase this is + // a CR wake mid-flight, not a migration: disengage, don't repaint. + if cs.Status.Phase != cocoonv1.CocoonSetPhaseMigrating { + return false, ctrl.Result{}, nil + } + return r.markMigrating(ctx, cs, classified) + + default: + // Restored with a fresh VMID: drop the snapshot; next pass settles to Running. + logger.Infof(ctx, "migrate %s/%s: restored on %s, dropping hibernate snapshot", cs.Namespace, cs.Name, desired) + if err := r.Registry.DeleteManifest(ctx, vmName, meta.HibernateSnapshotTag); err != nil { + return true, ctrl.Result{}, fmt.Errorf("migrate: drop hibernate snapshot %s: %w", vmName, err) + } + return r.markMigrating(ctx, cs, classified) + } +} + +func (r *Reconciler) markMigrating(ctx context.Context, cs *cocoonv1.CocoonSet, classified classifiedPods) (bool, ctrl.Result, error) { + if err := r.patchStatus(ctx, cs, buildStatus(cs, classified, cocoonv1.CocoonSetPhaseMigrating)); err != nil { + return true, ctrl.Result{}, fmt.Errorf("migrate: patch migrating status %s/%s: %w", cs.Namespace, cs.Name, err) + } + return true, ctrl.Result{RequeueAfter: requeueMigratePoll}, nil +} + +// podSettledOn reports the steady state of a pinned CocoonSet: the main pod +// runs on the desired node with a live VM and no pending quiesce. +func podSettledOn(main *corev1.Pod, desired string) bool { + return main != nil && main.Spec.NodeName == desired && + vmLive(main) && !bool(meta.ReadHibernateState(main)) +} + +// vmLive reports a running container with a vk-assigned VMID. Both checks are +// load-bearing: containerStatuses can momentarily report Running before vk +// pulls the snapshot (same gate as the hibernation wake's vmClonedAndRunning). +func vmLive(pod *corev1.Pod) bool { + return meta.ParseVMRuntime(pod).VMID != "" && meta.IsContainerRunning(pod) +} diff --git a/cocoonset/migrate_test.go b/cocoonset/migrate_test.go new file mode 100644 index 0000000..6dc91e8 --- /dev/null +++ b/cocoonset/migrate_test.go @@ -0,0 +1,333 @@ +package cocoonset + +import ( + "errors" + "slices" + "testing" + + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + ctrlfake "sigs.k8s.io/controller-runtime/pkg/client/fake" + + cocoonv1 "github.com/cocoonstack/cocoon-common/apis/v1" + "github.com/cocoonstack/cocoon-common/meta" +) + +var migVMName = meta.VMNameForDeployment("ns", "demo", 0) + +func TestMigrationNoopWithoutNodeName(t *testing.T) { + cs := newCocoonSet("demo") + main := migMainPod(t, cs, "node-a", "vmid-1", true) + r := &Reconciler{Scheme: testScheme(t), Registry: &fakeRegistry{}} + handled, _, err := r.reconcileMigration(t.Context(), cs, classifiedPods{main: main}) + if err != nil { + t.Fatalf("reconcileMigration: %v", err) + } + if handled { + t.Error("no spec.nodeName => migration must not engage") + } +} + +func TestMigrationNoopWhenSettledOnTarget(t *testing.T) { + cs := migCocoonSet("node-b") + main := migMainPod(t, cs, "node-b", "vmid-1", true) + r := &Reconciler{Scheme: testScheme(t), Registry: &fakeRegistry{}} // fast-path: registry never consulted + handled, _, err := r.reconcileMigration(t.Context(), cs, classifiedPods{main: main}) + if err != nil { + t.Fatalf("reconcileMigration: %v", err) + } + if handled { + t.Error("main already on target with no snapshot => not migrating") + } +} + +func TestMigrationStartsHibernateOnWrongNode(t *testing.T) { + cs := migCocoonSet("node-b") + main := migMainPod(t, cs, "node-a", "vmid-1", true) + cli := ctrlfake.NewClientBuilder().WithScheme(testScheme(t)). + WithObjects(cs, main).WithStatusSubresource(&cocoonv1.CocoonSet{}).Build() + r := &Reconciler{Client: cli, Scheme: testScheme(t), Registry: &fakeRegistry{}} + + handled, _, err := r.reconcileMigration(t.Context(), cs, classifiedPods{main: main}) + if err != nil || !handled { + t.Fatalf("expected handled migration, 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("get main: %v", err) + } + if !meta.ReadHibernateState(&got) { + t.Error("migration start must set hibernate annotation on the old pod") + } +} + +func TestMigrationDeletesOldPodAfterSnapshotLands(t *testing.T) { + cs := migCocoonSet("node-b") + main := migMainPod(t, cs, "node-a", "vmid-1", true) + meta.HibernateState(true).Apply(main) // quiesced by the migration start pass + reg := &fakeRegistry{present: map[string]bool{migVMName + ":" + meta.HibernateSnapshotTag: true}} + cli := ctrlfake.NewClientBuilder().WithScheme(testScheme(t)). + WithObjects(cs, main).WithStatusSubresource(&cocoonv1.CocoonSet{}).Build() + r := &Reconciler{Client: cli, Scheme: testScheme(t), Registry: reg} + + handled, _, err := r.reconcileMigration(t.Context(), cs, classifiedPods{main: main}) + 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); !apierrors.IsNotFound(err) { + t.Errorf("old pod must be deleted, got err=%v", err) + } + // Ordering gate: snapshot must NOT be dropped while tearing down the old pod. + if len(reg.deleted) != 0 { + t.Errorf("snapshot dropped too early: %v", reg.deleted) + } +} + +func TestMigrationRecreatesOnTargetWithRestoreAnnotation(t *testing.T) { + cs := migCocoonSet("node-b") + reg := &fakeRegistry{present: map[string]bool{migVMName + ":" + meta.HibernateSnapshotTag: true}} + cli := ctrlfake.NewClientBuilder().WithScheme(testScheme(t)). + WithObjects(cs).WithStatusSubresource(&cocoonv1.CocoonSet{}).Build() + r := &Reconciler{Client: cli, Scheme: testScheme(t), Registry: reg} + + handled, _, err := r.reconcileMigration(t.Context(), cs, classifiedPods{}) // main absent + 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 the restore-from-hibernate annotation") + } + na := got.Spec.Affinity + if na == nil || na.NodeAffinity == nil || + na.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms[0].MatchExpressions[0].Values[0] != "node-b" { + t.Errorf("recreated pod must have hostname affinity to node-b, got %+v", na) + } +} + +func TestMigrationWaitsWhileRestoring(t *testing.T) { + cs := migCocoonSet("node-b") + cs.Status.Phase = cocoonv1.CocoonSetPhaseMigrating // real migrations persist the phase up front + main := migMainPod(t, cs, "node-b", "", false) // on target, no VMID yet + reg := &fakeRegistry{present: map[string]bool{migVMName + ":" + meta.HibernateSnapshotTag: true}} + cli := ctrlfake.NewClientBuilder().WithScheme(testScheme(t)). + WithObjects(cs, main).WithStatusSubresource(&cocoonv1.CocoonSet{}).Build() + r := &Reconciler{Client: cli, Scheme: testScheme(t), Registry: reg} + + handled, _, err := r.reconcileMigration(t.Context(), cs, classifiedPods{main: main}) + if err != nil || !handled { + t.Fatalf("handled=%v err=%v", handled, err) + } + if len(reg.deleted) != 0 { + t.Errorf("must not drop snapshot before the restored VM has a VMID: %v", reg.deleted) + } +} + +func TestMigrationDropsSnapshotWhenRestored(t *testing.T) { + cs := migCocoonSet("node-b") + cs.Status.Phase = cocoonv1.CocoonSetPhaseMigrating // the settled fast-path defers to the in-flight migration + main := migMainPod(t, cs, "node-b", "vmid-new", true) // restored on target + reg := &fakeRegistry{present: map[string]bool{migVMName + ":" + meta.HibernateSnapshotTag: true}} + cli := ctrlfake.NewClientBuilder().WithScheme(testScheme(t)). + WithObjects(cs, main).WithStatusSubresource(&cocoonv1.CocoonSet{}).Build() + r := &Reconciler{Client: cli, Scheme: testScheme(t), Registry: reg} + + handled, _, err := r.reconcileMigration(t.Context(), cs, classifiedPods{main: main}) + if err != nil || !handled { + t.Fatalf("handled=%v err=%v", handled, err) + } + if !slices.Contains(reg.deleted, migVMName+":"+meta.HibernateSnapshotTag) { + t.Errorf("restored VM must drop its hibernate snapshot, deleted=%v", reg.deleted) + } +} + +func TestMigrationDoesNotDeleteRecreatedRestorePod(t *testing.T) { + cs := migCocoonSet("node-b") + cs.Status.Phase = cocoonv1.CocoonSetPhaseMigrating + // Freshly recreated restore pod: NodeName empty (awaiting scheduling), no VMID. + main := migMainPod(t, cs, "", "", false) + reg := &fakeRegistry{present: map[string]bool{migVMName + ":" + meta.HibernateSnapshotTag: true}} + cli := ctrlfake.NewClientBuilder().WithScheme(testScheme(t)). + WithObjects(cs, main).WithStatusSubresource(&cocoonv1.CocoonSet{}).Build() + r := &Reconciler{Client: cli, Scheme: testScheme(t), Registry: reg} + + handled, _, err := r.reconcileMigration(t.Context(), cs, classifiedPods{main: main}) + if err != nil || !handled { + t.Fatalf("handled=%v err=%v", handled, err) + } + // An unscheduled restore pod must survive, else the snapshot branch loops + // delete/recreate forever. + var got corev1.Pod + if err := cli.Get(t.Context(), types.NamespacedName{Namespace: "ns", Name: "demo-0"}, &got); err != nil { + t.Errorf("recreated restore pod must not be deleted while NodeName is empty: %v", err) + } +} + +func TestMigrationProbeErrorIsHandled(t *testing.T) { + cs := migCocoonSet("node-b") + main := migMainPod(t, cs, "node-a", "vmid-1", true) + r := &Reconciler{Scheme: testScheme(t), Registry: &fakeRegistry{probeErr: errors.New("boom")}} + handled, _, err := r.reconcileMigration(t.Context(), cs, classifiedPods{main: main}) + if !handled || err == nil { + t.Errorf("probe failure must own the reconcile, handled=%v err=%v: falling through would unwind the migration", handled, err) + } +} + +func TestMigrationDropsStaleTagInsteadOfDeletingLivePod(t *testing.T) { + cs := migCocoonSet("node-b") + main := migMainPod(t, cs, "node-a", "vmid-1", true) // live, never quiesced + reg := &fakeRegistry{present: map[string]bool{migVMName + ":" + meta.HibernateSnapshotTag: true}} + cli := ctrlfake.NewClientBuilder().WithScheme(testScheme(t)). + WithObjects(cs, main).WithStatusSubresource(&cocoonv1.CocoonSet{}).Build() + r := &Reconciler{Client: cli, Scheme: testScheme(t), Registry: reg} + + handled, _, err := r.reconcileMigration(t.Context(), cs, classifiedPods{main: main}) + 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.Errorf("live pod must survive a stale tag: %v", err) + } + if !slices.Contains(reg.deleted, migVMName+":"+meta.HibernateSnapshotTag) { + t.Errorf("stale tag must be dropped before migrating, deleted=%v", reg.deleted) + } +} + +func TestMigrationWakesInPlaceOnRetargetBack(t *testing.T) { + cs := migCocoonSet("node-b") + main := migMainPod(t, cs, "node-b", "", false) // quiesced on the (re-)target + meta.HibernateState(true).Apply(main) + reg := &fakeRegistry{present: map[string]bool{migVMName + ":" + meta.HibernateSnapshotTag: true}} + cli := ctrlfake.NewClientBuilder().WithScheme(testScheme(t)). + WithObjects(cs, main).WithStatusSubresource(&cocoonv1.CocoonSet{}).Build() + r := &Reconciler{Client: cli, Scheme: testScheme(t), Registry: reg} + + handled, _, err := r.reconcileMigration(t.Context(), cs, classifiedPods{main: main}) + 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("get main: %v", err) + } + if meta.ReadHibernateState(&got) { + t.Error("re-target back must wake the pod in place, not deadlock waiting for a restore") + } + if len(reg.deleted) != 0 { + t.Errorf("tag must survive until the VM runs again: %v", reg.deleted) + } +} + +func TestMigrationLeavesCRHibernationAlone(t *testing.T) { + cs := migCocoonSet("node-b") + main := migMainPod(t, cs, "node-b", "", false) + meta.HibernateState(true).Apply(main) + hib := &cocoonv1.CocoonHibernation{ + ObjectMeta: metav1.ObjectMeta{Name: "h", Namespace: "ns"}, + Spec: cocoonv1.CocoonHibernationSpec{ + PodRef: cocoonv1.HibernationPodRef{Name: main.Name}, + Desire: cocoonv1.HibernationDesireHibernate, + }, + } + // probeErr proves CR-owned hibernation short-circuits before the registry probe. + reg := &fakeRegistry{probeErr: errors.New("boom")} + cli := ctrlfake.NewClientBuilder().WithScheme(testScheme(t)). + WithObjects(cs, main, hib).WithStatusSubresource(&cocoonv1.CocoonSet{}).Build() + r := &Reconciler{Client: cli, Scheme: testScheme(t), Registry: reg} + + handled, _, err := r.reconcileMigration(t.Context(), cs, classifiedPods{main: main}) + if err != nil { + t.Fatalf("reconcileMigration: %v", err) + } + if handled { + t.Error("CR-owned hibernation on the target is not a migration") + } + var got corev1.Pod + if err := cli.Get(t.Context(), types.NamespacedName{Namespace: "ns", Name: "demo-0"}, &got); err != nil { + t.Fatalf("get main: %v", err) + } + if !meta.ReadHibernateState(&got) { + t.Error("must not wake a CR-hibernated pod") + } +} + +func TestMigrationFinishesAbortedRestore(t *testing.T) { + cs := migCocoonSet("") // nodeName cleared mid-flight + cs.Status.Phase = cocoonv1.CocoonSetPhaseMigrating + reg := &fakeRegistry{present: map[string]bool{migVMName + ":" + meta.HibernateSnapshotTag: true}} + cli := ctrlfake.NewClientBuilder().WithScheme(testScheme(t)). + WithObjects(cs).WithStatusSubresource(&cocoonv1.CocoonSet{}).Build() + r := &Reconciler{Client: cli, Scheme: testScheme(t), Registry: reg} + + handled, _, err := r.reconcileMigration(t.Context(), cs, classifiedPods{}) // old pod already deleted + 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("aborted migration must still recreate the restore pod: %v", err) + } + if !meta.ReadRestoreFromHibernate(&got) { + t.Error("recreated pod must restore from the snapshot, not fresh-boot over it") + } + if got.Spec.Affinity != nil { + t.Errorf("no nodeName => no affinity, got %+v", got.Spec.Affinity) + } +} + +func TestMigrationSkipsProbeWhenSettled(t *testing.T) { + cs := migCocoonSet("node-b") + main := migMainPod(t, cs, "node-b", "vmid-1", true) + // probeErr proves the registry is never consulted in steady state. + r := &Reconciler{Scheme: testScheme(t), Registry: &fakeRegistry{probeErr: errors.New("boom")}} + handled, _, err := r.reconcileMigration(t.Context(), cs, classifiedPods{main: main}) + if handled || err != nil { + t.Errorf("settled pinned set must skip the probe, handled=%v err=%v", handled, err) + } +} + +func TestMigrationDisengagesFromCRWakeWindow(t *testing.T) { + // CR wake mid-flight: annotation cleared, tag not yet dropped, VM not yet + // live, no Migrating phase — not a migration. + cs := migCocoonSet("node-b") + main := migMainPod(t, cs, "node-b", "", false) + reg := &fakeRegistry{present: map[string]bool{migVMName + ":" + meta.HibernateSnapshotTag: true}} + r := &Reconciler{Scheme: testScheme(t), Registry: reg} + + handled, _, err := r.reconcileMigration(t.Context(), cs, classifiedPods{main: main}) + if err != nil { + t.Fatalf("reconcileMigration: %v", err) + } + if handled { + t.Error("a CR wake without the Migrating phase must not be repainted as a migration") + } + if len(reg.deleted) != 0 { + t.Errorf("must not touch the wake's snapshot: %v", reg.deleted) + } +} + +func migCocoonSet(node string) *cocoonv1.CocoonSet { + return newCocoonSet("demo", func(cs *cocoonv1.CocoonSet) { cs.Spec.NodeName = node }) +} + +func migMainPod(t *testing.T, cs *cocoonv1.CocoonSet, node, vmid string, running bool) *corev1.Pod { + t.Helper() + pod := mustBuildAgentPod(t, cs, 0, "", "", testScheme(t)) + pod.Spec.NodeName = node + if vmid != "" { + meta.VMRuntime{VMID: vmid}.Apply(pod) + } + if running { + pod.Status.ContainerStatuses = []corev1.ContainerStatus{ + {State: corev1.ContainerState{Running: &corev1.ContainerStateRunning{}}}, + } + } + return pod +} diff --git a/cocoonset/pods.go b/cocoonset/pods.go index ed2cf49..0da3967 100644 --- a/cocoonset/pods.go +++ b/cocoonset/pods.go @@ -110,10 +110,30 @@ func buildAgentPod(cs *cocoonv1.CocoonSet, slot int32, mainVMName, bindNodeName } if bindNodeName != "" { pod.Spec.NodeName = bindNodeName + } else if slot == 0 && cs.Spec.NodeName != "" { + // Scheduler affinity, not a hard NodeName bind: the main lands only if + // it fits and the node is schedulable, else stays Pending. + pod.Spec.Affinity = hostnameAffinity(cs.Spec.NodeName) } return pod, nil } +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}, + }}, + }}, + }, + }, + } +} + 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/pods_test.go b/cocoonset/pods_test.go index bd9af50..d2b9e32 100644 --- a/cocoonset/pods_test.go +++ b/cocoonset/pods_test.go @@ -554,3 +554,61 @@ func TestPodSpecMatchesToolboxDetectsNodePoolDrift(t *testing.T) { t.Error("toolbox pod with old node pool should not match updated spec") } } + +func TestBuildAgentPodMainPinnedViaHostnameAffinity(t *testing.T) { + cs := newCocoonSet("demo", func(cs *cocoonv1.CocoonSet) { + cs.Spec.NodeName = "node-b" + }) + pod := mustBuildAgentPod(t, cs, 0, "", "", testScheme(t)) + + if pod.Spec.NodeName != "" { + t.Errorf("main must not be hard-bound; NodeName=%q", pod.Spec.NodeName) + } + if pod.Spec.Affinity == nil || pod.Spec.Affinity.NodeAffinity == nil { + t.Fatalf("expected node affinity, got %+v", pod.Spec.Affinity) + } + na := pod.Spec.Affinity.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution + if na == nil || len(na.NodeSelectorTerms) != 1 || len(na.NodeSelectorTerms[0].MatchExpressions) != 1 { + t.Fatalf("expected one hostname node-affinity term, got %+v", pod.Spec.Affinity) + } + req := na.NodeSelectorTerms[0].MatchExpressions[0] + if req.Key != corev1.LabelHostname || req.Operator != corev1.NodeSelectorOpIn || len(req.Values) != 1 || req.Values[0] != "node-b" { + t.Errorf("affinity req = %+v, want %s In [node-b]", req, corev1.LabelHostname) + } +} + +func TestBuildAgentPodNoAffinityWhenNodeNameEmpty(t *testing.T) { + cs := newCocoonSet("demo") + pod := mustBuildAgentPod(t, cs, 0, "", "", testScheme(t)) + if pod.Spec.Affinity != nil { + t.Errorf("no nodeName => no affinity, got %+v", pod.Spec.Affinity) + } +} + +func TestBuildAgentPodSubAgentIgnoresNodeNameAffinity(t *testing.T) { + cs := newCocoonSet("demo", func(cs *cocoonv1.CocoonSet) { + cs.Spec.Agent.Replicas = 2 + cs.Spec.NodeName = "node-b" + }) + pod := mustBuildAgentPod(t, cs, 1, "vk-ns-demo-0", "node-a", testScheme(t)) + if pod.Spec.NodeName != "node-a" { + t.Errorf("sub-agent must hard-bind to main's node; NodeName=%q want node-a", pod.Spec.NodeName) + } + if pod.Spec.Affinity != nil { + t.Errorf("sub-agent must not get hostname affinity, got %+v", pod.Spec.Affinity) + } +} + +// Affinity/nodeName are placement-only: if podSpecMatchesAgent ever compared +// them, every pinned CocoonSet would drift into a delete/recreate loop. +func TestPodSpecMatchesAgentIgnoresAffinity(t *testing.T) { + cs := newCocoonSet("demo", func(cs *cocoonv1.CocoonSet) { + cs.Spec.NodeName = "node-b" + }) + // A pre-migration pod: built before nodeName was set, so no affinity. + pre := newCocoonSet("demo") + pod := mustBuildAgentPod(t, pre, 0, "", "", testScheme(t)) + if !podSpecMatchesAgent(pod, cs, 0) { + t.Error("setting spec.nodeName must not drift-delete the existing pod") + } +} diff --git a/cocoonset/reconciler.go b/cocoonset/reconciler.go index ca403e3..6964c7b 100644 --- a/cocoonset/reconciler.go +++ b/cocoonset/reconciler.go @@ -26,6 +26,7 @@ const ( finalizerName = "cocoonset.cocoonstack.io/finalizer" requeueWaitForMain = 5 * time.Second requeueSuspendPoll = 5 * time.Second + requeueMigratePoll = 5 * time.Second ) // Reconciler watches CocoonSet resources and manages the lifecycle of agent @@ -103,6 +104,12 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu return r.reconcileSuspend(ctx, &cs, classified) } + // Cross-node migration owns the reconcile while in flight; runs before + // applyUnsuspend so its internal hibernate annotation is not cleared. + if handled, res, err := r.reconcileMigration(ctx, &cs, classified); handled { + return res, err + } + // Clear stale hibernate annotations from a prior suspend pass. if err := r.applyUnsuspend(ctx, cs.Namespace, classified); err != nil { return ctrl.Result{}, err diff --git a/config/crd/bases/cocoonset.cocoonstack.io_cocoonsets.yaml b/config/crd/bases/cocoonset.cocoonstack.io_cocoonsets.yaml index 037fba6..2c1f6a5 100644 --- a/config/crd/bases/cocoonset.cocoonstack.io_cocoonsets.yaml +++ b/config/crd/bases/cocoonset.cocoonstack.io_cocoonsets.yaml @@ -160,6 +160,7 @@ spec: - linux - windows - android + - macos type: string probePort: description: ProbePort overrides the default ICMP probe with a @@ -247,6 +248,12 @@ spec: required: - image type: object + nodeName: + description: |- + NodeName pins the VM to a node (cross-node migrate). Empty = let the + scheduler place it within NodePool and leave it alone; a value adds a + hostname nodeAffinity so the pod lands there or stays Pending if it won't fit. + type: string nodePool: default: default type: string @@ -322,6 +329,7 @@ spec: - linux - windows - android + - macos type: string probePort: description: ProbePort overrides the default ICMP probe with @@ -522,6 +530,7 @@ spec: - Scaling - Suspending - Suspended + - Migrating - Failed type: string readyAgents: From e12f600bd25d3874b6bcce95ad9af3f484072879 Mon Sep 17 00:00:00 2001 From: CMGS Date: Thu, 2 Jul 2026 15:51:38 +0800 Subject: [PATCH 2/2] fix(main): surface controller-runtime errors through core/log MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit crlog was set to logr.Discard(), so every reconcile error controller-runtime retried (returned by reconcilers, not logged at call sites) vanished — the migration E2E's registry 403s were invisible. Bridge logr to core/log: errors always forwarded (nil-err anomaly reports downgrade to Warn since core/log drops nil-err Error lines), V(0) info kept, V(1)+ internals chatter dropped. --- logbridge.go | 81 +++++++++++++++++++++++++++++++++++++++++++++++ logbridge_test.go | 61 +++++++++++++++++++++++++++++++++++ main.go | 6 ++-- 3 files changed, 145 insertions(+), 3 deletions(-) create mode 100644 logbridge.go create mode 100644 logbridge_test.go diff --git a/logbridge.go b/logbridge.go new file mode 100644 index 0000000..169042d --- /dev/null +++ b/logbridge.go @@ -0,0 +1,81 @@ +package main + +import ( + "context" + "fmt" + "strings" + + "github.com/go-logr/logr" + "github.com/projecteru2/core/log" +) + +// crSink forwards controller-runtime's logr output to core/log: reconcile +// errors surface only there, so discarding them (the old setup) hid them all. +type crSink struct { + ctx context.Context + name string + kv []any +} + +// newCRLogger wraps a crSink for crlog.SetLogger. The root name stays empty; +// controller-runtime adds its own via WithName — a fixed root would double up. +func newCRLogger(ctx context.Context) logr.Logger { + return logr.New(&crSink{ctx: ctx}) +} + +func (s *crSink) Init(logr.RuntimeInfo) {} + +// Enabled passes V(0) only; errors bypass this gate entirely (logr contract). +func (s *crSink) Enabled(level int) bool { return level == 0 } + +func (s *crSink) Info(_ int, msg string, kvs ...any) { + log.WithFunc(s.funcName()).Info(s.ctx, s.line(msg, kvs)) +} + +func (s *crSink) Error(err error, msg string, kvs ...any) { + if err == nil { + // logr allows Error(nil, ...) for anomaly reports, but core/log drops + // nil-err Error lines entirely; keep them visible as warnings. + log.WithFunc(s.funcName()).Warn(s.ctx, s.line(msg, kvs)) + return + } + log.WithFunc(s.funcName()).Error(s.ctx, err, s.line(msg, kvs)) +} + +func (s *crSink) WithValues(kvs ...any) logr.LogSink { + next := *s + next.kv = append(append([]any{}, s.kv...), kvs...) + return &next +} + +func (s *crSink) WithName(name string) logr.LogSink { + next := *s + if s.name == "" { + next.name = name + } else { + next.name = s.name + "." + name + } + return &next +} + +// funcName labels unnamed root output so its origin stays identifiable. +func (s *crSink) funcName() string { + if s.name == "" { + return "controller-runtime" + } + return s.name +} + +func (s *crSink) line(msg string, kvs []any) string { + pairs := make([]any, 0, len(s.kv)+len(kvs)) + pairs = append(append(pairs, s.kv...), kvs...) + if len(pairs) == 0 { + return msg + } + var b strings.Builder + b.WriteString(msg) + for i := 0; i+1 < len(pairs); i += 2 { + fmt.Fprintf(&b, " %v=%v", pairs[i], pairs[i+1]) + } + return b.String() +} diff --git a/logbridge_test.go b/logbridge_test.go new file mode 100644 index 0000000..fe92c68 --- /dev/null +++ b/logbridge_test.go @@ -0,0 +1,61 @@ +package main + +import ( + "testing" +) + +func TestCRSinkLine(t *testing.T) { + cases := []struct { + name string + base []any + kvs []any + want string + }{ + {"no pairs", nil, nil, "msg"}, + {"call pairs", nil, []any{"controller", "cocoonset"}, "msg controller=cocoonset"}, + {"accumulated then call", []any{"ns", "e2e"}, []any{"pod", "demo-0"}, "msg ns=e2e pod=demo-0"}, + {"dangling key dropped", nil, []any{"lone"}, "msg"}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + s := &crSink{ctx: t.Context(), kv: c.base} + if got := s.line("msg", c.kvs); got != c.want { + t.Errorf("line = %q, want %q", got, c.want) + } + }) + } +} + +func TestCRSinkWithNameAndValuesDoNotMutateParent(t *testing.T) { + parent := &crSink{ctx: t.Context(), name: "controller-runtime"} + child := parent.WithName("manager").(*crSink) + grandchild := child.WithValues("k", "v").(*crSink) + if parent.name != "controller-runtime" || len(parent.kv) != 0 { + t.Errorf("parent mutated: name=%q kv=%v", parent.name, parent.kv) + } + if child.name != "controller-runtime.manager" { + t.Errorf("child name = %q", child.name) + } + if grandchild.line("m", nil) != "m k=v" { + t.Errorf("grandchild line = %q", grandchild.line("m", nil)) + } +} + +func TestCRSinkEnabledOnlyV0(t *testing.T) { + s := &crSink{ctx: t.Context()} + if !s.Enabled(0) || s.Enabled(1) || s.Enabled(4) { + t.Error("Enabled must pass V(0) only") + } +} + +func TestCRSinkErrorNilDoesNotPanic(t *testing.T) { + // logr permits Error(nil, ...); core/log drops nil-err Error lines, so the + // sink reroutes to Warn. Output capture is impractical — pin no-panic. + s := &crSink{ctx: t.Context(), name: "controller-runtime"} + s.Error(nil, "update event has no old object", "type", "pod") + s.Error(assertErr{}, "real error path") +} + +type assertErr struct{} + +func (assertErr) Error() string { return "boom" } diff --git a/main.go b/main.go index 5593b41..073b68e 100644 --- a/main.go +++ b/main.go @@ -11,7 +11,6 @@ import ( "os/signal" "syscall" - "github.com/go-logr/logr" "github.com/google/go-containerregistry/pkg/authn" "github.com/google/go-containerregistry/pkg/v1/google" "github.com/projecteru2/core/log" @@ -62,8 +61,9 @@ func main() { fmt.Fprintf(os.Stderr, "setup log: %v\n", err) os.Exit(1) } - // Silence controller-runtime's own logger; we use core/log instead. - crlog.SetLogger(logr.Discard()) + // Route controller-runtime's logger through core/log: reconcile errors + // surface only there, so discarding it hides every one of them. + crlog.SetLogger(newCRLogger(ctx)) logger := log.WithFunc("main") logger.Infof(ctx, "cocoon-operator %s starting (rev=%s built=%s)",