From 25ce21f496bb4da77512afb14204b81ab001464c Mon Sep 17 00:00:00 2001
From: Prasanna Kumar Kalever <prasanna.kalever@redhat.com>
Date: Tue, 24 May 2022 19:38:12 +0530
Subject: [PATCH] e2e: add test cases for subvolume metadata validation

create a PVC and check PVC/PV metadata on cephFS subvolume

Fixes: #2875
Signed-off-by: Prasanna Kumar Kalever <prasanna.kalever@redhat.com>
---
 e2e/cephfs.go        | 58 ++++++++++++++++++++++++++++++++++++++++++++
 e2e/cephfs_helper.go | 32 ++++++++++++++++++++++++
 2 files changed, 90 insertions(+)

diff --git a/e2e/cephfs.go b/e2e/cephfs.go
index 4520c8adc..38a930481 100644
--- a/e2e/cephfs.go
+++ b/e2e/cephfs.go
@@ -397,6 +397,64 @@ var _ = Describe(cephfsType, func() {
 				}
 			})
 
+			By("create a PVC and check PVC/PV metadata on CephFS subvolume", func() {
+				err := createCephfsStorageClass(f.ClientSet, f, true, nil)
+				if err != nil {
+					e2elog.Failf("failed to create CephFS storageclass: %v", err)
+				}
+				pvc, err := loadPVC(pvcPath)
+				if err != nil {
+					e2elog.Failf("failed to load PVC: %v", err)
+				}
+				pvc.Namespace = f.UniqueName
+
+				err = createPVCAndvalidatePV(f.ClientSet, pvc, deployTimeout)
+				if err != nil {
+					e2elog.Failf("failed to create PVC: %v", err)
+				}
+
+				validateSubvolumeCount(f, 1, fileSystemName, subvolumegroup)
+				validateOmapCount(f, 1, cephfsType, metadataPool, volumesType)
+
+				pvcObj, err := getPersistentVolumeClaim(c, pvc.Namespace, pvc.Name)
+				if err != nil {
+					e2elog.Logf("error getting pvc %q in namespace %q: %v", pvc.Name, pvc.Namespace, err)
+				}
+				if pvcObj.Spec.VolumeName == "" {
+					e2elog.Logf("pv name is empty %q in namespace %q: %v", pvc.Name, pvc.Namespace, err)
+				}
+				subvol, err := listCephFSSubVolumes(f, fileSystemName, subvolumegroup)
+				if err != nil {
+					e2elog.Failf("failed to list CephFS subvolumes: %v", err)
+				}
+				if len(subvol) == 0 {
+					e2elog.Failf("cephFS subvolumes list is empty %s", fileSystemName)
+				}
+				metadata, err := listCephFSSubvolumeMetadata(f, fileSystemName, subvol[0].Name, subvolumegroup)
+				if err != nil {
+					e2elog.Failf("failed to list subvolume metadata: %v", err)
+				}
+
+				if metadata.PVCNameKey != pvc.Name {
+					e2elog.Failf("expected pvcName %q got %q", pvc.Name, metadata.PVCNameKey)
+				} else if metadata.PVCNamespaceKey != pvc.Namespace {
+					e2elog.Failf("expected pvcNamespace %q got %q", pvc.Namespace, metadata.PVCNamespaceKey)
+				} else if metadata.PVNameKey != pvcObj.Spec.VolumeName {
+					e2elog.Failf("expected pvName %q got %q", pvcObj.Spec.VolumeName, metadata.PVNameKey)
+				}
+
+				err = deletePVCAndValidatePV(f.ClientSet, pvc, deployTimeout)
+				if err != nil {
+					e2elog.Failf("failed to delete PVC: %v", err)
+				}
+				validateSubvolumeCount(f, 0, fileSystemName, subvolumegroup)
+				validateOmapCount(f, 0, cephfsType, metadataPool, volumesType)
+				err = deleteResource(cephFSExamplePath + "storageclass.yaml")
+				if err != nil {
+					e2elog.Failf("failed to delete CephFS storageclass: %v", err)
+				}
+			})
+
 			By("create PVC in storageClass with volumeNamePrefix", func() {
 				volumeNamePrefix := "foo-bar-"
 				err := createCephfsStorageClass(
diff --git a/e2e/cephfs_helper.go b/e2e/cephfs_helper.go
index 4cdd25d0e..2d2553b7e 100644
--- a/e2e/cephfs_helper.go
+++ b/e2e/cephfs_helper.go
@@ -212,6 +212,38 @@ func listCephFSSubVolumes(f *framework.Framework, filesystem, groupname string)
 	return subVols, nil
 }
 
+type cephfsSubvolumeMetadata struct {
+	PVCNameKey      string `json:"csi.storage.k8s.io/pvc/name"`
+	PVCNamespaceKey string `json:"csi.storage.k8s.io/pvc/namespace"`
+	PVNameKey       string `json:"csi.storage.k8s.io/pv/name"`
+}
+
+func listCephFSSubvolumeMetadata(
+	f *framework.Framework,
+	filesystem,
+	subvolume,
+	groupname string,
+) (cephfsSubvolumeMetadata, error) {
+	var metadata cephfsSubvolumeMetadata
+	stdout, stdErr, err := execCommandInToolBoxPod(
+		f,
+		fmt.Sprintf("ceph fs subvolume metadata ls %s %s --group_name=%s --format=json", filesystem, subvolume, groupname),
+		rookNamespace)
+	if err != nil {
+		return metadata, err
+	}
+	if stdErr != "" {
+		return metadata, fmt.Errorf("error listing subvolume metadata %v", stdErr)
+	}
+
+	err = json.Unmarshal([]byte(stdout), &metadata)
+	if err != nil {
+		return metadata, err
+	}
+
+	return metadata, nil
+}
+
 // getSubvolumepath validates whether subvolumegroup is present.
 func getSubvolumePath(f *framework.Framework, filesystem, subvolgrp, subvolume string) (string, error) {
 	cmd := fmt.Sprintf("ceph fs subvolume getpath %s %s --group_name=%s", filesystem, subvolume, subvolgrp)