Sorting functions for common use (createCloneByPV)

createCloneByPV is create RBD clone volume by PV's snapshot

Signed-off-by: Yuji Ito <llamerada.jp@gmail.com>

Author: llamerada-jp

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,55 @@ import (
 
 var errEmptyClusterID error = errors.New("cluster ID is empty")
 
+// createCloneByPV creates a clone RBD image from the snapshot of the volume bound to PV.
+// If the clone image already exists and is a clone of the snapshot, it does nothing.
+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
+		} else {
+			// 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)