add PersistentVolumeReconciler to remove RBD images by PV's finalizer

Signed-off-by: Ryotaro Banno <[email protected]>

PROJECT

@@ -35,4 +35,8 @@ resources:
   kind: MantleBackupConfig
   path: github.com/cybozu-go/mantle/api/v1
   version: v1
+- controller: true
+  domain: cybozu.io
+  kind: PersistentVolume
+  version: v1
 version: "3"

charts/mantle-cluster-wide/templates/clusterrole.yaml

@@ -27,6 +27,20 @@ rules:
       - patch
       - update
       - watch
+  - apiGroups:
+      - ""
+    resources:
+      - persistentvolumes/finalizers
+    verbs:
+      - update
+  - apiGroups:
+      - ""
+    resources:
+      - persistentvolumes/status
+    verbs:
+      - get
+      - patch
+      - update
   - apiGroups:
       - batch
     resources:

cmd/controller/main.go

@@ -192,6 +192,16 @@ func setupReconcilers(mgr manager.Manager, primarySettings *controller.PrimarySe
 		setupLog.Error(err, "unable to create controller", "controller", "MantleBackupConfig")
 		return err
 	}
+
+	if err := controller.NewPersistentVolumeReconciler(
+		mgr.GetClient(),
+		mgr.GetScheme(),
+		managedCephClusterID,
+	).SetupWithManager(mgr); err != nil {
+		setupLog.Error(err, "unable to create controller", "controller", "PersistentVolumeReconciler")
+		return err
+	}
+
 	//+kubebuilder:scaffold:builder
 
 	return nil

config/rbac/role.yaml

@@ -27,6 +27,20 @@ rules:
   - patch
   - update
   - watch
+- apiGroups:
+  - ""
+  resources:
+  - persistentvolumes/finalizers
+  verbs:
+  - update
+- apiGroups:
+  - ""
+  resources:
+  - persistentvolumes/status
+  verbs:
+  - get
+  - patch
+  - update
 - apiGroups:
   - batch
   resources:

internal/controller/mantlerestore_controller.go

@@ -340,12 +340,6 @@ func (r *MantleRestoreReconciler) cleanup(ctx context.Context, restore *mantlev1
 		return ctrl.Result{}, err
 	}
 
-	// delete the clone image
-	if err := r.removeRBDImage(ctx, restore); err != nil {
-		logger.Error(err, "failed to remove image")
-		return ctrl.Result{}, err
-	}
-
 	// remove the finalizer
 	controllerutil.RemoveFinalizer(restore, MantleRestoreFinalizerName)
 	err = r.client.Update(ctx, restore)
@@ -419,23 +413,6 @@ func (r *MantleRestoreReconciler) deleteRestoringPV(ctx context.Context, restore
 	}
 }
 
-func (r *MantleRestoreReconciler) removeRBDImage(ctx context.Context, restore *mantlev1.MantleRestore) error {
-	logger := log.FromContext(ctx)
-	image := r.restoringRBDImageName(restore)
-	pool := restore.Status.Pool
-	logger.Info("removing image", "pool", pool, "image", image)
-	images, err := r.ceph.RBDLs(pool)
-	if err != nil {
-		return fmt.Errorf("failed to list RBD images: %v", err)
-	}
-
-	if !slices.Contains(images, image) {
-		return nil
-	}
-
-	return r.ceph.RBDRm(pool, image)
-}
-
 // SetupWithManager sets up the controller with the Manager.
 func (r *MantleRestoreReconciler) SetupWithManager(mgr ctrl.Manager) error {
 	return ctrl.NewControllerManagedBy(mgr).

internal/controller/mantlerestore_controller_e2e.go

@@ -25,7 +25,3 @@ func NewMantleRestoreReconcilerE2E(managedCephClusterID, toolsNamespace string)
 func (r *MantleRestoreReconcilerE2E) CloneImageFromBackup(ctx context.Context, restore *mantlev1.MantleRestore, backup *mantlev1.MantleBackup) error {
 	return r.cloneImageFromBackup(ctx, restore, backup)
 }
-
-func (r *MantleRestoreReconcilerE2E) RemoveRBDImage(ctx context.Context, restore *mantlev1.MantleRestore) error {
-	return r.removeRBDImage(ctx, restore)
-}

internal/controller/persistentvolume_controller.go

@@ -0,0 +1,143 @@
+package controller
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"slices"
+
+	"github.com/cybozu-go/mantle/internal/ceph"
+	corev1 "k8s.io/api/core/v1"
+	aerrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/runtime"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
+	"sigs.k8s.io/controller-runtime/pkg/event"
+	"sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/controller-runtime/pkg/predicate"
+)
+
+// PersistentVolumeReconciler reconciles a PersistentVolume object
+type PersistentVolumeReconciler struct {
+	client               client.Client
+	Scheme               *runtime.Scheme
+	ceph                 ceph.CephCmd
+	managedCephClusterID string
+}
+
+// +kubebuilder:rbac:groups="",resources=persistentvolumes,verbs=get;list;watch;create;update;patch;delete
+// +kubebuilder:rbac:groups="",resources=persistentvolumes/status,verbs=get;update;patch
+// +kubebuilder:rbac:groups="",resources=persistentvolumes/finalizers,verbs=update
+
+func NewPersistentVolumeReconciler(
+	client client.Client,
+	scheme *runtime.Scheme,
+	managedCephClusterID string,
+) *PersistentVolumeReconciler {
+	return &PersistentVolumeReconciler{
+		client:               client,
+		Scheme:               scheme,
+		ceph:                 ceph.NewCephCmd(),
+		managedCephClusterID: managedCephClusterID,
+	}
+}
+
+// Reconcile is part of the main kubernetes reconciliation loop which aims to
+// move the current state of the cluster closer to the desired state.
+// TODO(user): Modify the Reconcile function to compare the state specified by
+// the PersistentVolume object against the actual cluster state, and then
+// perform operations to make the cluster state reflect the state specified by
+// the user.
+//
+// For more details, check Reconcile and its Result here:
+// - https://pkg.go.dev/sigs.k8s.io/[email protected]/pkg/reconcile
+func (r *PersistentVolumeReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
+	logger := log.FromContext(ctx)
+
+	// Get the PV being reconciled.
+	var pv corev1.PersistentVolume
+	if err := r.client.Get(ctx, req.NamespacedName, &pv); err != nil {
+		if aerrors.IsNotFound(err) {
+			return ctrl.Result{}, nil
+		}
+		return ctrl.Result{}, fmt.Errorf("failed to get PersistentVolume: %w", err)
+	}
+
+	// Check if the PV is managed by the target Ceph cluster.
+	clusterID, err := getCephClusterIDFromSCName(ctx, r.client, pv.Spec.StorageClassName)
+	if err != nil {
+		if errors.Is(err, errEmptyClusterID) {
+			return ctrl.Result{}, nil
+		}
+		return ctrl.Result{}, err
+	}
+	if clusterID != r.managedCephClusterID {
+		logger.Info("PV is not provisioned by the target Ceph cluster", "pv", pv.Name, "clusterID", clusterID)
+		return ctrl.Result{}, nil
+	}
+
+	// Make sure the PV has the finalizer.
+	if !controllerutil.ContainsFinalizer(&pv, RestoringPVFinalizerName) {
+		return ctrl.Result{}, nil
+	}
+
+	// Make sure the PV has a deletionTimestamp.
+	if pv.GetDeletionTimestamp().IsZero() {
+		return ctrl.Result{}, nil
+	}
+
+	// Wait until the PV's status becomes Released.
+	if pv.Status.Phase != corev1.VolumeReleased {
+		return ctrl.Result{Requeue: true}, nil
+	}
+
+	// Delete the RBD clone image.
+	if err := r.removeRBDImage(ctx, &pv); err != nil {
+		return ctrl.Result{}, fmt.Errorf("failed to remove RBD image: %s: %w", pv.Name, err)
+	}
+
+	// Remove the finalizer of the PV.
+	controllerutil.RemoveFinalizer(&pv, RestoringPVFinalizerName)
+	if err := r.client.Update(ctx, &pv); err != nil {
+		return ctrl.Result{}, fmt.Errorf("failed to remove finalizer from PersistentVolume: %s: %s: %w", RestoringPVFinalizerName, pv.Name, err)
+	}
+
+	logger.Info("finalize PV successfully", "pvName", pv.Name)
+
+	return ctrl.Result{}, nil
+}
+
+// SetupWithManager sets up the controller with the Manager.
+func (r *PersistentVolumeReconciler) SetupWithManager(mgr ctrl.Manager) error {
+	return ctrl.NewControllerManagedBy(mgr).
+		For(&corev1.PersistentVolume{}).
+		WithEventFilter(predicate.Funcs{
+			CreateFunc:  func(event.CreateEvent) bool { return true },
+			UpdateFunc:  func(event.UpdateEvent) bool { return true },
+			GenericFunc: func(event.GenericEvent) bool { return true },
+			DeleteFunc: func(ev event.DeleteEvent) bool {
+				return !controllerutil.ContainsFinalizer(ev.Object, RestoringPVFinalizerName)
+			},
+		}).
+		Complete(r)
+}
+
+func (r *PersistentVolumeReconciler) removeRBDImage(ctx context.Context, pv *corev1.PersistentVolume) error {
+	logger := log.FromContext(ctx)
+
+	image := pv.Spec.CSI.VolumeHandle
+	pool := pv.Spec.CSI.VolumeAttributes["pool"]
+	logger.Info("removing image", "pool", pool, "image", image)
+
+	images, err := r.ceph.RBDLs(pool)
+	if err != nil {
+		return fmt.Errorf("failed to list RBD images: %v", err)
+	}
+
+	if !slices.Contains(images, image) {
+		return nil
+	}
+
+	return r.ceph.RBDRm(pool, image)
+}

internal/controller/persistentvolume_controller_e2e.go

@@ -0,0 +1,26 @@
+package controller
+
+import (
+	"context"
+
+	"github.com/cybozu-go/mantle/internal/ceph"
+	corev1 "k8s.io/api/core/v1"
+)
+
+// PersistentVolumeReconcilerE2E is a wrapper of PersistentVolumeReconciler.
+// This module is used to test removeRBDImage in e2e tests.
+type PersistentVolumeReconcilerE2E struct {
+	PersistentVolumeReconciler
+}
+
+func NewPersistentVolumeReconcilerE2E(toolsNamespace string) *PersistentVolumeReconcilerE2E {
+	return &PersistentVolumeReconcilerE2E{
+		PersistentVolumeReconciler{
+			ceph: ceph.NewCephCmdWithTools(toolsNamespace),
+		},
+	}
+}
+
+func (r *PersistentVolumeReconcilerE2E) RemoveRBDImage(ctx context.Context, pv *corev1.PersistentVolume) error {
+	return r.removeRBDImage(ctx, pv)
+}

internal/controller/util.go

@@ -2,6 +2,8 @@ package controller
 
 import (
 	"context"
+	"errors"
+	"fmt"
 	"strings"
 
 	batchv1 "k8s.io/api/batch/v1"
@@ -12,30 +14,46 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/log"
 )
 
-func getCephClusterIDFromPVC(ctx context.Context, k8sClient client.Client, pvc *corev1.PersistentVolumeClaim) (string, error) {
-	logger := log.FromContext(ctx)
-	storageClassName := pvc.Spec.StorageClassName
-	if storageClassName == nil {
-		logger.Info("not managed storage class", "namespace", pvc.Namespace, "pvc", pvc.Name)
-		return "", nil
-	}
+var errEmptyClusterID error = errors.New("cluster ID is empty")
+
+func getCephClusterIDFromSCName(ctx context.Context, k8sClient client.Client, storageClassName string) (string, error) {
 	var storageClass storagev1.StorageClass
-	err := k8sClient.Get(ctx, types.NamespacedName{Name: *storageClassName}, &storageClass)
+	err := k8sClient.Get(ctx, types.NamespacedName{Name: storageClassName}, &storageClass)
 	if err != nil {
-		return "", err
+		return "", fmt.Errorf("failed to get StorageClass: %s: %w", storageClassName, err)
 	}
 
 	// Check if the MantleBackup resource being reconciled is managed by the CephCluster we are in charge of.
 	if !strings.HasSuffix(storageClass.Provisioner, ".rbd.csi.ceph.com") {
-		logger.Info("SC is not managed by RBD", "namespace", pvc.Namespace, "pvc", pvc.Name, "storageClassName", *storageClassName)
-		return "", nil
+		return "", fmt.Errorf("SC is not managed by RBD: %s: %w", storageClassName, errEmptyClusterID)
 	}
 	clusterID, ok := storageClass.Parameters["clusterID"]
 	if !ok {
-		logger.Info("clusterID not found", "namespace", pvc.Namespace, "pvc", pvc.Name, "storageClassName", *storageClassName)
+		return "", fmt.Errorf("clusterID not found: %s: %w", storageClassName, errEmptyClusterID)
+	}
+
+	return clusterID, nil
+}
+
+func getCephClusterIDFromPVC(ctx context.Context, k8sClient client.Client, pvc *corev1.PersistentVolumeClaim) (string, error) {
+	logger := log.FromContext(ctx)
+
+	storageClassName := pvc.Spec.StorageClassName
+	if storageClassName == nil {
+		logger.Info("not managed storage class", "namespace", pvc.Namespace, "pvc", pvc.Name)
 		return "", nil
 	}
 
+	clusterID, err := getCephClusterIDFromSCName(ctx, k8sClient, *storageClassName)
+	if err != nil {
+		logger.Info("failed to get ceph cluster ID from StorageClass name",
+			"error", err, "namespace", pvc.Namespace, "pvc", pvc.Name, "storageClassName", *storageClassName)
+		if errors.Is(err, errEmptyClusterID) {
+			return "", nil
+		}
+		return "", err
+	}
+
 	return clusterID, nil
 }
 

test/e2e/singlek8s/restore_test.go

@@ -555,18 +555,17 @@ func (test *restoreTest) testCloneImageFromBackup() {
 
 func (test *restoreTest) testRemoveImage() {
 	cloneImageName := fmt.Sprintf("mantle-%s-%s", test.tenantNamespace, test.mantleRestoreName1)
-	reconciler := controller.NewMantleRestoreReconcilerE2E(cephCluster1Namespace, cephCluster1Namespace)
-	restore := &mantlev1.MantleRestore{
-		ObjectMeta: metav1.ObjectMeta{
-			Name:      test.mantleRestoreName1,
-			Namespace: test.tenantNamespace,
-		},
-		Spec: mantlev1.MantleRestoreSpec{
-			Backup: test.mantleBackupName1,
-		},
-		Status: mantlev1.MantleRestoreStatus{
-			ClusterID: cephCluster1Namespace,
-			Pool:      test.poolName,
+	pvReconciler := controller.NewPersistentVolumeReconcilerE2E(cephCluster1Namespace)
+	pv := &corev1.PersistentVolume{
+		Spec: corev1.PersistentVolumeSpec{
+			PersistentVolumeSource: corev1.PersistentVolumeSource{
+				CSI: &corev1.CSIPersistentVolumeSource{
+					VolumeHandle: cloneImageName,
+					VolumeAttributes: map[string]string{
+						"pool": test.poolName,
+					},
+				},
+			},
 		},
 	}
 
@@ -575,7 +574,7 @@ func (test *restoreTest) testRemoveImage() {
 			_, err := getRBDInfo(cephCluster1Namespace, test.poolName, cloneImageName)
 			Expect(err).NotTo(HaveOccurred())
 
-			err = reconciler.RemoveRBDImage(ctx, restore)
+			err = pvReconciler.RemoveRBDImage(ctx, pv)
 			Expect(err).NotTo(HaveOccurred())
 
 			// should get an error since the image is removed
@@ -584,7 +583,7 @@ func (test *restoreTest) testRemoveImage() {
 		})
 
 		It("should skip removing the image if it does not exist", func(ctx SpecContext) {
-			err := reconciler.RemoveRBDImage(ctx, restore)
+			err := pvReconciler.RemoveRBDImage(ctx, pv)
 			Expect(err).NotTo(HaveOccurred())
 		})
 	})