test: skip StorageClassMigration if msg 'Will be reverted'

Signed-off-by: Nikita Korolev <nikita.korolev@flant.com>

Author: universal-itengineer

test/e2e/controller/controller.go

@@ -24,6 +24,7 @@ import (
 const (
 	VirtualizationController = "virtualization-controller"
 	VirtualizationNamespace  = "d8-virtualization"
+	LeaderElectionID         = "d8-virt-operator-leader-election-helper"
 )
 
 var (

test/e2e/internal/util/vm.go

@@ -19,7 +19,10 @@ package util
 import (
 	"context"
 	"fmt"
+	"io"
 	"regexp"
+	"slices"
+	"strings"
 	"time"
 
 	. "github.com/onsi/ginkgo/v2"
@@ -28,13 +31,15 @@ import (
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/wait"
 	virtv1 "kubevirt.io/api/core/v1"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	vmopbuilder "github.com/deckhouse/virtualization-controller/pkg/builder/vmop"
 	"github.com/deckhouse/virtualization-controller/pkg/controller/conditions"
 	"github.com/deckhouse/virtualization/api/core/v1alpha2"
 	"github.com/deckhouse/virtualization/api/core/v1alpha2/vmcondition"
+	"github.com/deckhouse/virtualization/test/e2e/controller"
 	"github.com/deckhouse/virtualization/test/e2e/internal/framework"
 	"github.com/deckhouse/virtualization/test/e2e/internal/rewrite"
 )
@@ -46,6 +51,18 @@ const (
 
 var knownKubeVirtClientSocketClosedRe = regexp.MustCompile(`(?is)virError\(Code=1,.*internal error:\s*client\s+socket\s+is\s+closed`)
 
+var knownVDMigrationControllerRevertMessages = []string{
+	"VirtualMachine is not running. Will be reverted.",
+	"VirtualMachine is not migrating. Will be reverted.",
+	"Target PersistentVolumeClaim is not found. Revert old PersistentVolumeClaim and remove migration condition.",
+	"Target PersistentVolumeClaim is not bound. Revert old PersistentVolumeClaim and remove migration condition.",
+}
+
+type controllerLogMatch struct {
+	PodName string
+	Line    string
+}
+
 func IsKnownKubeVirtClientSocketClosedFailureReason(reason string) bool {
 	return knownKubeVirtClientSocketClosedRe.MatchString(reason)
 }
@@ -106,6 +123,188 @@ func SkipIfKnownMigrationFailure(vm *v1alpha2.VirtualMachine) {
 	SkipIfKnownVolumesUpdateMigrationFailure(vm)
 }
 
+func WaitUntilConditionOrSkipKnownVDMigrationControllerRevert(timeout time.Duration, namespace string, condition func() error) {
+	GinkgoHelper()
+
+	waitStartedAt := time.Now()
+	var lastErr error
+
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+
+	err := wait.PollUntilContextTimeout(ctx, time.Second, timeout, true, func(context.Context) (bool, error) {
+		lastErr = condition()
+		return lastErr == nil, nil
+	})
+	if err == nil {
+		return
+	}
+
+	if ctx.Err() == context.DeadlineExceeded {
+		SkipIfKnownVDMigrationControllerRevertOnTimeout(namespace, waitStartedAt)
+	}
+
+	if lastErr != nil {
+		Fail(fmt.Sprintf("timed out waiting for condition: %v", lastErr))
+	}
+
+	Expect(err).NotTo(HaveOccurred())
+}
+
+func SkipIfKnownVDMigrationControllerRevertOnTimeout(namespace string, since time.Time) {
+	GinkgoHelper()
+
+	match, err := findKnownVDMigrationControllerRevertLog(namespace, since)
+	if err != nil {
+		GinkgoWriter.Printf("Failed to inspect virtualization-controller logs for namespace %q: %v\n", namespace, err)
+		return
+	}
+	if match == nil {
+		return
+	}
+
+	Skip(fmt.Sprintf(
+		"skip due to known virtualization-controller volume migration revert for namespace %s in pod %s: %s",
+		namespace, match.PodName, match.Line,
+	))
+}
+
+func findKnownVDMigrationControllerRevertLog(namespace string, since time.Time) (*controllerLogMatch, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), framework.ShortTimeout)
+	defer cancel()
+
+	pods, err := framework.GetClients().KubeClient().CoreV1().Pods(controller.VirtualizationNamespace).List(ctx, metav1.ListOptions{
+		LabelSelector: fmt.Sprintf("app=%s", controller.VirtualizationController),
+	})
+	if err != nil {
+		return nil, fmt.Errorf("list virtualization-controller pods: %w", err)
+	}
+	orderedPods, err := orderVirtualizationControllerPodsByLeader(ctx, pods.Items)
+	if err != nil {
+		GinkgoWriter.Printf("Failed to resolve virtualization-controller leader pod, fallback to all pods: %v\n", err)
+		orderedPods = pods.Items
+	}
+
+	sinceTime := metav1.NewTime(since.Add(-5 * time.Second))
+	for _, pod := range orderedPods {
+		stream, err := framework.GetClients().KubeClient().CoreV1().Pods(pod.Namespace).GetLogs(pod.Name, &corev1.PodLogOptions{
+			Container: controller.VirtualizationController,
+			SinceTime: &sinceTime,
+		}).Stream(ctx)
+		if err != nil {
+			GinkgoWriter.Printf("Failed to read virtualization-controller logs from pod %s: %v\n", pod.Name, err)
+			continue
+		}
+
+		logs, readErr := io.ReadAll(stream)
+		closeErr := stream.Close()
+		if readErr != nil {
+			return nil, fmt.Errorf("read virtualization-controller logs from pod %s: %w", pod.Name, readErr)
+		}
+		if closeErr != nil {
+			GinkgoWriter.Printf("Failed to close virtualization-controller log stream for pod %s: %v\n", pod.Name, closeErr)
+		}
+
+		if line := findKnownVDMigrationControllerRevertLine(string(logs), namespace); line != "" {
+			return &controllerLogMatch{
+				PodName: pod.Name,
+				Line:    line,
+			}, nil
+		}
+	}
+
+	return nil, nil
+}
+
+func orderVirtualizationControllerPodsByLeader(ctx context.Context, pods []corev1.Pod) ([]corev1.Pod, error) {
+	if len(pods) <= 1 {
+		return pods, nil
+	}
+	if !isVirtualizationControllerLeaderElectionEnabled(pods) {
+		return pods, nil
+	}
+
+	lease, err := framework.GetClients().KubeClient().CoordinationV1().Leases(controller.VirtualizationNamespace).Get(ctx, controller.LeaderElectionID, metav1.GetOptions{})
+	if err != nil {
+		if k8serrors.IsNotFound(err) {
+			return pods, nil
+		}
+		return nil, fmt.Errorf("get leader election lease %q: %w", controller.LeaderElectionID, err)
+	}
+	if lease.Spec.HolderIdentity == nil || *lease.Spec.HolderIdentity == "" {
+		return pods, nil
+	}
+
+	holderIdentity := *lease.Spec.HolderIdentity
+	leaderIdx := slices.IndexFunc(pods, func(pod corev1.Pod) bool {
+		return pod.Name == holderIdentity || strings.HasPrefix(holderIdentity, pod.Name+"_")
+	})
+	if leaderIdx == -1 {
+		GinkgoWriter.Printf("Virtualization-controller leader lease holder %q does not match listed pods; fallback to all pods\n", holderIdentity)
+		return pods, nil
+	}
+
+	orderedPods := make([]corev1.Pod, 0, len(pods))
+	orderedPods = append(orderedPods, pods[leaderIdx])
+	for i, pod := range pods {
+		if i == leaderIdx {
+			continue
+		}
+		orderedPods = append(orderedPods, pod)
+	}
+
+	return orderedPods, nil
+}
+
+func isVirtualizationControllerLeaderElectionEnabled(pods []corev1.Pod) bool {
+	for _, pod := range pods {
+		for _, container := range pod.Spec.Containers {
+			if container.Name != controller.VirtualizationController {
+				continue
+			}
+			return isLeaderElectionEnabledByArgs(container.Args)
+		}
+	}
+
+	// The controller uses a default value of true when the flag is not passed.
+	return true
+}
+
+func isLeaderElectionEnabledByArgs(args []string) bool {
+	enabled := true
+
+	for i, arg := range args {
+		switch {
+		case arg == "--leader-election" && i+1 < len(args) && !strings.HasPrefix(args[i+1], "--"):
+			enabled = args[i+1] != "false"
+		case arg == "--leader-election":
+			enabled = true
+		case arg == "--leader-election=true":
+			enabled = true
+		case arg == "--leader-election=false":
+			enabled = false
+		case strings.HasPrefix(arg, "--leader-election="):
+			enabled = strings.TrimPrefix(arg, "--leader-election=") != "false"
+		}
+	}
+
+	return enabled
+}
+
+func findKnownVDMigrationControllerRevertLine(logs, namespace string) string {
+	for _, line := range strings.Split(logs, "\n") {
+		if !strings.Contains(line, namespace) {
+			continue
+		}
+		for _, message := range knownVDMigrationControllerRevertMessages {
+			if strings.Contains(line, message) {
+				return strings.TrimSpace(line)
+			}
+		}
+	}
+	return ""
+}
+
 func getInternalVirtualMachineInstance(vm *v1alpha2.VirtualMachine) (*virtv1.VirtualMachineInstance, error) {
 	GinkgoHelper()
 
@@ -152,7 +351,7 @@ func UntilSSHReady(f *framework.Framework, vm *v1alpha2.VirtualMachine, timeout
 func UntilVMMigrationSucceeded(key client.ObjectKey, timeout time.Duration) {
 	GinkgoHelper()
 
-	Eventually(func() error {
+	WaitUntilConditionOrSkipKnownVDMigrationControllerRevert(timeout, key.Namespace, func() error {
 		vm, err := framework.GetClients().VirtClient().VirtualMachines(key.Namespace).Get(context.Background(), key.Name, metav1.GetOptions{})
 		if err != nil {
 			return err
@@ -177,7 +376,7 @@ func UntilVMMigrationSucceeded(key client.ObjectKey, timeout time.Duration) {
 		}
 
 		return nil
-	}).WithTimeout(timeout).WithPolling(time.Second).Should(Succeed())
+	})
 }
 
 func MigrateVirtualMachine(f *framework.Framework, vm *v1alpha2.VirtualMachine, options ...vmopbuilder.Option) {

test/e2e/vm/util.go

@@ -129,33 +129,60 @@ func untilVirtualDisksMigrationsSucceeded(f *framework.Framework) {
 	GinkgoHelper()
 
 	By("Wait until VirtualDisks migrations succeeded")
-	Eventually(func(g Gomega) {
+	e2eutil.WaitUntilConditionOrSkipKnownVDMigrationControllerRevert(framework.MaxTimeout, f.Namespace().Name, func() error {
 		vms, err := f.VirtClient().VirtualMachines(f.Namespace().Name).List(context.Background(), metav1.ListOptions{})
-		g.Expect(err).NotTo(HaveOccurred())
+		if err != nil {
+			return err
+		}
 		for _, vm := range vms.Items {
 			// TODO: remove temporary migration skip logic when both known issues are fixed:
 			// kubevirt "client socket is closed" and Volume(s)UpdateError.
 			e2eutil.SkipIfKnownMigrationFailure(&vm)
 		}
 
 		vds, err := f.VirtClient().VirtualDisks(f.Namespace().Name).List(context.Background(), metav1.ListOptions{})
-		g.Expect(err).NotTo(HaveOccurred())
+		if err != nil {
+			return err
+		}
 
-		g.Expect(vds.Items).ShouldNot(BeEmpty())
+		if len(vds.Items) == 0 {
+			return fmt.Errorf("virtual disk list is empty")
+		}
 		for _, vd := range vds.Items {
-			g.Expect(vd.Status.Phase).To(Equal(v1alpha2.DiskReady))
-			g.Expect(vd.Status.Target.PersistentVolumeClaim).ShouldNot(BeEmpty())
+			if vd.Status.Phase != v1alpha2.DiskReady {
+				return fmt.Errorf("vd %s phase is %s, expected %s", vd.Name, vd.Status.Phase, v1alpha2.DiskReady)
+			}
+			if vd.Status.Target.PersistentVolumeClaim == "" {
+				return fmt.Errorf("vd %s target pvc is empty", vd.Name)
+			}
 
 			if vd.Status.MigrationState.StartTimestamp.IsZero() {
 				// Skip the disks that are not migrated
 				continue
 			}
 
-			g.Expect(vd.Status.MigrationState.EndTimestamp.IsZero()).Should(BeFalse(), "migration is not ended for vd %s", vd.Name)
-			g.Expect(vd.Status.Target.PersistentVolumeClaim).To(Equal(vd.Status.MigrationState.TargetPVC))
-			g.Expect(vd.Status.MigrationState.Result).To(Equal(v1alpha2.VirtualDiskMigrationResultSucceeded))
+			if vd.Status.MigrationState.EndTimestamp.IsZero() {
+				return fmt.Errorf("migration is not ended for vd %s", vd.Name)
+			}
+			if vd.Status.Target.PersistentVolumeClaim != vd.Status.MigrationState.TargetPVC {
+				return fmt.Errorf(
+					"vd %s target pvc mismatch: current=%s migration-target=%s",
+					vd.Name,
+					vd.Status.Target.PersistentVolumeClaim,
+					vd.Status.MigrationState.TargetPVC,
+				)
+			}
+			if vd.Status.MigrationState.Result != v1alpha2.VirtualDiskMigrationResultSucceeded {
+				return fmt.Errorf(
+					"vd %s migration result is %s, expected %s",
+					vd.Name,
+					vd.Status.MigrationState.Result,
+					v1alpha2.VirtualDiskMigrationResultSucceeded,
+				)
+			}
 		}
-	}).WithTimeout(framework.MaxTimeout).WithPolling(time.Second).Should(Succeed())
+		return nil
+	})
 }
 
 func untilVirtualDisksMigrationsFailed(f *framework.Framework) {

test/e2e/vm/volume_migration_storage_class_changed.go

@@ -236,7 +236,7 @@ var _ = Describe("StorageClassMigration", decoratorsForVolumeMigrations(), func(
 			err = patchStorageClassName(context.Background(), f, sc, vdForMigration)
 			Expect(err).NotTo(HaveOccurred())
 
-			Eventually(func() error {
+			util.WaitUntilConditionOrSkipKnownVDMigrationControllerRevert(framework.MaxTimeout, ns, func() error {
 				vm, err = f.VirtClient().VirtualMachines(ns).Get(context.Background(), vm.GetName(), metav1.GetOptions{})
 				if err != nil {
 					return err
@@ -247,7 +247,9 @@ var _ = Describe("StorageClassMigration", decoratorsForVolumeMigrations(), func(
 
 				var lastVMOP *v1alpha2.VirtualMachineOperation
 				vmops, err := f.VirtClient().VirtualMachineOperations(ns).List(context.Background(), metav1.ListOptions{})
-				Expect(err).NotTo(HaveOccurred())
+				if err != nil {
+					return err
+				}
 
 				for _, vmop := range vmops.Items {
 					if vmop.Spec.VirtualMachine == vm.Name {
@@ -271,7 +273,7 @@ var _ = Describe("StorageClassMigration", decoratorsForVolumeMigrations(), func(
 				}
 
 				return fmt.Errorf("migration is not completed")
-			}).WithTimeout(framework.MaxTimeout).WithPolling(time.Second).Should(Succeed())
+			})
 
 			By("Wait until VM migration succeeded")
 			util.UntilVMMigrationSucceeded(crclient.ObjectKeyFromObject(vm), framework.MaxTimeout)