Internal ItemBlockAction plugins

This PR implements the internal ItemBlockAction plugins needed for pod, PVC, and SA.

Signed-off-by: Scott Seago <sseago@redhat.com>

changelogs/unreleased/8054-sseago

@@ -0,0 +1 @@
+Internal ItemBlockAction plugins

pkg/backup/actions/backup_pv_action.go

@@ -26,8 +26,8 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 
 	v1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
-	"github.com/vmware-tanzu/velero/pkg/kuberesource"
 	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
+	"github.com/vmware-tanzu/velero/pkg/util/actionhelpers"
 )
 
 // PVCAction inspects a PersistentVolumeClaim for the PersistentVolume
@@ -51,7 +51,7 @@ func (a *PVCAction) AppliesTo() (velero.ResourceSelector, error) {
 func (a *PVCAction) Execute(item runtime.Unstructured, backup *v1.Backup) (runtime.Unstructured, []velero.ResourceIdentifier, error) {
 	a.log.Info("Executing PVCAction")
 
-	var pvc corev1api.PersistentVolumeClaim
+	pvc := new(corev1api.PersistentVolumeClaim)
 	if err := runtime.DefaultUnstructuredConverter.FromUnstructured(item.UnstructuredContent(), &pvc); err != nil {
 		return nil, nil, errors.Wrap(err, "unable to convert unstructured item to persistent volume claim")
 	}
@@ -60,10 +60,6 @@ func (a *PVCAction) Execute(item runtime.Unstructured, backup *v1.Backup) (runti
 		return item, nil, nil
 	}
 
-	pv := velero.ResourceIdentifier{
-		GroupResource: kuberesource.PersistentVolumes,
-		Name:          pvc.Spec.VolumeName,
-	}
 	// remove dataSource if exists from prior restored CSI volumes
 	if pvc.Spec.DataSource != nil {
 		pvc.Spec.DataSource = nil
@@ -94,5 +90,5 @@ func (a *PVCAction) Execute(item runtime.Unstructured, backup *v1.Backup) (runti
 		return nil, nil, errors.Wrap(err, "unable to convert pvc to unstructured item")
 	}
 
-	return &unstructured.Unstructured{Object: pvcMap}, []velero.ResourceIdentifier{pv}, nil
+	return &unstructured.Unstructured{Object: pvcMap}, actionhelpers.RelatedItemsForPVC(pvc, a.log), nil
 }

pkg/backup/actions/pod_action.go

@@ -23,8 +23,8 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 
 	v1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
-	"github.com/vmware-tanzu/velero/pkg/kuberesource"
 	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
+	"github.com/vmware-tanzu/velero/pkg/util/actionhelpers"
 )
 
 // PodAction implements ItemAction.
@@ -55,32 +55,5 @@ func (a *PodAction) Execute(item runtime.Unstructured, backup *v1.Backup) (runti
 	if err := runtime.DefaultUnstructuredConverter.FromUnstructured(item.UnstructuredContent(), pod); err != nil {
 		return nil, nil, errors.WithStack(err)
 	}
-
-	var additionalItems []velero.ResourceIdentifier
-	if pod.Spec.PriorityClassName != "" {
-		a.log.Infof("Adding priorityclass %s to additionalItems", pod.Spec.PriorityClassName)
-		additionalItems = append(additionalItems, velero.ResourceIdentifier{
-			GroupResource: kuberesource.PriorityClasses,
-			Name:          pod.Spec.PriorityClassName,
-		})
-	}
-
-	if len(pod.Spec.Volumes) == 0 {
-		a.log.Info("pod has no volumes")
-		return item, additionalItems, nil
-	}
-
-	for _, volume := range pod.Spec.Volumes {
-		if volume.PersistentVolumeClaim != nil && volume.PersistentVolumeClaim.ClaimName != "" {
-			a.log.Infof("Adding pvc %s to additionalItems", volume.PersistentVolumeClaim.ClaimName)
-
-			additionalItems = append(additionalItems, velero.ResourceIdentifier{
-				GroupResource: kuberesource.PersistentVolumeClaims,
-				Namespace:     pod.Namespace,
-				Name:          volume.PersistentVolumeClaim.ClaimName,
-			})
-		}
-	}
-
-	return item, additionalItems, nil
+	return item, actionhelpers.RelatedItemsForPod(pod, a.log), nil
 }

pkg/backup/actions/service_account_action.go

@@ -19,40 +19,24 @@ package actions
 import (
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
-	rbac "k8s.io/api/rbac/v1"
 	"k8s.io/apimachinery/pkg/api/meta"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
-	"k8s.io/apimachinery/pkg/util/sets"
 
 	v1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	velerodiscovery "github.com/vmware-tanzu/velero/pkg/discovery"
-	"github.com/vmware-tanzu/velero/pkg/kuberesource"
 	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
+	"github.com/vmware-tanzu/velero/pkg/util/actionhelpers"
 )
 
 // ServiceAccountAction implements ItemAction.
 type ServiceAccountAction struct {
 	log                 logrus.FieldLogger
-	clusterRoleBindings []ClusterRoleBinding
+	clusterRoleBindings []actionhelpers.ClusterRoleBinding
 }
 
 // NewServiceAccountAction creates a new ItemAction for service accounts.
-func NewServiceAccountAction(logger logrus.FieldLogger, clusterRoleBindingListers map[string]ClusterRoleBindingLister, discoveryHelper velerodiscovery.Helper) (*ServiceAccountAction, error) {
-	// Look up the supported RBAC version
-	var supportedAPI metav1.GroupVersionForDiscovery
-	for _, ag := range discoveryHelper.APIGroups() {
-		if ag.Name == rbac.GroupName {
-			supportedAPI = ag.PreferredVersion
-			break
-		}
-	}
-
-	crbLister := clusterRoleBindingListers[supportedAPI.Version]
-
-	// This should be safe because the List call will return a 0-item slice
-	// if there's no matching API version.
-	crbs, err := crbLister.List()
+func NewServiceAccountAction(logger logrus.FieldLogger, clusterRoleBindingListers map[string]actionhelpers.ClusterRoleBindingLister, discoveryHelper velerodiscovery.Helper) (*ServiceAccountAction, error) {
+	crbs, err := actionhelpers.ClusterRoleBindingsForAction(clusterRoleBindingListers, discoveryHelper)
 	if err != nil {
 		return nil, err
 	}
@@ -82,40 +66,5 @@ func (a *ServiceAccountAction) Execute(item runtime.Unstructured, backup *v1.Bac
 		return nil, nil, errors.WithStack(err)
 	}
 
-	var (
-		namespace = objectMeta.GetNamespace()
-		name      = objectMeta.GetName()
-		bindings  = sets.NewString()
-		roles     = sets.NewString()
-	)
-
-	for _, crb := range a.clusterRoleBindings {
-		for _, s := range crb.ServiceAccountSubjects(namespace) {
-			if s == name {
-				a.log.Infof("Adding clusterrole %s and clusterrolebinding %s to additionalItems since serviceaccount %s/%s is a subject",
-					crb.RoleRefName(), crb.Name(), namespace, name)
-
-				bindings.Insert(crb.Name())
-				roles.Insert(crb.RoleRefName())
-				break
-			}
-		}
-	}
-
-	var additionalItems []velero.ResourceIdentifier
-	for binding := range bindings {
-		additionalItems = append(additionalItems, velero.ResourceIdentifier{
-			GroupResource: kuberesource.ClusterRoleBindings,
-			Name:          binding,
-		})
-	}
-
-	for role := range roles {
-		additionalItems = append(additionalItems, velero.ResourceIdentifier{
-			GroupResource: kuberesource.ClusterRoles,
-			Name:          role,
-		})
-	}
-
-	return item, additionalItems, nil
+	return item, actionhelpers.RelatedItemsForServiceAccount(objectMeta, a.clusterRoleBindings, a.log), nil
 }

pkg/backup/actions/service_account_action_test.go

@@ -31,21 +31,22 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/kuberesource"
 	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
 	velerotest "github.com/vmware-tanzu/velero/pkg/test"
+	"github.com/vmware-tanzu/velero/pkg/util/actionhelpers"
 )
 
-func newV1ClusterRoleBindingList(rbacCRBList []rbac.ClusterRoleBinding) []ClusterRoleBinding {
-	var crbs []ClusterRoleBinding
+func newV1ClusterRoleBindingList(rbacCRBList []rbac.ClusterRoleBinding) []actionhelpers.ClusterRoleBinding {
+	var crbs []actionhelpers.ClusterRoleBinding
 	for _, c := range rbacCRBList {
-		crbs = append(crbs, v1ClusterRoleBinding{crb: c})
+		crbs = append(crbs, actionhelpers.V1ClusterRoleBinding{Crb: c})
 	}
 
 	return crbs
 }
 
-func newV1beta1ClusterRoleBindingList(rbacCRBList []rbacbeta.ClusterRoleBinding) []ClusterRoleBinding {
-	var crbs []ClusterRoleBinding
+func newV1beta1ClusterRoleBindingList(rbacCRBList []rbacbeta.ClusterRoleBinding) []actionhelpers.ClusterRoleBinding {
+	var crbs []actionhelpers.ClusterRoleBinding
 	for _, c := range rbacCRBList {
-		crbs = append(crbs, v1beta1ClusterRoleBinding{crb: c})
+		crbs = append(crbs, actionhelpers.V1beta1ClusterRoleBinding{Crb: c})
 	}
 
 	return crbs
@@ -55,10 +56,10 @@ type FakeV1ClusterRoleBindingLister struct {
 	v1crbs []rbac.ClusterRoleBinding
 }
 
-func (f FakeV1ClusterRoleBindingLister) List() ([]ClusterRoleBinding, error) {
-	var crbs []ClusterRoleBinding
+func (f FakeV1ClusterRoleBindingLister) List() ([]actionhelpers.ClusterRoleBinding, error) {
+	var crbs []actionhelpers.ClusterRoleBinding
 	for _, c := range f.v1crbs {
-		crbs = append(crbs, v1ClusterRoleBinding{crb: c})
+		crbs = append(crbs, actionhelpers.V1ClusterRoleBinding{Crb: c})
 	}
 	return crbs, nil
 }
@@ -67,10 +68,10 @@ type FakeV1beta1ClusterRoleBindingLister struct {
 	v1beta1crbs []rbacbeta.ClusterRoleBinding
 }
 
-func (f FakeV1beta1ClusterRoleBindingLister) List() ([]ClusterRoleBinding, error) {
-	var crbs []ClusterRoleBinding
+func (f FakeV1beta1ClusterRoleBindingLister) List() ([]actionhelpers.ClusterRoleBinding, error) {
+	var crbs []actionhelpers.ClusterRoleBinding
 	for _, c := range f.v1beta1crbs {
-		crbs = append(crbs, v1beta1ClusterRoleBinding{crb: c})
+		crbs = append(crbs, actionhelpers.V1beta1ClusterRoleBinding{Crb: c})
 	}
 	return crbs, nil
 }
@@ -93,21 +94,21 @@ func TestNewServiceAccountAction(t *testing.T) {
 	tests := []struct {
 		name         string
 		version      string
-		expectedCRBs []ClusterRoleBinding
+		expectedCRBs []actionhelpers.ClusterRoleBinding
 	}{
 		{
 			name:    "rbac v1 API instantiates an saAction",
 			version: rbac.SchemeGroupVersion.Version,
-			expectedCRBs: []ClusterRoleBinding{
-				v1ClusterRoleBinding{
-					crb: rbac.ClusterRoleBinding{
+			expectedCRBs: []actionhelpers.ClusterRoleBinding{
+				actionhelpers.V1ClusterRoleBinding{
+					Crb: rbac.ClusterRoleBinding{
 						ObjectMeta: metav1.ObjectMeta{
 							Name: "v1crb-1",
 						},
 					},
 				},
-				v1ClusterRoleBinding{
-					crb: rbac.ClusterRoleBinding{
+				actionhelpers.V1ClusterRoleBinding{
+					Crb: rbac.ClusterRoleBinding{
 						ObjectMeta: metav1.ObjectMeta{
 							Name: "v1crb-2",
 						},
@@ -118,16 +119,16 @@ func TestNewServiceAccountAction(t *testing.T) {
 		{
 			name:    "rbac v1beta1 API instantiates an saAction",
 			version: rbacbeta.SchemeGroupVersion.Version,
-			expectedCRBs: []ClusterRoleBinding{
-				v1beta1ClusterRoleBinding{
-					crb: rbacbeta.ClusterRoleBinding{
+			expectedCRBs: []actionhelpers.ClusterRoleBinding{
+				actionhelpers.V1beta1ClusterRoleBinding{
+					Crb: rbacbeta.ClusterRoleBinding{
 						ObjectMeta: metav1.ObjectMeta{
 							Name: "v1beta1crb-1",
 						},
 					},
 				},
-				v1beta1ClusterRoleBinding{
-					crb: rbacbeta.ClusterRoleBinding{
+				actionhelpers.V1beta1ClusterRoleBinding{
+					Crb: rbacbeta.ClusterRoleBinding{
 						ObjectMeta: metav1.ObjectMeta{
 							Name: "v1beta1crb-2",
 						},
@@ -138,7 +139,7 @@ func TestNewServiceAccountAction(t *testing.T) {
 		{
 			name:         "no RBAC API instantiates an saAction with empty slice",
 			version:      "",
-			expectedCRBs: []ClusterRoleBinding{},
+			expectedCRBs: []actionhelpers.ClusterRoleBinding{},
 		},
 	}
 	// Set up all of our fakes outside the test loop
@@ -171,10 +172,10 @@ func TestNewServiceAccountAction(t *testing.T) {
 		},
 	}
 
-	clusterRoleBindingListers := map[string]ClusterRoleBindingLister{
+	clusterRoleBindingListers := map[string]actionhelpers.ClusterRoleBindingLister{
 		rbac.SchemeGroupVersion.Version:     FakeV1ClusterRoleBindingLister{v1crbs: v1crbs},
 		rbacbeta.SchemeGroupVersion.Version: FakeV1beta1ClusterRoleBindingLister{v1beta1crbs: v1beta1crbs},
-		"":                                  noopClusterRoleBindingLister{},
+		"":                                  actionhelpers.NoopClusterRoleBindingLister{},
 	}
 
 	for _, test := range tests {

pkg/cmd/server/plugin/plugin.go

@@ -30,10 +30,12 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/client"
 	velerodiscovery "github.com/vmware-tanzu/velero/pkg/discovery"
 	"github.com/vmware-tanzu/velero/pkg/features"
+	iba "github.com/vmware-tanzu/velero/pkg/itemblock/actions"
 	veleroplugin "github.com/vmware-tanzu/velero/pkg/plugin/framework"
 	plugincommon "github.com/vmware-tanzu/velero/pkg/plugin/framework/common"
 	ria "github.com/vmware-tanzu/velero/pkg/restore/actions"
 	csiria "github.com/vmware-tanzu/velero/pkg/restore/actions/csi"
+	"github.com/vmware-tanzu/velero/pkg/util/actionhelpers"
 )
 
 func NewCommand(f client.Factory) *cobra.Command {
@@ -171,6 +173,18 @@ func NewCommand(f client.Factory) *cobra.Command {
 				RegisterRestoreItemActionV2(
 					"velero.io/csi-volumesnapshotclass-restorer",
 					newVolumeSnapshotClassRestoreItemAction,
+				).
+				RegisterItemBlockAction(
+					"velero.io/pvc",
+					newPVCItemBlockAction(f),
+				).
+				RegisterItemBlockAction(
+					"velero.io/pod",
+					newPodItemBlockAction,
+				).
+				RegisterItemBlockAction(
+					"velero.io/service-account",
+					newServiceAccountItemBlockAction(f),
 				)
 
 			if !features.IsEnabled(velerov1api.APIGroupVersionsFeatureFlag) {
@@ -211,7 +225,7 @@ func newServiceAccountBackupItemAction(f client.Factory) plugincommon.HandlerIni
 
 		action, err := bia.NewServiceAccountAction(
 			logger,
-			bia.NewClusterRoleBindingListerMap(clientset),
+			actionhelpers.NewClusterRoleBindingListerMap(clientset),
 			discoveryHelper)
 		if err != nil {
 			return nil, err
@@ -431,3 +445,38 @@ func newVolumeSnapshotContentRestoreItemAction(logger logrus.FieldLogger) (inter
 func newVolumeSnapshotClassRestoreItemAction(logger logrus.FieldLogger) (interface{}, error) {
 	return csiria.NewVolumeSnapshotClassRestoreItemAction(logger)
 }
+
+// ItemBlockAction plugins
+
+func newPVCItemBlockAction(f client.Factory) plugincommon.HandlerInitializer {
+	return iba.NewPVCAction(f)
+}
+
+func newPodItemBlockAction(logger logrus.FieldLogger) (interface{}, error) {
+	return iba.NewPodAction(logger), nil
+}
+
+func newServiceAccountItemBlockAction(f client.Factory) plugincommon.HandlerInitializer {
+	return func(logger logrus.FieldLogger) (interface{}, error) {
+		// TODO(ncdc): consider a k8s style WantsKubernetesClientSet initialization approach
+		clientset, err := f.KubeClient()
+		if err != nil {
+			return nil, err
+		}
+
+		discoveryHelper, err := velerodiscovery.NewHelper(clientset.Discovery(), logger)
+		if err != nil {
+			return nil, err
+		}
+
+		action, err := iba.NewServiceAccountAction(
+			logger,
+			actionhelpers.NewClusterRoleBindingListerMap(clientset),
+			discoveryHelper)
+		if err != nil {
+			return nil, err
+		}
+
+		return action, nil
+	}
+}

pkg/itemblock/actions/pod_action.go

@@ -0,0 +1,63 @@
+/*
+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 actions
+
+import (
+	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
+	corev1api "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+
+	v1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
+	"github.com/vmware-tanzu/velero/pkg/util/actionhelpers"
+)
+
+// PodAction implements ItemBlockAction.
+type PodAction struct {
+	log logrus.FieldLogger
+}
+
+// NewPodAction creates a new ItemBlockAction for pods.
+func NewPodAction(logger logrus.FieldLogger) *PodAction {
+	return &PodAction{log: logger}
+}
+
+// AppliesTo returns a ResourceSelector that applies only to pods.
+func (a *PodAction) AppliesTo() (velero.ResourceSelector, error) {
+	return velero.ResourceSelector{
+		IncludedResources: []string{"pods"},
+	}, nil
+}
+
+// GetRelatedItems scans the pod's spec.volumes for persistentVolumeClaim volumes and returns a
+// ResourceIdentifier list containing references to all of the persistentVolumeClaim volumes used by
+// the pod. This ensures that when a pod is backed up, all referenced PVCs are backed up along with the pod.
+func (a *PodAction) GetRelatedItems(item runtime.Unstructured, backup *v1.Backup) ([]velero.ResourceIdentifier, error) {
+	a.log.Info("Executing pod ItemBlockAction")
+	defer a.log.Info("Done executing pod ItemBlockAction")
+
+	pod := new(corev1api.Pod)
+	if err := runtime.DefaultUnstructuredConverter.FromUnstructured(item.UnstructuredContent(), pod); err != nil {
+		return nil, errors.WithStack(err)
+	}
+	return actionhelpers.RelatedItemsForPod(pod, a.log), nil
+}
+
+func (a *PodAction) Name() string {
+	return "PodItemBlockAction"
+}

pkg/itemblock/actions/pod_action_test.go

@@ -0,0 +1,148 @@
+/*
+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 actions
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"k8s.io/apimachinery/pkg/runtime"
+
+	"github.com/vmware-tanzu/velero/pkg/kuberesource"
+	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
+	velerotest "github.com/vmware-tanzu/velero/pkg/test"
+)
+
+func TestPodActionAppliesTo(t *testing.T) {
+	a := NewPodAction(velerotest.NewLogger())
+
+	actual, err := a.AppliesTo()
+	require.NoError(t, err)
+
+	expected := velero.ResourceSelector{
+		IncludedResources: []string{"pods"},
+	}
+	assert.Equal(t, expected, actual)
+}
+
+func TestPodActionGetRelatedItems(t *testing.T) {
+	tests := []struct {
+		name     string
+		pod      runtime.Unstructured
+		expected []velero.ResourceIdentifier
+	}{
+		{
+			name: "no spec.volumes",
+			pod: velerotest.UnstructuredOrDie(`
+			{
+				"apiVersion": "v1",
+				"kind": "Pod",
+				"metadata": {
+					"namespace": "foo",
+					"name": "bar"
+				}
+			}
+			`),
+		},
+		{
+			name: "persistentVolumeClaim without claimName",
+			pod: velerotest.UnstructuredOrDie(`
+			{
+				"apiVersion": "v1",
+				"kind": "Pod",
+				"metadata": {
+					"namespace": "foo",
+					"name": "bar"
+				},
+				"spec": {
+					"volumes": [
+						{
+							"persistentVolumeClaim": {}
+						}
+					]
+				}
+			}
+			`),
+		},
+		{
+			name: "full test, mix of volume types",
+			pod: velerotest.UnstructuredOrDie(`
+			{
+				"apiVersion": "v1",
+				"kind": "Pod",
+				"metadata": {
+					"namespace": "foo",
+					"name": "bar"
+				},
+				"spec": {
+					"volumes": [
+						{
+							"persistentVolumeClaim": {}
+						},
+						{
+							"emptyDir": {}
+						},
+						{
+							"persistentVolumeClaim": {"claimName": "claim1"}
+						},
+						{
+							"emptyDir": {}
+						},
+						{
+							"persistentVolumeClaim": {"claimName": "claim2"}
+						}
+					]
+				}
+			}
+			`),
+			expected: []velero.ResourceIdentifier{
+				{GroupResource: kuberesource.PersistentVolumeClaims, Namespace: "foo", Name: "claim1"},
+				{GroupResource: kuberesource.PersistentVolumeClaims, Namespace: "foo", Name: "claim2"},
+			},
+		},
+		{
+			name: "test priority class",
+			pod: velerotest.UnstructuredOrDie(`
+			{
+				"apiVersion": "v1",
+				"kind": "Pod",
+				"metadata": {
+					"namespace": "foo",
+					"name": "bar"
+				},
+				"spec": {
+					"priorityClassName": "testPriorityClass"
+				}
+			}
+			`),
+			expected: []velero.ResourceIdentifier{
+				{GroupResource: kuberesource.PriorityClasses, Name: "testPriorityClass"},
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			a := NewPodAction(velerotest.NewLogger())
+
+			relatedItems, err := a.GetRelatedItems(test.pod, nil)
+			require.NoError(t, err)
+			assert.Equal(t, test.expected, relatedItems)
+		})
+	}
+}

pkg/itemblock/actions/pvc_action.go

@@ -0,0 +1,113 @@
+/*
+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 actions
+
+import (
+	"context"
+
+	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
+	corev1api "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	crclient "sigs.k8s.io/controller-runtime/pkg/client"
+
+	v1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	"github.com/vmware-tanzu/velero/pkg/client"
+	"github.com/vmware-tanzu/velero/pkg/kuberesource"
+	plugincommon "github.com/vmware-tanzu/velero/pkg/plugin/framework/common"
+	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
+	"github.com/vmware-tanzu/velero/pkg/util/actionhelpers"
+	"github.com/vmware-tanzu/velero/pkg/util/kube"
+)
+
+// PVCAction inspects a PersistentVolumeClaim for the PersistentVolume
+// that it references and backs it up
+type PVCAction struct {
+	log      logrus.FieldLogger
+	crClient crclient.Client
+}
+
+func NewPVCAction(f client.Factory) plugincommon.HandlerInitializer {
+	return func(logger logrus.FieldLogger) (interface{}, error) {
+		crClient, err := f.KubebuilderClient()
+		if err != nil {
+			return nil, errors.WithStack(err)
+		}
+
+		return &PVCAction{
+			log:      logger,
+			crClient: crClient,
+		}, nil
+	}
+}
+
+func (a *PVCAction) AppliesTo() (velero.ResourceSelector, error) {
+	return velero.ResourceSelector{
+		IncludedResources: []string{"persistentvolumeclaims"},
+	}, nil
+}
+
+func (a *PVCAction) GetRelatedItems(item runtime.Unstructured, backup *v1.Backup) ([]velero.ResourceIdentifier, error) {
+	a.log.Info("Executing PVC ItemBlockAction")
+	defer a.log.Info("Done executing PVC ItemBlockAction")
+
+	pvc := new(corev1api.PersistentVolumeClaim)
+	if err := runtime.DefaultUnstructuredConverter.FromUnstructured(item.UnstructuredContent(), &pvc); err != nil {
+		return nil, errors.Wrap(err, "unable to convert unstructured item to persistent volume claim")
+	}
+
+	if pvc.Status.Phase != corev1api.ClaimBound || pvc.Spec.VolumeName == "" {
+		return nil, nil
+	}
+	// returns the PV for the PVC (shared with BIA additionalItems)
+	relatedItems := actionhelpers.RelatedItemsForPVC(pvc, a.log)
+
+	// Adds pods mounting this PVC to ensure that multiple pods mounting the same RWX
+	// volume get backed up together.
+	pods := new(corev1api.PodList)
+	err := a.crClient.List(context.Background(), pods, crclient.InNamespace(pvc.Namespace))
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to list pods")
+	}
+
+	for i := range pods.Items {
+		for _, volume := range pods.Items[i].Spec.Volumes {
+			if volume.VolumeSource.PersistentVolumeClaim == nil {
+				continue
+			}
+			if volume.PersistentVolumeClaim.ClaimName == pvc.Name {
+				if kube.IsPodRunning(&pods.Items[i]) != nil {
+					a.log.Infof("Related pod %s is not running, not adding to ItemBlock for PVC %s", pods.Items[i].Name, pvc.Name)
+				} else {
+					a.log.Infof("Adding related Pod %s to PVC %s", pods.Items[i].Name, pvc.Name)
+					relatedItems = append(relatedItems, velero.ResourceIdentifier{
+						GroupResource: kuberesource.Pods,
+						Namespace:     pods.Items[i].Namespace,
+						Name:          pods.Items[i].Name,
+					})
+				}
+				break
+			}
+		}
+	}
+
+	return relatedItems, nil
+}
+
+func (a *PVCAction) Name() string {
+	return "PodItemBlockAction"
+}

pkg/itemblock/actions/pvc_action_test.go

@@ -0,0 +1,167 @@
+/*
+Copyright 2017 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 actions
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/sirupsen/logrus"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	corev1api "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime"
+
+	v1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	"github.com/vmware-tanzu/velero/pkg/builder"
+	factorymocks "github.com/vmware-tanzu/velero/pkg/client/mocks"
+	"github.com/vmware-tanzu/velero/pkg/kuberesource"
+	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
+	velerotest "github.com/vmware-tanzu/velero/pkg/test"
+)
+
+func TestBackupPVAction(t *testing.T) {
+	tests := []struct {
+		name            string
+		pvc             *corev1api.PersistentVolumeClaim
+		pods            []*corev1api.Pod
+		expectedErr     error
+		expectedRelated []velero.ResourceIdentifier
+	}{
+		{
+			name:            "Test no volumeName",
+			pvc:             builder.ForPersistentVolumeClaim("velero", "testPVC").Phase(corev1api.ClaimBound).Result(),
+			expectedErr:     nil,
+			expectedRelated: nil,
+		},
+		{
+			name:            "Test empty volumeName",
+			pvc:             builder.ForPersistentVolumeClaim("velero", "testPVC").VolumeName("").Phase(corev1api.ClaimBound).Result(),
+			expectedErr:     nil,
+			expectedRelated: nil,
+		},
+		{
+			name:            "Test no status phase",
+			pvc:             builder.ForPersistentVolumeClaim("velero", "testPVC").VolumeName("testPV").Result(),
+			expectedErr:     nil,
+			expectedRelated: nil,
+		},
+		{
+			name:            "Test pending status phase",
+			pvc:             builder.ForPersistentVolumeClaim("velero", "testPVC").VolumeName("testPV").Phase(corev1api.ClaimPending).Result(),
+			expectedErr:     nil,
+			expectedRelated: nil,
+		},
+		{
+			name:            "Test lost status phase",
+			pvc:             builder.ForPersistentVolumeClaim("velero", "testPVC").VolumeName("testPV").Phase(corev1api.ClaimLost).Result(),
+			expectedErr:     nil,
+			expectedRelated: nil,
+		},
+		{
+			name:        "Test with volume",
+			pvc:         builder.ForPersistentVolumeClaim("velero", "testPVC").VolumeName("testPV").Phase(corev1api.ClaimBound).Result(),
+			expectedErr: nil,
+			expectedRelated: []velero.ResourceIdentifier{
+				{GroupResource: kuberesource.PersistentVolumes, Name: "testPV"},
+			},
+		},
+		{
+			name: "Test with volume and one running pod",
+			pvc:  builder.ForPersistentVolumeClaim("velero", "testPVC").VolumeName("testPV").Phase(corev1api.ClaimBound).Result(),
+			pods: []*corev1api.Pod{
+				builder.ForPod("velero", "testPod1").Volumes(builder.ForVolume("testPVC").PersistentVolumeClaimSource("testPVC").Result()).NodeName("velero").Phase(corev1api.PodRunning).Result(),
+			},
+			expectedErr: nil,
+			expectedRelated: []velero.ResourceIdentifier{
+				{GroupResource: kuberesource.PersistentVolumes, Name: "testPV"},
+				{GroupResource: kuberesource.Pods, Namespace: "velero", Name: "testPod1"},
+			},
+		},
+		{
+			name: "Test with volume and multiple running pods",
+			pvc:  builder.ForPersistentVolumeClaim("velero", "testPVC").VolumeName("testPV").Phase(corev1api.ClaimBound).Result(),
+			pods: []*corev1api.Pod{
+				builder.ForPod("velero", "testPod1").Volumes(builder.ForVolume("testPVC").PersistentVolumeClaimSource("testPVC").Result()).NodeName("velero").Phase(corev1api.PodRunning).Result(),
+				builder.ForPod("velero", "testPod2").Volumes(builder.ForVolume("testPVC").PersistentVolumeClaimSource("testPVC").Result()).NodeName("velero").Phase(corev1api.PodRunning).Result(),
+				builder.ForPod("velero", "testPod3").Volumes(builder.ForVolume("testPVC").PersistentVolumeClaimSource("testPVC").Result()).NodeName("velero").Phase(corev1api.PodRunning).Result(),
+			},
+			expectedErr: nil,
+			expectedRelated: []velero.ResourceIdentifier{
+				{GroupResource: kuberesource.PersistentVolumes, Name: "testPV"},
+				{GroupResource: kuberesource.Pods, Namespace: "velero", Name: "testPod1"},
+				{GroupResource: kuberesource.Pods, Namespace: "velero", Name: "testPod2"},
+				{GroupResource: kuberesource.Pods, Namespace: "velero", Name: "testPod3"},
+			},
+		},
+		{
+			name: "Test with volume and multiple running pods, some not running",
+			pvc:  builder.ForPersistentVolumeClaim("velero", "testPVC").VolumeName("testPV").Phase(corev1api.ClaimBound).Result(),
+			pods: []*corev1api.Pod{
+				builder.ForPod("velero", "testPod1").Volumes(builder.ForVolume("testPVC").PersistentVolumeClaimSource("testPVC").Result()).NodeName("velero").Phase(corev1api.PodSucceeded).Result(),
+				builder.ForPod("velero", "testPod2").Volumes(builder.ForVolume("testPVC").PersistentVolumeClaimSource("testPVC").Result()).NodeName("velero").Phase(corev1api.PodRunning).Result(),
+				builder.ForPod("velero", "testPod3").Volumes(builder.ForVolume("testPVC").PersistentVolumeClaimSource("testPVC").Result()).Phase(corev1api.PodRunning).Result(),
+			},
+			expectedErr: nil,
+			expectedRelated: []velero.ResourceIdentifier{
+				{GroupResource: kuberesource.PersistentVolumes, Name: "testPV"},
+				{GroupResource: kuberesource.Pods, Namespace: "velero", Name: "testPod2"},
+			},
+		},
+	}
+
+	backup := &v1.Backup{}
+	logger := logrus.New()
+
+	f := &factorymocks.Factory{}
+	f.On("KubebuilderClient").Return(nil, fmt.Errorf(""))
+	plugin := NewPVCAction(f)
+	_, err := plugin(logger)
+	require.Error(t, err)
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(*testing.T) {
+			crClient := velerotest.NewFakeControllerRuntimeClient(t)
+			f := &factorymocks.Factory{}
+			f.On("KubebuilderClient").Return(crClient, nil)
+			plugin := NewPVCAction(f)
+			i, err := plugin(logger)
+			require.NoError(t, err)
+			a := i.(*PVCAction)
+
+			if tc.pvc != nil {
+				require.NoError(t, crClient.Create(context.Background(), tc.pvc))
+			}
+			for _, pod := range tc.pods {
+				require.NoError(t, crClient.Create(context.Background(), pod))
+			}
+
+			pvcMap, err := runtime.DefaultUnstructuredConverter.ToUnstructured(&tc.pvc)
+			require.NoError(t, err)
+
+			relatedItems, err := a.GetRelatedItems(&unstructured.Unstructured{Object: pvcMap}, backup)
+			if tc.expectedErr != nil {
+				require.EqualError(t, err, tc.expectedErr.Error())
+			} else {
+				require.NoError(t, err)
+			}
+			assert.Equal(t, tc.expectedRelated, relatedItems)
+		})
+	}
+}

pkg/itemblock/actions/service_account_action.go

@@ -0,0 +1,73 @@
+/*
+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 actions
+
+import (
+	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
+	"k8s.io/apimachinery/pkg/api/meta"
+	"k8s.io/apimachinery/pkg/runtime"
+
+	v1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	velerodiscovery "github.com/vmware-tanzu/velero/pkg/discovery"
+	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
+	"github.com/vmware-tanzu/velero/pkg/util/actionhelpers"
+)
+
+// ServiceAccountAction implements ItemBlockAction.
+type ServiceAccountAction struct {
+	log                 logrus.FieldLogger
+	clusterRoleBindings []actionhelpers.ClusterRoleBinding
+}
+
+// NewServiceAccountAction creates a new ItemBlockAction for service accounts.
+func NewServiceAccountAction(logger logrus.FieldLogger, clusterRoleBindingListers map[string]actionhelpers.ClusterRoleBindingLister, discoveryHelper velerodiscovery.Helper) (*ServiceAccountAction, error) {
+	crbs, err := actionhelpers.ClusterRoleBindingsForAction(clusterRoleBindingListers, discoveryHelper)
+	if err != nil {
+		return nil, err
+	}
+
+	return &ServiceAccountAction{
+		log:                 logger,
+		clusterRoleBindings: crbs,
+	}, nil
+}
+
+// AppliesTo returns a ResourceSelector that applies only to service accounts.
+func (a *ServiceAccountAction) AppliesTo() (velero.ResourceSelector, error) {
+	return velero.ResourceSelector{
+		IncludedResources: []string{"serviceaccounts"},
+	}, nil
+}
+
+// GetRelatedItems checks for any ClusterRoleBindings that have this service account as a subject, and
+// returns the ClusterRoleBinding and associated ClusterRole.
+func (a *ServiceAccountAction) GetRelatedItems(item runtime.Unstructured, backup *v1.Backup) ([]velero.ResourceIdentifier, error) {
+	a.log.Info("Running ServiceAccount ItemBlockAction")
+	defer a.log.Info("Done running ServiceAccount ItemBlockAction")
+
+	objectMeta, err := meta.Accessor(item)
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+
+	return actionhelpers.RelatedItemsForServiceAccount(objectMeta, a.clusterRoleBindings, a.log), nil
+}
+
+func (a *ServiceAccountAction) Name() string {
+	return "ServiceAccountItemBlockAction"
+}

pkg/itemblock/actions/service_account_action_test.go

@@ -0,0 +1,609 @@
+/*
+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 actions
+
+import (
+	"fmt"
+	"sort"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	rbac "k8s.io/api/rbac/v1"
+	rbacbeta "k8s.io/api/rbac/v1beta1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+
+	"github.com/vmware-tanzu/velero/pkg/kuberesource"
+	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
+	velerotest "github.com/vmware-tanzu/velero/pkg/test"
+	"github.com/vmware-tanzu/velero/pkg/util/actionhelpers"
+)
+
+func newV1ClusterRoleBindingList(rbacCRBList []rbac.ClusterRoleBinding) []actionhelpers.ClusterRoleBinding {
+	var crbs []actionhelpers.ClusterRoleBinding
+	for _, c := range rbacCRBList {
+		crbs = append(crbs, actionhelpers.V1ClusterRoleBinding{Crb: c})
+	}
+
+	return crbs
+}
+
+func newV1beta1ClusterRoleBindingList(rbacCRBList []rbacbeta.ClusterRoleBinding) []actionhelpers.ClusterRoleBinding {
+	var crbs []actionhelpers.ClusterRoleBinding
+	for _, c := range rbacCRBList {
+		crbs = append(crbs, actionhelpers.V1beta1ClusterRoleBinding{Crb: c})
+	}
+
+	return crbs
+}
+
+type FakeV1ClusterRoleBindingLister struct {
+	v1crbs []rbac.ClusterRoleBinding
+}
+
+func (f FakeV1ClusterRoleBindingLister) List() ([]actionhelpers.ClusterRoleBinding, error) {
+	var crbs []actionhelpers.ClusterRoleBinding
+	for _, c := range f.v1crbs {
+		crbs = append(crbs, actionhelpers.V1ClusterRoleBinding{Crb: c})
+	}
+	return crbs, nil
+}
+
+type FakeV1beta1ClusterRoleBindingLister struct {
+	v1beta1crbs []rbacbeta.ClusterRoleBinding
+}
+
+func (f FakeV1beta1ClusterRoleBindingLister) List() ([]actionhelpers.ClusterRoleBinding, error) {
+	var crbs []actionhelpers.ClusterRoleBinding
+	for _, c := range f.v1beta1crbs {
+		crbs = append(crbs, actionhelpers.V1beta1ClusterRoleBinding{Crb: c})
+	}
+	return crbs, nil
+}
+
+func TestServiceAccountActionAppliesTo(t *testing.T) {
+	// Instantiating the struct directly since using
+	// NewServiceAccountAction requires a full Kubernetes clientset
+	a := &ServiceAccountAction{}
+
+	actual, err := a.AppliesTo()
+	require.NoError(t, err)
+
+	expected := velero.ResourceSelector{
+		IncludedResources: []string{"serviceaccounts"},
+	}
+	assert.Equal(t, expected, actual)
+}
+
+func TestNewServiceAccountAction(t *testing.T) {
+	tests := []struct {
+		name         string
+		version      string
+		expectedCRBs []actionhelpers.ClusterRoleBinding
+	}{
+		{
+			name:    "rbac v1 API instantiates an saAction",
+			version: rbac.SchemeGroupVersion.Version,
+			expectedCRBs: []actionhelpers.ClusterRoleBinding{
+				actionhelpers.V1ClusterRoleBinding{
+					Crb: rbac.ClusterRoleBinding{
+						ObjectMeta: metav1.ObjectMeta{
+							Name: "v1crb-1",
+						},
+					},
+				},
+				actionhelpers.V1ClusterRoleBinding{
+					Crb: rbac.ClusterRoleBinding{
+						ObjectMeta: metav1.ObjectMeta{
+							Name: "v1crb-2",
+						},
+					},
+				},
+			},
+		},
+		{
+			name:    "rbac v1beta1 API instantiates an saAction",
+			version: rbacbeta.SchemeGroupVersion.Version,
+			expectedCRBs: []actionhelpers.ClusterRoleBinding{
+				actionhelpers.V1beta1ClusterRoleBinding{
+					Crb: rbacbeta.ClusterRoleBinding{
+						ObjectMeta: metav1.ObjectMeta{
+							Name: "v1beta1crb-1",
+						},
+					},
+				},
+				actionhelpers.V1beta1ClusterRoleBinding{
+					Crb: rbacbeta.ClusterRoleBinding{
+						ObjectMeta: metav1.ObjectMeta{
+							Name: "v1beta1crb-2",
+						},
+					},
+				},
+			},
+		},
+		{
+			name:         "no RBAC API instantiates an saAction with empty slice",
+			version:      "",
+			expectedCRBs: []actionhelpers.ClusterRoleBinding{},
+		},
+	}
+	// Set up all of our fakes outside the test loop
+	discoveryHelper := velerotest.FakeDiscoveryHelper{}
+	logger := velerotest.NewLogger()
+
+	v1crbs := []rbac.ClusterRoleBinding{
+		{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "v1crb-1",
+			},
+		},
+		{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "v1crb-2",
+			},
+		},
+	}
+
+	v1beta1crbs := []rbacbeta.ClusterRoleBinding{
+		{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "v1beta1crb-1",
+			},
+		},
+		{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "v1beta1crb-2",
+			},
+		},
+	}
+
+	clusterRoleBindingListers := map[string]actionhelpers.ClusterRoleBindingLister{
+		rbac.SchemeGroupVersion.Version:     FakeV1ClusterRoleBindingLister{v1crbs: v1crbs},
+		rbacbeta.SchemeGroupVersion.Version: FakeV1beta1ClusterRoleBindingLister{v1beta1crbs: v1beta1crbs},
+		"":                                  actionhelpers.NoopClusterRoleBindingLister{},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			// We only care about the preferred version, nothing else in the list
+			discoveryHelper.APIGroupsList = []metav1.APIGroup{
+				{
+					Name: rbac.GroupName,
+					PreferredVersion: metav1.GroupVersionForDiscovery{
+						Version: test.version,
+					},
+				},
+			}
+			action, err := NewServiceAccountAction(logger, clusterRoleBindingListers, &discoveryHelper)
+			require.NoError(t, err)
+			assert.Equal(t, test.expectedCRBs, action.clusterRoleBindings)
+		})
+	}
+}
+
+func TestServiceAccountActionExecute(t *testing.T) {
+	tests := []struct {
+		name                    string
+		serviceAccount          runtime.Unstructured
+		crbs                    []rbac.ClusterRoleBinding
+		expectedAdditionalItems []velero.ResourceIdentifier
+	}{
+		{
+			name: "no crbs",
+			serviceAccount: velerotest.UnstructuredOrDie(`
+			{
+				"apiVersion": "v1",
+				"kind": "ServiceAccount",
+				"metadata": {
+					"namespace": "velero",
+					"name": "velero"
+				}
+			}
+			`),
+			crbs:                    nil,
+			expectedAdditionalItems: nil,
+		},
+		{
+			name: "no matching crbs",
+			serviceAccount: velerotest.UnstructuredOrDie(`
+			{
+				"apiVersion": "v1",
+				"kind": "ServiceAccount",
+				"metadata": {
+					"namespace": "velero",
+					"name": "velero"
+				}
+			}
+			`),
+			crbs: []rbac.ClusterRoleBinding{
+				{
+					Subjects: []rbac.Subject{
+						{
+							Kind:      "non-matching-kind",
+							Namespace: "non-matching-ns",
+							Name:      "non-matching-name",
+						},
+						{
+							Kind:      "non-matching-kind",
+							Namespace: "velero",
+							Name:      "velero",
+						},
+						{
+							Kind:      rbac.ServiceAccountKind,
+							Namespace: "non-matching-ns",
+							Name:      "velero",
+						},
+						{
+							Kind:      rbac.ServiceAccountKind,
+							Namespace: "velero",
+							Name:      "non-matching-name",
+						},
+					},
+					RoleRef: rbac.RoleRef{
+						Name: "role",
+					},
+				},
+			},
+			expectedAdditionalItems: nil,
+		},
+		{
+			name: "some matching crbs",
+			serviceAccount: velerotest.UnstructuredOrDie(`
+			{
+				"apiVersion": "v1",
+				"kind": "ServiceAccount",
+				"metadata": {
+					"namespace": "velero",
+					"name": "velero"
+				}
+			}
+			`),
+			crbs: []rbac.ClusterRoleBinding{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "crb-1",
+					},
+					Subjects: []rbac.Subject{
+						{
+							Kind:      "non-matching-kind",
+							Namespace: "non-matching-ns",
+							Name:      "non-matching-name",
+						},
+					},
+					RoleRef: rbac.RoleRef{
+						Name: "role-1",
+					},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "crb-2",
+					},
+					Subjects: []rbac.Subject{
+						{
+							Kind:      "non-matching-kind",
+							Namespace: "non-matching-ns",
+							Name:      "non-matching-name",
+						},
+						{
+							Kind:      rbac.ServiceAccountKind,
+							Namespace: "velero",
+							Name:      "velero",
+						},
+					},
+					RoleRef: rbac.RoleRef{
+						Name: "role-2",
+					},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "crb-3",
+					},
+					Subjects: []rbac.Subject{
+						{
+							Kind:      rbac.ServiceAccountKind,
+							Namespace: "velero",
+							Name:      "velero",
+						},
+					},
+					RoleRef: rbac.RoleRef{
+						Name: "role-3",
+					},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "crb-4",
+					},
+					Subjects: []rbac.Subject{
+						{
+							Kind:      rbac.ServiceAccountKind,
+							Namespace: "velero",
+							Name:      "velero",
+						},
+						{
+							Kind:      "non-matching-kind",
+							Namespace: "non-matching-ns",
+							Name:      "non-matching-name",
+						},
+					},
+					RoleRef: rbac.RoleRef{
+						Name: "role-4",
+					},
+				},
+			},
+			expectedAdditionalItems: []velero.ResourceIdentifier{
+				{
+					GroupResource: kuberesource.ClusterRoleBindings,
+					Name:          "crb-2",
+				},
+				{
+					GroupResource: kuberesource.ClusterRoleBindings,
+					Name:          "crb-3",
+				},
+				{
+					GroupResource: kuberesource.ClusterRoleBindings,
+					Name:          "crb-4",
+				},
+				{
+					GroupResource: kuberesource.ClusterRoles,
+					Name:          "role-2",
+				},
+				{
+					GroupResource: kuberesource.ClusterRoles,
+					Name:          "role-3",
+				},
+				{
+					GroupResource: kuberesource.ClusterRoles,
+					Name:          "role-4",
+				},
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			// Create the action struct directly so we don't need to mock a clientset
+			action := &ServiceAccountAction{
+				log:                 velerotest.NewLogger(),
+				clusterRoleBindings: newV1ClusterRoleBindingList(test.crbs),
+			}
+
+			additional, err := action.GetRelatedItems(test.serviceAccount, nil)
+
+			assert.NoError(t, err)
+
+			// ensure slices are ordered for valid comparison
+			sort.Slice(test.expectedAdditionalItems, func(i, j int) bool {
+				return fmt.Sprintf("%s.%s", test.expectedAdditionalItems[i].GroupResource.String(), test.expectedAdditionalItems[i].Name) <
+					fmt.Sprintf("%s.%s", test.expectedAdditionalItems[j].GroupResource.String(), test.expectedAdditionalItems[j].Name)
+			})
+
+			sort.Slice(additional, func(i, j int) bool {
+				return fmt.Sprintf("%s.%s", additional[i].GroupResource.String(), additional[i].Name) <
+					fmt.Sprintf("%s.%s", additional[j].GroupResource.String(), additional[j].Name)
+			})
+
+			assert.Equal(t, test.expectedAdditionalItems, additional)
+		})
+	}
+}
+
+func TestServiceAccountActionExecuteOnBeta1(t *testing.T) {
+	tests := []struct {
+		name                    string
+		serviceAccount          runtime.Unstructured
+		crbs                    []rbacbeta.ClusterRoleBinding
+		expectedAdditionalItems []velero.ResourceIdentifier
+	}{
+		{
+			name: "no crbs",
+			serviceAccount: velerotest.UnstructuredOrDie(`
+			{
+				"apiVersion": "v1",
+				"kind": "ServiceAccount",
+				"metadata": {
+					"namespace": "velero",
+					"name": "velero"
+				}
+			}
+			`),
+			crbs:                    nil,
+			expectedAdditionalItems: nil,
+		},
+		{
+			name: "no matching crbs",
+			serviceAccount: velerotest.UnstructuredOrDie(`
+			{
+				"apiVersion": "v1",
+				"kind": "ServiceAccount",
+				"metadata": {
+					"namespace": "velero",
+					"name": "velero"
+				}
+			}
+			`),
+			crbs: []rbacbeta.ClusterRoleBinding{
+				{
+					Subjects: []rbacbeta.Subject{
+						{
+							Kind:      "non-matching-kind",
+							Namespace: "non-matching-ns",
+							Name:      "non-matching-name",
+						},
+						{
+							Kind:      "non-matching-kind",
+							Namespace: "velero",
+							Name:      "velero",
+						},
+						{
+							Kind:      rbacbeta.ServiceAccountKind,
+							Namespace: "non-matching-ns",
+							Name:      "velero",
+						},
+						{
+							Kind:      rbacbeta.ServiceAccountKind,
+							Namespace: "velero",
+							Name:      "non-matching-name",
+						},
+					},
+					RoleRef: rbacbeta.RoleRef{
+						Name: "role",
+					},
+				},
+			},
+			expectedAdditionalItems: nil,
+		},
+		{
+			name: "some matching crbs",
+			serviceAccount: velerotest.UnstructuredOrDie(`
+			{
+				"apiVersion": "v1",
+				"kind": "ServiceAccount",
+				"metadata": {
+					"namespace": "velero",
+					"name": "velero"
+				}
+			}
+			`),
+			crbs: []rbacbeta.ClusterRoleBinding{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "crb-1",
+					},
+					Subjects: []rbacbeta.Subject{
+						{
+							Kind:      "non-matching-kind",
+							Namespace: "non-matching-ns",
+							Name:      "non-matching-name",
+						},
+					},
+					RoleRef: rbacbeta.RoleRef{
+						Name: "role-1",
+					},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "crb-2",
+					},
+					Subjects: []rbacbeta.Subject{
+						{
+							Kind:      "non-matching-kind",
+							Namespace: "non-matching-ns",
+							Name:      "non-matching-name",
+						},
+						{
+							Kind:      rbacbeta.ServiceAccountKind,
+							Namespace: "velero",
+							Name:      "velero",
+						},
+					},
+					RoleRef: rbacbeta.RoleRef{
+						Name: "role-2",
+					},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "crb-3",
+					},
+					Subjects: []rbacbeta.Subject{
+						{
+							Kind:      rbacbeta.ServiceAccountKind,
+							Namespace: "velero",
+							Name:      "velero",
+						},
+					},
+					RoleRef: rbacbeta.RoleRef{
+						Name: "role-3",
+					},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "crb-4",
+					},
+					Subjects: []rbacbeta.Subject{
+						{
+							Kind:      rbacbeta.ServiceAccountKind,
+							Namespace: "velero",
+							Name:      "velero",
+						},
+						{
+							Kind:      "non-matching-kind",
+							Namespace: "non-matching-ns",
+							Name:      "non-matching-name",
+						},
+					},
+					RoleRef: rbacbeta.RoleRef{
+						Name: "role-4",
+					},
+				},
+			},
+			expectedAdditionalItems: []velero.ResourceIdentifier{
+				{
+					GroupResource: kuberesource.ClusterRoleBindings,
+					Name:          "crb-2",
+				},
+				{
+					GroupResource: kuberesource.ClusterRoleBindings,
+					Name:          "crb-3",
+				},
+				{
+					GroupResource: kuberesource.ClusterRoleBindings,
+					Name:          "crb-4",
+				},
+				{
+					GroupResource: kuberesource.ClusterRoles,
+					Name:          "role-2",
+				},
+				{
+					GroupResource: kuberesource.ClusterRoles,
+					Name:          "role-3",
+				},
+				{
+					GroupResource: kuberesource.ClusterRoles,
+					Name:          "role-4",
+				},
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			// Create the action struct directly so we don't need to mock a clientset
+			action := &ServiceAccountAction{
+				log:                 velerotest.NewLogger(),
+				clusterRoleBindings: newV1beta1ClusterRoleBindingList(test.crbs),
+			}
+
+			additional, err := action.GetRelatedItems(test.serviceAccount, nil)
+
+			assert.NoError(t, err)
+
+			// ensure slices are ordered for valid comparison
+			sort.Slice(test.expectedAdditionalItems, func(i, j int) bool {
+				return fmt.Sprintf("%s.%s", test.expectedAdditionalItems[i].GroupResource.String(), test.expectedAdditionalItems[i].Name) <
+					fmt.Sprintf("%s.%s", test.expectedAdditionalItems[j].GroupResource.String(), test.expectedAdditionalItems[j].Name)
+			})
+
+			sort.Slice(additional, func(i, j int) bool {
+				return fmt.Sprintf("%s.%s", additional[i].GroupResource.String(), additional[i].Name) <
+					fmt.Sprintf("%s.%s", additional[j].GroupResource.String(), additional[j].Name)
+			})
+
+			assert.Equal(t, test.expectedAdditionalItems, additional)
+		})
+	}
+}

pkg/util/actionhelpers/pod_helper.go

@@ -0,0 +1,53 @@
+/*
+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 actionhelpers
+
+import (
+	"github.com/sirupsen/logrus"
+	corev1api "k8s.io/api/core/v1"
+
+	"github.com/vmware-tanzu/velero/pkg/kuberesource"
+	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
+)
+
+func RelatedItemsForPod(pod *corev1api.Pod, log logrus.FieldLogger) []velero.ResourceIdentifier {
+	var additionalItems []velero.ResourceIdentifier
+	if pod.Spec.PriorityClassName != "" {
+		log.Infof("Adding priorityclass %s to additionalItems", pod.Spec.PriorityClassName)
+		additionalItems = append(additionalItems, velero.ResourceIdentifier{
+			GroupResource: kuberesource.PriorityClasses,
+			Name:          pod.Spec.PriorityClassName,
+		})
+	}
+
+	if len(pod.Spec.Volumes) == 0 {
+		log.Info("pod has no volumes")
+	}
+
+	for _, volume := range pod.Spec.Volumes {
+		if volume.PersistentVolumeClaim != nil && volume.PersistentVolumeClaim.ClaimName != "" {
+			log.Infof("Adding pvc %s to additionalItems", volume.PersistentVolumeClaim.ClaimName)
+
+			additionalItems = append(additionalItems, velero.ResourceIdentifier{
+				GroupResource: kuberesource.PersistentVolumeClaims,
+				Namespace:     pod.Namespace,
+				Name:          volume.PersistentVolumeClaim.ClaimName,
+			})
+		}
+	}
+	return additionalItems
+}

pkg/util/actionhelpers/pvc_helper.go

@@ -0,0 +1,34 @@
+/*
+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 actionhelpers
+
+import (
+	"github.com/sirupsen/logrus"
+	corev1api "k8s.io/api/core/v1"
+
+	"github.com/vmware-tanzu/velero/pkg/kuberesource"
+	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
+)
+
+func RelatedItemsForPVC(pvc *corev1api.PersistentVolumeClaim, log logrus.FieldLogger) []velero.ResourceIdentifier {
+	return []velero.ResourceIdentifier{
+		{
+			GroupResource: kuberesource.PersistentVolumes,
+			Name:          pvc.Spec.VolumeName,
+		},
+	}
+}

pkg/util/actionhelpers/rbac.go

@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package actions
+package actionhelpers
 
 import (
 	"context"
@@ -35,42 +35,42 @@ type ClusterRoleBindingLister interface {
 }
 
 // noopClusterRoleBindingLister exists to handle clusters where RBAC is disabled.
-type noopClusterRoleBindingLister struct {
+type NoopClusterRoleBindingLister struct {
 }
 
-func (noop noopClusterRoleBindingLister) List() ([]ClusterRoleBinding, error) {
+func (noop NoopClusterRoleBindingLister) List() ([]ClusterRoleBinding, error) {
 	return []ClusterRoleBinding{}, nil
 }
 
-type v1ClusterRoleBindingLister struct {
+type V1ClusterRoleBindingLister struct {
 	client rbacclient.ClusterRoleBindingInterface
 }
 
-func (v1 v1ClusterRoleBindingLister) List() ([]ClusterRoleBinding, error) {
+func (v1 V1ClusterRoleBindingLister) List() ([]ClusterRoleBinding, error) {
 	crbList, err := v1.client.List(context.TODO(), metav1.ListOptions{})
 	if err != nil {
 		return nil, errors.WithStack(err)
 	}
 	var crbs []ClusterRoleBinding
 	for _, crb := range crbList.Items {
-		crbs = append(crbs, v1ClusterRoleBinding{crb: crb})
+		crbs = append(crbs, V1ClusterRoleBinding{Crb: crb})
 	}
 
 	return crbs, nil
 }
 
-type v1beta1ClusterRoleBindingLister struct {
+type V1beta1ClusterRoleBindingLister struct {
 	client rbacbetaclient.ClusterRoleBindingInterface
 }
 
-func (v1beta1 v1beta1ClusterRoleBindingLister) List() ([]ClusterRoleBinding, error) {
+func (v1beta1 V1beta1ClusterRoleBindingLister) List() ([]ClusterRoleBinding, error) {
 	crbList, err := v1beta1.client.List(context.TODO(), metav1.ListOptions{})
 	if err != nil {
 		return nil, errors.WithStack(err)
 	}
 	var crbs []ClusterRoleBinding
 	for _, crb := range crbList.Items {
-		crbs = append(crbs, v1beta1ClusterRoleBinding{crb: crb})
+		crbs = append(crbs, V1beta1ClusterRoleBinding{Crb: crb})
 	}
 
 	return crbs, nil
@@ -81,9 +81,9 @@ func (v1beta1 v1beta1ClusterRoleBindingLister) List() ([]ClusterRoleBinding, err
 // Necessary so that callers to the ClusterRoleBindingLister interfaces don't need the kubernetes.Interface.
 func NewClusterRoleBindingListerMap(clientset kubernetes.Interface) map[string]ClusterRoleBindingLister {
 	return map[string]ClusterRoleBindingLister{
-		rbac.SchemeGroupVersion.Version:     v1ClusterRoleBindingLister{client: clientset.RbacV1().ClusterRoleBindings()},
-		rbacbeta.SchemeGroupVersion.Version: v1beta1ClusterRoleBindingLister{client: clientset.RbacV1beta1().ClusterRoleBindings()},
-		"":                                  noopClusterRoleBindingLister{},
+		rbac.SchemeGroupVersion.Version:     V1ClusterRoleBindingLister{client: clientset.RbacV1().ClusterRoleBindings()},
+		rbacbeta.SchemeGroupVersion.Version: V1beta1ClusterRoleBindingLister{client: clientset.RbacV1beta1().ClusterRoleBindings()},
+		"":                                  NoopClusterRoleBindingLister{},
 	}
 }
 
@@ -97,21 +97,21 @@ type ClusterRoleBinding interface {
 	RoleRefName() string
 }
 
-type v1ClusterRoleBinding struct {
-	crb rbac.ClusterRoleBinding
+type V1ClusterRoleBinding struct {
+	Crb rbac.ClusterRoleBinding
 }
 
-func (c v1ClusterRoleBinding) Name() string {
-	return c.crb.Name
+func (c V1ClusterRoleBinding) Name() string {
+	return c.Crb.Name
 }
 
-func (c v1ClusterRoleBinding) RoleRefName() string {
-	return c.crb.RoleRef.Name
+func (c V1ClusterRoleBinding) RoleRefName() string {
+	return c.Crb.RoleRef.Name
 }
 
-func (c v1ClusterRoleBinding) ServiceAccountSubjects(namespace string) []string {
+func (c V1ClusterRoleBinding) ServiceAccountSubjects(namespace string) []string {
 	var saSubjects []string
-	for _, s := range c.crb.Subjects {
+	for _, s := range c.Crb.Subjects {
 		if s.Kind == rbac.ServiceAccountKind && s.Namespace == namespace {
 			saSubjects = append(saSubjects, s.Name)
 		}
@@ -119,21 +119,21 @@ func (c v1ClusterRoleBinding) ServiceAccountSubjects(namespace string) []string
 	return saSubjects
 }
 
-type v1beta1ClusterRoleBinding struct {
-	crb rbacbeta.ClusterRoleBinding
+type V1beta1ClusterRoleBinding struct {
+	Crb rbacbeta.ClusterRoleBinding
 }
 
-func (c v1beta1ClusterRoleBinding) Name() string {
-	return c.crb.Name
+func (c V1beta1ClusterRoleBinding) Name() string {
+	return c.Crb.Name
 }
 
-func (c v1beta1ClusterRoleBinding) RoleRefName() string {
-	return c.crb.RoleRef.Name
+func (c V1beta1ClusterRoleBinding) RoleRefName() string {
+	return c.Crb.RoleRef.Name
 }
 
-func (c v1beta1ClusterRoleBinding) ServiceAccountSubjects(namespace string) []string {
+func (c V1beta1ClusterRoleBinding) ServiceAccountSubjects(namespace string) []string {
 	var saSubjects []string
-	for _, s := range c.crb.Subjects {
+	for _, s := range c.Crb.Subjects {
 		if s.Kind == rbac.ServiceAccountKind && s.Namespace == namespace {
 			saSubjects = append(saSubjects, s.Name)
 		}

pkg/util/actionhelpers/service_account_helper.go

@@ -0,0 +1,84 @@
+/*
+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 actionhelpers
+
+import (
+	"github.com/sirupsen/logrus"
+	rbac "k8s.io/api/rbac/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/sets"
+
+	velerodiscovery "github.com/vmware-tanzu/velero/pkg/discovery"
+	"github.com/vmware-tanzu/velero/pkg/kuberesource"
+	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
+)
+
+func ClusterRoleBindingsForAction(clusterRoleBindingListers map[string]ClusterRoleBindingLister, discoveryHelper velerodiscovery.Helper) ([]ClusterRoleBinding, error) {
+	// Look up the supported RBAC version
+	var supportedAPI metav1.GroupVersionForDiscovery
+	for _, ag := range discoveryHelper.APIGroups() {
+		if ag.Name == rbac.GroupName {
+			supportedAPI = ag.PreferredVersion
+			break
+		}
+	}
+
+	crbLister := clusterRoleBindingListers[supportedAPI.Version]
+
+	// This should be safe because the List call will return a 0-item slice
+	// if there's no matching API version.
+	return crbLister.List()
+}
+
+func RelatedItemsForServiceAccount(objectMeta metav1.Object, clusterRoleBindings []ClusterRoleBinding, log logrus.FieldLogger) []velero.ResourceIdentifier {
+	var (
+		namespace = objectMeta.GetNamespace()
+		name      = objectMeta.GetName()
+		bindings  = sets.NewString()
+		roles     = sets.NewString()
+	)
+
+	for _, crb := range clusterRoleBindings {
+		for _, s := range crb.ServiceAccountSubjects(namespace) {
+			if s == name {
+				log.Infof("Adding clusterrole %s and clusterrolebinding %s to relatedItems since serviceaccount %s/%s is a subject",
+					crb.RoleRefName(), crb.Name(), namespace, name)
+
+				bindings.Insert(crb.Name())
+				roles.Insert(crb.RoleRefName())
+				break
+			}
+		}
+	}
+
+	var relatedItems []velero.ResourceIdentifier
+	for binding := range bindings {
+		relatedItems = append(relatedItems, velero.ResourceIdentifier{
+			GroupResource: kuberesource.ClusterRoleBindings,
+			Name:          binding,
+		})
+	}
+
+	for role := range roles {
+		relatedItems = append(relatedItems, velero.ResourceIdentifier{
+			GroupResource: kuberesource.ClusterRoles,
+			Name:          role,
+		})
+	}
+
+	return relatedItems
+}