From b3f9201df71717eae735c0889dcfb9b686974bdd Mon Sep 17 00:00:00 2001 From: Giovanni Ferri Date: Wed, 8 Jul 2026 00:08:30 +0100 Subject: [PATCH] refactor(phase2): dedup VTEP registration and suspend-state cleanup Post-merge cleanup pass over the Phase 2 suspend-on-idle diff: - Extract ensureVTEPAndFDB helper; route handleScheduled and handleResuming through it. Resume now registers the VTEP under a vtep_register child span (previously omitted) and log wording is unified. - handleTerminating: compute the suspend dir once instead of calling suspendDirFor twice. - scheduler: rename NodeInfo.Used{VCPU,MemoryMiB} to Resident{VCPU,MemoryMiB} to match its documented meaning; inline the single-use oversubscribed local. - Extract clearSuspendState helper for the two node-loss reschedule branches. --- internal/agent/reconciler.go | 64 ++++++++++++------------- internal/controller/impvm_controller.go | 18 +++---- internal/controller/impvm_scheduler.go | 4 +- internal/controller/scheduler.go | 24 +++++----- internal/controller/scheduler_test.go | 34 ++++++------- 5 files changed, 73 insertions(+), 71 deletions(-) diff --git a/internal/agent/reconciler.go b/internal/agent/reconciler.go index 208686a..09fd303 100644 --- a/internal/agent/reconciler.go +++ b/internal/agent/reconciler.go @@ -206,33 +206,39 @@ func (r *ImpVMReconciler) handleScheduled(ctx context.Context, vm *impdevv1alpha } // Register VTEP entry so the operator and other nodes know where this VM lives. - if vm.Spec.NetworkRef != nil && state.IP != "" && r.NodeIP != "" { - macAddr := network.MACAddr(vm.Namespace + "/" + vm.Name) - { - vCtx, vSpan := otel.Tracer("imp.agent").Start(ctx, "agent.impvm.vtep_register", - trace.WithAttributes( - attribute.String("vm.name", vm.Name), - attribute.String("vm.ip", state.IP), - ), - ) - vtepErr := r.registerVTEP(vCtx, vm, state.IP, macAddr) - tracing.RecordError(vSpan, vtepErr) - vSpan.End() - if vtepErr != nil { - log.Error(vtepErr, "registerVTEP failed — FDB sync may be incomplete") - } else { - // Sync local FDB now that this node has a VTEP entry. - if err := r.syncFDB(ctx, vm); err != nil { - log.Error(err, "syncFDB after registerVTEP failed") - } - } - } - } + r.ensureVTEPAndFDB(ctx, vm, state.IP) log.Info("VM started", "pid", pid, "ip", state.IP) return ctrl.Result{}, nil } +// ensureVTEPAndFDB registers vm's VTEP entry (under a child span) and syncs the +// local FDB so the operator and other nodes route to it. No-op when the VM has +// no network, no IP, or this node has no IP. Best-effort: errors are logged. +func (r *ImpVMReconciler) ensureVTEPAndFDB(ctx context.Context, vm *impdevv1alpha1.ImpVM, ip string) { + if vm.Spec.NetworkRef == nil || ip == "" || r.NodeIP == "" { + return + } + log := logf.FromContext(ctx) + macAddr := network.MACAddr(vm.Namespace + "/" + vm.Name) + vCtx, vSpan := otel.Tracer("imp.agent").Start(ctx, "agent.impvm.vtep_register", + trace.WithAttributes( + attribute.String("vm.name", vm.Name), + attribute.String("vm.ip", ip), + ), + ) + vtepErr := r.registerVTEP(vCtx, vm, ip, macAddr) + tracing.RecordError(vSpan, vtepErr) + vSpan.End() + if vtepErr != nil { + log.Error(vtepErr, "Failed to register VTEP — FDB sync may be incomplete") + return + } + if err := r.syncFDB(ctx, vm); err != nil { + log.Error(err, "Failed to sync FDB after registering VTEP") + } +} + func (r *ImpVMReconciler) handleRunning(ctx context.Context, vm *impdevv1alpha1.ImpVM) (ctrl.Result, error) { log := logf.FromContext(ctx) @@ -358,8 +364,9 @@ func (r *ImpVMReconciler) handleTerminating(ctx context.Context, vm *impdevv1alp } // Remove any node-local suspend snapshot so it does not outlive the VM. - if err := os.RemoveAll(r.suspendDirFor(vm)); err != nil { - log.Error(err, "Failed to remove suspend snapshot dir", "dir", r.suspendDirFor(vm)) + dir := r.suspendDirFor(vm) + if err := os.RemoveAll(dir); err != nil { + log.Error(err, "Failed to remove suspend snapshot dir", "dir", dir) } return r.clearOwnership(ctx, vm) @@ -495,14 +502,7 @@ func (r *ImpVMReconciler) handleResuming(ctx context.Context, vm *impdevv1alpha1 } // Re-register VTEP + sync FDB so other nodes route to the resumed VM. - if vm.Spec.NetworkRef != nil && state.IP != "" && r.NodeIP != "" { - macAddr := network.MACAddr(vm.Namespace + "/" + vm.Name) - if vtepErr := r.registerVTEP(ctx, vm, state.IP, macAddr); vtepErr != nil { - log.Error(vtepErr, "Failed to register VTEP after resume — FDB sync may be incomplete") - } else if fdbErr := r.syncFDB(ctx, vm); fdbErr != nil { - log.Error(fdbErr, "Failed to sync FDB after resume") - } - } + r.ensureVTEPAndFDB(ctx, vm, state.IP) log.Info("VM resumed", "pid", pid, "ip", state.IP) return ctrl.Result{}, nil diff --git a/internal/controller/impvm_controller.go b/internal/controller/impvm_controller.go index 1deac08..a2d4f5a 100644 --- a/internal/controller/impvm_controller.go +++ b/internal/controller/impvm_controller.go @@ -225,6 +225,14 @@ func (r *ImpVMReconciler) reconcileExpiry(ctx context.Context, vm *impdevv1alpha return remaining, false, nil } +// clearSuspendState drops the node-local suspend snapshot reference so a VM +// rescheduled after node loss cold-boots instead of failing to resume from a +// path that died with the node. +func clearSuspendState(vm *impdevv1alpha1.ImpVM) { + vm.Status.SuspendSnapshotPath = "" + vm.Status.SuspendedAt = nil +} + func (r *ImpVMReconciler) syncStatus(ctx context.Context, vm *impdevv1alpha1.ImpVM) (ctrl.Result, error) { log := logf.FromContext(ctx) @@ -255,10 +263,7 @@ func (r *ImpVMReconciler) syncStatus(ctx context.Context, vm *impdevv1alpha1.Imp setNodeUnhealthy(vm, reason) vm.Status.Phase = impdevv1alpha1.VMPhasePending setUnscheduled(vm) - // The node-local suspend snapshot died with the node; clear it so the - // rescheduled VM cold-boots instead of failing to resume from a lost path. - vm.Status.SuspendSnapshotPath = "" - vm.Status.SuspendedAt = nil + clearSuspendState(vm) if err2 := r.Status().Patch(ctx, vm, client.MergeFrom(vmCopy)); err2 != nil { return ctrl.Result{}, err2 } @@ -282,10 +287,7 @@ func (r *ImpVMReconciler) syncStatus(ctx context.Context, vm *impdevv1alpha1.Imp setNodeUnhealthy(vm, reason) vm.Status.Phase = impdevv1alpha1.VMPhasePending setUnscheduled(vm) - // Node-local suspend snapshot is gone with the node; clear it so the - // rescheduled VM cold-boots instead of failing to resume. - vm.Status.SuspendSnapshotPath = "" - vm.Status.SuspendedAt = nil + clearSuspendState(vm) if err2 := r.Status().Patch(ctx, vm, client.MergeFrom(vmCopy)); err2 != nil { return ctrl.Result{}, err2 } diff --git a/internal/controller/impvm_scheduler.go b/internal/controller/impvm_scheduler.go index e5dc564..4ce1242 100644 --- a/internal/controller/impvm_scheduler.go +++ b/internal/controller/impvm_scheduler.go @@ -152,8 +152,8 @@ func (r *ImpVMReconciler) schedule(ctx context.Context, vm *impdevv1alpha1.ImpVM NodeName: node.Name, VCPUCapacity: profile.Spec.VCPUCapacity, MemoryMiB: profile.Spec.MemoryMiB, - UsedVCPU: used.residentVCPU, - UsedMemoryMiB: used.residentMem, + ResidentVCPU: used.residentVCPU, + ResidentMemoryMiB: used.residentMem, ReservedVCPU: used.reservedVCPU, ReservedMemoryMiB: used.reservedMem, }) diff --git a/internal/controller/scheduler.go b/internal/controller/scheduler.go index 428e6b2..76ee59a 100644 --- a/internal/controller/scheduler.go +++ b/internal/controller/scheduler.go @@ -15,12 +15,13 @@ type NodeInfo struct { NodeName string VCPUCapacity int32 MemoryMiB int64 - // UsedVCPU/UsedMemoryMiB is the RESIDENT usage: capacity consumed by VMs whose - // memory is actually present on the node. Suspended VMs are excluded (their - // memory is freed to node-local disk). This is the hard scheduling constraint - // and the basis of overcommit — new VMs pack into capacity freed by suspension. - UsedVCPU int32 - UsedMemoryMiB int64 + // ResidentVCPU/ResidentMemoryMiB is the RESIDENT usage: capacity consumed by + // VMs whose memory is actually present on the node. Suspended VMs are excluded + // (their memory is freed to node-local disk). This is the hard scheduling + // constraint and the basis of overcommit — new VMs pack into capacity freed by + // suspension. + ResidentVCPU int32 + ResidentMemoryMiB int64 // ReservedVCPU/ReservedMemoryMiB additionally includes Suspended VMs, which // retain a logical claim on the node. Tracked for observability only; it may // exceed capacity (intentional, accepted overcommit — see the Phase 2 plan @@ -45,19 +46,18 @@ func Schedule(log logr.Logger, vcpu int32, memMiB int64, nodes []NodeInfo) (stri var candidates []candidate for _, n := range nodes { - freeVCPU := n.VCPUCapacity - n.UsedVCPU - freeMemMiB := n.MemoryMiB - n.UsedMemoryMiB + freeVCPU := n.VCPUCapacity - n.ResidentVCPU + freeMemMiB := n.MemoryMiB - n.ResidentMemoryMiB fits := freeVCPU >= vcpu && freeMemMiB >= memMiB - // oversubscribed is informational: reserved (incl. suspended VMs) exceeds - // capacity, so a mass resume could exhaust the node. - oversubscribed := n.ReservedMemoryMiB > n.MemoryMiB || n.ReservedVCPU > n.VCPUCapacity log.V(1).Info("Scheduling candidate", "node", n.NodeName, "freeVCPU", freeVCPU, "freeMemMiB", freeMemMiB, "reservedVCPU", n.ReservedVCPU, "reservedMemMiB", n.ReservedMemoryMiB, - "oversubscribed", oversubscribed, + // oversubscribed is informational: reserved (incl. suspended VMs) exceeds + // capacity, so a mass resume could exhaust the node. + "oversubscribed", n.ReservedMemoryMiB > n.MemoryMiB || n.ReservedVCPU > n.VCPUCapacity, "required.vcpu", vcpu, "required.memMiB", memMiB, "fits", fits, diff --git a/internal/controller/scheduler_test.go b/internal/controller/scheduler_test.go index 929523a..f22b545 100644 --- a/internal/controller/scheduler_test.go +++ b/internal/controller/scheduler_test.go @@ -9,11 +9,11 @@ import ( func TestSchedule_singleNodeFits(t *testing.T) { nodes := []NodeInfo{{ - NodeName: "node1", - VCPUCapacity: 8, - MemoryMiB: 8192, - UsedVCPU: 2, - UsedMemoryMiB: 1024, + NodeName: "node1", + VCPUCapacity: 8, + MemoryMiB: 8192, + ResidentVCPU: 2, + ResidentMemoryMiB: 1024, }} got, err := Schedule(logr.Discard(), 4, 2048, nodes) if err != nil { @@ -26,11 +26,11 @@ func TestSchedule_singleNodeFits(t *testing.T) { func TestSchedule_noFit_returnsUnschedulable(t *testing.T) { nodes := []NodeInfo{{ - NodeName: "node1", - VCPUCapacity: 4, - MemoryMiB: 4096, - UsedVCPU: 3, - UsedMemoryMiB: 4000, + NodeName: "node1", + VCPUCapacity: 4, + MemoryMiB: 4096, + ResidentVCPU: 3, + ResidentMemoryMiB: 4000, }} _, err := Schedule(logr.Discard(), 2, 200, nodes) if !errors.Is(err, ErrUnschedulable) { @@ -47,9 +47,9 @@ func TestSchedule_emptyNodeList_returnsUnschedulable(t *testing.T) { func TestSchedule_tieBreak_picksHighestFreeMemory(t *testing.T) { nodes := []NodeInfo{ - {NodeName: "node-a", VCPUCapacity: 8, MemoryMiB: 8192, UsedVCPU: 2, UsedMemoryMiB: 2048}, // free: 6 cpu, 6144 mem - {NodeName: "node-b", VCPUCapacity: 8, MemoryMiB: 8192, UsedVCPU: 2, UsedMemoryMiB: 1024}, // free: 6 cpu, 7168 mem — wins - {NodeName: "node-c", VCPUCapacity: 8, MemoryMiB: 8192, UsedVCPU: 2, UsedMemoryMiB: 4096}, // free: 6 cpu, 4096 mem + {NodeName: "node-a", VCPUCapacity: 8, MemoryMiB: 8192, ResidentVCPU: 2, ResidentMemoryMiB: 2048}, // free: 6 cpu, 6144 mem + {NodeName: "node-b", VCPUCapacity: 8, MemoryMiB: 8192, ResidentVCPU: 2, ResidentMemoryMiB: 1024}, // free: 6 cpu, 7168 mem — wins + {NodeName: "node-c", VCPUCapacity: 8, MemoryMiB: 8192, ResidentVCPU: 2, ResidentMemoryMiB: 4096}, // free: 6 cpu, 4096 mem } got, err := Schedule(logr.Discard(), 2, 512, nodes) if err != nil { @@ -62,8 +62,8 @@ func TestSchedule_tieBreak_picksHighestFreeMemory(t *testing.T) { func TestSchedule_vcpuConstraintFiltersNode(t *testing.T) { nodes := []NodeInfo{ - {NodeName: "small", VCPUCapacity: 4, MemoryMiB: 8192, UsedVCPU: 3, UsedMemoryMiB: 0}, // only 1 free VCPU - {NodeName: "large", VCPUCapacity: 8, MemoryMiB: 8192, UsedVCPU: 2, UsedMemoryMiB: 0}, // 6 free VCPUs + {NodeName: "small", VCPUCapacity: 4, MemoryMiB: 8192, ResidentVCPU: 3, ResidentMemoryMiB: 0}, // only 1 free VCPU + {NodeName: "large", VCPUCapacity: 8, MemoryMiB: 8192, ResidentVCPU: 2, ResidentMemoryMiB: 0}, // 6 free VCPUs } got, err := Schedule(logr.Discard(), 4, 512, nodes) if err != nil { @@ -76,8 +76,8 @@ func TestSchedule_vcpuConstraintFiltersNode(t *testing.T) { func TestSchedule_memoryConstraintFiltersNode(t *testing.T) { nodes := []NodeInfo{ - {NodeName: "low-mem", VCPUCapacity: 8, MemoryMiB: 2048, UsedVCPU: 0, UsedMemoryMiB: 1900}, // only 148 MiB free - {NodeName: "hi-mem", VCPUCapacity: 8, MemoryMiB: 8192, UsedVCPU: 0, UsedMemoryMiB: 1024}, // 7168 MiB free + {NodeName: "low-mem", VCPUCapacity: 8, MemoryMiB: 2048, ResidentVCPU: 0, ResidentMemoryMiB: 1900}, // only 148 MiB free + {NodeName: "hi-mem", VCPUCapacity: 8, MemoryMiB: 8192, ResidentVCPU: 0, ResidentMemoryMiB: 1024}, // 7168 MiB free } got, err := Schedule(logr.Discard(), 1, 512, nodes) if err != nil {