Clean up completed virt-launcher pods after migration completes.

pkg/controller/directvolumemigration/task.go

@@ -48,6 +48,7 @@ const (
 	Completed                                   = "Completed"
 	MigrationFailed                             = "MigrationFailed"
 	VerifyVMs                                   = "VerifyVMs"
+	CleanupStaleVirtHandlerPods                 = "CleanupStaleVirtHandlerPods"
 	DeleteStaleVirtualMachineInstanceMigrations = "DeleteStaleVirtualMachineInstanceMigrations"
 )
 
@@ -139,6 +140,7 @@ var VolumeMigration = Itinerary{
 		{phase: RunRsyncOperations},
 		{phase: DeleteRsyncResources},
 		{phase: WaitForRsyncResourcesTerminated},
+		{phase: CleanupStaleVirtHandlerPods},
 		{phase: Completed},
 	},
 }
@@ -161,6 +163,7 @@ var RollbackItinerary = Itinerary{
 		{phase: DeleteStaleVirtualMachineInstanceMigrations},
 		{phase: WaitForStaleRsyncResourcesTerminated},
 		{phase: RunRsyncOperations},
+		{phase: CleanupStaleVirtHandlerPods},
 		{phase: Completed},
 	},
 }
@@ -299,6 +302,15 @@ func (t *Task) Run(ctx context.Context) error {
 		if err = t.next(); err != nil {
 			return liberr.Wrap(err)
 		}
+	case CleanupStaleVirtHandlerPods:
+		err := t.cleanupStaleVirtLauncherPods()
+		if err != nil {
+			return liberr.Wrap(err)
+		}
+		t.Requeue = NoReQ
+		if err = t.next(); err != nil {
+			return liberr.Wrap(err)
+		}
 	case CreateDestinationNamespaces:
 		// Create all of the namespaces the migrated PVCs are in are created on the
 		// destination

pkg/controller/directvolumemigration/vm.go

@@ -33,11 +33,13 @@ import (
 )
 
 const (
-	prometheusURLKey = "PROMETHEUS_URL"
-	prometheusRoute  = "prometheus-k8s"
-	progressQuery    = "kubevirt_vmi_migration_data_processed_bytes{name=\"%s\"} / (kubevirt_vmi_migration_data_processed_bytes{name=\"%s\"} + kubevirt_vmi_migration_data_remaining_bytes{name=\"%s\"}) * 100"
-	VMIKind          = "VirtualMachineInstance"
-	PodKind          = "Pod"
+	prometheusURLKey                  = "PROMETHEUS_URL"
+	prometheusRoute                   = "prometheus-k8s"
+	progressQuery                     = "kubevirt_vmi_migration_data_processed_bytes{name=\"%s\"} / (kubevirt_vmi_migration_data_processed_bytes{name=\"%s\"} + kubevirt_vmi_migration_data_remaining_bytes{name=\"%s\"}) * 100"
+	VMIKind                           = "VirtualMachineInstance"
+	PodKind                           = "Pod"
+	virtLauncherPodLabelSelectorKey   = "kubevirt.io"
+	virtLauncherPodLabelSelectorValue = "virt-launcher"
 )
 
 var (
@@ -341,6 +343,32 @@ func (t *Task) verifyVMs() error {
 	return nil
 }
 
+func (t *Task) cleanupStaleVirtLauncherPods() error {
+	t.Log.Info("Cleaning up stale virt-launcher pods")
+	sourceClient := t.sourceClient
+	namespaces := sets.NewString()
+	for _, pvc := range t.Owner.Spec.PersistentVolumeClaims {
+		namespaces.Insert(pvc.Namespace)
+	}
+	for _, namespace := range namespaces.List() {
+		podList := &corev1.PodList{}
+		labelSelector := map[string]string{virtLauncherPodLabelSelectorKey: virtLauncherPodLabelSelectorValue}
+		if err := sourceClient.List(context.TODO(), podList, k8sclient.InNamespace(namespace), k8sclient.MatchingLabels(labelSelector)); err != nil {
+			return err
+		}
+		for _, pod := range podList.Items {
+			if pod.Status.Phase == corev1.PodSucceeded {
+				if err := sourceClient.Delete(context.TODO(), &pod); err != nil {
+					if !k8serrors.IsNotFound(err) {
+						return err
+					}
+				}
+			}
+		}
+	}
+	return nil
+}
+
 func findVirtualMachineInstanceMigration(client k8sclient.Client, vmName, namespace string) (*virtv1.VirtualMachineInstanceMigration, error) {
 	vmimList := &virtv1.VirtualMachineInstanceMigrationList{}
 	if err := client.List(context.TODO(), vmimList, k8sclient.InNamespace(namespace)); err != nil {