rbd: consider ErrPermissionDenied for vol

Incase of RDR with restricted access the
ceph user will not have access to all the objects
or all the pools where mapping exists

This commits add a check to continue to get
the volume if there is a permission error

Signed-off-by: Madhu Rajanna <madhupr007@gmail.com>

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)
+			}
+		})
+	}
+}