nvmeof: add volume cloning capability

e2e/nvmeof.go

@@ -130,6 +130,7 @@ var _ = ginkgo.Describe("nvmeof", func() {
 	ginkgo.Context("Test NVMe CSI", ginkgo.Ordered, func() {
 
 		pvcPath := nvmeofExamplePath + "pvc.yaml"
+		pvcClonePath := nvmeofExamplePath + "pvc-clone.yaml"
 		appPath := nvmeofExamplePath + "pod.yaml"
 		rawPvcPath := nvmeofExamplePath + "raw-block-pvc.yaml"
 		rawAppPath := nvmeofExamplePath + "raw-block-pod.yaml"
@@ -432,5 +433,70 @@ var _ = ginkgo.Describe("nvmeof", func() {
 			validateRBDImageCount(f, 0, nvmeofPool)
 			validateOmapCount(f, 0, rbdType, nvmeofPool, volumesType)
 		})
+
+		ginkgo.It("Cloning nvmeof PVC", func() {
+			// This test validates cloning of NVMe-oF PVCs.
+			//
+			// Test flow:
+			// 1. Create a source PVC and bind it to an app
+			// 2. Create a clone of the source PVC and bind it to an app
+			// 3. Delete the clone app and PVC
+			// 4. Delete the source app and PVC
+
+			ginkgo.By("Creating a source PVC")
+			sourcePVC, err := loadPVC(pvcPath)
+			Expect(err).ShouldNot(HaveOccurred())
+
+			sourcePVC.Namespace = f.UniqueName
+			sourcePVC.Spec.StorageClassName = &nvmeofStorageClass
+
+			err = createPVCAndvalidatePV(f.ClientSet, sourcePVC, deployTimeout)
+			Expect(err).ShouldNot(HaveOccurred())
+
+			ginkgo.By("Binding source PVC to an application")
+			sourceApp, err := loadApp(appPath)
+			Expect(err).ShouldNot(HaveOccurred())
+
+			sourceApp.Namespace = f.UniqueName
+			sourceApp.Spec.Volumes[0].PersistentVolumeClaim.ClaimName = sourcePVC.Name
+
+			err = createApp(f.ClientSet, sourceApp, deployTimeout)
+			Expect(err).ShouldNot(HaveOccurred())
+
+			ginkgo.By("Creating a clone of the source PVC")
+			// Load clone PVC template with DataSource already configured
+			clonePVC, err := loadPVC(pvcClonePath)
+			Expect(err).ShouldNot(HaveOccurred())
+
+			clonePVC.Namespace = f.UniqueName
+			clonePVC.Spec.StorageClassName = &nvmeofStorageClass
+			clonePVC.Spec.DataSource.Name = sourcePVC.Name
+
+			err = createPVCAndvalidatePV(f.ClientSet, clonePVC, deployTimeout)
+			Expect(err).ShouldNot(HaveOccurred())
+
+			ginkgo.By("Binding clone PVC to an application")
+			cloneApp, err := loadApp(appPath)
+			Expect(err).ShouldNot(HaveOccurred())
+
+			cloneApp.Name = sourceApp.Name + "-clone"
+			cloneApp.Namespace = f.UniqueName
+			cloneApp.Spec.Volumes[0].PersistentVolumeClaim.ClaimName = clonePVC.Name
+
+			err = createApp(f.ClientSet, cloneApp, deployTimeout)
+			Expect(err).ShouldNot(HaveOccurred())
+
+			ginkgo.By("Deleting the clone application and PVC")
+			err = deletePVCAndApp("", f, clonePVC, cloneApp)
+			Expect(err).ShouldNot(HaveOccurred())
+
+			ginkgo.By("Deleting the source application and PVC")
+			err = deletePVCAndApp("", f, sourcePVC, sourceApp)
+			Expect(err).ShouldNot(HaveOccurred())
+
+			// validate all backend rbd images are cleaned up
+			validateRBDImageCount(f, 0, nvmeofPool)
+			validateOmapCount(f, 0, rbdType, nvmeofPool, volumesType)
+		})
 	})
 })

examples/nvmeof/pvc-clone.yaml

@@ -0,0 +1,15 @@
+---
+apiVersion: v1
+kind: PersistentVolumeClaim
+metadata:
+  name: nvmeof-pvc-clone
+spec:
+  storageClassName: csi-nvmeof-sc
+  dataSource:
+    name: nvmeof-pvc
+    kind: PersistentVolumeClaim
+  accessModes:
+    - ReadWriteOnce
+  resources:
+    requests:
+      storage: 64Mi

internal/nvmeof/controller/controllerserver.go

@@ -118,7 +118,37 @@ func (cs *Server) CreateVolume(
 	}
 	defer cs.volumeLocks.Release(req.GetName())
 
+	// Extract clone source if present (for locking)
+	// Note- there is no need for snapshot source locking because snapshots are not
+	// used in the NVMe-oF gateway and do not affect the NVMe-oF resources.
+	var sourceVolumeID, sourceSnapshotID string
+	if contentSource := req.GetVolumeContentSource(); contentSource != nil {
+		switch contentSource.GetType().(type) {
+		case *csi.VolumeContentSource_Snapshot:
+			if snapshot := contentSource.GetSnapshot(); snapshot != nil {
+				sourceSnapshotID = snapshot.GetSnapshotId()
+				log.DebugLog(ctx, "Creating volume from snapshot: %s", sourceSnapshotID)
+			}
+		case *csi.VolumeContentSource_Volume:
+			if vol := contentSource.GetVolume(); vol != nil {
+				sourceVolumeID = vol.GetVolumeId()
+				log.DebugLog(ctx, "Creating volume clone from volume: %s", sourceVolumeID)
+			}
+		}
+	}
+
+	// Lock source volume to prevent concurrent deletion
+	if sourceVolumeID != "" {
+		if acquired := cs.volumeLocks.TryAcquire(sourceVolumeID); !acquired {
+			log.ErrorLog(ctx, util.VolumeOperationAlreadyExistsFmt, sourceVolumeID)
+
+			return nil, status.Errorf(codes.Aborted, util.VolumeOperationAlreadyExistsFmt, sourceVolumeID)
+		}
+		defer cs.volumeLocks.Release(sourceVolumeID)
+	}
+
 	// Step 1: Create RBD volume through backend. if exists, it is ok.
+	// RBD backend automatically handles cloning when VolumeContentSource is present.
 	res, err := cs.backendServer.CreateVolume(ctx, req)
 	if err != nil {
 		log.ErrorLog(ctx, "failed to create RBD volume: %v", err)

internal/nvmeof/driver/driver.go

@@ -66,6 +66,7 @@ func (d *nvmeofDriver) Run(conf *util.Config) {
 			csi.ControllerServiceCapability_RPC_MODIFY_VOLUME,
 			csi.ControllerServiceCapability_RPC_EXPAND_VOLUME,
 			csi.ControllerServiceCapability_RPC_CREATE_DELETE_SNAPSHOT,
+			csi.ControllerServiceCapability_RPC_CLONE_VOLUME,
 		})
 
 		cd.AddVolumeCapabilityAccessModes([]csi.VolumeCapability_AccessMode_Mode{