cephfs: handle metadata op-failures with unsupported ceph versions

Signed-off-by: Prasanna Kumar Kalever <prasanna.kalever@redhat.com>

internal/cephfs/controllerserver.go

@@ -601,6 +601,7 @@ func cleanUpBackingVolume(
 		snapClient := core.NewSnapshot(
 			snapParentVolOptions.GetConnection(),
 			snapID.FsSnapshotName,
+			volOptions.ClusterID,
 			&snapParentVolOptions.SubVolume,
 		)
 
@@ -819,7 +820,11 @@ func (cs *ControllerServer) CreateSnapshot(
 	if sid != nil {
 		// check snapshot is protected
 		protected := true
-		snapClient := core.NewSnapshot(parentVolOptions.GetConnection(), sid.FsSnapshotName, &parentVolOptions.SubVolume)
+		snapClient := core.NewSnapshot(
+			parentVolOptions.GetConnection(),
+			sid.FsSnapshotName,
+			parentVolOptions.ClusterID,
+			&parentVolOptions.SubVolume)
 		if !(snapInfo.Protected == core.SnapshotIsProtected) {
 			err = snapClient.ProtectSnapshot(ctx)
 			if err != nil {
@@ -887,7 +892,7 @@ func doSnapshot(
 ) (core.SnapshotInfo, error) {
 	snapID := fsutil.VolumeID(snapshotName)
 	snap := core.SnapshotInfo{}
-	snapClient := core.NewSnapshot(volOpt.GetConnection(), snapshotName, &volOpt.SubVolume)
+	snapClient := core.NewSnapshot(volOpt.GetConnection(), snapshotName, volOpt.ClusterID, &volOpt.SubVolume)
 	err := snapClient.CreateSnapshot(ctx)
 	if err != nil {
 		log.ErrorLog(ctx, "failed to create snapshot %s %v", snapID, err)
@@ -1044,7 +1049,7 @@ func (cs *ControllerServer) DeleteSnapshot(
 	if snapInfo.HasPendingClones == "yes" {
 		return nil, status.Errorf(codes.FailedPrecondition, "snapshot %s has pending clones", snapshotID)
 	}
-	snapClient := core.NewSnapshot(volOpt.GetConnection(), sid.FsSnapshotName, &volOpt.SubVolume)
+	snapClient := core.NewSnapshot(volOpt.GetConnection(), sid.FsSnapshotName, volOpt.ClusterID, &volOpt.SubVolume)
 	if snapInfo.Protected == core.SnapshotIsProtected {
 		err = snapClient.UnprotectSnapshot(ctx)
 		if err != nil {

internal/cephfs/core/clone.go

@@ -66,7 +66,7 @@ func (s *subVolumeClient) CreateCloneFromSubvolume(
 	parentvolOpt *SubVolume,
 ) error {
 	snapshotID := s.VolID
-	snapClient := NewSnapshot(s.conn, snapshotID, parentvolOpt)
+	snapClient := NewSnapshot(s.conn, snapshotID, s.clusterID, parentvolOpt)
 	err := snapClient.CreateSnapshot(ctx)
 	if err != nil {
 		log.ErrorLog(ctx, "failed to create snapshot %s %v", snapshotID, err)
@@ -165,7 +165,7 @@ func (s *subVolumeClient) CleanupSnapshotFromSubvolume(
 	// snapshot name is same as clone name as we need a name which can be
 	// identified during PVC-PVC cloning.
 	snapShotID := s.VolID
-	snapClient := NewSnapshot(s.conn, snapShotID, parentVol)
+	snapClient := NewSnapshot(s.conn, snapShotID, s.clusterID, parentVol)
 	snapInfo, err := snapClient.GetSnapshotInfo(ctx)
 	if err != nil {
 		if errors.Is(err, cerrors.ErrSnapNotFound) {
@@ -198,7 +198,7 @@ func (s *subVolumeClient) CreateCloneFromSnapshot(
 	ctx context.Context, snap Snapshot,
 ) error {
 	snapID := snap.SnapshotID
-	snapClient := NewSnapshot(s.conn, snapID, snap.SubVolume)
+	snapClient := NewSnapshot(s.conn, snapID, s.clusterID, snap.SubVolume)
 	err := snapClient.CloneSnapshot(ctx, s.SubVolume)
 	if err != nil {
 		return err

internal/cephfs/core/snapshot.go

@@ -62,6 +62,7 @@ type SnapshotClient interface {
 // snapshotClient is the implementation of SnapshotClient interface.
 type snapshotClient struct {
 	*Snapshot                         // Embedded snapshot struct.
+	clusterID string                  // Cluster ID.
 	conn      *util.ClusterConnection // Cluster connection.
 }
 
@@ -72,12 +73,13 @@ type Snapshot struct {
 }
 
 // NewSnapshot creates a new snapshot client.
-func NewSnapshot(conn *util.ClusterConnection, snapshotID string, vol *SubVolume) SnapshotClient {
+func NewSnapshot(conn *util.ClusterConnection, snapshotID, clusterID string, vol *SubVolume) SnapshotClient {
 	return &snapshotClient{
 		Snapshot: &Snapshot{
 			SnapshotID: snapshotID,
 			SubVolume:  vol,
 		},
+		clusterID: clusterID,
 		conn:      conn,
 	}
 }

internal/cephfs/core/snapshot_metadata.go

@@ -17,30 +17,75 @@ limitations under the License.
 package core
 
 import (
+	"errors"
 	"fmt"
 	"strings"
+
+	fsAdmin "github.com/ceph/go-ceph/cephfs/admin"
 )
 
+// ErrSubVolSnapMetadataNotSupported is returned when set/get/list/remove
+// subvolume snapshot metadata options are not supported.
+var ErrSubVolSnapMetadataNotSupported = errors.New("subvolume snapshot metadata operations are not supported")
+
+func (s *snapshotClient) supportsSubVolSnapMetadata() bool {
+	if _, keyPresent := clusterAdditionalInfo[s.clusterID]; !keyPresent {
+		clusterAdditionalInfo[s.clusterID] = &localClusterState{}
+	}
+
+	return clusterAdditionalInfo[s.clusterID].subVolSnapshotMetadataState != unsupported
+}
+
+func (s *snapshotClient) isUnsupportedSubVolSnapMetadata(err error) bool {
+	var invalid fsAdmin.NotImplementedError
+	if err != nil && errors.Is(err, &invalid) {
+		// In case the error is other than invalid command return error to
+		// the caller.
+		clusterAdditionalInfo[s.clusterID].subVolSnapshotMetadataState = unsupported
+
+		return false
+	}
+	clusterAdditionalInfo[s.clusterID].subVolSnapshotMetadataState = supported
+
+	return true
+}
+
 // setSnapshotMetadata sets custom metadata on the subvolume snapshot in a
 // volume as a key-value pair.
 func (s *snapshotClient) setSnapshotMetadata(key, value string) error {
+	if !s.supportsSubVolSnapMetadata() {
+		return ErrSubVolSnapMetadataNotSupported
+	}
 	fsa, err := s.conn.GetFSAdmin()
 	if err != nil {
 		return err
 	}
 
-	return fsa.SetSnapshotMetadata(s.FsName, s.SubvolumeGroup, s.VolID, s.SnapshotID, key, value)
+	err = fsa.SetSnapshotMetadata(s.FsName, s.SubvolumeGroup, s.VolID, s.SnapshotID, key, value)
+	if !s.isUnsupportedSubVolSnapMetadata(err) {
+		return ErrSubVolSnapMetadataNotSupported
+	}
+
+	return err
 }
 
 // removeSnapshotMetadata removes custom metadata set on the subvolume
 // snapshot in a volume using the metadata key.
 func (s *snapshotClient) removeSnapshotMetadata(key string) error {
+	if !s.supportsSubVolSnapMetadata() {
+		return ErrSubVolSnapMetadataNotSupported
+	}
 	fsa, err := s.conn.GetFSAdmin()
 	if err != nil {
 		return err
 	}
 
-	return fsa.RemoveSnapshotMetadata(s.FsName, s.SubvolumeGroup, s.VolID, s.SnapshotID, key)
+	err = fsa.RemoveSnapshotMetadata(s.FsName, s.SubvolumeGroup, s.VolID, s.SnapshotID, key)
+	if !s.isUnsupportedSubVolSnapMetadata(err) {
+		return ErrSubVolSnapMetadataNotSupported
+	}
+
+	return err
 }
 
 // SetAllSnapshotMetadata set all the metadata from arg parameters on

internal/cephfs/core/volume.go

@@ -193,6 +193,7 @@ type localClusterState struct {
 	// unsupported as per the state of the cluster.
 	resizeState                 operationState
 	subVolMetadataState         operationState
+	subVolSnapshotMetadataState operationState
 	// set true once a subvolumegroup is created
 	// for corresponding cluster.
 	subVolumeGroupCreated bool

internal/cephfs/store/fsjournal.go

@@ -398,7 +398,7 @@ func CheckSnapExists(
 	snapUUID := snapData.ImageUUID
 	snapID := snapData.ImageAttributes.ImageName
 	sid.FsSnapshotName = snapData.ImageAttributes.ImageName
-	snapClient := core.NewSnapshot(volOptions.conn, snapID, &volOptions.SubVolume)
+	snapClient := core.NewSnapshot(volOptions.conn, snapID, volOptions.ClusterID, &volOptions.SubVolume)
 	snapInfo, err := snapClient.GetSnapshotInfo(ctx)
 	if err != nil {
 		if errors.Is(err, cerrors.ErrSnapNotFound) {

internal/cephfs/store/volumeoptions.go

@@ -735,7 +735,7 @@ func NewSnapshotOptionsFromID(
 	volOptions.Features = subvolInfo.Features
 	volOptions.Size = subvolInfo.BytesQuota
 	volOptions.RootPath = subvolInfo.Path
-	snap := core.NewSnapshot(volOptions.conn, sid.FsSnapshotName, &volOptions.SubVolume)
+	snap := core.NewSnapshot(volOptions.conn, sid.FsSnapshotName, volOptions.ClusterID, &volOptions.SubVolume)
 	info, err := snap.GetSnapshotInfo(ctx)
 	if err != nil {
 		return &volOptions, nil, &sid, err