Skip to content

refactor(phase2): dedup VTEP registration and suspend-state cleanup #27

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
syscod3 merged 1 commit into from
Jul 7, 2026
Merged

refactor(phase2): dedup VTEP registration and suspend-state cleanup #27

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
64 changes: 32 additions & 32 deletions internal/agent/reconciler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)

Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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
Expand Down
18 changes: 10 additions & 8 deletions internal/controller/impvm_controller.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)

Expand Down Expand Up @@ -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
}
Expand All @@ -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
}
Expand Down
4 changes: 2 additions & 2 deletions internal/controller/impvm_scheduler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
})
Expand Down
24 changes: 12 additions & 12 deletions internal/controller/scheduler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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,
Expand Down
34 changes: 17 additions & 17 deletions internal/controller/scheduler_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand All @@ -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) {
Expand All @@ -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 {
Expand All @@ -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 {
Expand All @@ -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 {
Expand Down
Loading