[virt-controller]: consider the ParallelOutboundMigrationsPerNode when evicting VMs

pkg/virt-controller/watch/drain/evacuation/evacuation.go

@@ -375,20 +375,20 @@ func (c *EvacuationController) sync(node *k8sv1.Node, vmisOnNode []*virtv1.Virtu
 	}
 
 	migrationCandidates, nonMigrateable := c.filterRunningNonMigratingVMIs(vmisToMigrate, activeMigrations)
-
-	// Don't create hundreds of pending migration objects.
-	// This is just best-effort and is *not* intended to not overload the cluster.
-	// It is possible that more migrations than the limit are created because of evacuations on other nodes.
-	// The migration controller needs to limit itself to a reasonable number of running migrations
+	activeMigrationsFromThisSourceNode := c.numOfVMIMForThisSourceNode(vmisOnNode, activeMigrations)
+	maxParallelMigrationsPerOutboundNode :=
+		int(*c.clusterConfig.GetMigrationConfiguration().ParallelOutboundMigrationsPerNode)
 	maxParallelMigrations := int(*c.clusterConfig.GetMigrationConfiguration().ParallelMigrationsPerCluster)
-	if len(activeMigrations) >= maxParallelMigrations {
-		// We have to re-enqueue if some work is left, since migrations from other controllers or workers` don't wake us up again
+	freeSpotsPerCluster := maxParallelMigrations - len(activeMigrations)
+	freeSpotsPerThisSourceNode := maxParallelMigrationsPerOutboundNode - activeMigrationsFromThisSourceNode
+	freeSpots := int(math.Min(float64(freeSpotsPerCluster), float64(freeSpotsPerThisSourceNode)))
+	if freeSpots <= 0 {
 		if len(migrationCandidates) > 0 || len(nonMigrateable) > 0 {
 			c.Queue.AddAfter(node.Name, 5*time.Second)
+			return nil
 		}
-		return nil
 	}
-	freeSpots := maxParallelMigrations - len(activeMigrations)
+
 	diff := int(math.Min(float64(freeSpots), float64(len(migrationCandidates))))
 	remaining := freeSpots - diff
 	remainingForNonMigrateableDiff := int(math.Min(float64(remaining), float64(len(nonMigrateable))))
@@ -527,3 +527,21 @@ func nodeHasTaint(taint *k8sv1.Taint, node *k8sv1.Node) bool {
 	}
 	return false
 }
+
+func (c *EvacuationController) numOfVMIMForThisSourceNode(
+	vmisOnNode []*virtv1.VirtualMachineInstance,
+	activeMigrations []*virtv1.VirtualMachineInstanceMigration) (activeMigrationsFromThisSourceNode int) {
+
+	vmiMap := make(map[string]bool)
+	for _, vmi := range vmisOnNode {
+		vmiMap[vmi.Name] = true
+	}
+
+	for _, vmim := range activeMigrations {
+		if _, ok := vmiMap[vmim.Spec.VMIName]; ok {
+			activeMigrationsFromThisSourceNode++
+		}
+	}
+
+	return
+}

pkg/virt-controller/watch/drain/evacuation/evacuation_test.go

@@ -1,6 +1,7 @@
 package evacuation_test
 
 import (
+	"fmt"
 	"time"
 
 	"github.com/golang/mock/gomock"
@@ -209,26 +210,29 @@ var _ = Describe("Evacuation", func() {
 			node.Spec.Taints = append(node.Spec.Taints, *newTaint())
 			addNode(node)
 
-			vmi := newVirtualMachine("testvm", node.Name)
-			vmi.Spec.EvictionStrategy = newEvictionStrategyLiveMigrate()
-			vmi1 := newVirtualMachine("testvm1", node.Name)
-			vmi1.Spec.EvictionStrategy = newEvictionStrategyLiveMigrate()
-			vmiFeeder.Add(vmi)
+			vmi1 := newVirtualMachineMarkedForEviction("testvmi1", node.Name)
+			migration1 := newMigration("mig1", vmi1.Name, v1.MigrationRunning)
 			vmiFeeder.Add(vmi1)
-			migration := newMigration("mig1", vmi.Name, v1.MigrationRunning)
+			migrationFeeder.Add(migration1)
 
-			migrationFeeder.Add(migration)
-			migrationFeeder.Add(newMigration("mig2", vmi.Name, v1.MigrationRunning))
-			migrationFeeder.Add(newMigration("mig3", vmi.Name, v1.MigrationRunning))
-			migrationFeeder.Add(newMigration("mig4", vmi.Name, v1.MigrationRunning))
-			migrationFeeder.Add(newMigration("mig5", vmi.Name, v1.MigrationRunning))
+			vmi2 := newVirtualMachineMarkedForEviction("testvmi2", node.Name)
+			migration2 := newMigration("mig2", vmi1.Name, v1.MigrationRunning)
+			vmiFeeder.Add(vmi2)
+			migrationFeeder.Add(migration2)
+
+			vmi3 := newVirtualMachineMarkedForEviction("testvmi3", node.Name)
+			vmiFeeder.Add(vmi3)
 
 			controller.Execute()
 
-			migration.Status.Phase = v1.MigrationSucceeded
-			migrationFeeder.Modify(migration)
+			migration2.Status.Phase = v1.MigrationSucceeded
+			migrationFeeder.Modify(migration2)
+
+			migrationInterface.
+				EXPECT().
+				Create(gomock.Any(), &v13.CreateOptions{}).
+				Return(&v1.VirtualMachineInstanceMigration{ObjectMeta: v13.ObjectMeta{Name: "something"}}, nil)
 
-			migrationInterface.EXPECT().Create(gomock.Any(), &v13.CreateOptions{}).Return(&v1.VirtualMachineInstanceMigration{ObjectMeta: v13.ObjectMeta{Name: "something"}}, nil)
 			controller.Execute()
 			testutils.ExpectEvent(recorder, evacuation.SuccessfulCreateVirtualMachineInstanceMigrationReason)
 		})
@@ -292,49 +296,6 @@ var _ = Describe("Evacuation", func() {
 			controller.Execute()
 		})
 
-		It("Should start a migration when we have a free spot", func() {
-			node := newNode("foo")
-			addNode(node)
-			vmi := newVirtualMachine("testvm", node.Name)
-			vmi.Status.Conditions = []v1.VirtualMachineInstanceCondition{
-				{
-					Type:   v1.VirtualMachineInstanceIsMigratable,
-					Status: v12.ConditionTrue,
-				},
-			}
-			vmi.Spec.EvictionStrategy = newEvictionStrategyLiveMigrate()
-			vmi.Status.EvacuationNodeName = node.Name
-			vmiFeeder.Add(vmi)
-
-			vmi1 := newVirtualMachine("testvm1", node.Name)
-			vmi1.Spec.EvictionStrategy = newEvictionStrategyLiveMigrate()
-			vmi1.Status.Conditions = []v1.VirtualMachineInstanceCondition{
-				{
-					Type:   v1.VirtualMachineInstanceIsMigratable,
-					Status: v12.ConditionTrue,
-				},
-			}
-			vmi1.Status.EvacuationNodeName = node.Name
-			vmiFeeder.Add(vmi1)
-
-			migration := newMigration("mig1", vmi.Name, v1.MigrationRunning)
-			migrationFeeder.Add(migration)
-			migrationFeeder.Add(newMigration("mig2", vmi.Name, v1.MigrationRunning))
-			migrationFeeder.Add(newMigration("mig3", vmi.Name, v1.MigrationRunning))
-			migrationFeeder.Add(newMigration("mig4", vmi.Name, v1.MigrationRunning))
-			migrationFeeder.Add(newMigration("mig5", vmi.Name, v1.MigrationRunning))
-
-			controller.Execute()
-
-			migration.Status.Phase = v1.MigrationSucceeded
-			migrationFeeder.Modify(migration)
-
-			migrationInterface.EXPECT().Create(gomock.Any(), &v13.CreateOptions{}).
-				Return(&v1.VirtualMachineInstanceMigration{ObjectMeta: v13.ObjectMeta{Name: "something"}}, nil)
-			controller.Execute()
-			testutils.ExpectEvent(recorder, evacuation.SuccessfulCreateVirtualMachineInstanceMigrationReason)
-		})
-
 		It("Should not create a migration if one is already in progress", func() {
 			node := newNode("foo")
 			addNode(node)
@@ -445,6 +406,56 @@ var _ = Describe("Evacuation", func() {
 
 			controller.Execute()
 		})
+
+		It("Should create new evictions up to the configured maximum migrations per outbound node", func() {
+			var maxParallelMigrationsPerCluster uint32 = 10
+			var maxParallelMigrationsPerOutboundNode uint32 = 5
+			var activeMigrationsFromThisSourceNode = 4
+			var migrationCandidatesFromThisSourceNode = 4
+			config, _, _ := testutils.NewFakeClusterConfigUsingKVConfig(&v1.KubeVirtConfiguration{
+				MigrationConfiguration: &v1.MigrationConfiguration{
+					ParallelMigrationsPerCluster:      &maxParallelMigrationsPerCluster,
+					ParallelOutboundMigrationsPerNode: &maxParallelMigrationsPerOutboundNode,
+				},
+			})
+
+			controller = evacuation.
+				NewEvacuationController(
+					vmiInformer,
+					migrationInformer,
+					nodeInformer,
+					podInformer,
+					recorder,
+					virtClient,
+					config)
+
+			nodeName := "node01"
+			addNode(newNode(nodeName))
+
+			By(fmt.Sprintf("Creating %d active migrations from source node %s", activeMigrationsFromThisSourceNode, nodeName))
+			for i := 1; i <= activeMigrationsFromThisSourceNode; i++ {
+				vmiName := fmt.Sprintf("testvmi%d", i)
+				vmiFeeder.Add(newVirtualMachineMarkedForEviction(vmiName, nodeName))
+				migrationFeeder.Add(newMigration(fmt.Sprintf("mig%d", i), vmiName, v1.MigrationRunning))
+			}
+
+			By(fmt.Sprintf("Creating %d migration candidates from source node %s", migrationCandidatesFromThisSourceNode, nodeName))
+			for i := 1; i <= migrationCandidatesFromThisSourceNode; i++ {
+				vmiName := fmt.Sprintf("testvmi%d", i+activeMigrationsFromThisSourceNode)
+				vmiFeeder.Add(newVirtualMachineMarkedForEviction(vmiName, nodeName))
+			}
+
+			By(fmt.Sprintf("Expect only one new migration from node %s although cluster capacity allows more candidates", nodeName))
+			migrationInterface.
+				EXPECT().
+				Create(gomock.Any(), &v13.CreateOptions{}).
+				Return(&v1.VirtualMachineInstanceMigration{ObjectMeta: v13.ObjectMeta{Name: "something"}}, nil).
+				Times(1)
+
+			controller.Execute()
+
+			testutils.ExpectEvent(recorder, evacuation.SuccessfulCreateVirtualMachineInstanceMigrationReason)
+		})
 	})
 
 	AfterEach(func() {
@@ -463,6 +474,20 @@ func newNode(name string) *v12.Node {
 	}
 }
 
+func newVirtualMachineMarkedForEviction(name string, nodeName string) *v1.VirtualMachineInstance {
+	vmi := newVirtualMachine(name, nodeName)
+	vmi.Status.Conditions = []v1.VirtualMachineInstanceCondition{
+		{
+			Type:   v1.VirtualMachineInstanceIsMigratable,
+			Status: v12.ConditionTrue,
+		},
+	}
+
+	vmi.Spec.EvictionStrategy = newEvictionStrategyLiveMigrate()
+	vmi.Status.EvacuationNodeName = nodeName
+	return vmi
+}
+
 func newVirtualMachine(name string, nodeName string) *v1.VirtualMachineInstance {
 	vmi := api.NewMinimalVMI("testvm")
 	vmi.Name = name