Add MustIncludeAdditionalItemPVCs to help track BIA added PVC's PVB creation. (#10501)

* Add MustIncludeAdditionalItemPVCs structure in backup. It's used to track PVCs returned by BIA with mustIncluded annotaion and PVC is excluded from backup by global filter.
* Modfiy the volumeHelper interface to add a parameter function for ShouldPerformFSBackup.
* Modify to support fine-grained backup filters.
* Modify according to comments. Use a read-only interface to replace the parameter function.

Signed-off-by: Xun Jiang <xun.jiang@broadcom.com>

changelogs/unreleased/10501-blackpiglet

@@ -0,0 +1 @@
+Add MustIncludeAdditionalItemPVCs to help track BIA added PVC's PVB creation.

internal/volumehelper/volume_policy_helper.go

@@ -38,6 +38,10 @@ type volumeHelperImpl struct {
 	// pvcPodCache provides cached PVC to Pod mappings for improved performance.
 	// When there are many PVCs and pods, using this cache avoids O(N*M) lookups.
 	pvcPodCache *podvolumeutil.PVCPodCache
+	// pvcMustInclusionTracker provides read-only checks for whether a PVC is included
+	// in the backup as BIA's additionalItems through annotation
+	// backup.velero.io/must-include-additional-items.
+	pvcMustInclusionTracker vhutil.PVCMustInclusionTracker
 }
 
 // NewVolumeHelperImpl creates a VolumeHelper without PVC-to-Pod caching.
@@ -52,6 +56,7 @@ func NewVolumeHelperImpl(
 	client crclient.Client,
 	defaultVolumesToFSBackup bool,
 	backupExcludePVC bool,
+	pvcMustInclusionTracker vhutil.PVCMustInclusionTracker,
 ) vhutil.VolumeHelper {
 	// Pass nil namespaces - no cache will be built, so this never fails.
 	// This is used by plugins that don't need the cache optimization.
@@ -63,6 +68,7 @@ func NewVolumeHelperImpl(
 		defaultVolumesToFSBackup,
 		backupExcludePVC,
 		nil,
+		pvcMustInclusionTracker,
 	)
 	return vh
 }
@@ -80,6 +86,7 @@ func NewVolumeHelperImplWithNamespaces(
 	defaultVolumesToFSBackup bool,
 	backupExcludePVC bool,
 	namespaces []string,
+	pvcMustInclusionTracker vhutil.PVCMustInclusionTracker,
 ) (vhutil.VolumeHelper, error) {
 	var pvcPodCache *podvolumeutil.PVCPodCache
 	if len(namespaces) > 0 {
@@ -98,6 +105,7 @@ func NewVolumeHelperImplWithNamespaces(
 		defaultVolumesToFSBackup: defaultVolumesToFSBackup,
 		backupExcludePVC:         backupExcludePVC,
 		pvcPodCache:              pvcPodCache,
+		pvcMustInclusionTracker:  pvcMustInclusionTracker,
 	}, nil
 }
 
@@ -109,6 +117,7 @@ func NewVolumeHelperImplWithCache(
 	client crclient.Client,
 	logger logrus.FieldLogger,
 	pvcPodCache *podvolumeutil.PVCPodCache,
+	pvcMustInclusionTracker vhutil.PVCMustInclusionTracker,
 ) (vhutil.VolumeHelper, error) {
 	resourcePolicies, err := resourcepolicies.GetResourcePoliciesFromBackup(backup, client, logger)
 	if err != nil {
@@ -123,6 +132,7 @@ func NewVolumeHelperImplWithCache(
 		defaultVolumesToFSBackup: boolptr.IsSetToTrue(backup.Spec.DefaultVolumesToFsBackup),
 		backupExcludePVC:         boolptr.IsSetToTrue(backup.Spec.SnapshotMoveData),
 		pvcPodCache:              pvcPodCache,
+		pvcMustInclusionTracker:  pvcMustInclusionTracker,
 	}, nil
 }
 
@@ -260,7 +270,7 @@ func (v *volumeHelperImpl) ShouldPerformSnapshot(obj runtime.Unstructured, group
 }
 
 func (v volumeHelperImpl) ShouldPerformFSBackup(volume corev1api.Volume, pod corev1api.Pod) (bool, error) {
-	if !v.shouldIncludeVolumeInBackup(volume) {
+	if !v.shouldIncludeVolumeInBackup(volume, pod) {
 		v.logger.Debugf("skip fs-backup action for pod %s's volume %s, due to not pass volume check.", pod.Namespace+"/"+pod.Name, volume.Name)
 		return false, nil
 	}
@@ -442,7 +452,7 @@ func (v *volumeHelperImpl) GetSnapshotClass(obj runtime.Unstructured, groupResou
 	return action.GetSnapshotClass()
 }
 
-func (v *volumeHelperImpl) shouldIncludeVolumeInBackup(vol corev1api.Volume) bool {
+func (v *volumeHelperImpl) shouldIncludeVolumeInBackup(vol corev1api.Volume, pod corev1api.Pod) bool {
 	includeVolumeInBackup := true
 	// cannot backup hostpath volumes as they are not mounted into /var/lib/kubelet/pods
 	// and therefore not accessible to the node agent daemon set.
@@ -465,8 +475,12 @@ func (v *volumeHelperImpl) shouldIncludeVolumeInBackup(vol corev1api.Volume) boo
 	if vol.DownwardAPI != nil {
 		includeVolumeInBackup = false
 	}
-	if vol.PersistentVolumeClaim != nil && v.backupExcludePVC {
-		includeVolumeInBackup = false
+	if vol.PersistentVolumeClaim != nil {
+		if v.backupExcludePVC {
+			if v.pvcMustInclusionTracker == nil || !v.pvcMustInclusionTracker.IsPVCIncluded(pod.Namespace, vol.PersistentVolumeClaim.ClaimName) {
+				includeVolumeInBackup = false
+			}
+		}
 	}
 	// don't include volumes that mount the default service account token.
 	if strings.HasPrefix(vol.Name, "default-token") {

internal/volumehelper/volume_policy_helper_test.go

@@ -35,6 +35,7 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/kuberesource"
 	velerotest "github.com/vmware-tanzu/velero/pkg/test"
 	podvolumeutil "github.com/vmware-tanzu/velero/pkg/util/podvolume"
+	vhutil "github.com/vmware-tanzu/velero/pkg/util/volumehelper"
 )
 
 func TestVolumeHelperImpl_ShouldPerformSnapshot(t *testing.T) {
@@ -329,6 +330,7 @@ func TestVolumeHelperImpl_ShouldPerformSnapshot(t *testing.T) {
 				fakeClient,
 				tc.defaultVolumesToFSBackup,
 				false,
+				nil,
 			)
 
 			obj, err := runtime.DefaultUnstructuredConverter.ToUnstructured(tc.inputObj)
@@ -345,11 +347,23 @@ func TestVolumeHelperImpl_ShouldPerformSnapshot(t *testing.T) {
 	}
 }
 
+type mockPVCMustInclusionTracker struct {
+	isPVCIncluded func(namespace, pvcName string) bool
+}
+
+func (m *mockPVCMustInclusionTracker) IsPVCIncluded(namespace, pvcName string) bool {
+	if m.isPVCIncluded == nil {
+		return false
+	}
+	return m.isPVCIncluded(namespace, pvcName)
+}
+
 func TestVolumeHelperImpl_ShouldIncludeVolumeInBackup(t *testing.T) {
 	testCases := []struct {
 		name             string
 		vol              corev1api.Volume
 		backupExcludePVC bool
+		isPVCIncluded    func(pvcName string) bool
 		shouldInclude    bool
 	}{
 		{
@@ -445,6 +459,38 @@ func TestVolumeHelperImpl_ShouldIncludeVolumeInBackup(t *testing.T) {
 			backupExcludePVC: true,
 			shouldInclude:    false,
 		},
+		{
+			name: "volume has pvc, backupExcludePVC is true, but isPVCIncluded returns true so include",
+			vol: corev1api.Volume{
+				Name: "sample-volume",
+				VolumeSource: corev1api.VolumeSource{
+					PersistentVolumeClaim: &corev1api.PersistentVolumeClaimVolumeSource{
+						ClaimName: "sample-pvc",
+					},
+				},
+			},
+			backupExcludePVC: true,
+			isPVCIncluded: func(pvcName string) bool {
+				return pvcName == "sample-pvc"
+			},
+			shouldInclude: true,
+		},
+		{
+			name: "volume has pvc, backupExcludePVC is false, isPVCIncluded returns false, but globally included so include",
+			vol: corev1api.Volume{
+				Name: "sample-volume",
+				VolumeSource: corev1api.VolumeSource{
+					PersistentVolumeClaim: &corev1api.PersistentVolumeClaimVolumeSource{
+						ClaimName: "sample-pvc",
+					},
+				},
+			},
+			backupExcludePVC: false,
+			isPVCIncluded: func(pvcName string) bool {
+				return false
+			},
+			shouldInclude: true,
+		},
 		{
 			name: "volume name has prefix default-token so do not include",
 			vol: corev1api.Volume{
@@ -480,13 +526,23 @@ func TestVolumeHelperImpl_ShouldIncludeVolumeInBackup(t *testing.T) {
 			if err != nil {
 				t.Fatalf("failed to build policy with error %v", err)
 			}
-			vh := &volumeHelperImpl{
-				volumePolicy:     p,
-				snapshotVolumes:  ptr.To(true),
-				logger:           velerotest.NewLogger(),
-				backupExcludePVC: tc.backupExcludePVC,
+			var tracker vhutil.PVCMustInclusionTracker
+			if tc.isPVCIncluded != nil {
+				tracker = &mockPVCMustInclusionTracker{
+					isPVCIncluded: func(ns, pvcName string) bool {
+						return tc.isPVCIncluded(pvcName)
+					},
+				}
 			}
-			actualShouldInclude := vh.shouldIncludeVolumeInBackup(tc.vol)
+			vh := &volumeHelperImpl{
+				volumePolicy:            p,
+				snapshotVolumes:         ptr.To(true),
+				logger:                  velerotest.NewLogger(),
+				backupExcludePVC:        tc.backupExcludePVC,
+				pvcMustInclusionTracker: tracker,
+			}
+			pod := corev1api.Pod{ObjectMeta: metav1.ObjectMeta{Namespace: "default"}}
+			actualShouldInclude := vh.shouldIncludeVolumeInBackup(tc.vol, pod)
 			assert.Equalf(t, actualShouldInclude, tc.shouldInclude, "Want shouldInclude as %v; Got actualShouldInclude as %v", tc.shouldInclude, actualShouldInclude)
 		})
 	}
@@ -694,6 +750,7 @@ func TestVolumeHelperImpl_ShouldPerformFSBackup(t *testing.T) {
 				fakeClient,
 				tc.defaultVolumesToFSBackup,
 				false,
+				nil,
 			)
 
 			actualShouldFSBackup, actualError := vh.ShouldPerformFSBackup(tc.pod.Spec.Volumes[0], *tc.pod)
@@ -889,6 +946,7 @@ func TestVolumeHelperImplWithCache_ShouldPerformSnapshot(t *testing.T) {
 				tc.defaultVolumesToFSBackup,
 				false,
 				namespaces,
+				nil,
 			)
 			require.NoError(t, err)
 
@@ -1041,6 +1099,7 @@ func TestVolumeHelperImplWithCache_ShouldPerformFSBackup(t *testing.T) {
 				tc.defaultVolumesToFSBackup,
 				false,
 				namespaces,
+				nil,
 			)
 			require.NoError(t, err)
 
@@ -1166,6 +1225,7 @@ volumePolicies:
 				fakeClient,
 				logrus.StandardLogger(),
 				cache,
+				nil,
 			)
 
 			if tc.expectError {
@@ -1221,7 +1281,7 @@ func TestNewVolumeHelperImplWithCache_UsesCache(t *testing.T) {
 		},
 	}
 
-	vh, err := NewVolumeHelperImplWithCache(backup, fakeClient, logrus.StandardLogger(), cache)
+	vh, err := NewVolumeHelperImplWithCache(backup, fakeClient, logrus.StandardLogger(), cache, nil)
 	require.NoError(t, err)
 
 	// Convert PV to unstructured
@@ -1353,6 +1413,7 @@ func TestVolumeHelperImpl_ShouldPerformSnapshot_UnboundPVC(t *testing.T) {
 				fakeClient,
 				false,
 				false,
+				nil,
 			)
 
 			obj, err := runtime.DefaultUnstructuredConverter.ToUnstructured(tc.inputPVC)
@@ -1530,6 +1591,7 @@ func TestVolumeHelperImpl_ShouldPerformFSBackup_UnboundPVC(t *testing.T) {
 				fakeClient,
 				false,
 				false,
+				nil,
 			)
 
 			actualShouldFSBackup, actualError := vh.ShouldPerformFSBackup(tc.pod.Spec.Volumes[0], *tc.pod)
@@ -1669,6 +1731,7 @@ func TestGetDataMoverFromActionParameters(t *testing.T) {
 				fakeClient,
 				false,
 				false,
+				nil,
 			)
 
 			obj, err := runtime.DefaultUnstructuredConverter.ToUnstructured(tc.inputObj)
@@ -1794,6 +1857,7 @@ func TestGetActionParameters(t *testing.T) {
 				fakeClient,
 				false,
 				false,
+				nil,
 			)
 
 			obj, err := runtime.DefaultUnstructuredConverter.ToUnstructured(tc.inputObj)
@@ -1980,6 +2044,7 @@ func TestShouldPerformCustomAction(t *testing.T) {
 				fakeClient,
 				false,
 				false,
+				nil,
 			)
 
 			obj, err := runtime.DefaultUnstructuredConverter.ToUnstructured(tc.inputObj)
@@ -2102,6 +2167,7 @@ func TestGetPVAndMatchAction(t *testing.T) {
 				fakeClient,
 				false,
 				false,
+				nil,
 			)
 
 			obj, err := runtime.DefaultUnstructuredConverter.ToUnstructured(tc.inputObj)

pkg/backup/actions/csi/pvc_action.go

@@ -136,6 +136,7 @@ func (p *pvcBackupItemAction) getVolumeHelperWithCache(backup *velerov1api.Backu
 		p.crClient,
 		p.log,
 		p.pvcPodCache,
+		nil,
 	)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to create VolumeHelper")

pkg/backup/backup.go

@@ -485,6 +485,8 @@ func (kb *kubernetesBackupper) BackupWithResolvers(
 		return err
 	}
 
+	pvcMustInclusionTracker := NewPVCMustInclusionTracker(backupRequest.MustIncludeAdditionalItemPVCs)
+
 	volumeHelperImpl, err := volumehelper.NewVolumeHelperImplWithNamespaces(
 		backupRequest.ResPolicies,
 		backupRequest.Spec.SnapshotVolumes,
@@ -493,6 +495,7 @@ func (kb *kubernetesBackupper) BackupWithResolvers(
 		boolptr.IsSetToTrue(backupRequest.Spec.DefaultVolumesToFsBackup),
 		!backupRequest.ResourceIncludesExcludes.ShouldInclude(kuberesource.PersistentVolumeClaims.String()),
 		namespaces,
+		pvcMustInclusionTracker,
 	)
 	if err != nil {
 		log.WithError(err).Error("Failed to build PVC-to-Pod cache for volume policy lookups")

pkg/backup/item_backupper.go

@@ -245,6 +245,7 @@ func (ib *itemBackupper) backupItemInternal(logger logrus.FieldLogger, obj runti
 			// where it's been backed up from another pod), since we don't need >1 backup per PVC.
 			for _, volume := range pod.Spec.Volumes {
 				shouldDoFSBackup, err := ib.volumeHelperImpl.ShouldPerformFSBackup(volume, *pod)
+
 				if err != nil {
 					backupErrs = append(backupErrs, errors.WithStack(err))
 				}
@@ -480,6 +481,26 @@ func (ib *itemBackupper) executeActions(
 		delete(u.GetAnnotations(), velerov1api.MustIncludeAdditionalItemAnnotation)
 		obj = u
 
+		// If the BIA specifies that additional items must be included, we track any PVCs returned as additional items.
+		// This tracking is necessary because the FSB (File System Backup) evaluation for a Pod
+		// happens before its PVCs are processed. By tracking these explicitly included PVCs here,
+		// the FSB logic can correctly determine that the PVC will be backed up and therefore
+		// a PodVolumeBackup should be created.
+		// We track this unconditionally when mustInclude is true, because fine-grained backup filters
+		// might exclude a PVC even if it's globally included, but mustInclude overrides those filters.
+		if mustInclude && ib.backupRequest.MustIncludeAdditionalItemPVCs != nil {
+			for _, additionalItem := range additionalItemIdentifiers {
+				if additionalItem.GroupResource == kuberesource.PersistentVolumeClaims {
+					key := itemKey{
+						resource:  additionalItem.GroupResource.String(),
+						namespace: additionalItem.Namespace,
+						name:      additionalItem.Name,
+					}
+					ib.backupRequest.MustIncludeAdditionalItemPVCs.AddItem(key)
+				}
+			}
+		}
+
 		// If async plugin started async operation, add it to the ItemOperations list
 		// ignore during finalize phase
 		if operationID != "" {

pkg/backup/pvc_must_inclusion_tracker.go

@@ -0,0 +1,50 @@
+/*
+Copyright the Velero contributors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package backup
+
+import (
+	"github.com/vmware-tanzu/velero/pkg/kuberesource"
+	vhutil "github.com/vmware-tanzu/velero/pkg/util/volumehelper"
+)
+
+// pvcMustInclusionTracker provides read-only checks for whether a PVC is included
+// in the backup as BIA's additionalItems through annotation
+// backup.velero.io/must-include-additional-items.
+type pvcMustInclusionTracker struct {
+	mustInclude *backedUpItemsMap
+}
+
+func NewPVCMustInclusionTracker(mustInclude *backedUpItemsMap) vhutil.PVCMustInclusionTracker {
+	return &pvcMustInclusionTracker{
+		mustInclude: mustInclude,
+	}
+}
+
+func (p *pvcMustInclusionTracker) IsPVCIncluded(namespace, pvcName string) bool {
+	pvcKey := itemKey{
+		resource:  kuberesource.PersistentVolumeClaims.String(),
+		namespace: namespace,
+		name:      pvcName,
+	}
+
+	// 1. If the PVC was explicitly forced into the backup by a BIA, it will be backed up.
+	if p.mustInclude != nil && p.mustInclude.Has(pvcKey) {
+		return true
+	}
+
+	return false
+}

pkg/backup/pvc_must_inclusion_tracker_test.go

@@ -0,0 +1,47 @@
+/*
+Copyright the Velero contributors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package backup
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/vmware-tanzu/velero/pkg/kuberesource"
+)
+
+func TestPVCMustInclusionTracker_IsPVCIncluded(t *testing.T) {
+	mustIncludeMap := NewBackedUpItemsMap()
+
+	tracker := NewPVCMustInclusionTracker(mustIncludeMap)
+
+	pvcKey1 := itemKey{
+		resource:  kuberesource.PersistentVolumeClaims.String(),
+		namespace: "ns-1",
+		name:      "pvc-1",
+	}
+
+	// Initially neither PVC is included
+	assert.False(t, tracker.IsPVCIncluded("ns-1", "pvc-1"))
+
+	// Add pvc-1 to mustInclude map
+	mustIncludeMap.AddItem(pvcKey1)
+	assert.True(t, tracker.IsPVCIncluded("ns-1", "pvc-1"))
+
+	// Check a PVC not in any map
+	assert.False(t, tracker.IsPVCIncluded("ns-1", "pvc-2"))
+}

pkg/backup/request.go

@@ -83,11 +83,16 @@ type Request struct {
 	VolumeSnapshots           SynchronizedVSList
 	PodVolumeBackups          []*velerov1api.PodVolumeBackup
 	BackedUpItems             *backedUpItemsMap
-	itemOperationsList        *[]*itemoperation.BackupOperation
-	ResPolicies               *resourcepolicies.Policies
-	SkippedPVTracker          *skipPVTracker
-	VolumesInformation        volume.BackupVolumesInformation
-	WorkerPool                *ItemBlockWorkerPool
+	// MustIncludeAdditionalItemPVCs keeps track of PVCs that are returned as additionalItems
+	// by a BackupItemAction plugin with the must-include annotation. This is specifically
+	// used to ensure PodVolumeBackups (FSB) are created for these PVCs even when PVCs are
+	// excluded by global or fine-grained backup resource filters.
+	MustIncludeAdditionalItemPVCs *backedUpItemsMap
+	itemOperationsList            *[]*itemoperation.BackupOperation
+	ResPolicies                   *resourcepolicies.Policies
+	SkippedPVTracker              *skipPVTracker
+	VolumesInformation            volume.BackupVolumesInformation
+	WorkerPool                    *ItemBlockWorkerPool
 
 	// ClusterScopedFilterMap holds resolved global filters for cluster-scoped resources.
 	// Key is the resolved group-resource string.

pkg/controller/backup_controller.go

@@ -395,10 +395,11 @@ func (b *backupReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctr
 
 func (b *backupReconciler) prepareBackupRequest(ctx context.Context, backup *velerov1api.Backup, logger logrus.FieldLogger) *pkgbackup.Request {
 	request := &pkgbackup.Request{
-		Backup:           backup.DeepCopy(), // don't modify items in the cache
-		SkippedPVTracker: pkgbackup.NewSkipPVTracker(),
-		BackedUpItems:    pkgbackup.NewBackedUpItemsMap(),
-		WorkerPool:       pkgbackup.StartItemBlockWorkerPool(ctx, b.itemBlockWorkerCount, logger),
+		Backup:                        backup.DeepCopy(), // don't modify items in the cache
+		SkippedPVTracker:              pkgbackup.NewSkipPVTracker(),
+		BackedUpItems:                 pkgbackup.NewBackedUpItemsMap(),
+		MustIncludeAdditionalItemPVCs: pkgbackup.NewBackedUpItemsMap(),
+		WorkerPool:                    pkgbackup.StartItemBlockWorkerPool(ctx, b.itemBlockWorkerCount, logger),
 	}
 	request.VolumesInformation.Init()
 

pkg/controller/backup_finalizer_controller.go

@@ -158,10 +158,11 @@ func (r *backupFinalizerReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 	}
 
 	backupRequest := &pkgbackup.Request{
-		Backup:           backup,
-		StorageLocation:  location,
-		SkippedPVTracker: pkgbackup.NewSkipPVTracker(),
-		BackedUpItems:    pkgbackup.NewBackedUpItemsMap(),
+		Backup:                        backup,
+		StorageLocation:               location,
+		SkippedPVTracker:              pkgbackup.NewSkipPVTracker(),
+		BackedUpItems:                 pkgbackup.NewBackedUpItemsMap(),
+		MustIncludeAdditionalItemPVCs: pkgbackup.NewBackedUpItemsMap(),
 	}
 	var outBackupFile *os.File
 	if len(operations) > 0 {

pkg/plugin/utils/volumehelper/volume_policy_helper.go

@@ -93,6 +93,7 @@ func ShouldPerformSnapshotWithVolumeHelper(
 		crClient,
 		boolptr.IsSetToTrue(backup.Spec.DefaultVolumesToFsBackup),
 		true,
+		nil,
 	)
 
 	return volumeHelperImpl.ShouldPerformSnapshot(unstructured, groupResource)
@@ -111,6 +112,7 @@ func NewVolumeHelperWithNamespaces(
 	defaultVolumesToFSBackup bool,
 	backupExcludePVC bool,
 	namespaces []string,
+	pvcMustInclusionTracker vhutil.PVCMustInclusionTracker,
 ) (vhutil.VolumeHelper, error) {
 	return volumehelper.NewVolumeHelperImplWithNamespaces(
 		volumePolicy,
@@ -120,6 +122,7 @@ func NewVolumeHelperWithNamespaces(
 		defaultVolumesToFSBackup,
 		backupExcludePVC,
 		namespaces,
+		pvcMustInclusionTracker,
 	)
 }
 
@@ -131,11 +134,13 @@ func NewVolumeHelperWithCache(
 	client crclient.Client,
 	logger logrus.FieldLogger,
 	pvcPodCache *podvolumeutil.PVCPodCache,
+	pvcMustInclusionTracker vhutil.PVCMustInclusionTracker,
 ) (vhutil.VolumeHelper, error) {
 	return volumehelper.NewVolumeHelperImplWithCache(
 		backup,
 		client,
 		logger,
 		pvcPodCache,
+		pvcMustInclusionTracker,
 	)
 }

pkg/plugin/utils/volumehelper/volume_policy_helper_test.go

@@ -300,6 +300,7 @@ func TestShouldPerformSnapshotWithNonNilVolumeHelper(t *testing.T) {
 		false, // defaultVolumesToFSBackup
 		true,  // backupExcludePVC
 		[]string{"default"},
+		nil,
 	)
 	require.NoError(t, err)
 	require.NotNil(t, vh)

pkg/util/volumehelper/volume_policy_helper.go

@@ -22,6 +22,13 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 )
 
+// PVCMustInclusionTracker provides read-only checks for whether a PVC is included
+// in the backup as BIA's additionalItems through annotation
+// backup.velero.io/must-include-additional-items.
+type PVCMustInclusionTracker interface {
+	IsPVCIncluded(namespace, pvcName string) bool
+}
+
 type VolumeHelper interface {
 	ShouldPerformSnapshot(obj runtime.Unstructured, groupResource schema.GroupResource) (bool, error)
 	ShouldPerformFSBackup(volume corev1api.Volume, pod corev1api.Pod) (bool, error)

site/content/docs/main/custom-plugins.md

@@ -65,6 +65,23 @@ order in which item action plugins are invoked. However, if a single binary impl
 they may be invoked in the order in which they are registered but it is best to not depend on this
 implementation. This is not guaranteed officially and the implementation can change at any time.
 
+### Must-include additional items (Backup Item Actions)
+
+Backup Item Actions may return `AdditionalItems` that Velero backs up as dependencies of the current item.
+By default those additional items must still pass the backup's global resource and namespace include/exclude filters.
+
+To force-backup hard dependencies despite those filters, set the following annotation on the `UpdatedItem` returned from `Execute()`:
+
+```
+backup.velero.io/must-include-additional-items: "true"
+```
+
+Behavior:
+- Only the string value `"true"` enables the bypass.
+- The annotation applies blanket to all `AdditionalItems` from that BIA invocation (not per-item).
+- Velero strips the annotation before saving the item to the backup tarball.
+- **Important Note for File System Backup (FSB):** If your plugin returns both a Pod and its associated PersistentVolumeClaims (PVCs) as `AdditionalItems`, and you expect Velero to create PodVolumeBackups (PVBs) for those PVCs using File System Backup, using this annotation ensures Velero correctly evaluates the PVCs for FSB. Velero explicitly tracks PVCs returned as additional items with this annotation, guaranteeing that PVBs are created even if the PVCs are excluded by global or fine-grained backup filters, regardless of the order they are returned in the `AdditionalItems` slice.
+
 ### Must-include additional items (Restore Item Actions)
 
 Restore Item Actions may return `AdditionalItems` that Velero restores as dependencies of the current item.