cephfs: remove subvolumegroup creation

Signed-off-by: Praveen M <[email protected]>

e2e/cephfs.go

@@ -215,6 +215,9 @@ var _ = Describe(cephfsType, func() {
 		if err != nil {
 			framework.Failf("timeout waiting for deployment update %s/%s: %v", cephCSINamespace, cephFSDeploymentName, err)
 		}
+
+		// create subvolumegroup
+		createSubvolumegroup(f)
 	})
 
 	AfterEach(func() {
@@ -263,6 +266,9 @@ var _ = Describe(cephfsType, func() {
 				}
 			}
 		}
+
+		// delete subvolumegroup
+		deleteSubvolumegroup(f)
 	})
 
 	Context("Test CephFS CSI", func() {

e2e/cephfs_helper.go

@@ -568,3 +568,32 @@ func validateFscryptClone(
 		}
 	}
 }
+
+// validateSubvolumegroup validates whether subvolumegroup is present.
+func createSubvolumegroup(f *framework.Framework) error {
+	cmd := fmt.Sprintf("ceph fs subvolumegroup create %s %s", fileSystemName, subvolumegroup)
+	_, stdErr, err := execCommandInToolBoxPod(f, cmd, rookNamespace)
+	if err != nil {
+		return fmt.Errorf("failed to exec command in toolbox: %w", err)
+	}
+	if stdErr != "" {
+		return fmt.Errorf("failed to create subvolumegroup %s : %v", subvolumegroup, stdErr)
+	}
+
+	return nil
+}
+
+// validateSubvolumegroup validates whether subvolumegroup is present.
+func deleteSubvolumegroup(f *framework.Framework) error {
+	cmd := fmt.Sprintf("ceph fs subvolumegroup rm %s %s", fileSystemName, subvolumegroup)
+	_, stdErr, err := execCommandInToolBoxPod(f, cmd, rookNamespace)
+	if err != nil {
+		return fmt.Errorf("failed to exec command in toolbox: %w", err)
+	}
+	if stdErr != "" {
+		return fmt.Errorf("failed to remove subvolumegroup %s : %v", subvolumegroup, stdErr)
+	}
+
+	return nil
+}
+

internal/cephfs/core/metadata.go

@@ -51,32 +51,6 @@ func (s *subVolumeClient) isUnsupportedSubVolMetadata(err error) bool {
 	return true
 }
 
-// isSubVolumeGroupCreated returns true if subvolume group is created.
-func (s *subVolumeClient) isSubVolumeGroupCreated() bool {
-	newLocalClusterState(s.clusterID)
-	clusterAdditionalInfo[s.clusterID].subVolumeGroupsRWMutex.RLock()
-	defer clusterAdditionalInfo[s.clusterID].subVolumeGroupsRWMutex.RUnlock()
-
-	if clusterAdditionalInfo[s.clusterID].subVolumeGroupsCreated == nil {
-		return false
-	}
-
-	return clusterAdditionalInfo[s.clusterID].subVolumeGroupsCreated[s.SubvolumeGroup]
-}
-
-// updateSubVolumeGroupCreated updates subvolume group created map.
-// If the map is nil, it creates a new map and updates it.
-func (s *subVolumeClient) updateSubVolumeGroupCreated(state bool) {
-	clusterAdditionalInfo[s.clusterID].subVolumeGroupsRWMutex.Lock()
-	defer clusterAdditionalInfo[s.clusterID].subVolumeGroupsRWMutex.Unlock()
-
-	if clusterAdditionalInfo[s.clusterID].subVolumeGroupsCreated == nil {
-		clusterAdditionalInfo[s.clusterID].subVolumeGroupsCreated = make(map[string]bool)
-	}
-
-	clusterAdditionalInfo[s.clusterID].subVolumeGroupsCreated[s.SubvolumeGroup] = state
-}
-
 // setMetadata sets custom metadata on the subvolume in a volume as a
 // key-value pair.
 func (s *subVolumeClient) setMetadata(key, value string) error {

internal/cephfs/core/volume.go

@@ -210,14 +210,6 @@ type localClusterState struct {
 	resizeState                 operationState
 	subVolMetadataState         operationState
 	subVolSnapshotMetadataState operationState
-	// A cluster can have multiple filesystem for that we need to have a map of
-	// subvolumegroups to check filesystem is created nor not.
-	// set true once a subvolumegroup is created
-	// for corresponding filesystem in a cluster.
-	subVolumeGroupsCreated map[string]bool
-	// subVolumeGroupsRWMutex is used to protect subVolumeGroupsCreated map
-	//  against concurrent writes while allowing multiple readers.
-	subVolumeGroupsRWMutex sync.RWMutex
 }
 
 func newLocalClusterState(clusterID string) {
@@ -241,24 +233,6 @@ func (s *subVolumeClient) CreateVolume(ctx context.Context) error {
 		return err
 	}
 
-	// create subvolumegroup if not already created for the cluster.
-	if !s.isSubVolumeGroupCreated() {
-		opts := fsAdmin.SubVolumeGroupOptions{}
-		err = ca.CreateSubVolumeGroup(s.FsName, s.SubvolumeGroup, &opts)
-		if err != nil {
-			log.ErrorLog(
-				ctx,
-				"failed to create subvolume group %s, for the vol %s: %s",
-				s.SubvolumeGroup,
-				s.VolID,
-				err)
-
-			return err
-		}
-		log.DebugLog(ctx, "cephfs: created subvolume group %s", s.SubvolumeGroup)
-		s.updateSubVolumeGroupCreated(true)
-	}
-
 	opts := fsAdmin.SubVolumeOptions{
 		Size: fsAdmin.ByteCount(s.Size),
 	}
@@ -271,12 +245,6 @@ func (s *subVolumeClient) CreateVolume(ctx context.Context) error {
 	if err != nil {
 		log.ErrorLog(ctx, "failed to create subvolume %s in fs %s: %s", s.VolID, s.FsName, err)
 
-		if errors.Is(err, rados.ErrNotFound) {
-			// Reset the subVolumeGroupsCreated so that we can try again to create the
-			// subvolumegroup in next request if the error is Not Found.
-			s.updateSubVolumeGroupCreated(false)
-		}
-
 		return err
 	}