diff --git a/pkg/apis/ark/v1/restore.go b/pkg/apis/ark/v1/restore.go
index 43b649db9..17835966f 100644
--- a/pkg/apis/ark/v1/restore.go
+++ b/pkg/apis/ark/v1/restore.go
@@ -32,6 +32,14 @@ type RestoreSpec struct {
 	// included in the restore.
 	ExcludedNamespaces []string `json:"excludedNamespaces"`
 
+	// IncludedResources is a slice of resource names to include
+	// in the restore. If empty, all resources in the backup are included.
+	IncludedResources []string `json:"includedResources"`
+
+	// ExcludedResources is a slice of resource names that are not
+	// included in the restore.
+	ExcludedResources []string `json:"excludedResources"`
+
 	// NamespaceMapping is a map of source namespace names
 	// to target namespace names to restore into. Any source
 	// namespaces not included in the map will be restored into
diff --git a/pkg/cmd/cli/restore/create.go b/pkg/cmd/cli/restore/create.go
index 2a0df90d6..4badeab55 100644
--- a/pkg/cmd/cli/restore/create.go
+++ b/pkg/cmd/cli/restore/create.go
@@ -59,6 +59,8 @@ type CreateOptions struct {
 	Labels            flag.Map
 	IncludeNamespaces flag.StringArray
 	ExcludeNamespaces flag.StringArray
+	IncludeResources  flag.StringArray
+	ExcludeResources  flag.StringArray
 	NamespaceMappings flag.Map
 	Selector          flag.LabelSelector
 }
@@ -77,6 +79,8 @@ func (o *CreateOptions) BindFlags(flags *pflag.FlagSet) {
 	flags.Var(&o.ExcludeNamespaces, "exclude-namespaces", "namespaces to exclude from the restore")
 	flags.Var(&o.NamespaceMappings, "namespace-mappings", "namespace mappings from name in the backup to desired restored name in the form src1:dst1,src2:dst2,...")
 	flags.Var(&o.Labels, "labels", "labels to apply to the restore")
+	flags.Var(&o.IncludeResources, "include-resources", "resources to include in the restore, formatted as resource.group, such as storageclasses.storage.k8s.io (use '*' for all resources)")
+	flags.Var(&o.ExcludeResources, "exclude-resources", "resources to exclude from the restore, formatted as resource.group, such as storageclasses.storage.k8s.io")
 	flags.VarP(&o.Selector, "selector", "l", "only restore resources matching this label selector")
 	f := flags.VarPF(&o.RestoreVolumes, "restore-volumes", "", "whether to restore volumes from snapshots")
 	// this allows the user to just specify "--restore-volumes" as shorthand for "--restore-volumes=true"
@@ -117,6 +121,8 @@ func (o *CreateOptions) Run(c *cobra.Command, f client.Factory) error {
 			BackupName:         o.BackupName,
 			IncludedNamespaces: o.IncludeNamespaces,
 			ExcludedNamespaces: o.ExcludeNamespaces,
+			IncludedResources:  o.IncludeResources,
+			ExcludedResources:  o.ExcludeResources,
 			NamespaceMapping:   o.NamespaceMappings.Data(),
 			LabelSelector:      o.Selector.LabelSelector,
 			RestorePVs:         o.RestoreVolumes.Value,
diff --git a/pkg/controller/restore_controller.go b/pkg/controller/restore_controller.go
index b140de7fd..739b92c79 100644
--- a/pkg/controller/restore_controller.go
+++ b/pkg/controller/restore_controller.go
@@ -221,9 +221,13 @@ func (controller *restoreController) processRestore(key string) error {
 		return err
 	}
 
+	// defaulting
 	if len(restore.Spec.IncludedNamespaces) == 0 {
 		restore.Spec.IncludedNamespaces = []string{"*"}
 	}
+	if len(restore.Spec.IncludedResources) == 0 {
+		restore.Spec.IncludedResources = []string{"*"}
+	}
 
 	// validation
 	if restore.Status.ValidationErrors = controller.getValidationErrors(restore); len(restore.Status.ValidationErrors) > 0 {
@@ -284,6 +288,10 @@ func (controller *restoreController) getValidationErrors(itm *api.Restore) []str
 		validationErrors = append(validationErrors, fmt.Sprintf("Invalid included/excluded namespace lists: %v", err))
 	}
 
+	for _, err := range collections.ValidateIncludesExcludes(itm.Spec.IncludedResources, itm.Spec.ExcludedResources) {
+		validationErrors = append(validationErrors, fmt.Sprintf("Invalid included/excluded resource lists: %v", err))
+	}
+
 	if !controller.pvProviderExists && itm.Spec.RestorePVs != nil && *itm.Spec.RestorePVs {
 		validationErrors = append(validationErrors, "Server is not configured for PV snapshot restores")
 	}
diff --git a/pkg/controller/restore_controller_test.go b/pkg/controller/restore_controller_test.go
index da567a5e0..8b30991dc 100644
--- a/pkg/controller/restore_controller_test.go
+++ b/pkg/controller/restore_controller_test.go
@@ -119,7 +119,6 @@ func TestProcessRestore(t *testing.T) {
 		expectedRestoreUpdates      []*api.Restore
 		expectedRestorerCall        *api.Restore
 		backupServiceGetBackupError error
-		expectRestore               bool
 	}{
 		{
 			name:        "invalid key returns error",
@@ -148,37 +147,45 @@ func TestProcessRestore(t *testing.T) {
 		},
 		{
 			name:        "restore with both namespace in both includedNamespaces and excludedNamespaces fails validation",
-			restore:     NewTestRestore("foo", "bar", api.RestorePhaseNew).WithBackup("backup-1").WithIncludedNamespace("another-1").WithExcludedNamespace("another-1").Restore,
+			restore:     NewRestore("foo", "bar", "backup-1", "another-1", "*", api.RestorePhaseNew).WithExcludedNamespace("another-1").Restore,
 			backup:      NewTestBackup().WithName("backup-1").Backup,
 			expectedErr: false,
 			expectedRestoreUpdates: []*api.Restore{
-				NewTestRestore("foo", "bar", api.RestorePhaseFailedValidation).
-					WithBackup("backup-1").
-					WithIncludedNamespace("another-1").
-					WithExcludedNamespace("another-1").
-					WithValidationError("Invalid included/excluded namespace lists: excludes list cannot contain an item in the includes list: another-1").Restore,
+				NewRestore("foo", "bar", "backup-1", "another-1", "*", api.RestorePhaseFailedValidation).WithExcludedNamespace("another-1").
+					WithValidationError("Invalid included/excluded namespace lists: excludes list cannot contain an item in the includes list: another-1").
+					Restore,
+			},
+		},
+		{
+			name:        "restore with resource in both includedResources and excludedResources fails validation",
+			restore:     NewRestore("foo", "bar", "backup-1", "*", "a-resource", api.RestorePhaseNew).WithExcludedResource("a-resource").Restore,
+			backup:      NewTestBackup().WithName("backup-1").Backup,
+			expectedErr: false,
+			expectedRestoreUpdates: []*api.Restore{
+				NewRestore("foo", "bar", "backup-1", "*", "a-resource", api.RestorePhaseFailedValidation).WithExcludedResource("a-resource").
+					WithValidationError("Invalid included/excluded resource lists: excludes list cannot contain an item in the includes list: a-resource").
+					Restore,
 			},
 		},
 		{
 			name:        "new restore with empty backup name fails validation",
-			restore:     NewTestRestore("foo", "bar", api.RestorePhaseNew).WithIncludedNamespace("ns-1").Restore,
+			restore:     NewRestore("foo", "bar", "", "ns-1", "", api.RestorePhaseNew).Restore,
 			expectedErr: false,
 			expectedRestoreUpdates: []*api.Restore{
-				NewTestRestore("foo", "bar", api.RestorePhaseFailedValidation).
-					WithIncludedNamespace("ns-1").
-					WithValidationError("BackupName must be non-empty and correspond to the name of a backup in object storage.").Restore,
+				NewRestore("foo", "bar", "", "ns-1", "*", api.RestorePhaseFailedValidation).
+					WithValidationError("BackupName must be non-empty and correspond to the name of a backup in object storage.").
+					Restore,
 			},
 		},
+
 		{
 			name:                        "restore with non-existent backup name fails",
 			restore:                     NewTestRestore("foo", "bar", api.RestorePhaseNew).WithBackup("backup-1").WithIncludedNamespace("ns-1").Restore,
 			expectedErr:                 false,
 			backupServiceGetBackupError: errors.New("no backup here"),
 			expectedRestoreUpdates: []*api.Restore{
-				NewTestRestore("foo", "bar", api.RestorePhaseInProgress).WithBackup("backup-1").WithIncludedNamespace("ns-1").Restore,
-				NewTestRestore("foo", "bar", api.RestorePhaseCompleted).
-					WithBackup("backup-1").
-					WithIncludedNamespace("ns-1").
+				NewRestore("foo", "bar", "backup-1", "ns-1", "*", api.RestorePhaseInProgress).Restore,
+				NewRestore("foo", "bar", "backup-1", "ns-1", "*", api.RestorePhaseCompleted).
 					WithErrors(api.RestoreResult{
 						Cluster: []string{"no backup here"},
 					}).
@@ -187,69 +194,66 @@ func TestProcessRestore(t *testing.T) {
 		},
 		{
 			name:          "restorer throwing an error causes the restore to fail",
-			restore:       NewTestRestore("foo", "bar", api.RestorePhaseNew).WithBackup("backup-1").WithIncludedNamespace("ns-1").Restore,
+			restore:       NewRestore("foo", "bar", "backup-1", "ns-1", "", api.RestorePhaseNew).Restore,
 			backup:        NewTestBackup().WithName("backup-1").Backup,
-			expectRestore: true,
 			restorerError: errors.New("blarg"),
 			expectedErr:   false,
 			expectedRestoreUpdates: []*api.Restore{
-				NewTestRestore("foo", "bar", api.RestorePhaseInProgress).WithBackup("backup-1").WithIncludedNamespace("ns-1").Restore,
-				NewTestRestore("foo", "bar", api.RestorePhaseCompleted).
-					WithBackup("backup-1").
-					WithIncludedNamespace("ns-1").
+				NewRestore("foo", "bar", "backup-1", "ns-1", "*", api.RestorePhaseInProgress).Restore,
+				NewRestore("foo", "bar", "backup-1", "ns-1", "*", api.RestorePhaseCompleted).
 					WithErrors(api.RestoreResult{
 						Namespaces: map[string][]string{
 							"ns-1": {"blarg"},
 						},
-					}).Restore,
+					}).
+					Restore,
 			},
-			expectedRestorerCall: NewTestRestore("foo", "bar", api.RestorePhaseInProgress).WithBackup("backup-1").WithIncludedNamespace("ns-1").Restore,
+			expectedRestorerCall: NewRestore("foo", "bar", "backup-1", "ns-1", "*", api.RestorePhaseInProgress).Restore,
 		},
 		{
-			name:          "valid restore gets executed",
-			restore:       NewTestRestore("foo", "bar", api.RestorePhaseNew).WithBackup("backup-1").WithIncludedNamespace("ns-1").Restore,
-			backup:        NewTestBackup().WithName("backup-1").Backup,
-			expectRestore: true,
-			expectedErr:   false,
-			expectedRestoreUpdates: []*api.Restore{
-				NewTestRestore("foo", "bar", api.RestorePhaseInProgress).WithBackup("backup-1").WithIncludedNamespace("ns-1").Restore,
-				NewTestRestore("foo", "bar", api.RestorePhaseCompleted).WithBackup("backup-1").WithIncludedNamespace("ns-1").Restore,
-			},
-			expectedRestorerCall: NewTestRestore("foo", "bar", api.RestorePhaseInProgress).WithBackup("backup-1").WithIncludedNamespace("ns-1").Restore,
-		},
-		{
-			name:          "restore with no restorable namespaces gets defaulted to *",
-			restore:       NewTestRestore("foo", "bar", api.RestorePhaseNew).WithBackup("backup-1").Restore,
-			backup:        NewTestBackup().WithName("backup-1").Backup,
-			expectRestore: true,
-			expectedErr:   false,
-			expectedRestoreUpdates: []*api.Restore{
-				NewTestRestore("foo", "bar", api.RestorePhaseInProgress).WithBackup("backup-1").WithIncludedNamespace("*").Restore,
-				NewTestRestore("foo", "bar", api.RestorePhaseCompleted).WithBackup("backup-1").WithIncludedNamespace("*").Restore,
-			},
-			expectedRestorerCall: NewTestRestore("foo", "bar", api.RestorePhaseInProgress).WithBackup("backup-1").WithIncludedNamespace("*").Restore,
-		},
-		{
-			name:                  "valid restore with RestorePVs=true gets executed when allowRestoreSnapshots=true",
-			restore:               NewTestRestore("foo", "bar", api.RestorePhaseNew).WithBackup("backup-1").WithIncludedNamespace("ns-1").WithRestorePVs(true).Restore,
-			backup:                NewTestBackup().WithName("backup-1").Backup,
-			expectRestore:         true,
-			allowRestoreSnapshots: true,
-			expectedErr:           false,
-			expectedRestoreUpdates: []*api.Restore{
-				NewTestRestore("foo", "bar", api.RestorePhaseInProgress).WithBackup("backup-1").WithIncludedNamespace("ns-1").WithRestorePVs(true).Restore,
-				NewTestRestore("foo", "bar", api.RestorePhaseCompleted).WithBackup("backup-1").WithIncludedNamespace("ns-1").WithRestorePVs(true).Restore,
-			},
-			expectedRestorerCall: NewTestRestore("foo", "bar", api.RestorePhaseInProgress).WithBackup("backup-1").WithIncludedNamespace("ns-1").WithRestorePVs(true).Restore,
-		},
-		{
-			name:        "restore with RestorePVs=true fails validation when allowRestoreSnapshots=false",
-			restore:     NewTestRestore("foo", "bar", api.RestorePhaseNew).WithBackup("backup-1").WithIncludedNamespace("ns-1").WithRestorePVs(true).Restore,
+			name:        "valid restore gets executed",
+			restore:     NewRestore("foo", "bar", "backup-1", "ns-1", "", api.RestorePhaseNew).Restore,
 			backup:      NewTestBackup().WithName("backup-1").Backup,
 			expectedErr: false,
 			expectedRestoreUpdates: []*api.Restore{
-				NewTestRestore("foo", "bar", api.RestorePhaseFailedValidation).WithBackup("backup-1").WithIncludedNamespace("ns-1").WithRestorePVs(true).
-					WithValidationError("Server is not configured for PV snapshot restores").Restore,
+				NewRestore("foo", "bar", "backup-1", "ns-1", "*", api.RestorePhaseInProgress).Restore,
+				NewRestore("foo", "bar", "backup-1", "ns-1", "*", api.RestorePhaseCompleted).Restore,
+			},
+			expectedRestorerCall: NewRestore("foo", "bar", "backup-1", "ns-1", "*", api.RestorePhaseInProgress).Restore,
+		},
+		{
+			name:        "restore with no restorable namespaces gets defaulted to *",
+			restore:     NewRestore("foo", "bar", "backup-1", "", "", api.RestorePhaseNew).Restore,
+			backup:      NewTestBackup().WithName("backup-1").Backup,
+			expectedErr: false,
+			expectedRestoreUpdates: []*api.Restore{
+				NewRestore("foo", "bar", "backup-1", "*", "*", api.RestorePhaseInProgress).Restore,
+				NewRestore("foo", "bar", "backup-1", "*", "*", api.RestorePhaseCompleted).Restore,
+			},
+			expectedRestorerCall: NewRestore("foo", "bar", "backup-1", "*", "*", api.RestorePhaseInProgress).Restore,
+		},
+		{
+			name:                  "valid restore with RestorePVs=true gets executed when allowRestoreSnapshots=true",
+			restore:               NewRestore("foo", "bar", "backup-1", "ns-1", "", api.RestorePhaseNew).WithRestorePVs(true).Restore,
+			backup:                NewTestBackup().WithName("backup-1").Backup,
+			allowRestoreSnapshots: true,
+			expectedErr:           false,
+			expectedRestoreUpdates: []*api.Restore{
+				NewRestore("foo", "bar", "backup-1", "ns-1", "*", api.RestorePhaseInProgress).WithRestorePVs(true).Restore,
+				NewRestore("foo", "bar", "backup-1", "ns-1", "*", api.RestorePhaseCompleted).WithRestorePVs(true).Restore,
+			},
+			expectedRestorerCall: NewRestore("foo", "bar", "backup-1", "ns-1", "*", api.RestorePhaseInProgress).WithRestorePVs(true).Restore,
+		},
+		{
+			name:        "restore with RestorePVs=true fails validation when allowRestoreSnapshots=false",
+			restore:     NewRestore("foo", "bar", "backup-1", "ns-1", "", api.RestorePhaseNew).WithRestorePVs(true).Restore,
+			backup:      NewTestBackup().WithName("backup-1").Backup,
+			expectedErr: false,
+			expectedRestoreUpdates: []*api.Restore{
+				NewRestore("foo", "bar", "backup-1", "ns-1", "*", api.RestorePhaseFailedValidation).
+					WithRestorePVs(true).
+					WithValidationError("Server is not configured for PV snapshot restores").
+					Restore,
 			},
 		},
 	}
@@ -299,7 +303,7 @@ func TestProcessRestore(t *testing.T) {
 			if test.restorerError != nil {
 				errors.Namespaces = map[string][]string{"ns-1": {test.restorerError.Error()}}
 			}
-			if test.expectRestore {
+			if test.expectedRestorerCall != nil {
 				downloadedBackup := ioutil.NopCloser(bytes.NewReader([]byte("hello world")))
 				backupSvc.On("DownloadBackup", mock.Anything, mock.Anything).Return(downloadedBackup, nil)
 				restorer.On("Restore", mock.Anything, mock.Anything, mock.Anything).Return(warnings, errors)
@@ -356,6 +360,20 @@ func TestProcessRestore(t *testing.T) {
 	}
 }
 
+func NewRestore(ns, name, backup, includeNS, includeResource string, phase api.RestorePhase) *TestRestore {
+	restore := NewTestRestore(ns, name, phase).WithBackup(backup)
+
+	if includeNS != "" {
+		restore = restore.WithIncludedNamespace(includeNS)
+	}
+
+	if includeResource != "" {
+		restore = restore.WithIncludedResource(includeResource)
+	}
+
+	return restore
+}
+
 type fakeRestorer struct {
 	mock.Mock
 	calledWithArg api.Restore
diff --git a/pkg/restore/restore.go b/pkg/restore/restore.go
index aec0c56ac..617c6b172 100644
--- a/pkg/restore/restore.go
+++ b/pkg/restore/restore.go
@@ -30,7 +30,6 @@ import (
 	"github.com/golang/glog"
 
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
-	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/labels"
@@ -45,8 +44,8 @@ import (
 	"github.com/heptio/ark/pkg/discovery"
 	arkv1client "github.com/heptio/ark/pkg/generated/clientset/typed/ark/v1"
 	"github.com/heptio/ark/pkg/restore/restorers"
-	"github.com/heptio/ark/pkg/util/kube"
 	"github.com/heptio/ark/pkg/util/collections"
+	"github.com/heptio/ark/pkg/util/kube"
 )
 
 // Restorer knows how to restore a backup.
@@ -75,7 +74,7 @@ type kubernetesRestorer struct {
 // prioritizeResources takes a list of pre-prioritized resources and a full list of resources to restore,
 // and returns an ordered list of GroupResource-resolved resources in the order that they should be
 // restored.
-func prioritizeResources(mapper meta.RESTMapper, priorities []string, resources []*metav1.APIResourceList) ([]schema.GroupResource, error) {
+func prioritizeResources(helper discovery.Helper, priorities []string, includedResources *collections.IncludesExcludes) ([]schema.GroupResource, error) {
 	var ret []schema.GroupResource
 
 	// set keeps track of resolved GroupResource names
@@ -83,19 +82,23 @@ func prioritizeResources(mapper meta.RESTMapper, priorities []string, resources
 
 	// start by resolving priorities into GroupResources and adding them to ret
 	for _, r := range priorities {
-		gr := schema.ParseGroupResource(r)
-		gvr, err := mapper.ResourceFor(gr.WithVersion(""))
+		gr, err := helper.ResolveGroupResource(r)
 		if err != nil {
 			return nil, err
 		}
-		gr = gvr.GroupResource()
+
+		if !includedResources.ShouldInclude(gr.String()) {
+			glog.Infof("Not including resource %v", gr)
+			continue
+		}
+
 		ret = append(ret, gr)
 		set.Insert(gr.String())
 	}
 
 	// go through everything we got from discovery and add anything not in "set" to byName
 	var byName []schema.GroupResource
-	for _, resourceGroup := range resources {
+	for _, resourceGroup := range helper.Resources() {
 		// will be something like storage.k8s.io/v1
 		groupVersion, err := schema.ParseGroupVersion(resourceGroup.GroupVersion)
 		if err != nil {
@@ -104,6 +107,12 @@ func prioritizeResources(mapper meta.RESTMapper, priorities []string, resources
 
 		for _, resource := range resourceGroup.APIResources {
 			gr := groupVersion.WithResource(resource.Name).GroupResource()
+
+			if !includedResources.ShouldInclude(gr.String()) {
+				glog.Infof("Not including resource %v", gr)
+				continue
+			}
+
 			if !set.Has(gr.String()) {
 				byName = append(byName, gr)
 			}
@@ -131,14 +140,13 @@ func NewKubernetesRestorer(
 	backupClient arkv1client.BackupsGetter,
 	namespaceClient corev1.NamespaceInterface,
 ) (Restorer, error) {
-	mapper := discoveryHelper.Mapper()
 	r := make(map[schema.GroupResource]restorers.ResourceRestorer)
 	for gr, restorer := range customRestorers {
-		gvr, err := mapper.ResourceFor(schema.ParseGroupResource(gr).WithVersion(""))
+		resolved, err := discoveryHelper.ResolveGroupResource(gr)
 		if err != nil {
 			return nil, err
 		}
-		r[gvr.GroupResource()] = restorer
+		r[resolved] = restorer
 	}
 
 	return &kubernetesRestorer{
@@ -171,7 +179,21 @@ func (kr *kubernetesRestorer) Restore(restore *api.Restore, backup *api.Backup,
 		return api.RestoreResult{}, api.RestoreResult{Ark: []string{err.Error()}}
 	}
 
-	prioritizedResources, err := prioritizeResources(kr.discoveryHelper.Mapper(), kr.resourcePriorities, kr.discoveryHelper.Resources())
+	// get resource includes-excludes
+	resourceIncludesExcludes := collections.GenerateIncludesExcludes(
+		restore.Spec.IncludedResources,
+		restore.Spec.ExcludedResources,
+		func(item string) (string, error) {
+			gr, err := kr.discoveryHelper.ResolveGroupResource(item)
+			if err != nil {
+				return "", err
+			}
+
+			return gr.String(), nil
+		},
+	)
+
+	prioritizedResources, err := prioritizeResources(kr.discoveryHelper, kr.resourcePriorities, resourceIncludesExcludes)
 	if err != nil {
 		return api.RestoreResult{}, api.RestoreResult{Ark: []string{err.Error()}}
 	}
@@ -389,7 +411,7 @@ func (kr *kubernetesRestorer) restoreResourceForNamespace(
 		if restorer == nil {
 			// initialize client & restorer for this Resource. we need
 			// metadata from an object to do this.
-			glog.Infof("Getting client for %s", obj.GroupVersionKind().String())
+			glog.Infof("Getting client for %v", obj.GroupVersionKind())
 
 			resource := metav1.APIResource{
 				Namespaced: len(namespace) > 0,
@@ -399,22 +421,22 @@ func (kr *kubernetesRestorer) restoreResourceForNamespace(
 			var err error
 			resourceClient, err = kr.dynamicFactory.ClientForGroupVersionKind(obj.GroupVersionKind(), resource, namespace)
 			if err != nil {
-				addArkError(&errors, fmt.Errorf("error getting resource client for namespace %q, resource %q: %v", namespace, groupResource.String(), err))
+				addArkError(&errors, fmt.Errorf("error getting resource client for namespace %q, resource %q: %v", namespace, groupResource, err))
 				return warnings, errors
 			}
 
 			restorer = kr.restorers[groupResource]
 			if restorer == nil {
-				glog.Infof("Using default restorer for %s", groupResource.String())
+				glog.Infof("Using default restorer for %v", groupResource)
 				restorer = restorers.NewBasicRestorer(true)
 			} else {
-				glog.Infof("Using custom restorer for %s", groupResource.String())
+				glog.Infof("Using custom restorer for %v", groupResource)
 			}
 
 			if restorer.Wait() {
 				itmWatch, err := resourceClient.Watch(metav1.ListOptions{})
 				if err != nil {
-					addArkError(&errors, fmt.Errorf("error watching for namespace %q, resource %q: %v", namespace, groupResource.String(), err))
+					addArkError(&errors, fmt.Errorf("error watching for namespace %q, resource %q: %v", namespace, groupResource, err))
 					return warnings, errors
 				}
 				watchChan := itmWatch.ResultChan()
@@ -473,7 +495,7 @@ func (kr *kubernetesRestorer) restoreResourceForNamespace(
 
 	if waiter != nil {
 		if err := waiter.Wait(); err != nil {
-			addArkError(&errors, fmt.Errorf("error waiting for all %s resources to be created in namespace %s: %v", groupResource.String(), namespace, err))
+			addArkError(&errors, fmt.Errorf("error waiting for all %v resources to be created in namespace %s: %v", groupResource, namespace, err))
 		}
 	}
 
diff --git a/pkg/restore/restore_test.go b/pkg/restore/restore_test.go
index 879732145..001053db7 100644
--- a/pkg/restore/restore_test.go
+++ b/pkg/restore/restore_test.go
@@ -24,6 +24,7 @@ import (
 
 	"github.com/spf13/afero"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
@@ -40,39 +41,71 @@ import (
 )
 
 func TestPrioritizeResources(t *testing.T) {
-	mapper := &FakeMapper{AutoReturnResource: true}
-	priorities := []string{"namespaces", "configmaps", "pods"}
-	resources := []*metav1.APIResourceList{
+	tests := []struct {
+		name         string
+		apiResources map[string][]string
+		priorities   []string
+		includes     []string
+		excludes     []string
+		expected     []string
+	}{
 		{
-			GroupVersion: "v1",
-			APIResources: []metav1.APIResource{
-				{Name: "aaa"},
-				{Name: "bbb"},
-				{Name: "configmaps"},
-				{Name: "ddd"},
-				{Name: "namespaces"},
-				{Name: "ooo"},
-				{Name: "pods"},
-				{Name: "sss"},
+			name: "priorities & ordering are correctly applied",
+			apiResources: map[string][]string{
+				"v1": []string{"aaa", "bbb", "configmaps", "ddd", "namespaces", "ooo", "pods", "sss"},
 			},
+			priorities: []string{"namespaces", "configmaps", "pods"},
+			includes:   []string{"*"},
+			expected:   []string{"namespaces", "configmaps", "pods", "aaa", "bbb", "ddd", "ooo", "sss"},
+		},
+		{
+			name: "includes are correctly applied",
+			apiResources: map[string][]string{
+				"v1": []string{"aaa", "bbb", "configmaps", "ddd", "namespaces", "ooo", "pods", "sss"},
+			},
+			priorities: []string{"namespaces", "configmaps", "pods"},
+			includes:   []string{"namespaces", "aaa", "sss"},
+			expected:   []string{"namespaces", "aaa", "sss"},
+		},
+		{
+			name: "excludes are correctly applied",
+			apiResources: map[string][]string{
+				"v1": []string{"aaa", "bbb", "configmaps", "ddd", "namespaces", "ooo", "pods", "sss"},
+			},
+			priorities: []string{"namespaces", "configmaps", "pods"},
+			includes:   []string{"*"},
+			excludes:   []string{"ooo", "pods"},
+			expected:   []string{"namespaces", "configmaps", "aaa", "bbb", "ddd", "sss"},
 		},
 	}
 
-	result, err := prioritizeResources(mapper, priorities, resources)
-	if err != nil {
-		t.Fatalf("unexpected error: %v", err)
-	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			helper := &FakeDiscoveryHelper{RESTMapper: &FakeMapper{AutoReturnResource: true}}
 
-	expected := []string{"namespaces", "configmaps", "pods", "aaa", "bbb", "ddd", "ooo", "sss"}
-	for i := range result {
-		if len(expected) < i+1 {
-			t.Errorf("result is too small: %v", result)
-			break
-		}
+			for gv, resources := range test.apiResources {
+				resourceList := &metav1.APIResourceList{GroupVersion: gv}
+				for _, resource := range resources {
+					resourceList.APIResources = append(resourceList.APIResources, metav1.APIResource{Name: resource})
+				}
+				helper.ResourceList = append(helper.ResourceList, resourceList)
+			}
 
-		if e, a := expected[i], result[i].Resource; e != a {
-			t.Errorf("index %d, expected %s, got %s", i, e, a)
-		}
+			includesExcludes := collections.NewIncludesExcludes().Includes(test.includes...).Excludes(test.excludes...)
+
+			result, err := prioritizeResources(helper, test.priorities, includesExcludes)
+			if err != nil {
+				t.Fatalf("unexpected error: %v", err)
+			}
+
+			require.Equal(t, len(test.expected), len(result))
+
+			for i := range result {
+				if e, a := test.expected[i], result[i].Resource; e != a {
+					t.Errorf("index %d, expected %s, got %s", i, e, a)
+				}
+			}
+		})
 	}
 }
 
diff --git a/pkg/util/collections/includes_excludes.go b/pkg/util/collections/includes_excludes.go
index 00cd48eb1..71d00a2bc 100644
--- a/pkg/util/collections/includes_excludes.go
+++ b/pkg/util/collections/includes_excludes.go
@@ -76,6 +76,8 @@ func (ie *IncludesExcludes) ShouldInclude(s string) bool {
 	return ie.includes.Has("*") || ie.includes.Has(s)
 }
 
+// ValidateIncludesExcludes checks provided lists of included and excluded
+// items to ensure they are a valid set of IncludesExcludes data.
 func ValidateIncludesExcludes(includesList, excludesList []string) []error {
 	// TODO we should not allow an IncludesExcludes object to be created that
 	// does not meet these criteria. Do a more significant refactoring to embed
@@ -100,20 +102,24 @@ func ValidateIncludesExcludes(includesList, excludesList []string) []error {
 
 	for _, itm := range excludes.List() {
 		if includes.Has(itm) {
-			errs = append(errs, errors.New(fmt.Sprintf("excludes list cannot contain an item in the includes list: %v", itm)))
+			errs = append(errs, fmt.Errorf("excludes list cannot contain an item in the includes list: %v", itm))
 		}
 	}
 
 	return errs
 }
 
-func GenerateIncludesExcludes(includes []string, excludes []string, mapFunc func(string) (string, error)) *IncludesExcludes {
+// GenerateIncludesExcludes constructs an IncludesExcludes struct by taking the provided
+// include/exclude slices, applying the specified mapping function to each item in them,
+// and adding the output of the function to the new struct. If the mapping function returns
+// an error for an item, it is omitted from the result.
+func GenerateIncludesExcludes(includes, excludes []string, mapFunc func(string) (string, error)) *IncludesExcludes {
 	res := NewIncludesExcludes()
 
 	for _, item := range includes {
 		if item == "*" {
 			res.Includes(item)
-			return res
+			continue
 		}
 
 		key, err := mapFunc(item)
diff --git a/pkg/util/test/test_restore.go b/pkg/util/test/test_restore.go
index 57ca9a094..3dfb3d99c 100644
--- a/pkg/util/test/test_restore.go
+++ b/pkg/util/test/test_restore.go
@@ -82,3 +82,13 @@ func (r *TestRestore) WithMappedNamespace(from string, to string) *TestRestore {
 	r.Spec.NamespaceMapping[from] = to
 	return r
 }
+
+func (r *TestRestore) WithIncludedResource(resource string) *TestRestore {
+	r.Spec.IncludedResources = append(r.Spec.IncludedResources, resource)
+	return r
+}
+
+func (r *TestRestore) WithExcludedResource(resource string) *TestRestore {
+	r.Spec.ExcludedResources = append(r.Spec.ExcludedResources, resource)
+	return r
+}