Merge pull request #399 from OdedViner/skip_subvol_get_info_err

csi: skip not-ready cephfs subvolumes in ls and summarize errors

Author: subhamkrai

pkg/filesystem/subvolume.go

@@ -21,6 +21,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	osexec "os/exec"
 	"strconv"
 	"strings"
 	"syscall"
@@ -250,6 +251,9 @@ func listCephFSSubvolumes(ctx context.Context, clientsets *k8sutil.Clientsets, o
 	}
 	fmt.Println("Filesystem  Subvolume  SubvolumeGroup  State")
 
+	// collect subvolumes that are not ready (EAGAIN) to show a summary at the end
+	var notReadyErrors []string
+
 	// this iterates over the filesystems and subvolumegroup to get the list of subvolumes that exist
 	for _, fs := range fsstruct {
 		// gets the subvolumegroup in the filesystem
@@ -276,10 +280,9 @@ func listCephFSSubvolumes(ctx context.Context, clientsets *k8sutil.Clientsets, o
 				// subvolume info returns error in case of pending clone or if it is not ready
 				// it is suggested to delete the pvc before deleting the subvolume.
 				if err != nil {
-					if errors.Is(err, syscall.EAGAIN) {
-						logging.Warning("Found pending clone: %q", sv.Name)
-						logging.Warning("Please delete the pending pv if any before deleting the subvolume %s", sv.Name)
-						logging.Warning("To avoid stale resources, please scale down the cephfs deployment before deleting the subvolume.")
+					if isSubvolumeNotReady(err) {
+						// Skip listing this subvolume but remember to report it later
+						notReadyErrors = append(notReadyErrors, fmt.Sprintf("%s/%s/%s: %v", fs.Name, svg.Name, sv.Name, err))
 						continue
 					}
 					logging.Fatal(fmt.Errorf("failed to get subvolume state: %q %q", sv.Name, err))
@@ -317,6 +320,15 @@ func listCephFSSubvolumes(ctx context.Context, clientsets *k8sutil.Clientsets, o
 			}
 		}
 	}
+
+	// After listing, show a concise summary of skipped subvolumes due to not-ready state
+	if len(notReadyErrors) > 0 {
+		logging.Warning("%d subvolumes were skipped because they are not ready (pending clone or in progress):", len(notReadyErrors))
+		for _, errMsg := range notReadyErrors {
+			logging.Warning("  %s", errMsg)
+		}
+		logging.Warning("To avoid stale resources, you may scale down the cephfs deployment before deleting such subvolumes.")
+	}
 }
 
 // getSubvolumeState returns the state of the subvolume
@@ -326,7 +338,8 @@ func getSubvolumeState(ctx context.Context, clientsets *k8sutil.Clientsets, oper
 
 	subVolumeInfo, errvol := runCommand(ctx, clientsets, operatorNamespace, clusterNamespace, cmd, args)
 	if errvol != nil {
-		logging.Error(errvol, "failed to get subvolume info")
+		// Avoid fmt EXTRA artifacts by formatting the error ourselves
+		logging.Error(fmt.Errorf("failed to get subvolume info for %s/%s/%s: %v", fsName, SubvolumeGroup, SubVol, errvol))
 		return "", errvol
 	}
 	var info map[string]interface{}
@@ -341,6 +354,59 @@ func getSubvolumeState(ctx context.Context, clientsets *k8sutil.Clientsets, oper
 	return state, nil
 }
 
+// exitCodeFromError extracts exit code from wrapped errors if possible.
+func exitCodeFromError(err error) (int, bool) {
+	// errno-style errors
+	var errno syscall.Errno
+	if errors.As(err, &errno) {
+		return int(errno), true
+	}
+	// errors with ExitStatus()
+	type exitStatuser interface{ ExitStatus() int }
+	var es exitStatuser
+	if errors.As(err, &es) {
+		return es.ExitStatus(), true
+	}
+	// os/exec errors
+	var ee *osexec.ExitError
+	if errors.As(err, &ee) {
+		if status, ok := ee.Sys().(syscall.WaitStatus); ok {
+			return status.ExitStatus(), true
+		}
+	}
+	return 0, false
+}
+
+// isSubvolumeNotReady detects Ceph EAGAIN "not ready" conditions
+func isSubvolumeNotReady(err error) bool {
+	if err == nil {
+		return false
+	}
+
+	// Best check: proper error type match
+	if errors.Is(err, syscall.EAGAIN) {
+		return true
+	}
+
+	// Next best: exit codes 11 or -11 (common Ceph EAGAIN)
+	if code, ok := exitCodeFromError(err); ok {
+		if code == 11 || code == -11 {
+			return true
+		}
+	}
+
+	// Fallback: minimal string detection
+	msg := err.Error()
+	if strings.Contains(msg, "EAGAIN") {
+		return true
+	}
+	if strings.Contains(msg, "exit code 11") || strings.Contains(msg, "exit status 11") || strings.Contains(msg, "exit status -11") {
+		return true
+	}
+
+	return false
+}
+
 // gets list of filesystem
 func getFileSystem(ctx context.Context, clientsets *k8sutil.Clientsets, operatorNamespace, clusterNamespace string) ([]fsStruct, error) {