GC only if allowed to update DV owner finalizers

pkg/controller/BUILD.bazel

@@ -43,6 +43,7 @@ go_library(
         "//vendor/github.com/openshift/api/route/v1:go_default_library",
         "//vendor/github.com/pkg/errors:go_default_library",
         "//vendor/github.com/prometheus/client_golang/prometheus:go_default_library",
+        "//vendor/k8s.io/api/authorization/v1:go_default_library",
         "//vendor/k8s.io/api/batch/v1:go_default_library",
         "//vendor/k8s.io/api/core/v1:go_default_library",
         "//vendor/k8s.io/api/networking/v1:go_default_library",

pkg/controller/datavolume-controller.go

@@ -34,6 +34,7 @@ import (
 	snapshotv1 "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
 	"github.com/pkg/errors"
 
+	authorizationv1 "k8s.io/api/authorization/v1"
 	corev1 "k8s.io/api/core/v1"
 	storagev1 "k8s.io/api/storage/v1"
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
@@ -2783,6 +2784,10 @@ func (r *DatavolumeReconciler) garbageCollect(dataVolume *cdiv1.DataVolume, pvc
 	if dataVolume.Status.Phase != cdiv1.Succeeded {
 		return nil, nil
 	}
+	if allowed, err := r.isGarbageCollectionAllowed(dataVolume); !allowed || err != nil {
+		log.Info("Garbage Collection is not allowed, due to owner finalizers update RBAC")
+		return nil, err
+	}
 	cdiConfig := &cdiv1.CDIConfig{}
 	if err := r.client.Get(context.TODO(), types.NamespacedName{Name: common.ConfigName}, cdiConfig); err != nil {
 		return nil, err
@@ -2802,6 +2807,41 @@ func (r *DatavolumeReconciler) garbageCollect(dataVolume *cdiv1.DataVolume, pvc
 	return &reconcile.Result{}, nil
 }
 
+func (r *DatavolumeReconciler) isGarbageCollectionAllowed(dv *cdiv1.DataVolume) (bool, error) {
+	for _, ref := range dv.OwnerReferences {
+		if ref.BlockOwnerDeletion == nil || *ref.BlockOwnerDeletion == false {
+			continue
+		}
+		if allowed, err := r.canUpdateFinalizers(ref); !allowed || err != nil {
+			return false, err
+		}
+	}
+	return true, nil
+}
+
+func (r *DatavolumeReconciler) canUpdateFinalizers(ownerRef metav1.OwnerReference) (bool, error) {
+	gvk := schema.FromAPIVersionAndKind(ownerRef.APIVersion, ownerRef.Kind)
+	mapping, err := r.client.RESTMapper().RESTMapping(gvk.GroupKind(), gvk.Version)
+	if err != nil {
+		return false, err
+	}
+	ssar := &authorizationv1.SelfSubjectAccessReview{
+		Spec: authorizationv1.SelfSubjectAccessReviewSpec{
+			ResourceAttributes: &authorizationv1.ResourceAttributes{
+				Group:       gvk.Group,
+				Version:     gvk.Version,
+				Resource:    mapping.Resource.Resource,
+				Subresource: "finalizers",
+				Verb:        "update",
+			},
+		},
+	}
+	if err := r.client.Create(context.TODO(), ssar); err != nil {
+		return false, err
+	}
+	return ssar.Status.Allowed, nil
+}
+
 func (r *DatavolumeReconciler) detachPvcDeleteDv(pvc *corev1.PersistentVolumeClaim, dv *cdiv1.DataVolume, log logr.Logger) error {
 	if dv.Status.Phase != cdiv1.Succeeded {
 		return nil
@@ -2824,13 +2864,24 @@ func (r *DatavolumeReconciler) detachPvcDeleteDv(pvc *corev1.PersistentVolumeCla
 
 func updatePvcOwnerRefs(pvc *corev1.PersistentVolumeClaim, dv *cdiv1.DataVolume) {
 	refs := pvc.OwnerReferences
-	for i, r := range refs {
-		if r.UID == dv.UID {
-			pvc.OwnerReferences = append(refs[:i], refs[i+1:]...)
-			break
+	if i := findOwnerRef(refs, dv.UID); i >= 0 {
+		refs = append(refs[:i], refs[i+1:]...)
+	}
+	for _, ref := range dv.OwnerReferences {
+		if findOwnerRef(refs, ref.UID) == -1 {
+			refs = append(refs, ref)
+		}
+	}
+	pvc.OwnerReferences = refs
+}
+
+func findOwnerRef(refs []metav1.OwnerReference, uid types.UID) int {
+	for i, ref := range refs {
+		if ref.UID == uid {
+			return i
 		}
 	}
-	pvc.OwnerReferences = append(pvc.OwnerReferences, dv.OwnerReferences...)
+	return -1
 }
 
 func getDeltaTTL(dv *cdiv1.DataVolume, ttl int32) time.Duration {

pkg/controller/datavolume-controller_test.go

@@ -2101,6 +2101,26 @@ var _ = Describe("All DataVolume Tests", func() {
 		Entry("40Gi virtual size, default overhead to be 40Gi if <= 1Gi and 41Gi if > 40Gi", 40*Gi, defaultOverhead),
 		Entry("40Gi virtual size, large overhead to be 40Gi if <= 40Gi and 41Gi if > 40Gi", 40*Gi, largeOverhead),
 	)
+
+	Describe("DataVolume garbage collection", func() {
+		It("updatePvcOwnerRefs should correctly update PVC owner refs", func() {
+			ref := func(uid string) metav1.OwnerReference {
+				return metav1.OwnerReference{UID: types.UID(uid)}
+			}
+			dv := newImportDataVolume("test-dv")
+			vmOwnerRef := metav1.OwnerReference{Kind: "VirtualMachine", Name: "test-vm", UID: "test-vm-uid"}
+			dv.OwnerReferences = append(dv.OwnerReferences, ref("3"), vmOwnerRef, ref("2"), ref("1"))
+
+			pvc := createPvc("test-dv", metav1.NamespaceDefault, nil, nil)
+			dvOwnerRef := metav1.OwnerReference{Kind: "DataVolume", Name: "test-dv", UID: dv.UID}
+			pvc.OwnerReferences = append(pvc.OwnerReferences, ref("1"), dvOwnerRef, ref("2"))
+
+			updatePvcOwnerRefs(pvc, dv)
+			Expect(pvc.OwnerReferences).To(HaveLen(4))
+			Expect(pvc.OwnerReferences).To(Equal([]metav1.OwnerReference{ref("1"), ref("2"), ref("3"), vmOwnerRef}))
+		})
+	})
+
 })
 
 func createStorageSpec() *cdiv1.StorageSpec {

pkg/operator/resources/cluster/controller.go

@@ -234,6 +234,17 @@ func getControllerClusterPolicyRules() []rbacv1.PolicyRule {
 				"watch",
 			},
 		},
+		{
+			APIGroups: []string{
+				"kubevirt.io",
+			},
+			Resources: []string{
+				"virtualmachines/finalizers",
+			},
+			Verbs: []string{
+				"update",
+			},
+		},
 	}
 }