[x-port] [9.1] Watch PVCs for VM Spec.Volumes in volume batch controller

controllers/virtualmachine/volumebatch/volumebatch_controller.go

@@ -79,6 +79,25 @@ func AddToManager(ctx *pkgctx.ControllerManagerContext, mgr manager.Manager) err
 		return err
 	}
 
+	// Set up field index for VirtualMachine by ClaimName to efficiently query VMs
+	// referencing a PVC.
+	if err := mgr.GetFieldIndexer().IndexField(
+		ctx,
+		&vmopv1.VirtualMachine{},
+		"spec.volumes.persistentVolumeClaim.claimName",
+		func(rawObj client.Object) []string {
+			vm := rawObj.(*vmopv1.VirtualMachine)
+			pvcs := make([]string, 0, len(vm.Spec.Volumes))
+			for _, volume := range vm.Spec.Volumes {
+				if pvc := volume.PersistentVolumeClaim; pvc != nil && pvc.ClaimName != "" {
+					pvcs = append(pvcs, pvc.ClaimName)
+				}
+			}
+			return pvcs
+		}); err != nil {
+		return err
+	}
+
 	r := NewReconciler(
 		ctx,
 		mgr.GetClient(),
@@ -159,6 +178,22 @@ func AddToManager(ctx *pkgctx.ControllerManagerContext, mgr manager.Manager) err
 				"failed to start VirtualMachine watch "+
 					"for PersistentVolumeClaim: %w", err)
 		}
+
+		// Watch for changes for PersistentVolumeClaim, and enqueue
+		// VirtualMachine that reference the PVC in their Spec.Volumes.
+		//
+		// TODO(BMV): This should cover every case that the above OwnerRef
+		// mapper does, and that can be removed later.
+		if err := c.Watch(source.Kind(
+			mgr.GetCache(),
+			&corev1.PersistentVolumeClaim{},
+			handler.TypedEnqueueRequestsFromMapFunc(
+				vmopv1util.PVCToVirtualMachineVolumeClaimNameMapper(ctx, r.Client)),
+		)); err != nil {
+			return fmt.Errorf(
+				"failed to start VirtualMachine claim names watch "+
+					"for PersistentVolumeClaim: %w", err)
+		}
 	}
 
 	return nil

controllers/virtualmachine/volumebatch/volumebatch_controller_intg_test.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/google/uuid"
 	corev1 "k8s.io/api/core/v1"
+	storagev1 "k8s.io/api/storage/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -229,6 +230,97 @@ func intgTestsReconcile() {
 			})
 		})
 
+		It("Reconciles VirtualMachine Spec.Volumes when PVC is updated", func() {
+			Expect(ctx.Client.Create(ctx, vm)).To(Succeed())
+
+			vm = getVirtualMachine(vmKey)
+			Expect(vm).ToNot(BeNil())
+
+			By("VM has no volumes", func() {
+				Expect(vm.Spec.Volumes).To(BeEmpty())
+				Expect(vm.Status.Volumes).To(BeEmpty())
+			})
+
+			By("Assign VM BiosUUID and InstanceUUID", func() {
+				vm.Status.BiosUUID = dummyBiosUUID
+				vm.Status.InstanceUUID = dummyInstanceUUID
+				vm.Status.Hardware = &vmopv1.VirtualMachineHardwareStatus{
+					Controllers: []vmopv1.VirtualControllerStatus{
+						{
+							Type:      "SCSI",
+							BusNumber: 0,
+							DeviceKey: 1000,
+						},
+					},
+				}
+				Expect(ctx.Client.Status().Update(ctx, vm)).To(Succeed())
+			})
+
+			By("Add CNS volume to Spec.Volumes", func() {
+				sc := &storagev1.StorageClass{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "storage-class-1",
+					},
+					Provisioner: "kubernetes.io/dummy-volume-driver",
+				}
+
+				// Unbound PVC needs a storage class assigned so doesn't fail the
+				// WFFC check.
+				By("Create Immediate StorageClass", func() {
+					Expect(ctx.Client.Create(ctx, sc)).To(Succeed())
+				})
+
+				By("Create PVC and Pending", func() {
+					pvc1.Spec.StorageClassName = ptr.To(sc.Name)
+					Expect(ctx.Client.Create(ctx, pvc1)).To(Succeed())
+					pvc1.Status.Phase = corev1.ClaimPending
+					Expect(ctx.Client.Status().Update(ctx, pvc1)).To(Succeed())
+				})
+				vm.Spec.Volumes = append(vm.Spec.Volumes, vmVolume1)
+				Expect(ctx.Client.Update(ctx, vm)).To(Succeed())
+			})
+
+			By("CnsNodeVMBatchAttachment should be created but empty", func() {
+				Eventually(func(g Gomega) {
+					attachment := getCnsNodeVMBatchAttachment(vm)
+					g.Expect(attachment).ToNot(BeNil())
+					g.Expect(attachment.Spec.InstanceUUID).To(Equal(dummyInstanceUUID))
+					g.Expect(attachment.Spec.Volumes).To(BeEmpty())
+				}).Should(Succeed())
+			})
+
+			// Ideally we'd have a way to assert that the batch controller quiesced here.
+			By("Mark PVC as Bound", func() {
+				Expect(ctx.Client.Get(ctx, client.ObjectKeyFromObject(pvc1), pvc1)).To(Succeed())
+				pvc1.Status.Phase = corev1.ClaimBound
+				Expect(ctx.Client.Status().Update(ctx, pvc1)).To(Succeed())
+			})
+
+			By("CnsNodeVMBatchAttachment should be updated with volume", func() {
+				var attachment *cnsv1alpha1.CnsNodeVMBatchAttachment
+				Eventually(func(g Gomega) {
+					attachment = getCnsNodeVMBatchAttachment(vm)
+					g.Expect(attachment).ToNot(BeNil())
+					g.Expect(attachment.Spec.InstanceUUID).To(Equal(dummyInstanceUUID))
+					g.Expect(attachment.Spec.Volumes).To(HaveLen(1))
+				}).Should(Succeed())
+				Expect(attachment.Spec.Volumes[0].Name).To(Equal(vmVolume1.Name))
+			})
+
+			By("VM Status.Volume should have entry for volume", func() {
+				var vm *vmopv1.VirtualMachine
+				Eventually(func(g Gomega) {
+					vm = getVirtualMachine(vmKey)
+					g.Expect(vm).ToNot(BeNil())
+					g.Expect(vm.Status.Volumes).To(HaveLen(1))
+				}).Should(Succeed())
+
+				volStatus := vm.Status.Volumes[0]
+				Expect(volStatus.Name).To(Equal(vmVolume1.Name))
+				Expect(volStatus.Attached).To(BeFalse())
+			})
+		})
+
 		It("Reconciles VirtualMachine Spec.Volumes", func() {
 			Expect(ctx.Client.Create(ctx, vm)).To(Succeed())
 

pkg/util/vmopv1/vm.go

@@ -12,6 +12,7 @@ import (
 
 	"github.com/go-logr/logr"
 	vimtypes "github.com/vmware/govmomi/vim25/types"
+	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
@@ -443,6 +444,53 @@ func CnsRegisterVolumeToVirtualMachineMapper(
 	}
 }
 
+// PVCToVirtualMachineVolumeClaimNameMapper returns a mapper function used to enqueue
+// reconcile requests for VirtualMachines that reference the Spec.Volumes.
+func PVCToVirtualMachineVolumeClaimNameMapper(
+	_ context.Context,
+	k8sClient client.Client) handler.TypedMapFunc[*corev1.PersistentVolumeClaim, reconcile.Request] {
+
+	if k8sClient == nil {
+		panic("k8sClient is nil")
+	}
+
+	return func(ctx context.Context, pvc *corev1.PersistentVolumeClaim) []reconcile.Request {
+		logger := pkglog.FromContextOrDefault(ctx).
+			WithName("PVCToVirtualMachineVolumeClaimNameMapper").
+			WithValues("name", pvc.Name, "namespace", pvc.Namespace)
+		logger.V(4).Info("Reconciling VMs by ClaimName due to PVC event")
+
+		list := &vmopv1.VirtualMachineList{}
+		err := k8sClient.List(
+			ctx,
+			list,
+			client.InNamespace(pvc.Namespace),
+			client.MatchingFields{"spec.volumes.persistentVolumeClaim.claimName": pvc.Name})
+		if err != nil {
+			logger.Error(err, "Failed to list VirtualMachines by ClaimName for PVC")
+			return nil
+		}
+
+		requests := make([]reconcile.Request, 0, len(list.Items))
+		for _, vm := range list.Items {
+			requests = append(requests, reconcile.Request{
+				NamespacedName: client.ObjectKey{
+					Namespace: vm.Namespace,
+					Name:      vm.Name,
+				},
+			})
+		}
+
+		if len(requests) > 0 {
+			logger.V(4).Info(
+				"Reconciling VMs due to PVC watch",
+				"requests", requests)
+		}
+
+		return requests
+	}
+}
+
 // KubernetesNodeLabelKey is the name of the label key used to identify a
 // Kubernetes cluster node.
 const KubernetesNodeLabelKey = "cluster.x-k8s.io/cluster-name"