rbd: consider ErrPermissionDenied for vol

internal/rbd/rbd_util.go

@@ -1214,8 +1214,7 @@ func GenVolFromVolID(
 	}
 
 	vol, err = generateVolumeFromVolumeID(ctx, volumeID, vi, cr, secrets)
-	if !errors.Is(err, util.ErrKeyNotFound) && !errors.Is(err, util.ErrPoolNotFound) &&
-		!errors.Is(err, ErrImageNotFound) {
+	if !shouldRetryVolumeGeneration(err) {
 		return vol, err
 	}
 
@@ -1226,8 +1225,7 @@ func GenVolFromVolID(
 	}
 	if mapping != nil {
 		rbdVol, vErr := generateVolumeFromMapping(ctx, mapping, volumeID, vi, cr, secrets)
-		if !errors.Is(vErr, util.ErrKeyNotFound) && !errors.Is(vErr, util.ErrPoolNotFound) &&
-			!errors.Is(vErr, ErrImageNotFound) {
+		if !shouldRetryVolumeGeneration(vErr) {
 			return rbdVol, vErr
 		}
 	}
@@ -1280,8 +1278,7 @@ func generateVolumeFromMapping(
 					// Add mapping poolID to Identifier
 					nvi.LocationID = pID
 					vol, err = generateVolumeFromVolumeID(ctx, volumeID, nvi, cr, secrets)
-					if !errors.Is(err, util.ErrKeyNotFound) && !errors.Is(err, util.ErrPoolNotFound) &&
-						!errors.Is(err, ErrImageNotFound) {
+					if !shouldRetryVolumeGeneration(err) {
 						return vol, err
 					}
 				}
@@ -1292,6 +1289,33 @@ func generateVolumeFromMapping(
 	return vol, util.ErrPoolNotFound
 }
 
+// shouldRetryVolumeGeneration determines whether the process of finding or generating
+// volumes should continue based on the type of error encountered.
+//
+// It checks if the given error matches any of the following known errors:
+//   - util.ErrKeyNotFound: The key required to locate the volume is missing in Rados omap.
+//   - util.ErrPoolNotFound: The rbd pool where the volume/omap is expected doesn't exist.
+//   - ErrImageNotFound: The image doesn't exist in the rbd pool.
+//   - rados.ErrPermissionDenied: Permissions to access the pool is denied.
+//
+// If any of these errors are encountered, the function returns `true`, indicating
+// that the volume search should continue because of known error. Otherwise, it
+// returns `false`, meaning the search should stop.
+//
+// This helper function is used in scenarios where multiple attempts may be made
+// to retrieve or generate volume information, and we want to gracefully handle
+// specific failure cases while retrying for others.
+func shouldRetryVolumeGeneration(err error) bool {
+	if err == nil {
+		return false // No error, do not retry
+	}
+	// Continue searching for specific known errors
+	return (errors.Is(err, util.ErrKeyNotFound) ||
+		errors.Is(err, util.ErrPoolNotFound) ||
+		errors.Is(err, ErrImageNotFound) ||
+		errors.Is(err, rados.ErrPermissionDenied))
+}
+
 func genVolFromVolumeOptions(
 	ctx context.Context,
 	volOptions map[string]string,

internal/rbd/rbd_util_test.go

@@ -23,8 +23,11 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/ceph/go-ceph/rados"
 	librbd "github.com/ceph/go-ceph/rbd"
 	"github.com/stretchr/testify/require"
+
+	"github.com/ceph/ceph-csi/internal/util"
 )
 
 func TestHasSnapshotFeature(t *testing.T) {
@@ -387,3 +390,54 @@ func Test_checkValidImageFeatures(t *testing.T) {
 		})
 	}
 }
+
+func Test_shouldRetryVolumeGeneration(t *testing.T) {
+	t.Parallel()
+	type args struct {
+		err error
+	}
+	tests := []struct {
+		name string
+		args args
+		want bool
+	}{
+		{
+			name: "No error (stop searching)",
+			args: args{err: nil},
+			want: false, // No error, stop searching
+		},
+		{
+			name: "ErrKeyNotFound (continue searching)",
+			args: args{err: util.ErrKeyNotFound},
+			want: true, // Known error, continue searching
+		},
+		{
+			name: "ErrPoolNotFound (continue searching)",
+			args: args{err: util.ErrPoolNotFound},
+			want: true, // Known error, continue searching
+		},
+		{
+			name: "ErrImageNotFound (continue searching)",
+			args: args{err: ErrImageNotFound},
+			want: true, // Known error, continue searching
+		},
+		{
+			name: "ErrPermissionDenied (continue searching)",
+			args: args{err: rados.ErrPermissionDenied},
+			want: true, // Known error, continue searching
+		},
+		{
+			name: "Different error (stop searching)",
+			args: args{err: errors.New("unknown error")},
+			want: false, // Unknown error, stop searching
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			if got := shouldRetryVolumeGeneration(tt.args.err); got != tt.want {
+				t.Errorf("shouldRetryVolumeGeneration() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}