cephfs: report detailed error message on clone failure

go-ceph provides a new GetFailure() method to retrieve details errors
when cloning failed. This is now included in the `cephFSCloneState`
struct, which was a simple string before.

While modifying the `cephFSCloneState` struct, the constants have been
removed, as go-ceph provides them as well.

Fixes: #3140
Signed-off-by: Niels de Vos <ndevos@redhat.com>

internal/cephfs/core/clone.go

@@ -19,43 +19,42 @@ package core
 import (
 	"context"
 	"errors"
+	"fmt"
 
 	cerrors "github.com/ceph/ceph-csi/internal/cephfs/errors"
 	"github.com/ceph/ceph-csi/internal/util/log"
+
+	"github.com/ceph/go-ceph/cephfs/admin"
 )
 
 // cephFSCloneState describes the status of the clone.
-type cephFSCloneState string
+type cephFSCloneState struct {
+	state    admin.CloneState
+	errno    string
+	errorMsg string
+}
 
 const (
-	// CephFSCloneError indicates that fetching the clone state returned an error.
-	CephFSCloneError = cephFSCloneState("")
-	// CephFSCloneFailed indicates that clone is in failed state.
-	CephFSCloneFailed = cephFSCloneState("failed")
-	// CephFSClonePending indicates that clone is in pending state.
-	CephFSClonePending = cephFSCloneState("pending")
-	// CephFSCloneInprogress indicates that clone is in in-progress state.
-	CephFSCloneInprogress = cephFSCloneState("in-progress")
-	// CephFSCloneComplete indicates that clone is in complete state.
-	CephFSCloneComplete = cephFSCloneState("complete")
-
 	// SnapshotIsProtected string indicates that the snapshot is currently protected.
 	SnapshotIsProtected = "yes"
 )
 
-// toError checks the state of the clone if it's not cephFSCloneComplete.
-func (cs cephFSCloneState) toError() error {
-	switch cs {
-	case CephFSCloneComplete:
+// CephFSCloneError indicates that fetching the clone state returned an error.
+var CephFSCloneError = cephFSCloneState{}
+
+// ToError checks the state of the clone if it's not cephFSCloneComplete.
+func (cs cephFSCloneState) ToError() error {
+	switch cs.state {
+	case admin.CloneComplete:
 		return nil
-	case CephFSCloneError:
-		return cerrors.ErrInvalidClone
-	case CephFSCloneInprogress:
+	case CephFSCloneError.state:
+		return fmt.Errorf("%w: %s (%s)", cerrors.ErrInvalidClone, cs.errorMsg, cs.errno)
+	case admin.CloneInProgress:
 		return cerrors.ErrCloneInProgress
-	case CephFSClonePending:
+	case admin.ClonePending:
 		return cerrors.ErrClonePending
-	case CephFSCloneFailed:
-		return cerrors.ErrCloneFailed
+	case admin.CloneFailed:
+		return fmt.Errorf("%w: %s (%s)", cerrors.ErrCloneFailed, cs.errorMsg, cs.errno)
 	}
 
 	return nil
@@ -125,10 +124,11 @@ func (s *subVolumeClient) CreateCloneFromSubvolume(
 		return cloneErr
 	}
 
-	if cloneState != CephFSCloneComplete {
-		log.ErrorLog(ctx, "clone %s did not complete: %v", s.VolID, cloneState.toError())
+	err = cloneState.ToError()
+	if err != nil {
+		log.ErrorLog(ctx, "clone %s did not complete: %v", s.VolID, err)
 
-		return cloneState.toError()
+		return err
 	}
 
 	err = s.ExpandVolume(ctx, s.Size)
@@ -220,8 +220,9 @@ func (s *subVolumeClient) CreateCloneFromSnapshot(
 		return err
 	}
 
-	if cloneState != CephFSCloneComplete {
-		return cloneState.toError()
+	err = cloneState.ToError()
+	if err != nil {
+		return err
 	}
 
 	err = s.ExpandVolume(ctx, s.Size)
@@ -255,5 +256,18 @@ func (s *subVolumeClient) GetCloneState(ctx context.Context) (cephFSCloneState,
 		return CephFSCloneError, err
 	}
 
-	return cephFSCloneState(cs.State), nil
+	errno := ""
+	errStr := ""
+	if failure := cs.GetFailure(); failure != nil {
+		errno = failure.Errno
+		errStr = failure.ErrStr
+	}
+
+	state := cephFSCloneState{
+		state:    cs.State,
+		errno:    errno,
+		errorMsg: errStr,
+	}
+
+	return state, nil
 }

internal/cephfs/core/clone_test.go

@@ -17,23 +17,25 @@ limitations under the License.
 package core
 
 import (
+	"errors"
 	"testing"
 
 	cerrors "github.com/ceph/ceph-csi/internal/cephfs/errors"
 
+	fsa "github.com/ceph/go-ceph/cephfs/admin"
 	"github.com/stretchr/testify/assert"
 )
 
 func TestCloneStateToError(t *testing.T) {
 	t.Parallel()
 	errorState := make(map[cephFSCloneState]error)
-	errorState[CephFSCloneComplete] = nil
+	errorState[cephFSCloneState{fsa.CloneComplete, "", ""}] = nil
 	errorState[CephFSCloneError] = cerrors.ErrInvalidClone
-	errorState[CephFSCloneInprogress] = cerrors.ErrCloneInProgress
-	errorState[CephFSClonePending] = cerrors.ErrClonePending
-	errorState[CephFSCloneFailed] = cerrors.ErrCloneFailed
+	errorState[cephFSCloneState{fsa.CloneInProgress, "", ""}] = cerrors.ErrCloneInProgress
+	errorState[cephFSCloneState{fsa.ClonePending, "", ""}] = cerrors.ErrClonePending
+	errorState[cephFSCloneState{fsa.CloneFailed, "", ""}] = cerrors.ErrCloneFailed
 
 	for state, err := range errorState {
-		assert.Equal(t, state.toError(), err)
+		assert.True(t, errors.Is(state.ToError(), err))
 	}
 }

internal/cephfs/store/fsjournal.go

@@ -119,15 +119,17 @@ func CheckVolExists(ctx context.Context,
 
 			return nil, err
 		}
-		if cloneState == core.CephFSCloneInprogress {
-			return nil, cerrors.ErrCloneInProgress
+		err = cloneState.ToError()
+		if errors.Is(err, cerrors.ErrCloneInProgress) {
+			return nil, err
 		}
-		if cloneState == core.CephFSClonePending {
-			return nil, cerrors.ErrClonePending
+		if errors.Is(err, cerrors.ErrClonePending) {
+			return nil, err
 		}
-		if cloneState == core.CephFSCloneFailed {
+		if errors.Is(err, cerrors.ErrCloneFailed) {
 			log.ErrorLog(ctx,
-				"clone failed, deleting subvolume clone. vol=%s, subvol=%s subvolgroup=%s",
+				"clone failed (%v), deleting subvolume clone. vol=%s, subvol=%s subvolgroup=%s",
+				err,
 				volOptions.FsName,
 				vid.FsSubvolName,
 				volOptions.SubvolumeGroup)
@@ -149,8 +151,8 @@ func CheckVolExists(ctx context.Context,
 
 			return nil, err
 		}
-		if cloneState != core.CephFSCloneComplete {
-			return nil, fmt.Errorf("clone is not in complete state for %s", vid.FsSubvolName)
+		if err != nil {
+			return nil, fmt.Errorf("clone is not in complete state for %s: %w", vid.FsSubvolName, err)
 		}
 	}