Sorting functions for common use

internal/controller/mantlebackup_controller.go

@@ -1918,11 +1918,24 @@ func (r *MantleBackupReconciler) reconcileZeroOutJob(
 		return ctrl.Result{}, nil
 	}
 
-	if err := r.createOrUpdateZeroOutPV(ctx, backup, snapshotTarget.pv); err != nil {
+	if err := r.createOrUpdateStaticPV(
+		ctx,
+		snapshotTarget.pv,
+		snapshotTarget.pv.Spec.CSI.VolumeAttributes["imageName"],
+		snapshotTarget.pv.Spec.Capacity,
+		MakeZeroOutPVName(backup),
+		labelComponentZeroOutVolume,
+	); err != nil {
 		return ctrl.Result{}, err
 	}
 
-	if err := r.createOrUpdateZeroOutPVC(ctx, backup, snapshotTarget.pvc); err != nil {
+	if err := r.createOrUpdateStaticPVC(
+		ctx,
+		MakeZeroOutPVCName(backup),
+		MakeZeroOutPVName(backup),
+		labelComponentZeroOutVolume,
+		snapshotTarget.pvc.Spec.Resources,
+	); err != nil {
 		return ctrl.Result{}, err
 	}
 
@@ -1940,24 +1953,40 @@ func (r *MantleBackupReconciler) reconcileZeroOutJob(
 	return requeueReconciliation(), nil
 }
 
-func (r *MantleBackupReconciler) createOrUpdateZeroOutPV(
+// createOrUpdateStaticPV creates or updates a static PersistentVolume (PV) resource.
+// It copies relevant CSI and metadata fields from the basePV, sets the provided volume handle,
+// capacity, and component label, and ensures the PV is configured for static provisioning.
+//
+// Parameters:
+//
+//	ctx           - context for the API calls
+//	basePV        - source PV to copy CSI and metadata fields from
+//	volume        - volume handle to assign to the new PV
+//	capacity      - resource capacity for the PV
+//	newPvName     - name for the new or updated PV
+//	componentName - label value for the component
+//
+// Returns an error if creation or update fails.
+func (r *MantleBackupReconciler) createOrUpdateStaticPV(
 	ctx context.Context,
-	backup *mantlev1.MantleBackup,
-	targetPV *corev1.PersistentVolume,
+	basePV *corev1.PersistentVolume,
+	volume string,
+	capacity corev1.ResourceList,
+	newPvName, componentName string,
 ) error {
 	var pv corev1.PersistentVolume
-	pv.SetName(MakeZeroOutPVName(backup))
+	pv.SetName(newPvName)
 	_, err := ctrl.CreateOrUpdate(ctx, r.Client, &pv, func() error {
 		labels := pv.GetLabels()
 		if labels == nil {
 			labels = map[string]string{}
 		}
 		labels["app.kubernetes.io/name"] = labelAppNameValue
-		labels["app.kubernetes.io/component"] = labelComponentZeroOutVolume
+		labels["app.kubernetes.io/component"] = componentName
 		pv.SetLabels(labels)
 
 		pv.Spec.AccessModes = []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce}
-		pv.Spec.Capacity = targetPV.Spec.Capacity
+		pv.Spec.Capacity = capacity
 		pv.Spec.PersistentVolumeReclaimPolicy = corev1.PersistentVolumeReclaimRetain
 		pv.Spec.StorageClassName = ""
 
@@ -1967,47 +1996,60 @@ func (r *MantleBackupReconciler) createOrUpdateZeroOutPV(
 		if pv.Spec.CSI == nil {
 			pv.Spec.CSI = &corev1.CSIPersistentVolumeSource{}
 		}
-		pv.Spec.CSI.Driver = targetPV.Spec.CSI.Driver
-		pv.Spec.CSI.ControllerExpandSecretRef = targetPV.Spec.CSI.ControllerExpandSecretRef
-		pv.Spec.CSI.NodeStageSecretRef = targetPV.Spec.CSI.NodeStageSecretRef
-		pv.Spec.CSI.VolumeHandle = targetPV.Spec.CSI.VolumeAttributes["imageName"]
+		pv.Spec.CSI.Driver = basePV.Spec.CSI.Driver
+		pv.Spec.CSI.ControllerExpandSecretRef = basePV.Spec.CSI.ControllerExpandSecretRef
+		pv.Spec.CSI.NodeStageSecretRef = basePV.Spec.CSI.NodeStageSecretRef
+		pv.Spec.CSI.VolumeHandle = volume
 
 		if pv.Spec.CSI.VolumeAttributes == nil {
 			pv.Spec.CSI.VolumeAttributes = map[string]string{}
 		}
-		pv.Spec.CSI.VolumeAttributes["clusterID"] = targetPV.Spec.CSI.VolumeAttributes["clusterID"]
-		pv.Spec.CSI.VolumeAttributes["imageFeatures"] = targetPV.Spec.CSI.VolumeAttributes["imageFeatures"]
-		pv.Spec.CSI.VolumeAttributes["imageFormat"] = targetPV.Spec.CSI.VolumeAttributes["imageFormat"]
-		pv.Spec.CSI.VolumeAttributes["pool"] = targetPV.Spec.CSI.VolumeAttributes["pool"]
+		pv.Spec.CSI.VolumeAttributes["clusterID"] = basePV.Spec.CSI.VolumeAttributes["clusterID"]
+		pv.Spec.CSI.VolumeAttributes["imageFeatures"] = basePV.Spec.CSI.VolumeAttributes["imageFeatures"]
+		pv.Spec.CSI.VolumeAttributes["imageFormat"] = basePV.Spec.CSI.VolumeAttributes["imageFormat"]
+		pv.Spec.CSI.VolumeAttributes["pool"] = basePV.Spec.CSI.VolumeAttributes["pool"]
 		pv.Spec.CSI.VolumeAttributes["staticVolume"] = "true"
 
 		return nil
 	})
 	return err
 }
 
-func (r *MantleBackupReconciler) createOrUpdateZeroOutPVC(
+// createOrUpdateStaticPVC creates or updates a PersistentVolumeClaim (PVC) for binding to a static PersistentVolume (PV).
+// This function configures the PVC to reference the specified static PV, sets labels, access modes, storage class,
+// and resource requirements for static volume provisioning.
+//
+// Parameters:
+//
+//	ctx           - context for the API calls
+//	pvcName       - name for the new or updated PVC
+//	pvName        - name of the static PV to bind to
+//	componentName - label value for the component
+//	resources     - resource requirements for the PVC
+//
+// Returns an error if creation or update fails.
+func (r *MantleBackupReconciler) createOrUpdateStaticPVC(
 	ctx context.Context,
-	backup *mantlev1.MantleBackup,
-	targetPVC *corev1.PersistentVolumeClaim,
+	pvcName, pvName, componentName string,
+	resources corev1.VolumeResourceRequirements,
 ) error {
 	var pvc corev1.PersistentVolumeClaim
-	pvc.SetName(MakeZeroOutPVCName(backup))
+	pvc.SetName(pvcName)
 	pvc.SetNamespace(r.managedCephClusterID)
 	_, err := ctrl.CreateOrUpdate(ctx, r.Client, &pvc, func() error {
 		labels := pvc.GetLabels()
 		if labels == nil {
 			labels = map[string]string{}
 		}
 		labels["app.kubernetes.io/name"] = labelAppNameValue
-		labels["app.kubernetes.io/component"] = labelComponentZeroOutVolume
+		labels["app.kubernetes.io/component"] = componentName
 		pvc.SetLabels(labels)
 
 		storageClassName := ""
 		pvc.Spec.StorageClassName = &storageClassName
 		pvc.Spec.AccessModes = []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce}
-		pvc.Spec.Resources = targetPVC.Spec.Resources
-		pvc.Spec.VolumeName = MakeZeroOutPVName(backup)
+		pvc.Spec.Resources = resources
+		pvc.Spec.VolumeName = pvName
 
 		volumeMode := corev1.PersistentVolumeBlock
 		pvc.Spec.VolumeMode = &volumeMode

internal/controller/mantlerestore_controller.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"slices"
 
 	mantlev1 "github.com/cybozu-go/mantle/api/v1"
 	"github.com/cybozu-go/mantle/internal/ceph"
@@ -183,54 +182,13 @@ func (r *MantleRestoreReconciler) restoringPVName(restore *mantlev1.MantleRestor
 }
 
 func (r *MantleRestoreReconciler) cloneImageFromBackup(ctx context.Context, restore *mantlev1.MantleRestore, backup *mantlev1.MantleBackup) error {
-	logger := log.FromContext(ctx)
 	pv := corev1.PersistentVolume{}
 	err := json.Unmarshal([]byte(backup.Status.PVManifest), &pv)
 	if err != nil {
 		return fmt.Errorf("failed to unmarshal PV manifest: %w", err)
 	}
 
-	bkImage := pv.Spec.CSI.VolumeAttributes["imageName"]
-	if bkImage == "" {
-		return fmt.Errorf("imageName not found in PV manifest")
-	}
-	pool := pv.Spec.CSI.VolumeAttributes["pool"]
-	if pool == "" {
-		return fmt.Errorf("pool not found in PV manifest")
-	}
-
-	images, err := r.ceph.RBDLs(pool)
-	if err != nil {
-		return fmt.Errorf("failed to list RBD images: %w", err)
-	}
-
-	// check if the image already exists
-	if slices.Contains(images, r.restoringRBDImageName(restore)) {
-		info, err := r.ceph.RBDInfo(pool, r.restoringRBDImageName(restore))
-		if err != nil {
-			return fmt.Errorf("failed to get RBD info: %w", err)
-		}
-		if info.Parent == nil {
-			return fmt.Errorf("failed to get RBD info: parent field is empty")
-		}
-
-		if info.Parent.Pool == pool && info.Parent.Image == bkImage && info.Parent.Snapshot == backup.Name {
-			logger.Info("image already exists", "image", r.restoringRBDImageName(restore))
-			return nil
-		} else {
-			return fmt.Errorf("image already exists but not a clone of the backup: %s", r.restoringRBDImageName(restore))
-		}
-	}
-
-	features := pv.Spec.CSI.VolumeAttributes["imageFeatures"]
-	if features == "" {
-		features = "deep-flatten"
-	} else {
-		features += ",deep-flatten"
-	}
-
-	// create a clone image from the backup
-	return r.ceph.RBDClone(pool, bkImage, backup.Name, r.restoringRBDImageName(restore), features)
+	return createCloneByPV(ctx, r.ceph, &pv, backup.Name, r.restoringRBDImageName(restore))
 }
 
 func (r *MantleRestoreReconciler) createOrUpdateRestoringPV(ctx context.Context, restore *mantlev1.MantleRestore, backup *mantlev1.MantleBackup) error {

internal/controller/util.go

@@ -5,9 +5,11 @@ import (
 	"errors"
 	"fmt"
 	"os"
+	"slices"
 	"strings"
 	"time"
 
+	"github.com/cybozu-go/mantle/internal/ceph"
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
 	storagev1 "k8s.io/api/storage/v1"
@@ -19,6 +21,68 @@ import (
 
 var errEmptyClusterID error = errors.New("cluster ID is empty")
 
+// createCloneByPV creates a clone RBD image from a snapshot of the volume bound to the given PersistentVolume (PV).
+//
+// This function checks if the clone image already exists in the Ceph pool:
+//   - If the clone exists and is a clone of the specified snapshot, it does nothing and returns nil.
+//   - If the clone exists but is not a clone of the specified snapshot, it returns an error.
+//   - If the clone does not exist, it creates a new clone image from the snapshot using the specified image features.
+//
+// Parameters:
+//
+//	ctx         - context for logging and API calls
+//	cephCmd     - Ceph command interface for RBD operations
+//	pv          - PersistentVolume object containing CSI and image attributes
+//	snapshotName- name of the snapshot to clone from
+//	cloneName   - name for the new clone image
+//
+// Returns an error if required attributes are missing, if Ceph operations fail, or if a conflicting clone image exists.
+func createCloneByPV(ctx context.Context, cephCmd ceph.CephCmd, pv *corev1.PersistentVolume, snapshotName, cloneName string) error {
+	logger := log.FromContext(ctx)
+
+	bkImage := pv.Spec.CSI.VolumeAttributes["imageName"]
+	if bkImage == "" {
+		return fmt.Errorf("imageName not found in PV manifest")
+	}
+	pool := pv.Spec.CSI.VolumeAttributes["pool"]
+	if pool == "" {
+		return fmt.Errorf("pool not found in PV manifest")
+	}
+
+	images, err := cephCmd.RBDLs(pool)
+	if err != nil {
+		return fmt.Errorf("failed to list RBD images: %w", err)
+	}
+
+	// check if the image already exists
+	if slices.Contains(images, cloneName) {
+		info, err := cephCmd.RBDInfo(pool, cloneName)
+		if err != nil {
+			return fmt.Errorf("failed to get RBD info: %w", err)
+		}
+		if info.Parent == nil {
+			return fmt.Errorf("failed to get RBD info: parent field is empty")
+		}
+
+		if info.Parent.Pool == pool && info.Parent.Image == bkImage && info.Parent.Snapshot == snapshotName {
+			logger.Info("image already exists", "image", cloneName)
+			return nil
+		}
+		// If the clone image already exists but is not a clone of the snapshot, it returns an error.
+		return fmt.Errorf("image already exists but not a clone of the backup: %s", cloneName)
+	}
+
+	features := pv.Spec.CSI.VolumeAttributes["imageFeatures"]
+	if features == "" {
+		features = "deep-flatten"
+	} else {
+		features += ",deep-flatten"
+	}
+
+	// create a clone image from the backup
+	return cephCmd.RBDClone(pool, bkImage, snapshotName, cloneName, features)
+}
+
 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)