Address review feedback: move cache building to volumehelper

- Rename NewVolumeHelperImplWithCache to NewVolumeHelperImplWithNamespaces
- Move cache building logic from backup.go into volumehelper
- Return error from NewVolumeHelperImplWithNamespaces if cache build fails
- Remove fallback in main backup path - backup fails if cache build fails
- Update NewVolumeHelperImpl to call NewVolumeHelperImplWithNamespaces
- Add comments clarifying fallback is only used by plugins
- Update tests for new error return signature

This addresses review comments from @Lyndon-Li and @kaovilai:
- Cache building is now encapsulated in volumehelper
- No fallback in main backup path ensures predictable performance
- Code reuse between constructors

Fixes #9179

Signed-off-by: Shubham Pampattiwar <spampatt@redhat.com>

internal/volumehelper/volume_policy_helper.go

@@ -1,6 +1,7 @@
 package volumehelper
 
 import (
+	"context"
 	"fmt"
 	"strings"
 
@@ -46,7 +47,9 @@ func NewVolumeHelperImpl(
 	defaultVolumesToFSBackup bool,
 	backupExcludePVC bool,
 ) VolumeHelper {
-	return NewVolumeHelperImplWithCache(
+	// Pass nil namespaces - no cache will be built, so this never fails.
+	// This is used by plugins that don't need the cache optimization.
+	vh, _ := NewVolumeHelperImplWithNamespaces(
 		volumePolicy,
 		snapshotVolumes,
 		logger,
@@ -55,19 +58,32 @@ func NewVolumeHelperImpl(
 		backupExcludePVC,
 		nil,
 	)
+	return vh
 }
 
-// NewVolumeHelperImplWithCache creates a VolumeHelper with a PVC-to-Pod cache for improved performance.
-// The cache should be built before backup processing begins.
-func NewVolumeHelperImplWithCache(
+// NewVolumeHelperImplWithNamespaces creates a VolumeHelper with a PVC-to-Pod cache for improved performance.
+// The cache is built internally from the provided namespaces list.
+// This avoids O(N*M) complexity when there are many PVCs and pods.
+// See issue #9179 for details.
+// Returns an error if cache building fails - callers should not proceed with backup in this case.
+func NewVolumeHelperImplWithNamespaces(
 	volumePolicy *resourcepolicies.Policies,
 	snapshotVolumes *bool,
 	logger logrus.FieldLogger,
 	client crclient.Client,
 	defaultVolumesToFSBackup bool,
 	backupExcludePVC bool,
-	pvcPodCache *podvolumeutil.PVCPodCache,
-) VolumeHelper {
+	namespaces []string,
+) (VolumeHelper, error) {
+	var pvcPodCache *podvolumeutil.PVCPodCache
+	if len(namespaces) > 0 {
+		pvcPodCache = podvolumeutil.NewPVCPodCache()
+		if err := pvcPodCache.BuildCacheForNamespaces(context.Background(), namespaces, client); err != nil {
+			return nil, err
+		}
+		logger.Infof("Built PVC-to-Pod cache for %d namespaces", len(namespaces))
+	}
+
 	return &volumeHelperImpl{
 		volumePolicy:             volumePolicy,
 		snapshotVolumes:          snapshotVolumes,
@@ -76,7 +92,7 @@ func NewVolumeHelperImplWithCache(
 		defaultVolumesToFSBackup: defaultVolumesToFSBackup,
 		backupExcludePVC:         backupExcludePVC,
 		pvcPodCache:              pvcPodCache,
-	}
+	}, nil
 }
 
 func (v *volumeHelperImpl) ShouldPerformSnapshot(obj runtime.Unstructured, groupResource schema.GroupResource) (bool, error) {

internal/volumehelper/volume_policy_helper_test.go

@@ -34,7 +34,6 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/builder"
 	"github.com/vmware-tanzu/velero/pkg/kuberesource"
 	velerotest "github.com/vmware-tanzu/velero/pkg/test"
-	podvolumeutil "github.com/vmware-tanzu/velero/pkg/util/podvolume"
 )
 
 func TestVolumeHelperImpl_ShouldPerformSnapshot(t *testing.T) {
@@ -876,22 +875,21 @@ func TestVolumeHelperImplWithCache_ShouldPerformSnapshot(t *testing.T) {
 				require.NoError(t, err)
 			}
 
-			var cache *podvolumeutil.PVCPodCache
+			var namespaces []string
 			if tc.buildCache {
-				cache = podvolumeutil.NewPVCPodCache()
-				err := cache.BuildCacheForNamespaces(t.Context(), []string{"ns"}, fakeClient)
-				require.NoError(t, err)
+				namespaces = []string{"ns"}
 			}
 
-			vh := NewVolumeHelperImplWithCache(
+			vh, err := NewVolumeHelperImplWithNamespaces(
 				p,
 				tc.snapshotVolumesFlag,
 				logrus.StandardLogger(),
 				fakeClient,
 				tc.defaultVolumesToFSBackup,
 				false,
-				cache,
+				namespaces,
 			)
+			require.NoError(t, err)
 
 			obj, err := runtime.DefaultUnstructuredConverter.ToUnstructured(tc.inputObj)
 			require.NoError(t, err)
@@ -1029,22 +1027,21 @@ func TestVolumeHelperImplWithCache_ShouldPerformFSBackup(t *testing.T) {
 				require.NoError(t, err)
 			}
 
-			var cache *podvolumeutil.PVCPodCache
+			var namespaces []string
 			if tc.buildCache {
-				cache = podvolumeutil.NewPVCPodCache()
-				err := cache.BuildCacheForNamespaces(t.Context(), []string{"ns"}, fakeClient)
-				require.NoError(t, err)
+				namespaces = []string{"ns"}
 			}
 
-			vh := NewVolumeHelperImplWithCache(
+			vh, err := NewVolumeHelperImplWithNamespaces(
 				p,
 				tc.snapshotVolumesFlag,
 				logrus.StandardLogger(),
 				fakeClient,
 				tc.defaultVolumesToFSBackup,
 				false,
-				cache,
+				namespaces,
 			)
+			require.NoError(t, err)
 
 			actualShouldFSBackup, actualError := vh.ShouldPerformFSBackup(tc.pod.Spec.Volumes[0], *tc.pod)
 			if tc.expectedErr {

pkg/backup/backup.go

@@ -64,7 +64,6 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/util/boolptr"
 	"github.com/vmware-tanzu/velero/pkg/util/collections"
 	"github.com/vmware-tanzu/velero/pkg/util/kube"
-	podvolumeutil "github.com/vmware-tanzu/velero/pkg/util/podvolume"
 )
 
 // BackupVersion is the current backup major version for Velero.
@@ -409,22 +408,26 @@ func (kb *kubernetesBackupper) BackupWithResolvers(
 	}
 	backupRequest.Status.Progress = &velerov1api.BackupProgress{TotalItems: len(items)}
 
-	// Build PVC-to-Pod cache for improved volume policy lookup performance.
-	// This avoids O(N*M) complexity when there are many PVCs and pods.
+	// Resolve namespaces for PVC-to-Pod cache building in volumehelper.
 	// See issue #9179 for details.
-	pvcPodCache := podvolumeutil.NewPVCPodCache()
 	namespaces, err := backupRequest.NamespaceIncludesExcludes.ResolveNamespaceList()
 	if err != nil {
-		log.WithError(err).Warn("Failed to resolve namespace list for PVC-to-Pod cache, falling back to direct lookups")
-		pvcPodCache = nil
-	} else if len(namespaces) > 0 {
-		if err := pvcPodCache.BuildCacheForNamespaces(context.Background(), namespaces, kb.kbClient); err != nil {
-			// Log warning but continue - the cache will fall back to direct lookups
-			log.WithError(err).Warn("Failed to build PVC-to-Pod cache, falling back to direct lookups")
-			pvcPodCache = nil
-		} else {
-			log.Infof("Built PVC-to-Pod cache for %d namespaces", len(namespaces))
-		}
+		log.WithError(err).Error("Failed to resolve namespace list for PVC-to-Pod cache")
+		return err
+	}
+
+	volumeHelperImpl, err := volumehelper.NewVolumeHelperImplWithNamespaces(
+		backupRequest.ResPolicies,
+		backupRequest.Spec.SnapshotVolumes,
+		log,
+		kb.kbClient,
+		boolptr.IsSetToTrue(backupRequest.Spec.DefaultVolumesToFsBackup),
+		!backupRequest.ResourceIncludesExcludes.ShouldInclude(kuberesource.PersistentVolumeClaims.String()),
+		namespaces,
+	)
+	if err != nil {
+		log.WithError(err).Error("Failed to build PVC-to-Pod cache for volume policy lookups")
+		return err
 	}
 
 	itemBackupper := &itemBackupper{
@@ -440,16 +443,8 @@ func (kb *kubernetesBackupper) BackupWithResolvers(
 		itemHookHandler: &hook.DefaultItemHookHandler{
 			PodCommandExecutor: kb.podCommandExecutor,
 		},
-		hookTracker: hook.NewHookTracker(),
-		volumeHelperImpl: volumehelper.NewVolumeHelperImplWithCache(
-			backupRequest.ResPolicies,
-			backupRequest.Spec.SnapshotVolumes,
-			log,
-			kb.kbClient,
-			boolptr.IsSetToTrue(backupRequest.Spec.DefaultVolumesToFsBackup),
-			!backupRequest.ResourceIncludesExcludes.ShouldInclude(kuberesource.PersistentVolumeClaims.String()),
-			pvcPodCache,
-		),
+		hookTracker:         hook.NewHookTracker(),
+		volumeHelperImpl:    volumeHelperImpl,
 		kubernetesBackupper: kb,
 	}
 

pkg/util/podvolume/pod_volume.go

@@ -206,6 +206,8 @@ func IsPVCDefaultToFSBackup(pvcNamespace, pvcName string, crClient crclient.Clie
 
 // IsPVCDefaultToFSBackupWithCache is the cached version of IsPVCDefaultToFSBackup.
 // If cache is nil or not built, it falls back to the non-cached version.
+// Note: In the main backup path, the cache is always built (via NewVolumeHelperImplWithNamespaces),
+// so the fallback is only used by plugins that don't need cache optimization.
 func IsPVCDefaultToFSBackupWithCache(
 	pvcNamespace, pvcName string,
 	crClient crclient.Client,
@@ -287,6 +289,8 @@ func GetPodsUsingPVC(
 // GetPodsUsingPVCWithCache returns all pods that use the specified PVC.
 // If cache is available and built, it uses the cache for O(1) lookup.
 // Otherwise, it falls back to the original GetPodsUsingPVC function.
+// Note: In the main backup path, the cache is always built (via NewVolumeHelperImplWithNamespaces),
+// so the fallback is only used by plugins that don't need cache optimization.
 func GetPodsUsingPVCWithCache(
 	pvcNamespace, pvcName string,
 	crClient crclient.Client,