feat(core): detect not converging migration from remaining data stall

Signed-off-by: Daniil Antoshin <daniil.antoshin@flant.com>

Author: danilrwx

images/virtualization-artifact/pkg/controller/vmop/migration/internal/handler/lifecycle.go

@@ -318,6 +318,16 @@ func (h LifecycleHandler) syncOperationComplete(ctx context.Context, vmop *v1alp
 		return err
 	}
 
+	autoConverge := h.resolveAutoConverge(vmop)
+
+	if reason == vmopcondition.ReasonSyncing {
+		record := migrationprogress.BuildRecord(vmop, mig, autoConverge, time.Now())
+		if h.progressStrategy != nil && h.progressStrategy.IsNotConverging(record) {
+			reason = vmopcondition.ReasonNotConverging
+			msg = "Migration is not converging: data remaining is not decreasing at maximum throttle"
+		}
+	}
+
 	vmop.Status.Phase = v1alpha2.VMOPPhaseInProgress
 	if reason == vmopcondition.ReasonTargetScheduling {
 		vmop.Status.Phase = v1alpha2.VMOPPhasePending
@@ -603,8 +613,8 @@ func (h LifecycleHandler) calculateMigrationProgress(
 		return progressTargetPreparing
 	case vmopcondition.ReasonTargetDiskError:
 		return progressTargetPreparing
-	case vmopcondition.ReasonSyncing:
-		record := migrationprogress.BuildRecord(vmop, mig, time.Now())
+	case vmopcondition.ReasonSyncing, vmopcondition.ReasonNotConverging:
+		record := migrationprogress.BuildRecord(vmop, mig, h.resolveAutoConverge(vmop), time.Now())
 		return h.progressStrategy.SyncProgress(record)
 	case vmopcondition.ReasonSourceSuspended:
 		h.forgetProgress(vmop)
@@ -624,6 +634,16 @@ func (h LifecycleHandler) calculateMigrationProgress(
 	}
 }
 
+func (h LifecycleHandler) resolveAutoConverge(vmop *v1alpha2.VirtualMachineOperation) bool {
+	if vmop == nil {
+		return false
+	}
+	if vmop.Spec.Force != nil && *vmop.Spec.Force {
+		return true
+	}
+	return false
+}
+
 func (h LifecycleHandler) forgetProgress(vmop *v1alpha2.VirtualMachineOperation) {
 	if h.progressStrategy == nil || vmop == nil {
 		return

images/virtualization-artifact/pkg/controller/vmop/migration/internal/handler/lifecycle_test.go

@@ -55,6 +55,10 @@ func (s *progressStrategyStub) SyncProgress(_ migrationprogress.Record) int32 {
 	return s.value
 }
 
+func (s *progressStrategyStub) IsNotConverging(_ migrationprogress.Record) bool {
+	return false
+}
+
 func (s *progressStrategyStub) Forget(uid types.UID) {
 	s.forgotten = append(s.forgotten, uid)
 }

images/virtualization-artifact/pkg/controller/vmop/migration/internal/progress/mapper.go

@@ -26,11 +26,12 @@ import (
 
 const unknownMetric = -1.0
 
-func BuildRecord(vmop *v1alpha2.VirtualMachineOperation, mig *virtv1.VirtualMachineInstanceMigration, now time.Time) Record {
+func BuildRecord(vmop *v1alpha2.VirtualMachineOperation, mig *virtv1.VirtualMachineInstanceMigration, autoConverge bool, now time.Time) Record {
 	record := Record{
 		Now:              now,
 		StartedAt:        now,
 		PreviousProgress: previousProgress(vmop),
+		AutoConverge:     autoConverge,
 		DataTotalMiB:     unknownMetric,
 		DataProcessedMiB: unknownMetric,
 		DataRemainingMiB: unknownMetric,

images/virtualization-artifact/pkg/controller/vmop/migration/internal/progress/mapper_test.go

@@ -31,7 +31,7 @@ import (
 func TestBuildRecord_NilVMOPAndMigration(t *testing.T) {
 	now := time.Unix(1710000000, 0)
 
-	record := BuildRecord(nil, nil, now)
+	record := BuildRecord(nil, nil, false, now)
 
 	if !record.StartedAt.Equal(now) {
 		t.Fatalf("expected StartedAt=%v, got %v", now, record.StartedAt)
@@ -54,7 +54,7 @@ func TestBuildRecord_UsesVMOPCreationTimestampAndPreviousProgress(t *testing.T)
 		Status: v1alpha2.VirtualMachineOperationStatus{Progress: ptr.To[int32](42)},
 	}
 
-	record := BuildRecord(vmop, nil, now)
+	record := BuildRecord(vmop, nil, false, now)
 
 	if record.OperationUID != vmop.UID {
 		t.Fatalf("expected OperationUID=%s, got %s", vmop.UID, record.OperationUID)
@@ -90,7 +90,7 @@ func TestBuildRecord_UsesMigrationState(t *testing.T) {
 		},
 	}
 
-	record := BuildRecord(nil, mig, now)
+	record := BuildRecord(nil, mig, false, now)
 
 	if record.Phase != virtv1.MigrationRunning {
 		t.Fatalf("expected Phase=%s, got %s", virtv1.MigrationRunning, record.Phase)

images/virtualization-artifact/pkg/controller/vmop/migration/internal/progress/progress.go

@@ -33,6 +33,8 @@ const (
 	iterativeCeiling = 90.0
 	thresholdFactor  = 0.05
 
+	notConvergingWindow = 10 * time.Second
+
 	bulkTimeRate      = 0.55
 	iterBaseTimeRate  = 0.022
 	iterThrottleRate  = 0.0012
@@ -49,6 +51,7 @@ const (
 
 type Strategy interface {
 	SyncProgress(record Record) int32
+	IsNotConverging(record Record) bool
 	Forget(uid types.UID)
 }
 
@@ -64,6 +67,7 @@ type Record struct {
 	HasThrottle          bool
 	AutoConvergeThrottle uint32
 	Throttle             float64
+	AutoConverge         bool
 	DataTotalMiB         float64
 	DataProcessedMiB     float64
 	DataRemainingMiB     float64
@@ -149,6 +153,7 @@ func (p *Progress) SyncProgress(record Record) int32 {
 	}
 
 	updateMetricState(record, &state)
+	updateMinRemaining(record, &state)
 	state.Progress = next
 	state.LastUpdatedAt = record.Now
 	state.LastIteration = record.Iteration
@@ -161,6 +166,27 @@ func (p *Progress) SyncProgress(record Record) int32 {
 	return next
 }
 
+func (p *Progress) IsNotConverging(record Record) bool {
+	if p == nil || p.store == nil || record.OperationUID == "" {
+		return false
+	}
+
+	state, ok := p.store.Load(record.OperationUID)
+	if !ok || !state.Iterative {
+		return false
+	}
+
+	if !isAtMaxThrottle(record) {
+		return false
+	}
+
+	if state.MinRemaining <= 0 || state.MinRemainingAt.IsZero() {
+		return false
+	}
+
+	return record.Now.Sub(state.MinRemainingAt) >= notConvergingWindow
+}
+
 func (p *Progress) getState(record Record) State {
 	if p == nil || p.store == nil || record.OperationUID == "" {
 		return State{Progress: clampSyncRange(record.PreviousProgress), LastMetricAt: record.Now}
@@ -324,6 +350,24 @@ func maxRemaining(record Record) float64 {
 	return 0
 }
 
+func updateMinRemaining(record Record, state *State) {
+	remaining := maxRemaining(record)
+	if remaining <= 0 {
+		return
+	}
+	if state.MinRemaining <= 0 || remaining < state.MinRemaining {
+		state.MinRemaining = remaining
+		state.MinRemainingAt = record.Now
+	}
+}
+
+func isAtMaxThrottle(record Record) bool {
+	if !record.AutoConverge {
+		return true
+	}
+	return record.HasThrottle && record.Throttle >= 0.99
+}
+
 func updateMetricState(record Record, state *State) {
 	if !metricChanged(record, state) {
 		return

images/virtualization-artifact/pkg/controller/vmop/migration/internal/progress/store.go

@@ -42,6 +42,8 @@ type State struct {
 	SmoothedRemaining float64
 	Threshold         float64
 	LastIncreaseAt    time.Time
+	MinRemaining      float64
+	MinRemainingAt    time.Time
 }
 
 type Store struct {