wip

Signed-off-by: Yuji Ito <[email protected]>

ceph/test/cluster/kubectl.go

@@ -22,6 +22,8 @@ const (
 var (
 	//go:embed template/deployment.yaml
 	deploymentTemplate string
+	//go:embed template/pod_with_block.yaml
+	podWithBlockTemplate string
 	//go:embed template/pool.yaml
 	poolTemplate string
 	//go:embed template/pvc.yaml
@@ -92,6 +94,19 @@ func GetObjectList[T any](kind, namespace string) (*T, error) {
 	return &objList, nil
 }
 
+func DeleteObject(kind, namespace, name string) error {
+	var err error
+	if len(namespace) == 0 {
+		_, err = Kubectl("delete", kind, name)
+	} else {
+		_, err = Kubectl("delete", kind, "-n", namespace, name)
+	}
+	if err != nil {
+		return fmt.Errorf("failed to delete %s %s/%s: %w", kind, namespace, name, err)
+	}
+	return nil
+}
+
 func DeleteAllObjects(kind, namespace string) error {
 	_, err := Kubectl("delete", kind, "-n", namespace, "--all", "--wait=false")
 	if err != nil {
@@ -121,6 +136,19 @@ func CreateDeployment(namespace, deployName, pvcName string) error {
 	return nil
 }
 
+func CreatePodWithBlock(namespace, podName, pvcName string) error {
+	manifest := fmt.Sprintf(podWithBlockTemplate, namespace, podName, pvcName)
+	_, err := KubectlWithInput([]byte(manifest), "apply", "-f", "-")
+	if err != nil {
+		return fmt.Errorf("failed to create pod with block: %w", err)
+	}
+	_, err = Kubectl("wait", "--for=condition=ready", "-n", namespace, "pod/"+podName, "--timeout=60s")
+	if err != nil {
+		return fmt.Errorf("failed to wait for pod: %w", err)
+	}
+	return nil
+}
+
 func CreatePool(poolName string) error {
 	manifest := fmt.Sprintf(poolTemplate, poolName)
 	_, err := KubectlWithInput([]byte(manifest), "apply", "-f", "-")
@@ -212,12 +240,12 @@ func ScaleDeployment(namespace, deployName string, replicas int) error {
 		return fmt.Errorf("failed to scale deployment: %w", err)
 	}
 	if replicas == 0 {
-		_, err = Kubectl("wait", "--for=delete", "pod", "-n", namespace, "--selector=app="+deployName)
+		_, err = Kubectl("wait", "--for=delete", "pod", "-n", namespace, "--selector=app="+deployName, "--timeout=3m")
 		if err != nil {
 			return fmt.Errorf("failed to wait for pod deletion: %w", err)
 		}
 	} else {
-		_, err = Kubectl("wait", "--for=condition=available", "deploy", "-n", namespace, deployName)
+		_, err = Kubectl("wait", "--for=condition=available", "deploy", "-n", namespace, deployName, "--timeout=3m")
 		if err != nil {
 			return fmt.Errorf("failed to wait for deployment available: %w", err)
 		}

ceph/test/cluster/rbd.go

@@ -1,6 +1,14 @@
 package cluster
 
-import "fmt"
+import (
+	"encoding/json"
+	"fmt"
+	"sync"
+
+	"github.com/cybozu-go/mantle/test/util"
+	corev1 "k8s.io/api/core/v1"
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
 
 func Rbd(args ...string) ([]byte, error) {
 	return Kubectl(append([]string{"exec", "-n", ROOK_NAMESPACE, "deploy/rook-ceph-tools", "--", "rbd"}, args...)...)
@@ -20,19 +28,151 @@ func ExportDiff(filename string, args ...string) error {
 	return nil
 }
 
-func ImportDiff(filename, pool, image string) error {
-	stdout, err := Kubectl("exec", "-n", ROOK_NAMESPACE, "deploy/rook-ceph-tools", "--",
-		"sh", "-c", fmt.Sprintf("cat %s | rbd import-diff -p %s - %s", filename, pool, image))
+func ImportDiff(filename, pool, image, rollbackTo, namespace, deployName, pvcName string) error {
+	return RunWithStopPod(namespace, deployName, func() error {
+		if rollbackTo == "" {
+			if len(pvcName) == 0 {
+				return fmt.Errorf("rollbackTo or pvcName must be specified")
+			}
+			err := discardVolume(namespace, pvcName)
+			if err != nil {
+				return fmt.Errorf("failed to discard volume: %w", err)
+			}
+		} else {
+			stdout, err := Rbd("snap", "rollback", pool+"/"+image+"@"+rollbackTo)
+			if err != nil {
+				return fmt.Errorf("failed to rollback snapshot: %w, %s", err, string(stdout))
+			}
+		}
+
+		stdout, err := Kubectl("exec", "-n", ROOK_NAMESPACE, "deploy/rook-ceph-tools", "--",
+			"sh", "-c", fmt.Sprintf("cat %s | rbd import-diff -p %s - %s", filename, pool, image))
+		if err != nil {
+			return fmt.Errorf("failed to import diff: %w, %s", err, string(stdout))
+		}
+		return nil
+	})
+}
+
+var mtxDiscardVolume = &sync.Mutex{}
+
+func discardVolume(namespace, pvcName string) error {
+	mtxDiscardVolume.Lock()
+	defer mtxDiscardVolume.Unlock()
+
+	origPVCRaw, err := Kubectl("get", "-n", namespace, "pvc", pvcName, "-o", "json")
 	if err != nil {
-		return fmt.Errorf("failed to import diff: %w, %s", err, string(stdout))
+		return fmt.Errorf("failed to get PVC: %w", err)
+	}
+	var origPVC corev1.PersistentVolumeClaim
+	if err := json.Unmarshal(origPVCRaw, &origPVC); err != nil {
+		return fmt.Errorf("failed to unmarshal PVC: %w", err)
 	}
-	return nil
-}
 
-func ImportDiffWithStoppingPod(filename, pool, image, namespace, deployName string) error {
-	return RunWithStopPod(namespace, deployName, func() error {
-		return ImportDiff(filename, pool, image)
-	})
+	origPVRaw, err := Kubectl("get", "pv", origPVC.Spec.VolumeName, "-o", "json")
+	if err != nil {
+		return fmt.Errorf("failed to get PV: %w", err)
+	}
+	var origPV corev1.PersistentVolume
+	if err := json.Unmarshal(origPVRaw, &origPV); err != nil {
+		return fmt.Errorf("failed to unmarshal PV: %w", err)
+	}
+
+	discardPVName := util.GetUniqueName("discard-pv-")
+	discardPVCName := util.GetUniqueName("discard-pvc-")
+	discardPodName := util.GetUniqueName("discard-pod-")
+
+	volumeMode := corev1.PersistentVolumeBlock
+	discardPV := corev1.PersistentVolume{
+		TypeMeta: origPV.TypeMeta,
+		ObjectMeta: v1.ObjectMeta{
+			Name: discardPVName,
+		},
+		Spec: corev1.PersistentVolumeSpec{
+			AccessModes: []corev1.PersistentVolumeAccessMode{
+				corev1.ReadWriteOnce,
+			},
+			Capacity:                      origPV.Spec.Capacity,
+			PersistentVolumeReclaimPolicy: corev1.PersistentVolumeReclaimRetain,
+			VolumeMode:                    &volumeMode,
+			StorageClassName:              "",
+		},
+	}
+	discardPV.Spec.CSI = &corev1.CSIPersistentVolumeSource{
+		Driver:                    origPV.Spec.CSI.Driver,
+		ControllerExpandSecretRef: origPV.Spec.CSI.ControllerExpandSecretRef,
+		NodeStageSecretRef:        origPV.Spec.CSI.NodeStageSecretRef,
+		VolumeAttributes: map[string]string{
+			"clusterID":     origPV.Spec.CSI.VolumeAttributes["clusterID"],
+			"imageFeatures": origPV.Spec.CSI.VolumeAttributes["imageFeatures"],
+			"imageFormat":   origPV.Spec.CSI.VolumeAttributes["imageFormat"],
+			"pool":          origPV.Spec.CSI.VolumeAttributes["pool"],
+			"staticVolume":  "true",
+		},
+		VolumeHandle: origPV.Spec.CSI.VolumeAttributes["imageName"],
+	}
+	discardPVRaw, err := json.Marshal(discardPV)
+	if err != nil {
+		return fmt.Errorf("failed to marshal PV: %w", err)
+	}
+	_, err = KubectlWithInput(discardPVRaw, "apply", "-f", "-")
+	if err != nil {
+		return fmt.Errorf("failed to create PV: %w", err)
+	}
+
+	sc := ""
+	discardPVC := corev1.PersistentVolumeClaim{
+		TypeMeta: origPVC.TypeMeta,
+		ObjectMeta: v1.ObjectMeta{
+			Name:      discardPVCName,
+			Namespace: namespace,
+		},
+		Spec: corev1.PersistentVolumeClaimSpec{
+			StorageClassName: &sc,
+			AccessModes: []corev1.PersistentVolumeAccessMode{
+				corev1.ReadWriteOnce,
+			},
+			Resources:  origPVC.Spec.Resources,
+			VolumeMode: &volumeMode,
+			VolumeName: discardPVName,
+		},
+	}
+	discardPVCRaw, err := json.Marshal(discardPVC)
+	if err != nil {
+		return fmt.Errorf("failed to marshal PVC: %w", err)
+	}
+	_, err = KubectlWithInput(discardPVCRaw, "apply", "-f", "-")
+	if err != nil {
+		return fmt.Errorf("failed to create PVC: %w", err)
+	}
+
+	err = CreatePodWithBlock(namespace, discardPodName, discardPVCName)
+	if err != nil {
+		return fmt.Errorf("failed to create pod: %w", err)
+	}
+
+	_, err = Kubectl("exec", "-n", namespace, discardPodName, "--",
+		"blkdiscard", "/dev/rbd-device")
+	if err != nil {
+		return fmt.Errorf("failed to discard volume: %w", err)
+	}
+
+	err = DeleteObject("pod", namespace, discardPodName)
+	if err != nil {
+		return fmt.Errorf("failed to delete pod: %w", err)
+	}
+
+	err = DeleteObject("pvc", namespace, discardPVCName)
+	if err != nil {
+		return fmt.Errorf("failed to delete PVC: %w", err)
+	}
+
+	err = DeleteObject("pv", "", discardPVName)
+	if err != nil {
+		return fmt.Errorf("failed to delete PV: %w", err)
+	}
+
+	return nil
 }
 
 func SnapCreate(pool, image, snap string) error {

ceph/test/cluster/template/pod_with_block.yaml

@@ -0,0 +1,17 @@
+apiVersion: v1
+kind: Pod
+metadata:
+  namespace: "%s"
+  name: "%s"
+spec:
+  containers:
+    - name: ubuntu
+      image: ghcr.io/cybozu/ubuntu:22.04
+      command: ["pause"]
+      volumeDevices:
+        - name: rbd-device
+          devicePath: /dev/rbd-device
+  volumes:
+    - name: rbd-device
+      persistentVolumeClaim:
+        claimName: "%s"