From 88b964fe18d37af5b53cea00e8d785159fb61f59 Mon Sep 17 00:00:00 2001
From: Madhu Rajanna <madhupr007@gmail.com>
Date: Tue, 1 Oct 2024 17:34:09 +0200
Subject: [PATCH] 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      | 36 +++++++++++++++++++----
 internal/rbd/rbd_util_test.go | 54 +++++++++++++++++++++++++++++++++++
 2 files changed, 84 insertions(+), 6 deletions(-)

diff --git a/internal/rbd/rbd_util.go b/internal/rbd/rbd_util.go
index 1beb4f531..3b07d235d 100644
--- a/internal/rbd/rbd_util.go
+++ b/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,
diff --git a/internal/rbd/rbd_util_test.go b/internal/rbd/rbd_util_test.go
index 905e97707..5d14ed844 100644
--- a/internal/rbd/rbd_util_test.go
+++ b/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)
+			}
+		})
+	}
+}