Merge branch 'main' into rm-improvement

pkg/apis/velero/v1/labels_annotations.go

@@ -89,6 +89,14 @@ const (
 
 	// ResourceUsageLabel is the label key to explain the Velero resource usage.
 	ResourceUsageLabel = "velero.io/resource-usage"
+
+	// VolumesToBackupAnnotation is the annotation on a pod whose mounted volumes
+	// need to be backed up using pod volume backup.
+	VolumesToBackupAnnotation = "backup.velero.io/backup-volumes"
+
+	// VolumesToExcludeAnnotation is the annotation on a pod whose mounted volumes
+	// should be excluded from pod volume backup.
+	VolumesToExcludeAnnotation = "backup.velero.io/backup-volumes-excludes"
 )
 
 type AsyncOperationIDPrefix string

pkg/apis/velero/v1/pod_volume_backup_types.go

@@ -114,7 +114,6 @@ type PodVolumeBackupStatus struct {
 // +kubebuilder:printcolumn:name="Namespace",type="string",JSONPath=".spec.pod.namespace",description="Namespace of the pod containing the volume to be backed up"
 // +kubebuilder:printcolumn:name="Pod",type="string",JSONPath=".spec.pod.name",description="Name of the pod containing the volume to be backed up"
 // +kubebuilder:printcolumn:name="Volume",type="string",JSONPath=".spec.volume",description="Name of the volume to be backed up"
-// +kubebuilder:printcolumn:name="Repository ID",type="string",JSONPath=".spec.repoIdentifier",description="Backup repository identifier for this backup"
 // +kubebuilder:printcolumn:name="Uploader Type",type="string",JSONPath=".spec.uploaderType",description="The type of the uploader to handle data transfer"
 // +kubebuilder:printcolumn:name="Storage Location",type="string",JSONPath=".spec.backupStorageLocation",description="Name of the Backup Storage Location where this backup should be stored"
 // +kubebuilder:printcolumn:name="Age",type="date",JSONPath=".metadata.creationTimestamp"

pkg/apis/velero/v2alpha1/data_download_types.go

@@ -56,7 +56,7 @@ type DataDownloadSpec struct {
 	OperationTimeout metav1.Duration `json:"operationTimeout"`
 }
 
-// TargetPVCSpec is the specification for a target PVC.
+// TargetVolumeSpec is the specification for a target PVC.
 type TargetVolumeSpec struct {
 	// PVC is the name of the target PVC that is created by Velero restore
 	PVC string `json:"pvc"`

pkg/apis/velero/v2alpha1/zz_generated.deepcopy.go

@@ -1,34 +1,17 @@
 //go:build !ignore_autogenerated
 // +build !ignore_autogenerated
 
-/*
-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.
-*/
-
-// Code generated by deepcopy-gen. DO NOT EDIT.
+// Code generated by controller-gen. DO NOT EDIT.
 
 package v2alpha1
 
 import (
-	runtime "k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime"
 )
 
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *CSISnapshotSpec) DeepCopyInto(out *CSISnapshotSpec) {
 	*out = *in
-	return
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new CSISnapshotSpec.
@@ -48,7 +31,6 @@ func (in *DataDownload) DeepCopyInto(out *DataDownload) {
 	in.ObjectMeta.DeepCopyInto(&out.ObjectMeta)
 	in.Spec.DeepCopyInto(&out.Spec)
 	in.Status.DeepCopyInto(&out.Status)
-	return
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new DataDownload.
@@ -81,7 +63,6 @@ func (in *DataDownloadList) DeepCopyInto(out *DataDownloadList) {
 			(*in)[i].DeepCopyInto(&(*out)[i])
 		}
 	}
-	return
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new DataDownloadList.
@@ -114,7 +95,6 @@ func (in *DataDownloadSpec) DeepCopyInto(out *DataDownloadSpec) {
 		}
 	}
 	out.OperationTimeout = in.OperationTimeout
-	return
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new DataDownloadSpec.
@@ -139,7 +119,6 @@ func (in *DataDownloadStatus) DeepCopyInto(out *DataDownloadStatus) {
 		*out = (*in).DeepCopy()
 	}
 	out.Progress = in.Progress
-	return
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new DataDownloadStatus.
@@ -159,7 +138,6 @@ func (in *DataUpload) DeepCopyInto(out *DataUpload) {
 	in.ObjectMeta.DeepCopyInto(&out.ObjectMeta)
 	in.Spec.DeepCopyInto(&out.Spec)
 	in.Status.DeepCopyInto(&out.Status)
-	return
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new DataUpload.
@@ -192,7 +170,6 @@ func (in *DataUploadList) DeepCopyInto(out *DataUploadList) {
 			(*in)[i].DeepCopyInto(&(*out)[i])
 		}
 	}
-	return
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new DataUploadList.
@@ -227,7 +204,6 @@ func (in *DataUploadResult) DeepCopyInto(out *DataUploadResult) {
 			}
 		}
 	}
-	return
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new DataUploadResult.
@@ -260,7 +236,6 @@ func (in *DataUploadSpec) DeepCopyInto(out *DataUploadSpec) {
 		}
 	}
 	out.OperationTimeout = in.OperationTimeout
-	return
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new DataUploadSpec.
@@ -296,7 +271,6 @@ func (in *DataUploadStatus) DeepCopyInto(out *DataUploadStatus) {
 		*out = (*in).DeepCopy()
 	}
 	out.Progress = in.Progress
-	return
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new DataUploadStatus.
@@ -312,7 +286,6 @@ func (in *DataUploadStatus) DeepCopy() *DataUploadStatus {
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *TargetVolumeSpec) DeepCopyInto(out *TargetVolumeSpec) {
 	*out = *in
-	return
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new TargetVolumeSpec.

pkg/backup/item_backupper.go

@@ -52,6 +52,7 @@ import (
 	vsv1 "github.com/vmware-tanzu/velero/pkg/plugin/velero/volumesnapshotter/v1"
 	"github.com/vmware-tanzu/velero/pkg/podvolume"
 	"github.com/vmware-tanzu/velero/pkg/util/boolptr"
+	pdvolumeutil "github.com/vmware-tanzu/velero/pkg/util/podvolume"
 	"github.com/vmware-tanzu/velero/pkg/volume"
 )
 
@@ -200,7 +201,7 @@ func (ib *itemBackupper) backupItemInternal(logger logrus.FieldLogger, obj runti
 			// Get the list of volumes to back up using pod volume backup from the pod's annotations. Remove from this list
 			// any volumes that use a PVC that we've already backed up (this would be in a read-write-many scenario,
 			// where it's been backed up from another pod), since we don't need >1 backup per PVC.
-			includedVolumes, optedOutVolumes := podvolume.GetVolumesByPod(pod, boolptr.IsSetToTrue(ib.backupRequest.Spec.DefaultVolumesToFsBackup))
+			includedVolumes, optedOutVolumes := pdvolumeutil.GetVolumesByPod(pod, boolptr.IsSetToTrue(ib.backupRequest.Spec.DefaultVolumesToFsBackup))
 			for _, volume := range includedVolumes {
 				// track the volumes that are PVCs using the PVC snapshot tracker, so that when we backup PVCs/PVs
 				// via an item action in the next step, we don't snapshot PVs that will have their data backed up

pkg/builder/persistent_volume_builder.go

@@ -95,6 +95,12 @@ func (b *PersistentVolumeBuilder) StorageClass(name string) *PersistentVolumeBui
 	return b
 }
 
+// VolumeMode sets the PersistentVolume's volume mode.
+func (b *PersistentVolumeBuilder) VolumeMode(volMode corev1api.PersistentVolumeMode) *PersistentVolumeBuilder {
+	b.object.Spec.VolumeMode = &volMode
+	return b
+}
+
 // NodeAffinityRequired sets the PersistentVolume's NodeAffinity Requirement.
 func (b *PersistentVolumeBuilder) NodeAffinityRequired(req *corev1api.NodeSelector) *PersistentVolumeBuilder {
 	b.object.Spec.NodeAffinity = &corev1api.VolumeNodeAffinity{

pkg/cmd/cli/backup/create.go

@@ -95,6 +95,7 @@ type CreateOptions struct {
 	ExcludeNamespaceScopedResources flag.StringArray
 	Labels                          flag.Map
 	Selector                        flag.LabelSelector
+	OrSelector                      flag.OrLabelSelector
 	IncludeClusterResources         flag.OptionalBool
 	Wait                            bool
 	StorageLocation                 string
@@ -130,6 +131,7 @@ func (o *CreateOptions) BindFlags(flags *pflag.FlagSet) {
 	flags.StringVar(&o.StorageLocation, "storage-location", "", "Location in which to store the backup.")
 	flags.StringSliceVar(&o.SnapshotLocations, "volume-snapshot-locations", o.SnapshotLocations, "List of locations (at most one per provider) where volume snapshots should be stored.")
 	flags.VarP(&o.Selector, "selector", "l", "Only back up resources matching this label selector.")
+	flags.Var(&o.OrSelector, "or-selector", "Backup resources matching at least one of the label selector from the list. Label selectors should be separated by ' or '. For example, foo=bar or app=nginx")
 	flags.StringVar(&o.OrderedResources, "ordered-resources", "", "Mapping Kinds to an ordered list of specific resources of that Kind.  Resource names are separated by commas and their names are in format 'namespace/resourcename'. For cluster scope resource, simply use resource name. Key-value pairs in the mapping are separated by semi-colon.  Example: 'pods=ns1/pod1,ns1/pod2;persistentvolumeclaims=ns1/pvc4,ns1/pvc8'.  Optional.")
 	flags.DurationVar(&o.CSISnapshotTimeout, "csi-snapshot-timeout", o.CSISnapshotTimeout, "How long to wait for CSI snapshot creation before timeout.")
 	flags.DurationVar(&o.ItemOperationTimeout, "item-operation-timeout", o.ItemOperationTimeout, "How long to wait for async plugin operations before timeout.")
@@ -168,9 +170,8 @@ func (o *CreateOptions) Validate(c *cobra.Command, args []string, f client.Facto
 		return err
 	}
 
-	client, err := f.KubebuilderWatchClient()
-	if err != nil {
-		return err
+	if o.Selector.LabelSelector != nil && o.OrSelector.OrLabelSelectors != nil {
+		return fmt.Errorf("either a 'selector' or an 'or-selector' can be specified, but not both")
 	}
 
 	// Ensure that unless FromSchedule is set, args contains a backup name
@@ -191,7 +192,7 @@ func (o *CreateOptions) Validate(c *cobra.Command, args []string, f client.Facto
 
 	if o.StorageLocation != "" {
 		location := &velerov1api.BackupStorageLocation{}
-		if err := client.Get(context.Background(), kbclient.ObjectKey{
+		if err := o.client.Get(context.Background(), kbclient.ObjectKey{
 			Namespace: f.Namespace(),
 			Name:      o.StorageLocation,
 		}, location); err != nil {
@@ -201,7 +202,7 @@ func (o *CreateOptions) Validate(c *cobra.Command, args []string, f client.Facto
 
 	for _, loc := range o.SnapshotLocations {
 		snapshotLocation := new(velerov1api.VolumeSnapshotLocation)
-		if err := o.client.Get(context.TODO(), kbclient.ObjectKey{Namespace: f.Namespace(), Name: loc}, snapshotLocation); err != nil {
+		if err := o.client.Get(context.Background(), kbclient.ObjectKey{Namespace: f.Namespace(), Name: loc}, snapshotLocation); err != nil {
 			return err
 		}
 	}
@@ -365,6 +366,7 @@ func (o *CreateOptions) BuildBackup(namespace string) (*velerov1api.Backup, erro
 			IncludedNamespaceScopedResources(o.IncludeNamespaceScopedResources...).
 			ExcludedNamespaceScopedResources(o.ExcludeNamespaceScopedResources...).
 			LabelSelector(o.Selector.LabelSelector).
+			OrLabelSelector(o.OrSelector.OrLabelSelectors).
 			TTL(o.TTL).
 			StorageLocation(o.StorageLocation).
 			VolumeSnapshotLocations(o.SnapshotLocations...).

pkg/cmd/cli/backup/create_test.go

@@ -47,6 +47,15 @@ func TestCreateOptions_BuildBackup(t *testing.T) {
 	orders, err := ParseOrderedResources(o.OrderedResources)
 	o.CSISnapshotTimeout = 20 * time.Minute
 	o.ItemOperationTimeout = 20 * time.Minute
+	orLabelSelectors := []*metav1.LabelSelector{
+		{
+			MatchLabels: map[string]string{"k1": "v1", "k2": "v2"},
+		},
+		{
+			MatchLabels: map[string]string{"a1": "b1", "a2": "b2"},
+		},
+	}
+	o.OrSelector.OrLabelSelectors = orLabelSelectors
 	assert.NoError(t, err)
 
 	backup, err := o.BuildBackup(cmdtest.VeleroNameSpace)
@@ -58,6 +67,7 @@ func TestCreateOptions_BuildBackup(t *testing.T) {
 		SnapshotVolumes:         o.SnapshotVolumes.Value,
 		IncludeClusterResources: o.IncludeClusterResources.Value,
 		OrderedResources:        orders,
+		OrLabelSelectors:        orLabelSelectors,
 		CSISnapshotTimeout:      metav1.Duration{Duration: o.CSISnapshotTimeout},
 		ItemOperationTimeout:    metav1.Duration{Duration: o.ItemOperationTimeout},
 	}, backup.Spec)

pkg/cmd/cli/backup/describe_test.go

@@ -69,6 +69,7 @@ func TestNewDescribeCommand(t *testing.T) {
 
 	if err == nil {
 		assert.Contains(t, stdout, "Velero-Native Snapshots: <none included>")
+		assert.Contains(t, stdout, "Or label selector:  <none>")
 		assert.Contains(t, stdout, fmt.Sprintf("Name:         %s", backupName))
 		return
 	}

pkg/cmd/cli/install/install.go

@@ -66,6 +66,7 @@ type Options struct {
 	BackupStorageConfig       flag.Map
 	VolumeSnapshotConfig      flag.Map
 	UseNodeAgent              bool
+	PrivilegedNodeAgent       bool
 	//TODO remove UseRestic when migration test out of using it
 	UseRestic                       bool
 	Wait                            bool
@@ -110,6 +111,7 @@ func (o *Options) BindFlags(flags *pflag.FlagSet) {
 	flags.BoolVar(&o.RestoreOnly, "restore-only", o.RestoreOnly, "Run the server in restore-only mode. Optional.")
 	flags.BoolVar(&o.DryRun, "dry-run", o.DryRun, "Generate resources, but don't send them to the cluster. Use with -o. Optional.")
 	flags.BoolVar(&o.UseNodeAgent, "use-node-agent", o.UseNodeAgent, "Create Velero node-agent daemonset. Optional. Velero node-agent hosts Velero modules that need to run in one or more nodes(i.e. Restic, Kopia).")
+	flags.BoolVar(&o.PrivilegedNodeAgent, "privileged-node-agent", o.PrivilegedNodeAgent, "Use privileged mode for the node agent. Optional. Required to backup block devices.")
 	flags.BoolVar(&o.Wait, "wait", o.Wait, "Wait for Velero deployment to be ready. Optional.")
 	flags.DurationVar(&o.DefaultRepoMaintenanceFrequency, "default-repo-maintain-frequency", o.DefaultRepoMaintenanceFrequency, "How often 'maintain' is run for backup repositories by default. Optional.")
 	flags.DurationVar(&o.GarbageCollectionFrequency, "garbage-collection-frequency", o.GarbageCollectionFrequency, "How often the garbage collection runs for expired backups.(default 1h)")
@@ -198,6 +200,7 @@ func (o *Options) AsVeleroOptions() (*install.VeleroOptions, error) {
 		SecretData:                      secretData,
 		RestoreOnly:                     o.RestoreOnly,
 		UseNodeAgent:                    o.UseNodeAgent,
+		PrivilegedNodeAgent:             o.PrivilegedNodeAgent,
 		UseVolumeSnapshots:              o.UseVolumeSnapshots,
 		BSLConfig:                       o.BackupStorageConfig.Data(),
 		VSLConfig:                       o.VolumeSnapshotConfig.Data(),

pkg/cmd/cli/restore/create.go

@@ -92,6 +92,7 @@ type CreateOptions struct {
 	StatusExcludeResources    flag.StringArray
 	NamespaceMappings         flag.Map
 	Selector                  flag.LabelSelector
+	OrSelector                flag.OrLabelSelector
 	IncludeClusterResources   flag.OptionalBool
 	Wait                      bool
 	AllowPartiallyFailed      flag.OptionalBool
@@ -124,6 +125,7 @@ func (o *CreateOptions) BindFlags(flags *pflag.FlagSet) {
 	flags.Var(&o.StatusIncludeResources, "status-include-resources", "Resources to include in the restore status, formatted as resource.group, such as storageclasses.storage.k8s.io.")
 	flags.Var(&o.StatusExcludeResources, "status-exclude-resources", "Resources to exclude from the restore status, formatted as resource.group, such as storageclasses.storage.k8s.io.")
 	flags.VarP(&o.Selector, "selector", "l", "Only restore resources matching this label selector.")
+	flags.Var(&o.OrSelector, "or-selector", "Restore resources matching at least one of the label selector from the list. Label selectors should be separated by ' or '. For example, foo=bar or app=nginx")
 	flags.DurationVar(&o.ItemOperationTimeout, "item-operation-timeout", o.ItemOperationTimeout, "How long to wait for async plugin operations before timeout.")
 	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"
@@ -185,6 +187,10 @@ func (o *CreateOptions) Validate(c *cobra.Command, args []string, f client.Facto
 		return errors.New("Velero client is not set; unable to proceed")
 	}
 
+	if o.Selector.LabelSelector != nil && o.OrSelector.OrLabelSelectors != nil {
+		return errors.New("either a 'selector' or an 'or-selector' can be specified, but not both")
+	}
+
 	if len(o.ExistingResourcePolicy) > 0 && !isResourcePolicyValid(o.ExistingResourcePolicy) {
 		return errors.New("existing-resource-policy has invalid value, it accepts only none, update as value")
 	}
@@ -304,6 +310,7 @@ func (o *CreateOptions) Run(c *cobra.Command, f client.Factory) error {
 			ExistingResourcePolicy:  api.PolicyType(o.ExistingResourcePolicy),
 			NamespaceMapping:        o.NamespaceMappings.Data(),
 			LabelSelector:           o.Selector.LabelSelector,
+			OrLabelSelectors:        o.OrSelector.OrLabelSelectors,
 			RestorePVs:              o.RestoreVolumes.Value,
 			PreserveNodePorts:       o.PreserveNodePorts.Value,
 			IncludeClusterResources: o.IncludeClusterResources.Value,

pkg/cmd/cli/schedule/create.go

@@ -145,6 +145,7 @@ func (o *CreateOptions) Run(c *cobra.Command, f client.Factory) error {
 				ExcludedNamespaceScopedResources: o.BackupOptions.ExcludeNamespaceScopedResources,
 				IncludeClusterResources:          o.BackupOptions.IncludeClusterResources.Value,
 				LabelSelector:                    o.BackupOptions.Selector.LabelSelector,
+				OrLabelSelectors:                 o.BackupOptions.OrSelector.OrLabelSelectors,
 				SnapshotVolumes:                  o.BackupOptions.SnapshotVolumes.Value,
 				TTL:                              metav1.Duration{Duration: o.BackupOptions.TTL},
 				StorageLocation:                  o.BackupOptions.StorageLocation,

pkg/cmd/util/flag/orlabelselector.go

@@ -0,0 +1,61 @@
+/*
+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 flag
+
+import (
+	"strings"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+// OrLabelSelector is a Cobra-compatible wrapper for defining
+// a Kubernetes or-label-selector flag.
+type OrLabelSelector struct {
+	OrLabelSelectors []*metav1.LabelSelector
+}
+
+// String returns a string representation of the or-label
+// selector flag.
+func (ls *OrLabelSelector) String() string {
+	orLabels := []string{}
+	for _, v := range ls.OrLabelSelectors {
+		orLabels = append(orLabels, metav1.FormatLabelSelector(v))
+	}
+	return strings.Join(orLabels, " or ")
+}
+
+// Set parses the provided string and assigns the result
+// to the or-label-selector receiver. It returns an error if
+// the string is not parseable.
+func (ls *OrLabelSelector) Set(s string) error {
+	orItems := strings.Split(s, " or ")
+	ls.OrLabelSelectors = make([]*metav1.LabelSelector, 0)
+	for _, orItem := range orItems {
+		parsed, err := metav1.ParseToLabelSelector(orItem)
+		if err != nil {
+			return err
+		}
+		ls.OrLabelSelectors = append(ls.OrLabelSelectors, parsed)
+	}
+	return nil
+}
+
+// Type returns a string representation of the
+// OrLabelSelector type.
+func (ls *OrLabelSelector) Type() string {
+	return "orLabelSelector"
+}

pkg/cmd/util/flag/orlabelselector_test.go

@@ -0,0 +1,102 @@
+package flag
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func TestStringOfOrLabelSelector(t *testing.T) {
+	tests := []struct {
+		name            string
+		orLabelSelector *OrLabelSelector
+		expectedStr     string
+	}{
+		{
+			name: "or between two labels",
+			orLabelSelector: &OrLabelSelector{
+				OrLabelSelectors: []*metav1.LabelSelector{
+					{
+						MatchLabels: map[string]string{"k1": "v1"},
+					},
+					{
+						MatchLabels: map[string]string{"k2": "v2"},
+					},
+				},
+			},
+			expectedStr: "k1=v1 or k2=v2",
+		},
+		{
+			name: "or between two label groups",
+			orLabelSelector: &OrLabelSelector{
+				OrLabelSelectors: []*metav1.LabelSelector{
+					{
+						MatchLabels: map[string]string{"k1": "v1", "k2": "v2"},
+					},
+					{
+						MatchLabels: map[string]string{"a1": "b1", "a2": "b2"},
+					},
+				},
+			},
+			expectedStr: "k1=v1,k2=v2 or a1=b1,a2=b2",
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			assert.Equal(t, test.expectedStr, test.orLabelSelector.String())
+		})
+	}
+}
+
+func TestSetOfOrLabelSelector(t *testing.T) {
+	tests := []struct {
+		name             string
+		inputStr         string
+		expectedSelector *OrLabelSelector
+	}{
+		{
+			name:     "or between two labels",
+			inputStr: "k1=v1 or k2=v2",
+			expectedSelector: &OrLabelSelector{
+				OrLabelSelectors: []*metav1.LabelSelector{
+					{
+						MatchLabels: map[string]string{"k1": "v1"},
+					},
+					{
+						MatchLabels: map[string]string{"k2": "v2"},
+					},
+				},
+			},
+		},
+		{
+			name:     "or between two label groups",
+			inputStr: "k1=v1,k2=v2 or a1=b1,a2=b2",
+			expectedSelector: &OrLabelSelector{
+				OrLabelSelectors: []*metav1.LabelSelector{
+					{
+						MatchLabels: map[string]string{"k1": "v1", "k2": "v2"},
+					},
+					{
+						MatchLabels: map[string]string{"a1": "b1", "a2": "b2"},
+					},
+				},
+			},
+		},
+	}
+	selector := &OrLabelSelector{}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			require.Nil(t, selector.Set(test.inputStr))
+			assert.Equal(t, len(test.expectedSelector.OrLabelSelectors), len(selector.OrLabelSelectors))
+			assert.Equal(t, test.expectedSelector.String(), selector.String())
+		})
+	}
+}
+
+func TestTypeOfOrLabelSelector(t *testing.T) {
+	selector := &OrLabelSelector{}
+	assert.Equal(t, "orLabelSelector", selector.Type())
+}

pkg/cmd/util/output/backup_describer.go

@@ -199,6 +199,18 @@ func DescribeBackupSpec(d *Describer, spec velerov1api.BackupSpec) {
 	}
 	d.Printf("Label selector:\t%s\n", s)
 
+	d.Println()
+	if len(spec.OrLabelSelectors) == 0 {
+		s = emptyDisplay
+	} else {
+		orLabelSelectors := []string{}
+		for _, v := range spec.OrLabelSelectors {
+			orLabelSelectors = append(orLabelSelectors, metav1.FormatLabelSelector(v))
+		}
+		s = strings.Join(orLabelSelectors, " or ")
+	}
+	d.Printf("Or label selector:\t%s\n", s)
+
 	d.Println()
 	d.Printf("Storage Location:\t%s\n", spec.StorageLocation)
 

pkg/cmd/util/output/backup_describer_test.go

@@ -91,6 +91,8 @@ Resources:
 
 Label selector:  <none>
 
+Or label selector:  <none>
+
 Storage Location:  backup-location
 
 Velero-Native Snapshot PVs:  auto
@@ -153,6 +155,8 @@ Resources:
 
 Label selector:  <none>
 
+Or label selector:  <none>
+
 Storage Location:  backup-location
 
 Velero-Native Snapshot PVs:  auto
@@ -208,6 +212,8 @@ Resources:
 
 Label selector:  <none>
 
+Or label selector:  <none>
+
 Storage Location:  backup-location
 
 Velero-Native Snapshot PVs:  auto

pkg/cmd/util/output/restore_describer.go

@@ -146,6 +146,18 @@ func DescribeRestore(ctx context.Context, kbClient kbclient.Client, restore *vel
 		}
 		d.Printf("Label selector:\t%s\n", s)
 
+		d.Println()
+		if len(restore.Spec.OrLabelSelectors) == 0 {
+			s = emptyDisplay
+		} else {
+			orLabelSelectors := []string{}
+			for _, v := range restore.Spec.OrLabelSelectors {
+				orLabelSelectors = append(orLabelSelectors, metav1.FormatLabelSelector(v))
+			}
+			s = strings.Join(orLabelSelectors, " or ")
+		}
+		d.Printf("Or label selector:\t%s\n", s)
+
 		d.Println()
 		d.Printf("Restore PVs:\t%s\n", BoolPointerString(restore.Spec.RestorePVs, "false", "true", "auto"))
 

pkg/cmd/util/output/schedule_describe_test.go

@@ -38,6 +38,8 @@ Backup Template:
   
   Label selector:  <none>
   
+  Or label selector:  <none>
+  
   Storage Location:  
   
   Velero-Native Snapshot PVs:  auto
@@ -82,6 +84,8 @@ Backup Template:
   
   Label selector:  <none>
   
+  Or label selector:  <none>
+  
   Storage Location:  
   
   Velero-Native Snapshot PVs:  auto

pkg/controller/backup_controller.go

@@ -540,8 +540,8 @@ func (b *backupReconciler) validateAndGetSnapshotLocations(backup *velerov1api.B
 	if len(errors) > 0 {
 		return nil, errors
 	}
-	allLocations := &velerov1api.VolumeSnapshotLocationList{}
-	err := b.kbClient.List(context.Background(), allLocations, &kbclient.ListOptions{Namespace: backup.Namespace, LabelSelector: labels.Everything()})
+	volumeSnapshotLocations := &velerov1api.VolumeSnapshotLocationList{}
+	err := b.kbClient.List(context.Background(), volumeSnapshotLocations, &kbclient.ListOptions{Namespace: backup.Namespace, LabelSelector: labels.Everything()})
 	if err != nil {
 		errors = append(errors, fmt.Sprintf("error listing volume snapshot locations: %v", err))
 		return nil, errors
@@ -549,8 +549,8 @@ func (b *backupReconciler) validateAndGetSnapshotLocations(backup *velerov1api.B
 
 	// build a map of provider->list of all locations for the provider
 	allProviderLocations := make(map[string][]*velerov1api.VolumeSnapshotLocation)
-	for i := range allLocations.Items {
-		loc := allLocations.Items[i]
+	for i := range volumeSnapshotLocations.Items {
+		loc := volumeSnapshotLocations.Items[i]
 		allProviderLocations[loc.Spec.Provider] = append(allProviderLocations[loc.Spec.Provider], &loc)
 	}
 

pkg/controller/backup_deletion_controller.go

@@ -570,7 +570,7 @@ func (r *backupDeletionReconciler) patchDeleteBackupRequest(ctx context.Context,
 }
 
 func (r *backupDeletionReconciler) patchBackup(ctx context.Context, backup *velerov1api.Backup, mutate func(*velerov1api.Backup)) (*velerov1api.Backup, error) {
-	//TODO: The patchHelper can't be used here because the `backup/xxx/status` does not exist, until the bakcup resource is refactored
+	//TODO: The patchHelper can't be used here because the `backup/xxx/status` does not exist, until the backup resource is refactored
 
 	// Record original json
 	oldData, err := json.Marshal(backup)

pkg/controller/backup_sync_controller.go

@@ -331,7 +331,7 @@ func (b *backupSyncReconciler) deleteOrphanedBackups(ctx context.Context, locati
 
 	for i, backup := range backupList.Items {
 		log = log.WithField("backup", backup.Name)
-		if backup.Status.Phase != velerov1api.BackupPhaseCompleted || backupStoreBackups.Has(backup.Name) {
+		if !(backup.Status.Phase == velerov1api.BackupPhaseCompleted || backup.Status.Phase == velerov1api.BackupPhasePartiallyFailed) || backupStoreBackups.Has(backup.Name) {
 			continue
 		}
 

pkg/controller/backup_sync_controller_test.go

@@ -557,7 +557,7 @@ var _ = Describe("Backup Sync Reconciler", func() {
 				k8sBackups: []*velerov1api.Backup{
 					baseBuilder("backupA").Phase(velerov1api.BackupPhaseCompleted).Result(),
 					baseBuilder("backupB").Phase(velerov1api.BackupPhaseCompleted).Result(),
-					baseBuilder("backupC").Phase(velerov1api.BackupPhaseCompleted).Result(),
+					baseBuilder("backupC").Phase(velerov1api.BackupPhasePartiallyFailed).Result(),
 				},
 				expectedDeletes: sets.NewString("backupA", "backupB", "backupC"),
 			},
@@ -568,9 +568,10 @@ var _ = Describe("Backup Sync Reconciler", func() {
 				k8sBackups: []*velerov1api.Backup{
 					baseBuilder("backup-1").Phase(velerov1api.BackupPhaseCompleted).Result(),
 					baseBuilder("backup-2").Phase(velerov1api.BackupPhaseCompleted).Result(),
-					baseBuilder("backup-C").Phase(velerov1api.BackupPhaseCompleted).Result(),
+					baseBuilder("backup-B").Phase(velerov1api.BackupPhaseCompleted).Result(),
+					baseBuilder("backup-C").Phase(velerov1api.BackupPhasePartiallyFailed).Result(),
 				},
-				expectedDeletes: sets.NewString("backup-C"),
+				expectedDeletes: sets.NewString("backup-B", "backup-C"),
 			},
 			{
 				name:         "all overlapping backups",
@@ -579,7 +580,7 @@ var _ = Describe("Backup Sync Reconciler", func() {
 				k8sBackups: []*velerov1api.Backup{
 					baseBuilder("backup-1").Phase(velerov1api.BackupPhaseCompleted).Result(),
 					baseBuilder("backup-2").Phase(velerov1api.BackupPhaseCompleted).Result(),
-					baseBuilder("backup-3").Phase(velerov1api.BackupPhaseCompleted).Result(),
+					baseBuilder("backup-3").Phase(velerov1api.BackupPhasePartiallyFailed).Result(),
 				},
 				expectedDeletes: sets.NewString(),
 			},
@@ -589,13 +590,14 @@ var _ = Describe("Backup Sync Reconciler", func() {
 				cloudBackups: sets.NewString("backup-1", "backup-2", "backup-3"),
 				k8sBackups: []*velerov1api.Backup{
 					baseBuilder("backupA").Phase(velerov1api.BackupPhaseCompleted).Result(),
+					baseBuilder("backupB").Phase(velerov1api.BackupPhasePartiallyFailed).Result(),
 					baseBuilder("Deleting").Phase(velerov1api.BackupPhaseDeleting).Result(),
 					baseBuilder("Failed").Phase(velerov1api.BackupPhaseFailed).Result(),
 					baseBuilder("FailedValidation").Phase(velerov1api.BackupPhaseFailedValidation).Result(),
 					baseBuilder("InProgress").Phase(velerov1api.BackupPhaseInProgress).Result(),
 					baseBuilder("New").Phase(velerov1api.BackupPhaseNew).Result(),
 				},
-				expectedDeletes: sets.NewString("backupA"),
+				expectedDeletes: sets.NewString("backupA", "backupB"),
 			},
 			{
 				name:         "all overlapping backups and all backups that are not complete",
@@ -616,12 +618,13 @@ var _ = Describe("Backup Sync Reconciler", func() {
 					baseBuilder("backup-1").Phase(velerov1api.BackupPhaseCompleted).Result(),
 					baseBuilder("backup-2").Phase(velerov1api.BackupPhaseCompleted).Result(),
 					baseBuilder("backup-C").Phase(velerov1api.BackupPhaseCompleted).Result(),
+					baseBuilder("backup-D").Phase(velerov1api.BackupPhasePartiallyFailed).Result(),
 
 					baseBuilder("backup-4").ObjectMeta(builder.WithLabels(velerov1api.StorageLocationLabel, "alternate")).Phase(velerov1api.BackupPhaseCompleted).Result(),
 					baseBuilder("backup-5").ObjectMeta(builder.WithLabels(velerov1api.StorageLocationLabel, "alternate")).Phase(velerov1api.BackupPhaseCompleted).Result(),
-					baseBuilder("backup-6").ObjectMeta(builder.WithLabels(velerov1api.StorageLocationLabel, "alternate")).Phase(velerov1api.BackupPhaseCompleted).Result(),
+					baseBuilder("backup-6").ObjectMeta(builder.WithLabels(velerov1api.StorageLocationLabel, "alternate")).Phase(velerov1api.BackupPhasePartiallyFailed).Result(),
 				},
-				expectedDeletes: sets.NewString("backup-C"),
+				expectedDeletes: sets.NewString("backup-C", "backup-D"),
 			},
 			{
 				name:         "some overlapping backups",
@@ -646,8 +649,14 @@ var _ = Describe("Backup Sync Reconciler", func() {
 						).
 						Phase(velerov1api.BackupPhaseCompleted).
 						Result(),
+					builder.ForBackup("ns-1", "backup-D").
+						ObjectMeta(
+							builder.WithLabels(velerov1api.StorageLocationLabel, "the-really-long-location-name-that-is-much-more-than-63-c69e779"),
+						).
+						Phase(velerov1api.BackupPhasePartiallyFailed).
+						Result(),
 				},
-				expectedDeletes: sets.NewString("backup-C"),
+				expectedDeletes: sets.NewString("backup-C", "backup-D"),
 				useLongBSLName:  true,
 			},
 		}

pkg/controller/data_upload_controller.go

@@ -177,7 +177,10 @@ func (r *DataUploadReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 			return ctrl.Result{}, nil
 		}
 
-		exposeParam := r.setupExposeParam(du)
+		exposeParam, err := r.setupExposeParam(du)
+		if err != nil {
+			return r.errorOut(ctx, du, err, "failed to set exposer parameters", log)
+		}
 
 		// Expose() will trigger to create one pod whose volume is restored by a given volume snapshot,
 		// but the pod maybe is not in the same node of the current controller, so we need to return it here.
@@ -735,18 +738,33 @@ func (r *DataUploadReconciler) closeDataPath(ctx context.Context, duName string)
 	r.dataPathMgr.RemoveAsyncBR(duName)
 }
 
-func (r *DataUploadReconciler) setupExposeParam(du *velerov2alpha1api.DataUpload) interface{} {
+func (r *DataUploadReconciler) setupExposeParam(du *velerov2alpha1api.DataUpload) (interface{}, error) {
 	if du.Spec.SnapshotType == velerov2alpha1api.SnapshotTypeCSI {
+		pvc := &corev1.PersistentVolumeClaim{}
+		err := r.client.Get(context.Background(), types.NamespacedName{
+			Namespace: du.Spec.SourceNamespace,
+			Name:      du.Spec.SourcePVC,
+		}, pvc)
+
+		if err != nil {
+			return nil, errors.Wrapf(err, "failed to get PVC %s/%s", du.Spec.SourceNamespace, du.Spec.SourcePVC)
+		}
+
+		accessMode := exposer.AccessModeFileSystem
+		if pvc.Spec.VolumeMode != nil && *pvc.Spec.VolumeMode == corev1.PersistentVolumeBlock {
+			accessMode = exposer.AccessModeBlock
+		}
+
 		return &exposer.CSISnapshotExposeParam{
 			SnapshotName:     du.Spec.CSISnapshot.VolumeSnapshot,
 			SourceNamespace:  du.Spec.SourceNamespace,
 			StorageClass:     du.Spec.CSISnapshot.StorageClass,
 			HostingPodLabels: map[string]string{velerov1api.DataUploadLabel: du.Name},
-			AccessMode:       exposer.AccessModeFileSystem,
+			AccessMode:       accessMode,
 			Timeout:          du.Spec.OperationTimeout.Duration,
-		}
+		}, nil
 	}
-	return nil
+	return nil, nil
 }
 
 func (r *DataUploadReconciler) setupWaitExposePara(du *velerov2alpha1api.DataUpload) interface{} {

pkg/controller/data_upload_controller_test.go

@@ -306,6 +306,7 @@ func TestReconcile(t *testing.T) {
 		name                string
 		du                  *velerov2alpha1api.DataUpload
 		pod                 *corev1.Pod
+		pvc                 *corev1.PersistentVolumeClaim
 		snapshotExposerList map[velerov2alpha1api.SnapshotType]exposer.SnapshotExposer
 		dataMgr             *datapath.Manager
 		expectedProcessed   bool
@@ -345,11 +346,21 @@ func TestReconcile(t *testing.T) {
 		}, {
 			name:              "Dataupload should be accepted",
 			du:                dataUploadBuilder().Result(),
-			pod:               builder.ForPod(velerov1api.DefaultNamespace, dataUploadName).Volumes(&corev1.Volume{Name: "dataupload-1"}).Result(),
+			pod:               builder.ForPod("fake-ns", dataUploadName).Volumes(&corev1.Volume{Name: "test-pvc"}).Result(),
+			pvc:               builder.ForPersistentVolumeClaim("fake-ns", "test-pvc").Result(),
 			expectedProcessed: false,
 			expected:          dataUploadBuilder().Phase(velerov2alpha1api.DataUploadPhaseAccepted).Result(),
 			expectedRequeue:   ctrl.Result{},
 		},
+		{
+			name:              "Dataupload should fail to get PVC information",
+			du:                dataUploadBuilder().Result(),
+			pod:               builder.ForPod("fake-ns", dataUploadName).Volumes(&corev1.Volume{Name: "wrong-pvc"}).Result(),
+			expectedProcessed: true,
+			expected:          dataUploadBuilder().Phase(velerov2alpha1api.DataUploadPhaseFailed).Result(),
+			expectedRequeue:   ctrl.Result{},
+			expectedErrMsg:    "failed to get PVC",
+		},
 		{
 			name:              "Dataupload should be prepared",
 			du:                dataUploadBuilder().SnapshotType(fakeSnapshotType).Result(),
@@ -448,6 +459,11 @@ func TestReconcile(t *testing.T) {
 				require.NoError(t, err)
 			}
 
+			if test.pvc != nil {
+				err = r.client.Create(ctx, test.pvc)
+				require.NoError(t, err)
+			}
+
 			if test.dataMgr != nil {
 				r.dataPathMgr = test.dataMgr
 			} else {

pkg/controller/download_request_controller.go

@@ -144,7 +144,7 @@ func (r *downloadRequestReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 					log.WithError(err).Error("fail to get restore for DownloadRequest")
 					return ctrl.Result{}, nil
 				}
-				log.Warnf("Fail to get restore for DownloadRequest %s. Retry later.", err.Error())
+				log.Warnf("fail to get restore for DownloadRequest %s. Retry later.", err.Error())
 				return ctrl.Result{}, errors.WithStack(err)
 			}
 			backupName = restore.Spec.BackupName
@@ -172,7 +172,7 @@ func (r *downloadRequestReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 				log.Errorf("BSL for DownloadRequest cannot be found")
 				return ctrl.Result{}, nil
 			}
-			log.Warnf("Fail to get BSL for DownloadRequest: %s", err.Error())
+			log.Warnf("fail to get BSL for DownloadRequest: %s", err.Error())
 			return ctrl.Result{}, errors.WithStack(err)
 		}
 

pkg/controller/schedule_controller.go

@@ -94,6 +94,7 @@ func (c *scheduleReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 	if err := c.Get(ctx, req.NamespacedName, schedule); err != nil {
 		if apierrors.IsNotFound(err) {
 			log.WithError(err).Error("schedule not found")
+			c.metrics.RemoveSchedule(req.Name)
 			return ctrl.Result{}, nil
 		}
 		return ctrl.Result{}, errors.Wrapf(err, "error getting schedule %s", req.String())

pkg/datapath/file_system.go

@@ -133,10 +133,10 @@ func (fs *fileSystemBR) StartBackup(source AccessPoint, realSource string, paren
 	if !fs.initialized {
 		return errors.New("file system data path is not initialized")
 	}
-	volMode := getPersistentVolumeMode(source)
 
 	go func() {
-		snapshotID, emptySnapshot, err := fs.uploaderProv.RunBackup(fs.ctx, source.ByPath, realSource, tags, forceFull, parentSnapshot, volMode, fs)
+		snapshotID, emptySnapshot, err := fs.uploaderProv.RunBackup(fs.ctx, source.ByPath, realSource, tags, forceFull,
+			parentSnapshot, source.VolMode, fs)
 
 		if err == provider.ErrorCanceled {
 			fs.callbacks.OnCancelled(context.Background(), fs.namespace, fs.jobName)
@@ -155,10 +155,8 @@ func (fs *fileSystemBR) StartRestore(snapshotID string, target AccessPoint) erro
 		return errors.New("file system data path is not initialized")
 	}
 
-	volMode := getPersistentVolumeMode(target)
-
 	go func() {
-		err := fs.uploaderProv.RunRestore(fs.ctx, snapshotID, target.ByPath, volMode, fs)
+		err := fs.uploaderProv.RunRestore(fs.ctx, snapshotID, target.ByPath, target.VolMode, fs)
 
 		if err == provider.ErrorCanceled {
 			fs.callbacks.OnCancelled(context.Background(), fs.namespace, fs.jobName)
@@ -172,13 +170,6 @@ func (fs *fileSystemBR) StartRestore(snapshotID string, target AccessPoint) erro
 	return nil
 }
 
-func getPersistentVolumeMode(source AccessPoint) uploader.PersistentVolumeMode {
-	if source.ByBlock != "" {
-		return uploader.PersistentVolumeBlock
-	}
-	return uploader.PersistentVolumeFilesystem
-}
-
 // UpdateProgress which implement ProgressUpdater interface to update progress status
 func (fs *fileSystemBR) UpdateProgress(p *uploader.Progress) {
 	if fs.callbacks.OnProgress != nil {

pkg/datapath/types.go

@@ -53,7 +53,7 @@ type Callbacks struct {
 // AccessPoint represents an access point that has been exposed to a data path instance
 type AccessPoint struct {
 	ByPath  string
-	ByBlock string
+	VolMode uploader.PersistentVolumeMode
 }
 
 // AsyncBR is the interface for asynchronous data path methods

pkg/exposer/csi_snapshot.go

@@ -233,9 +233,12 @@ func (e *csiSnapshotExposer) CleanUp(ctx context.Context, ownerObject corev1.Obj
 }
 
 func getVolumeModeByAccessMode(accessMode string) (corev1.PersistentVolumeMode, error) {
-	if accessMode == AccessModeFileSystem {
+	switch accessMode {
+	case AccessModeFileSystem:
 		return corev1.PersistentVolumeFilesystem, nil
-	} else {
+	case AccessModeBlock:
+		return corev1.PersistentVolumeBlock, nil
+	default:
 		return "", errors.Errorf("unsupported access mode %s", accessMode)
 	}
 }
@@ -356,6 +359,13 @@ func (e *csiSnapshotExposer) createBackupPod(ctx context.Context, ownerObject co
 	}
 
 	var gracePeriod int64 = 0
+	volumeMounts, volumeDevices := kube.MakePodPVCAttachment(volumeName, backupPVC.Spec.VolumeMode)
+
+	if label == nil {
+		label = make(map[string]string)
+	}
+
+	label[podGroupLabel] = podGroupSnapshot
 
 	pod := &corev1.Pod{
 		ObjectMeta: metav1.ObjectMeta{
@@ -373,16 +383,26 @@ func (e *csiSnapshotExposer) createBackupPod(ctx context.Context, ownerObject co
 			Labels: label,
 		},
 		Spec: corev1.PodSpec{
+			TopologySpreadConstraints: []corev1.TopologySpreadConstraint{
+				{
+					MaxSkew:           1,
+					TopologyKey:       "kubernetes.io/hostname",
+					WhenUnsatisfiable: corev1.ScheduleAnyway,
+					LabelSelector: &metav1.LabelSelector{
+						MatchLabels: map[string]string{
+							podGroupLabel: podGroupSnapshot,
+						},
+					},
+				},
+			},
 			Containers: []corev1.Container{
 				{
 					Name:            containerName,
 					Image:           podInfo.image,
 					ImagePullPolicy: corev1.PullNever,
 					Command:         []string{"/velero-helper", "pause"},
-					VolumeMounts: []corev1.VolumeMount{{
-						Name:      volumeName,
-						MountPath: "/" + volumeName,
-					}},
+					VolumeMounts:    volumeMounts,
+					VolumeDevices:   volumeDevices,
 				},
 			},
 			ServiceAccountName:            podInfo.serviceAccount,

pkg/exposer/generic_restore.go

@@ -82,7 +82,7 @@ func (e *genericRestoreExposer) Expose(ctx context.Context, ownerObject corev1.O
 		return errors.Errorf("Target PVC %s/%s has already been bound, abort", sourceNamespace, targetPVCName)
 	}
 
-	restorePod, err := e.createRestorePod(ctx, ownerObject, hostingPodLabels, selectedNode)
+	restorePod, err := e.createRestorePod(ctx, ownerObject, targetPVC, hostingPodLabels, selectedNode)
 	if err != nil {
 		return errors.Wrapf(err, "error to create restore pod")
 	}
@@ -247,7 +247,8 @@ func (e *genericRestoreExposer) RebindVolume(ctx context.Context, ownerObject co
 	return nil
 }
 
-func (e *genericRestoreExposer) createRestorePod(ctx context.Context, ownerObject corev1.ObjectReference, label map[string]string, selectedNode string) (*corev1.Pod, error) {
+func (e *genericRestoreExposer) createRestorePod(ctx context.Context, ownerObject corev1.ObjectReference, targetPVC *corev1.PersistentVolumeClaim,
+	label map[string]string, selectedNode string) (*corev1.Pod, error) {
 	restorePodName := ownerObject.Name
 	restorePVCName := ownerObject.Name
 
@@ -260,6 +261,7 @@ func (e *genericRestoreExposer) createRestorePod(ctx context.Context, ownerObjec
 	}
 
 	var gracePeriod int64 = 0
+	volumeMounts, volumeDevices := kube.MakePodPVCAttachment(volumeName, targetPVC.Spec.VolumeMode)
 
 	pod := &corev1.Pod{
 		ObjectMeta: metav1.ObjectMeta{
@@ -283,10 +285,8 @@ func (e *genericRestoreExposer) createRestorePod(ctx context.Context, ownerObjec
 					Image:           podInfo.image,
 					ImagePullPolicy: corev1.PullNever,
 					Command:         []string{"/velero-helper", "pause"},
-					VolumeMounts: []corev1.VolumeMount{{
-						Name:      volumeName,
-						MountPath: "/" + volumeName,
-					}},
+					VolumeMounts:    volumeMounts,
+					VolumeDevices:   volumeDevices,
 				},
 			},
 			ServiceAccountName:            podInfo.serviceAccount,

pkg/exposer/host_path.go

@@ -26,11 +26,13 @@ import (
 	ctrlclient "sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/vmware-tanzu/velero/pkg/datapath"
+	"github.com/vmware-tanzu/velero/pkg/uploader"
 	"github.com/vmware-tanzu/velero/pkg/util/filesystem"
 	"github.com/vmware-tanzu/velero/pkg/util/kube"
 )
 
 var getVolumeDirectory = kube.GetVolumeDirectory
+var getVolumeMode = kube.GetVolumeMode
 var singlePathMatch = kube.SinglePathMatch
 
 // GetPodVolumeHostPath returns a path that can be accessed from the host for a given volume of a pod
@@ -45,7 +47,17 @@ func GetPodVolumeHostPath(ctx context.Context, pod *corev1.Pod, volumeName strin
 
 	logger.WithField("volDir", volDir).Info("Got volume dir")
 
-	pathGlob := fmt.Sprintf("/host_pods/%s/volumes/*/%s", string(pod.GetUID()), volDir)
+	volMode, err := getVolumeMode(ctx, logger, pod, volumeName, cli)
+	if err != nil {
+		return datapath.AccessPoint{}, errors.Wrapf(err, "error getting volume mode for volume %s in pod %s", volumeName, pod.Name)
+	}
+
+	volSubDir := "volumes"
+	if volMode == uploader.PersistentVolumeBlock {
+		volSubDir = "volumeDevices"
+	}
+
+	pathGlob := fmt.Sprintf("/host_pods/%s/%s/*/%s", string(pod.GetUID()), volSubDir, volDir)
 	logger.WithField("pathGlob", pathGlob).Debug("Looking for path matching glob")
 
 	path, err := singlePathMatch(pathGlob, fs, logger)
@@ -56,6 +68,7 @@ func GetPodVolumeHostPath(ctx context.Context, pod *corev1.Pod, volumeName strin
 	logger.WithField("path", path).Info("Found path matching glob")
 
 	return datapath.AccessPoint{
-		ByPath: path,
+		ByPath:  path,
+		VolMode: volMode,
 	}, nil
 }

pkg/exposer/host_path_test.go

@@ -29,17 +29,19 @@ import (
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/builder"
 	velerotest "github.com/vmware-tanzu/velero/pkg/test"
+	"github.com/vmware-tanzu/velero/pkg/uploader"
 	"github.com/vmware-tanzu/velero/pkg/util/filesystem"
 )
 
 func TestGetPodVolumeHostPath(t *testing.T) {
 	tests := []struct {
-		name             string
-		getVolumeDirFunc func(context.Context, logrus.FieldLogger, *corev1.Pod, string, ctrlclient.Client) (string, error)
-		pathMatchFunc    func(string, filesystem.Interface, logrus.FieldLogger) (string, error)
-		pod              *corev1.Pod
-		pvc              string
-		err              string
+		name              string
+		getVolumeDirFunc  func(context.Context, logrus.FieldLogger, *corev1.Pod, string, ctrlclient.Client) (string, error)
+		getVolumeModeFunc func(context.Context, logrus.FieldLogger, *corev1.Pod, string, ctrlclient.Client) (uploader.PersistentVolumeMode, error)
+		pathMatchFunc     func(string, filesystem.Interface, logrus.FieldLogger) (string, error)
+		pod               *corev1.Pod
+		pvc               string
+		err               string
 	}{
 		{
 			name: "get volume dir fail",
@@ -55,6 +57,9 @@ func TestGetPodVolumeHostPath(t *testing.T) {
 			getVolumeDirFunc: func(context.Context, logrus.FieldLogger, *corev1.Pod, string, ctrlclient.Client) (string, error) {
 				return "", nil
 			},
+			getVolumeModeFunc: func(context.Context, logrus.FieldLogger, *corev1.Pod, string, ctrlclient.Client) (uploader.PersistentVolumeMode, error) {
+				return uploader.PersistentVolumeFilesystem, nil
+			},
 			pathMatchFunc: func(string, filesystem.Interface, logrus.FieldLogger) (string, error) {
 				return "", errors.New("fake-error-2")
 			},
@@ -62,6 +67,18 @@ func TestGetPodVolumeHostPath(t *testing.T) {
 			pvc: "fake-pvc-1",
 			err: "error identifying unique volume path on host for volume fake-pvc-1 in pod fake-pod-2: fake-error-2",
 		},
+		{
+			name: "get block volume dir success",
+			getVolumeDirFunc: func(context.Context, logrus.FieldLogger, *corev1.Pod, string, ctrlclient.Client) (
+				string, error) {
+				return "fake-pvc-1", nil
+			},
+			pathMatchFunc: func(string, filesystem.Interface, logrus.FieldLogger) (string, error) {
+				return "/host_pods/fake-pod-1-id/volumeDevices/kubernetes.io~csi/fake-pvc-1-id", nil
+			},
+			pod: builder.ForPod(velerov1api.DefaultNamespace, "fake-pod-1").Result(),
+			pvc: "fake-pvc-1",
+		},
 	}
 
 	for _, test := range tests {
@@ -70,12 +87,18 @@ func TestGetPodVolumeHostPath(t *testing.T) {
 				getVolumeDirectory = test.getVolumeDirFunc
 			}
 
+			if test.getVolumeModeFunc != nil {
+				getVolumeMode = test.getVolumeModeFunc
+			}
+
 			if test.pathMatchFunc != nil {
 				singlePathMatch = test.pathMatchFunc
 			}
 
 			_, err := GetPodVolumeHostPath(context.Background(), test.pod, test.pvc, nil, nil, velerotest.NewLogger())
-			assert.EqualError(t, err, test.err)
+			if test.err != "" || err != nil {
+				assert.EqualError(t, err, test.err)
+			}
 		})
 	}
 }

pkg/exposer/types.go

@@ -22,6 +22,9 @@ import (
 
 const (
 	AccessModeFileSystem = "by-file-system"
+	AccessModeBlock      = "by-block-device"
+	podGroupLabel        = "velero.io/exposer-pod-group"
+	podGroupSnapshot     = "snapshot-exposer"
 )
 
 // ExposeResult defines the result of expose.

pkg/install/daemonset.go

@@ -86,6 +86,14 @@ func DaemonSet(namespace string, opts ...podTemplateOption) *appsv1.DaemonSet {
 								},
 							},
 						},
+						{
+							Name: "host-plugins",
+							VolumeSource: corev1.VolumeSource{
+								HostPath: &corev1.HostPathVolumeSource{
+									Path: "/var/lib/kubelet/plugins",
+								},
+							},
+						},
 						{
 							Name: "scratch",
 							VolumeSource: corev1.VolumeSource{
@@ -102,13 +110,20 @@ func DaemonSet(namespace string, opts ...podTemplateOption) *appsv1.DaemonSet {
 								"/velero",
 							},
 							Args: daemonSetArgs,
-
+							SecurityContext: &corev1.SecurityContext{
+								Privileged: &c.privilegedNodeAgent,
+							},
 							VolumeMounts: []corev1.VolumeMount{
 								{
 									Name:             "host-pods",
 									MountPath:        "/host_pods",
 									MountPropagation: &mountPropagationMode,
 								},
+								{
+									Name:             "host-plugins",
+									MountPath:        "/var/lib/kubelet/plugins",
+									MountPropagation: &mountPropagationMode,
+								},
 								{
 									Name:      "scratch",
 									MountPath: "/scratch",

pkg/install/daemonset_test.go

@@ -35,7 +35,7 @@ func TestDaemonSet(t *testing.T) {
 
 	ds = DaemonSet("velero", WithSecret(true))
 	assert.Equal(t, 7, len(ds.Spec.Template.Spec.Containers[0].Env))
-	assert.Equal(t, 3, len(ds.Spec.Template.Spec.Volumes))
+	assert.Equal(t, 4, len(ds.Spec.Template.Spec.Volumes))
 
 	ds = DaemonSet("velero", WithFeatures([]string{"foo,bar,baz"}))
 	assert.Len(t, ds.Spec.Template.Spec.Containers[0].Args, 3)

pkg/install/deployment.go

@@ -47,6 +47,7 @@ type podTemplateConfig struct {
 	serviceAccountName              string
 	uploaderType                    string
 	defaultSnapshotMoveData         bool
+	privilegedNodeAgent             bool
 }
 
 func WithImage(image string) podTemplateOption {
@@ -149,6 +150,12 @@ func WithServiceAccountName(sa string) podTemplateOption {
 	}
 }
 
+func WithPrivilegedNodeAgent() podTemplateOption {
+	return func(c *podTemplateConfig) {
+		c.privilegedNodeAgent = true
+	}
+}
+
 func Deployment(namespace string, opts ...podTemplateOption) *appsv1.Deployment {
 	// TODO: Add support for server args
 	c := &podTemplateConfig{

pkg/install/resources.go

@@ -240,6 +240,7 @@ type VeleroOptions struct {
 	SecretData                      []byte
 	RestoreOnly                     bool
 	UseNodeAgent                    bool
+	PrivilegedNodeAgent             bool
 	UseVolumeSnapshots              bool
 	BSLConfig                       map[string]string
 	VSLConfig                       map[string]string
@@ -374,6 +375,9 @@ func AllResources(o *VeleroOptions) *unstructured.UnstructuredList {
 		if len(o.Features) > 0 {
 			dsOpts = append(dsOpts, WithFeatures(o.Features))
 		}
+		if o.PrivilegedNodeAgent {
+			dsOpts = append(dsOpts, WithPrivilegedNodeAgent())
+		}
 		ds := DaemonSet(o.Namespace, dsOpts...)
 		if err := appendUnstructured(resources, ds); err != nil {
 			fmt.Printf("error appending DaemonSet %s: %s\n", ds.GetName(), err.Error())

pkg/metrics/metrics.go

@@ -505,6 +505,88 @@ func (m *ServerMetrics) InitSchedule(scheduleName string) {
 	}
 }
 
+// RemoveSchedule removes metrics associated with a specified schedule.
+func (m *ServerMetrics) RemoveSchedule(scheduleName string) {
+	if g, ok := m.metrics[backupTarballSizeBytesGauge].(*prometheus.GaugeVec); ok {
+		g.DeleteLabelValues(scheduleName)
+	}
+	if c, ok := m.metrics[backupAttemptTotal].(*prometheus.CounterVec); ok {
+		c.DeleteLabelValues(scheduleName)
+	}
+	if c, ok := m.metrics[backupSuccessTotal].(*prometheus.CounterVec); ok {
+		c.DeleteLabelValues(scheduleName)
+	}
+	if c, ok := m.metrics[backupPartialFailureTotal].(*prometheus.CounterVec); ok {
+		c.DeleteLabelValues(scheduleName)
+	}
+	if c, ok := m.metrics[backupFailureTotal].(*prometheus.CounterVec); ok {
+		c.DeleteLabelValues(scheduleName)
+	}
+	if c, ok := m.metrics[backupValidationFailureTotal].(*prometheus.CounterVec); ok {
+		c.DeleteLabelValues(scheduleName)
+	}
+	if h, ok := m.metrics[backupDurationSeconds].(*prometheus.HistogramVec); ok {
+		h.DeleteLabelValues(scheduleName)
+	}
+	if c, ok := m.metrics[backupDeletionAttemptTotal].(*prometheus.CounterVec); ok {
+		c.DeleteLabelValues(scheduleName)
+	}
+	if c, ok := m.metrics[backupDeletionSuccessTotal].(*prometheus.CounterVec); ok {
+		c.DeleteLabelValues(scheduleName)
+	}
+	if c, ok := m.metrics[backupDeletionFailureTotal].(*prometheus.CounterVec); ok {
+		c.DeleteLabelValues(scheduleName)
+	}
+	if g, ok := m.metrics[backupLastSuccessfulTimestamp].(*prometheus.GaugeVec); ok {
+		g.DeleteLabelValues(scheduleName)
+	}
+	if c, ok := m.metrics[backupItemsTotalGauge].(*prometheus.GaugeVec); ok {
+		c.DeleteLabelValues(scheduleName)
+	}
+	if c, ok := m.metrics[backupItemsErrorsGauge].(*prometheus.GaugeVec); ok {
+		c.DeleteLabelValues(scheduleName)
+	}
+	if c, ok := m.metrics[backupWarningTotal].(*prometheus.CounterVec); ok {
+		c.DeleteLabelValues(scheduleName)
+	}
+	if c, ok := m.metrics[backupLastStatus].(*prometheus.GaugeVec); ok {
+		c.DeleteLabelValues(scheduleName)
+	}
+	if c, ok := m.metrics[restoreAttemptTotal].(*prometheus.CounterVec); ok {
+		c.DeleteLabelValues(scheduleName)
+	}
+	if c, ok := m.metrics[restorePartialFailureTotal].(*prometheus.CounterVec); ok {
+		c.DeleteLabelValues(scheduleName)
+	}
+	if c, ok := m.metrics[restoreFailedTotal].(*prometheus.CounterVec); ok {
+		c.DeleteLabelValues(scheduleName)
+	}
+	if c, ok := m.metrics[restoreSuccessTotal].(*prometheus.CounterVec); ok {
+		c.DeleteLabelValues(scheduleName)
+	}
+	if c, ok := m.metrics[restoreValidationFailedTotal].(*prometheus.CounterVec); ok {
+		c.DeleteLabelValues(scheduleName)
+	}
+	if c, ok := m.metrics[volumeSnapshotSuccessTotal].(*prometheus.CounterVec); ok {
+		c.DeleteLabelValues(scheduleName)
+	}
+	if c, ok := m.metrics[volumeSnapshotAttemptTotal].(*prometheus.CounterVec); ok {
+		c.DeleteLabelValues(scheduleName)
+	}
+	if c, ok := m.metrics[volumeSnapshotFailureTotal].(*prometheus.CounterVec); ok {
+		c.DeleteLabelValues(scheduleName)
+	}
+	if c, ok := m.metrics[csiSnapshotAttemptTotal].(*prometheus.CounterVec); ok {
+		c.DeleteLabelValues(scheduleName, "")
+	}
+	if c, ok := m.metrics[csiSnapshotSuccessTotal].(*prometheus.CounterVec); ok {
+		c.DeleteLabelValues(scheduleName, "")
+	}
+	if c, ok := m.metrics[csiSnapshotFailureTotal].(*prometheus.CounterVec); ok {
+		c.DeleteLabelValues(scheduleName, "")
+	}
+}
+
 // InitSchedule initializes counter metrics for a node.
 func (m *ServerMetrics) InitMetricsForNode(node string) {
 	if c, ok := m.metrics[podVolumeBackupEnqueueTotal].(*prometheus.CounterVec); ok {

pkg/podvolume/backupper.go

@@ -200,10 +200,11 @@ func (b *backupper) BackupPodVolumes(backup *velerov1api.Backup, pod *corev1api.
 	b.resultsLock.Unlock()
 
 	var (
-		errs              []error
-		podVolumeBackups  []*velerov1api.PodVolumeBackup
-		podVolumes        = make(map[string]corev1api.Volume)
-		mountedPodVolumes = sets.String{}
+		errs               []error
+		podVolumeBackups   []*velerov1api.PodVolumeBackup
+		podVolumes         = make(map[string]corev1api.Volume)
+		mountedPodVolumes  = sets.String{}
+		attachedPodDevices = sets.String{}
 	)
 	pvcSummary := NewPVCBackupSummary()
 
@@ -233,6 +234,14 @@ func (b *backupper) BackupPodVolumes(backup *velerov1api.Backup, pod *corev1api.
 		for _, volumeMount := range container.VolumeMounts {
 			mountedPodVolumes.Insert(volumeMount.Name)
 		}
+		for _, volumeDevice := range container.VolumeDevices {
+			attachedPodDevices.Insert(volumeDevice.Name)
+		}
+	}
+
+	repoIdentifier := ""
+	if repositoryType == velerov1api.BackupRepositoryTypeRestic {
+		repoIdentifier = repo.Spec.ResticIdentifier
 	}
 
 	var numVolumeSnapshots int
@@ -263,6 +272,15 @@ func (b *backupper) BackupPodVolumes(backup *velerov1api.Backup, pod *corev1api.
 			continue
 		}
 
+		// check if volume is a block volume
+		if attachedPodDevices.Has(volumeName) {
+			msg := fmt.Sprintf("volume %s declared in pod %s/%s is a block volume. Block volumes are not supported for fs backup, skipping",
+				volumeName, pod.Namespace, pod.Name)
+			log.Warn(msg)
+			pvcSummary.addSkipped(volumeName, msg)
+			continue
+		}
+
 		// volumes that are not mounted by any container should not be backed up, because
 		// its directory is not created
 		if !mountedPodVolumes.Has(volumeName) {
@@ -283,7 +301,7 @@ func (b *backupper) BackupPodVolumes(backup *velerov1api.Backup, pod *corev1api.
 			}
 		}
 
-		volumeBackup := newPodVolumeBackup(backup, pod, volume, repo.Spec.ResticIdentifier, b.uploaderType, pvc)
+		volumeBackup := newPodVolumeBackup(backup, pod, volume, repoIdentifier, b.uploaderType, pvc)
 		if _, err = b.veleroClient.VeleroV1().PodVolumeBackups(volumeBackup.Namespace).Create(context.TODO(), volumeBackup, metav1.CreateOptions{}); err != nil {
 			errs = append(errs, err)
 			continue

pkg/podvolume/restorer.go

@@ -157,6 +157,12 @@ func (r *restorer) RestorePodVolumes(data RestoreData) []error {
 	for _, podVolume := range data.Pod.Spec.Volumes {
 		podVolumes[podVolume.Name] = podVolume
 	}
+
+	repoIdentifier := ""
+	if repositoryType == velerov1api.BackupRepositoryTypeRestic {
+		repoIdentifier = repo.Spec.ResticIdentifier
+	}
+
 	for volume, backupInfo := range volumesToRestore {
 		volumeObj, ok := podVolumes[volume]
 		var pvc *corev1api.PersistentVolumeClaim
@@ -170,7 +176,7 @@ func (r *restorer) RestorePodVolumes(data RestoreData) []error {
 			}
 		}
 
-		volumeRestore := newPodVolumeRestore(data.Restore, data.Pod, data.BackupLocation, volume, backupInfo.snapshotID, repo.Spec.ResticIdentifier, backupInfo.uploaderType, data.SourceNamespace, pvc)
+		volumeRestore := newPodVolumeRestore(data.Restore, data.Pod, data.BackupLocation, volume, backupInfo.snapshotID, repoIdentifier, backupInfo.uploaderType, data.SourceNamespace, pvc)
 
 		if err := errorOnly(r.veleroClient.VeleroV1().PodVolumeRestores(volumeRestore.Namespace).Create(context.TODO(), volumeRestore, metav1.CreateOptions{})); err != nil {
 			errs = append(errs, errors.WithStack(err))

pkg/podvolume/util.go

@@ -37,14 +37,6 @@ const (
 	// TODO(2.0): remove
 	podAnnotationPrefix = "snapshot.velero.io/"
 
-	// VolumesToBackupAnnotation is the annotation on a pod whose mounted volumes
-	// need to be backed up using pod volume backup.
-	VolumesToBackupAnnotation = "backup.velero.io/backup-volumes"
-
-	// VolumesToExcludeAnnotation is the annotation on a pod whose mounted volumes
-	// should be excluded from pod volume backup.
-	VolumesToExcludeAnnotation = "backup.velero.io/backup-volumes-excludes"
-
 	// DefaultVolumesToFsBackup specifies whether pod volume backup should be used, by default, to
 	// take backup of all pod volumes.
 	DefaultVolumesToFsBackup = false
@@ -216,85 +208,3 @@ func getPodSnapshotAnnotations(obj metav1.Object) map[string]string {
 
 	return res
 }
-
-// GetVolumesToBackup returns a list of volume names to backup for
-// the provided pod.
-// Deprecated: Use GetVolumesByPod instead.
-func GetVolumesToBackup(obj metav1.Object) []string {
-	annotations := obj.GetAnnotations()
-	if annotations == nil {
-		return nil
-	}
-
-	backupsValue := annotations[VolumesToBackupAnnotation]
-	if backupsValue == "" {
-		return nil
-	}
-
-	return strings.Split(backupsValue, ",")
-}
-
-func getVolumesToExclude(obj metav1.Object) []string {
-	annotations := obj.GetAnnotations()
-	if annotations == nil {
-		return nil
-	}
-
-	return strings.Split(annotations[VolumesToExcludeAnnotation], ",")
-}
-
-func contains(list []string, k string) bool {
-	for _, i := range list {
-		if i == k {
-			return true
-		}
-	}
-	return false
-}
-
-// GetVolumesByPod returns a list of volume names to backup for the provided pod.
-func GetVolumesByPod(pod *corev1api.Pod, defaultVolumesToFsBackup bool) ([]string, []string) {
-	// tracks the volumes that have been explicitly opted out of backup via the annotation in the pod
-	optedOutVolumes := make([]string, 0)
-
-	if !defaultVolumesToFsBackup {
-		return GetVolumesToBackup(pod), optedOutVolumes
-	}
-
-	volsToExclude := getVolumesToExclude(pod)
-	podVolumes := []string{}
-	for _, pv := range pod.Spec.Volumes {
-		// cannot backup hostpath volumes as they are not mounted into /var/lib/kubelet/pods
-		// and therefore not accessible to the node agent daemon set.
-		if pv.HostPath != nil {
-			continue
-		}
-		// don't backup volumes mounting secrets. Secrets will be backed up separately.
-		if pv.Secret != nil {
-			continue
-		}
-		// don't backup volumes mounting config maps. Config maps will be backed up separately.
-		if pv.ConfigMap != nil {
-			continue
-		}
-		// don't backup volumes mounted as projected volumes, all data in those come from kube state.
-		if pv.Projected != nil {
-			continue
-		}
-		// don't backup DownwardAPI volumes, all data in those come from kube state.
-		if pv.DownwardAPI != nil {
-			continue
-		}
-		// don't backup volumes that are included in the exclude list.
-		if contains(volsToExclude, pv.Name) {
-			optedOutVolumes = append(optedOutVolumes, pv.Name)
-			continue
-		}
-		// don't include volumes that mount the default service account token.
-		if strings.HasPrefix(pv.Name, "default-token") {
-			continue
-		}
-		podVolumes = append(podVolumes, pv.Name)
-	}
-	return podVolumes, optedOutVolumes
-}

pkg/podvolume/util_test.go

@@ -17,12 +17,10 @@ limitations under the License.
 package podvolume
 
 import (
-	"sort"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
 	corev1api "k8s.io/api/core/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/builder"
@@ -303,322 +301,3 @@ func TestVolumeHasNonRestorableSource(t *testing.T) {
 
 	}
 }
-
-func TestGetVolumesToBackup(t *testing.T) {
-	tests := []struct {
-		name        string
-		annotations map[string]string
-		expected    []string
-	}{
-		{
-			name:        "nil annotations",
-			annotations: nil,
-			expected:    nil,
-		},
-		{
-			name:        "no volumes to backup",
-			annotations: map[string]string{"foo": "bar"},
-			expected:    nil,
-		},
-		{
-			name:        "one volume to backup",
-			annotations: map[string]string{"foo": "bar", VolumesToBackupAnnotation: "volume-1"},
-			expected:    []string{"volume-1"},
-		},
-		{
-			name:        "multiple volumes to backup",
-			annotations: map[string]string{"foo": "bar", VolumesToBackupAnnotation: "volume-1,volume-2,volume-3"},
-			expected:    []string{"volume-1", "volume-2", "volume-3"},
-		},
-	}
-
-	for _, test := range tests {
-		t.Run(test.name, func(t *testing.T) {
-			pod := &corev1api.Pod{}
-			pod.Annotations = test.annotations
-
-			res := GetVolumesToBackup(pod)
-
-			// sort to ensure good compare of slices
-			sort.Strings(test.expected)
-			sort.Strings(res)
-
-			assert.Equal(t, test.expected, res)
-		})
-	}
-}
-
-func TestGetVolumesByPod(t *testing.T) {
-	testCases := []struct {
-		name     string
-		pod      *corev1api.Pod
-		expected struct {
-			included []string
-			optedOut []string
-		}
-		defaultVolumesToFsBackup bool
-	}{
-		{
-			name:                     "should get PVs from VolumesToBackupAnnotation when defaultVolumesToFsBackup is false",
-			defaultVolumesToFsBackup: false,
-			pod: &corev1api.Pod{
-				ObjectMeta: metav1.ObjectMeta{
-					Annotations: map[string]string{
-						VolumesToBackupAnnotation: "pvbPV1,pvbPV2,pvbPV3",
-					},
-				},
-			},
-			expected: struct {
-				included []string
-				optedOut []string
-			}{
-				included: []string{"pvbPV1", "pvbPV2", "pvbPV3"},
-				optedOut: []string{},
-			},
-		},
-		{
-			name:                     "should get all pod volumes when defaultVolumesToFsBackup is true and no PVs are excluded",
-			defaultVolumesToFsBackup: true,
-			pod: &corev1api.Pod{
-				Spec: corev1api.PodSpec{
-					Volumes: []corev1api.Volume{
-						// PVB Volumes
-						{Name: "pvbPV1"}, {Name: "pvbPV2"}, {Name: "pvbPV3"},
-					},
-				},
-			},
-			expected: struct {
-				included []string
-				optedOut []string
-			}{
-				included: []string{"pvbPV1", "pvbPV2", "pvbPV3"},
-				optedOut: []string{},
-			},
-		},
-		{
-			name:                     "should get all pod volumes except ones excluded when defaultVolumesToFsBackup is true",
-			defaultVolumesToFsBackup: true,
-			pod: &corev1api.Pod{
-				ObjectMeta: metav1.ObjectMeta{
-					Annotations: map[string]string{
-						VolumesToExcludeAnnotation: "nonPvbPV1,nonPvbPV2,nonPvbPV3",
-					},
-				},
-				Spec: corev1api.PodSpec{
-					Volumes: []corev1api.Volume{
-						// PVB Volumes
-						{Name: "pvbPV1"}, {Name: "pvbPV2"}, {Name: "pvbPV3"},
-						/// Excluded from PVB through annotation
-						{Name: "nonPvbPV1"}, {Name: "nonPvbPV2"}, {Name: "nonPvbPV3"},
-					},
-				},
-			},
-			expected: struct {
-				included []string
-				optedOut []string
-			}{
-				included: []string{"pvbPV1", "pvbPV2", "pvbPV3"},
-				optedOut: []string{"nonPvbPV1", "nonPvbPV2", "nonPvbPV3"},
-			},
-		},
-		{
-			name:                     "should exclude default service account token from pod volume backup",
-			defaultVolumesToFsBackup: true,
-			pod: &corev1api.Pod{
-				Spec: corev1api.PodSpec{
-					Volumes: []corev1api.Volume{
-						// PVB Volumes
-						{Name: "pvbPV1"}, {Name: "pvbPV2"}, {Name: "pvbPV3"},
-						/// Excluded from PVB because colume mounting default service account token
-						{Name: "default-token-5xq45"},
-					},
-				},
-			},
-			expected: struct {
-				included []string
-				optedOut []string
-			}{
-				included: []string{"pvbPV1", "pvbPV2", "pvbPV3"},
-				optedOut: []string{},
-			},
-		},
-		{
-			name:                     "should exclude host path volumes from pod volume backups",
-			defaultVolumesToFsBackup: true,
-			pod: &corev1api.Pod{
-				ObjectMeta: metav1.ObjectMeta{
-					Annotations: map[string]string{
-						VolumesToExcludeAnnotation: "nonPvbPV1,nonPvbPV2,nonPvbPV3",
-					},
-				},
-				Spec: corev1api.PodSpec{
-					Volumes: []corev1api.Volume{
-						// PVB Volumes
-						{Name: "pvbPV1"}, {Name: "pvbPV2"}, {Name: "pvbPV3"},
-						/// Excluded from pod volume backup through annotation
-						{Name: "nonPvbPV1"}, {Name: "nonPvbPV2"}, {Name: "nonPvbPV3"},
-						// Excluded from pod volume backup because hostpath
-						{Name: "hostPath1", VolumeSource: corev1api.VolumeSource{HostPath: &corev1api.HostPathVolumeSource{Path: "/hostpathVol"}}},
-					},
-				},
-			},
-			expected: struct {
-				included []string
-				optedOut []string
-			}{
-				included: []string{"pvbPV1", "pvbPV2", "pvbPV3"},
-				optedOut: []string{"nonPvbPV1", "nonPvbPV2", "nonPvbPV3"},
-			},
-		},
-		{
-			name:                     "should exclude volumes mounting secrets",
-			defaultVolumesToFsBackup: true,
-			pod: &corev1api.Pod{
-				ObjectMeta: metav1.ObjectMeta{
-					Annotations: map[string]string{
-						VolumesToExcludeAnnotation: "nonPvbPV1,nonPvbPV2,nonPvbPV3",
-					},
-				},
-				Spec: corev1api.PodSpec{
-					Volumes: []corev1api.Volume{
-						// PVB Volumes
-						{Name: "pvbPV1"}, {Name: "pvbPV2"}, {Name: "pvbPV3"},
-						/// Excluded from pod volume backup through annotation
-						{Name: "nonPvbPV1"}, {Name: "nonPvbPV2"}, {Name: "nonPvbPV3"},
-						// Excluded from pod volume backup because hostpath
-						{Name: "superSecret", VolumeSource: corev1api.VolumeSource{Secret: &corev1api.SecretVolumeSource{SecretName: "super-secret"}}},
-					},
-				},
-			},
-			expected: struct {
-				included []string
-				optedOut []string
-			}{
-				included: []string{"pvbPV1", "pvbPV2", "pvbPV3"},
-				optedOut: []string{"nonPvbPV1", "nonPvbPV2", "nonPvbPV3"},
-			},
-		},
-		{
-			name:                     "should exclude volumes mounting config maps",
-			defaultVolumesToFsBackup: true,
-			pod: &corev1api.Pod{
-				ObjectMeta: metav1.ObjectMeta{
-					Annotations: map[string]string{
-						VolumesToExcludeAnnotation: "nonPvbPV1,nonPvbPV2,nonPvbPV3",
-					},
-				},
-				Spec: corev1api.PodSpec{
-					Volumes: []corev1api.Volume{
-						// PVB Volumes
-						{Name: "pvbPV1"}, {Name: "pvbPV2"}, {Name: "pvbPV3"},
-						/// Excluded from pod volume backup through annotation
-						{Name: "nonPvbPV1"}, {Name: "nonPvbPV2"}, {Name: "nonPvbPV3"},
-						// Excluded from pod volume backup because hostpath
-						{Name: "appCOnfig", VolumeSource: corev1api.VolumeSource{ConfigMap: &corev1api.ConfigMapVolumeSource{LocalObjectReference: corev1api.LocalObjectReference{Name: "app-config"}}}},
-					},
-				},
-			},
-			expected: struct {
-				included []string
-				optedOut []string
-			}{
-				included: []string{"pvbPV1", "pvbPV2", "pvbPV3"},
-				optedOut: []string{"nonPvbPV1", "nonPvbPV2", "nonPvbPV3"},
-			},
-		},
-		{
-			name:                     "should exclude projected volumes",
-			defaultVolumesToFsBackup: true,
-			pod: &corev1api.Pod{
-				ObjectMeta: metav1.ObjectMeta{
-					Annotations: map[string]string{
-						VolumesToExcludeAnnotation: "nonPvbPV1,nonPvbPV2,nonPvbPV3",
-					},
-				},
-				Spec: corev1api.PodSpec{
-					Volumes: []corev1api.Volume{
-						{Name: "pvbPV1"}, {Name: "pvbPV2"}, {Name: "pvbPV3"},
-						{
-							Name: "projected",
-							VolumeSource: corev1api.VolumeSource{
-								Projected: &corev1api.ProjectedVolumeSource{
-									Sources: []corev1api.VolumeProjection{{
-										Secret: &corev1api.SecretProjection{
-											LocalObjectReference: corev1api.LocalObjectReference{},
-											Items:                nil,
-											Optional:             nil,
-										},
-										DownwardAPI:         nil,
-										ConfigMap:           nil,
-										ServiceAccountToken: nil,
-									}},
-									DefaultMode: nil,
-								},
-							},
-						},
-					},
-				},
-			},
-			expected: struct {
-				included []string
-				optedOut []string
-			}{
-				included: []string{"pvbPV1", "pvbPV2", "pvbPV3"},
-				optedOut: []string{},
-			},
-		},
-		{
-			name:                     "should exclude DownwardAPI volumes",
-			defaultVolumesToFsBackup: true,
-			pod: &corev1api.Pod{
-				ObjectMeta: metav1.ObjectMeta{
-					Annotations: map[string]string{
-						VolumesToExcludeAnnotation: "nonPvbPV1,nonPvbPV2,nonPvbPV3",
-					},
-				},
-				Spec: corev1api.PodSpec{
-					Volumes: []corev1api.Volume{
-						{Name: "pvbPV1"}, {Name: "pvbPV2"}, {Name: "pvbPV3"},
-						{
-							Name: "downwardAPI",
-							VolumeSource: corev1api.VolumeSource{
-								DownwardAPI: &corev1api.DownwardAPIVolumeSource{
-									Items: []corev1api.DownwardAPIVolumeFile{
-										{
-											Path: "labels",
-											FieldRef: &corev1api.ObjectFieldSelector{
-												APIVersion: "v1",
-												FieldPath:  "metadata.labels",
-											},
-										},
-									},
-								},
-							},
-						},
-					},
-				},
-			},
-			expected: struct {
-				included []string
-				optedOut []string
-			}{
-				included: []string{"pvbPV1", "pvbPV2", "pvbPV3"},
-				optedOut: []string{},
-			},
-		},
-	}
-
-	for _, tc := range testCases {
-		t.Run(tc.name, func(t *testing.T) {
-			actualIncluded, actualOptedOut := GetVolumesByPod(tc.pod, tc.defaultVolumesToFsBackup)
-
-			sort.Strings(tc.expected.included)
-			sort.Strings(actualIncluded)
-			assert.Equal(t, tc.expected.included, actualIncluded)
-
-			sort.Strings(tc.expected.optedOut)
-			sort.Strings(actualOptedOut)
-			assert.Equal(t, tc.expected.optedOut, actualOptedOut)
-		})
-	}
-}

pkg/repository/provider/unified_repo.go

@@ -26,6 +26,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/kopia/kopia/repo"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 
@@ -188,11 +189,13 @@ func (urp *unifiedRepoProvider) PrepareRepo(ctx context.Context, param RepoParam
 		log.Debug("Repo has already been initialized remotely")
 		return nil
 	}
-	log.Infof("failed to connect to the repo: %v, will try to create it", err)
+	if !errors.Is(err, repo.ErrRepositoryNotInitialized) {
+		return errors.Wrap(err, "error to connect to backup repo")
+	}
 
 	err = urp.repoService.Init(ctx, *repoOption, true)
 	if err != nil {
-		return errors.Wrap(err, "error to init backup repo")
+		return errors.Wrap(err, "error to create backup repo")
 	}
 
 	log.Debug("Prepare repo complete")

pkg/repository/provider/unified_repo_test.go

@@ -23,6 +23,7 @@ import (
 	"testing"
 
 	awscredentials "github.com/aws/aws-sdk-go/aws/credentials"
+	"github.com/kopia/kopia/repo"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
@@ -648,7 +649,28 @@ func TestPrepareRepo(t *testing.T) {
 				}
 				return errors.New("fake-error-2")
 			},
-			expectedErr: "error to init backup repo: fake-error-2",
+			expectedErr: "error to connect to backup repo: fake-error-1",
+		},
+		{
+			name:            "not initialize",
+			getter:          new(credmock.SecretStore),
+			credStoreReturn: "fake-password",
+			funcTable: localFuncTable{
+				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string) (map[string]string, error) {
+					return map[string]string{}, nil
+				},
+				getStorageCredentials: func(*velerov1api.BackupStorageLocation, velerocredentials.FileStore) (map[string]string, error) {
+					return map[string]string{}, nil
+				},
+			},
+			repoService: new(reposervicenmocks.BackupRepoService),
+			retFuncInit: func(ctx context.Context, repoOption udmrepo.RepoOptions, createNew bool) error {
+				if !createNew {
+					return repo.ErrRepositoryNotInitialized
+				}
+				return errors.New("fake-error-2")
+			},
+			expectedErr: "error to create backup repo: fake-error-2",
 		},
 	}
 

pkg/uploader/kopia/block_backup.go

@@ -0,0 +1,55 @@
+/*
+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 kopia
+
+import (
+	"os"
+	"syscall"
+
+	"github.com/kopia/kopia/fs"
+	"github.com/kopia/kopia/fs/virtualfs"
+	"github.com/pkg/errors"
+)
+
+const ErrNotPermitted = "operation not permitted"
+
+func getLocalBlockEntry(sourcePath string) (fs.Entry, error) {
+	source, err := resolveSymlink(sourcePath)
+	if err != nil {
+		return nil, errors.Wrap(err, "resolveSymlink")
+	}
+
+	fileInfo, err := os.Lstat(source)
+	if err != nil {
+		return nil, errors.Wrapf(err, "unable to get the source device information %s", source)
+	}
+
+	if (fileInfo.Sys().(*syscall.Stat_t).Mode & syscall.S_IFMT) != syscall.S_IFBLK {
+		return nil, errors.Errorf("source path %s is not a block device", source)
+	}
+
+	device, err := os.Open(source)
+	if err != nil {
+		if os.IsPermission(err) || err.Error() == ErrNotPermitted {
+			return nil, errors.Wrapf(err, "no permission to open the source device %s, make sure that node agent is running in privileged mode", source)
+		}
+		return nil, errors.Wrapf(err, "unable to open the source device %s", source)
+	}
+
+	sf := virtualfs.StreamingFileFromReader(source, device)
+	return virtualfs.NewStaticDirectory(source, []fs.Entry{sf}), nil
+}

pkg/uploader/kopia/block_restore.go

@@ -0,0 +1,99 @@
+/*
+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 kopia
+
+import (
+	"context"
+	"io"
+	"os"
+	"path/filepath"
+	"syscall"
+
+	"github.com/kopia/kopia/fs"
+	"github.com/kopia/kopia/snapshot/restore"
+	"github.com/pkg/errors"
+)
+
+type BlockOutput struct {
+	*restore.FilesystemOutput
+
+	targetFileName string
+}
+
+var _ restore.Output = &BlockOutput{}
+
+const bufferSize = 128 * 1024
+
+func (o *BlockOutput) WriteFile(ctx context.Context, relativePath string, remoteFile fs.File) error {
+	remoteReader, err := remoteFile.Open(ctx)
+	if err != nil {
+		return errors.Wrapf(err, "failed to open remote file %s", remoteFile.Name())
+	}
+	defer remoteReader.Close()
+
+	targetFile, err := os.Create(o.targetFileName)
+	if err != nil {
+		return errors.Wrapf(err, "failed to open file %s", o.targetFileName)
+	}
+	defer targetFile.Close()
+
+	buffer := make([]byte, bufferSize)
+
+	readData := true
+	for readData {
+		bytesToWrite, err := remoteReader.Read(buffer)
+		if err != nil {
+			if err != io.EOF {
+				return errors.Wrapf(err, "failed to read data from remote file %s", o.targetFileName)
+			}
+			readData = false
+		}
+
+		if bytesToWrite > 0 {
+			offset := 0
+			for bytesToWrite > 0 {
+				if bytesWritten, err := targetFile.Write(buffer[offset:bytesToWrite]); err == nil {
+					bytesToWrite -= bytesWritten
+					offset += bytesWritten
+				} else {
+					return errors.Wrapf(err, "failed to write data to file %s", o.targetFileName)
+				}
+			}
+		}
+	}
+
+	return nil
+}
+
+func (o *BlockOutput) BeginDirectory(ctx context.Context, relativePath string, e fs.Directory) error {
+	var err error
+	o.targetFileName, err = filepath.EvalSymlinks(o.TargetPath)
+	if err != nil {
+		return errors.Wrapf(err, "unable to evaluate symlinks for %s", o.targetFileName)
+	}
+
+	fileInfo, err := os.Lstat(o.targetFileName)
+	if err != nil {
+		return errors.Wrapf(err, "unable to get the target device information for %s", o.TargetPath)
+	}
+
+	if (fileInfo.Sys().(*syscall.Stat_t).Mode & syscall.S_IFMT) != syscall.S_IFBLK {
+		return errors.Errorf("target file %s is not a block device", o.TargetPath)
+	}
+
+	return nil
+}

pkg/uploader/kopia/snapshot.go

@@ -68,6 +68,11 @@ func newOptionalInt(b int) *policy.OptionalInt {
 	return &ob
 }
 
+func newOptionalInt64(b int64) *policy.OptionalInt64 {
+	ob := policy.OptionalInt64(b)
+	return &ob
+}
+
 func newOptionalBool(b bool) *policy.OptionalBool {
 	ob := policy.OptionalBool(b)
 	return &ob
@@ -88,7 +93,7 @@ func getDefaultPolicy() *policy.Policy {
 		},
 		UploadPolicy: policy.UploadPolicy{
 			MaxParallelFileReads:    newOptionalInt(runtime.NumCPU()),
-			ParallelUploadAboveSize: nil,
+			ParallelUploadAboveSize: newOptionalInt64(math.MaxInt64),
 		},
 		SchedulingPolicy: policy.SchedulingPolicy{
 			Manual: true,
@@ -106,6 +111,11 @@ func setupDefaultPolicy(ctx context.Context, rep repo.RepositoryWriter, sourceIn
 		return nil, errors.Wrap(err, "error to set policy")
 	}
 
+	err = rep.Flush(ctx)
+	if err != nil {
+		return nil, errors.Wrap(err, "error to flush repo")
+	}
+
 	// retrieve policy from repo
 	policyTree, err := treeForSourceFunc(ctx, rep, sourceInfo)
 	if err != nil {
@@ -121,25 +131,22 @@ func Backup(ctx context.Context, fsUploader SnapshotUploader, repoWriter repo.Re
 	if fsUploader == nil {
 		return nil, false, errors.New("get empty kopia uploader")
 	}
-
-	if volMode == uploader.PersistentVolumeBlock {
-		return nil, false, errors.New("unable to handle block storage")
-	}
-
-	dir, err := filepath.Abs(sourcePath)
+	source, err := filepath.Abs(sourcePath)
 	if err != nil {
 		return nil, false, errors.Wrapf(err, "Invalid source path '%s'", sourcePath)
 	}
 
-	// to be consistent with restic when backup empty dir returns one error for upper logic handle
-	dirs, err := os.ReadDir(dir)
-	if err != nil {
-		return nil, false, errors.Wrapf(err, "Unable to read dir in path %s", dir)
-	} else if len(dirs) == 0 {
-		return nil, true, nil
+	if volMode == uploader.PersistentVolumeFilesystem {
+		// to be consistent with restic when backup empty dir returns one error for upper logic handle
+		dirs, err := os.ReadDir(source)
+		if err != nil {
+			return nil, false, errors.Wrapf(err, "Unable to read dir in path %s", source)
+		} else if len(dirs) == 0 {
+			return nil, true, nil
+		}
 	}
 
-	dir = filepath.Clean(dir)
+	source = filepath.Clean(source)
 
 	sourceInfo := snapshot.SourceInfo{
 		UserName: udmrepo.GetRepoUser(),
@@ -147,16 +154,25 @@ func Backup(ctx context.Context, fsUploader SnapshotUploader, repoWriter repo.Re
 		Path:     filepath.Clean(realSource),
 	}
 	if realSource == "" {
-		sourceInfo.Path = dir
+		sourceInfo.Path = source
 	}
 
-	rootDir, err := getLocalFSEntry(dir)
-	if err != nil {
-		return nil, false, errors.Wrap(err, "Unable to get local filesystem entry")
+	var sourceEntry fs.Entry
+
+	if volMode == uploader.PersistentVolumeBlock {
+		sourceEntry, err = getLocalBlockEntry(source)
+		if err != nil {
+			return nil, false, errors.Wrap(err, "unable to get local block device entry")
+		}
+	} else {
+		sourceEntry, err = getLocalFSEntry(source)
+		if err != nil {
+			return nil, false, errors.Wrap(err, "unable to get local filesystem entry")
+		}
 	}
 
 	kopiaCtx := kopia.SetupKopiaLog(ctx, log)
-	snapID, snapshotSize, err := SnapshotSource(kopiaCtx, repoWriter, fsUploader, sourceInfo, rootDir, forceFull, parentSnapshot, tags, log, "Kopia Uploader")
+	snapID, snapshotSize, err := SnapshotSource(kopiaCtx, repoWriter, fsUploader, sourceInfo, sourceEntry, forceFull, parentSnapshot, tags, log, "Kopia Uploader")
 	if err != nil {
 		return nil, false, err
 	}
@@ -338,7 +354,8 @@ func findPreviousSnapshotManifest(ctx context.Context, rep repo.Repository, sour
 }
 
 // Restore restore specific sourcePath with given snapshotID and update progress
-func Restore(ctx context.Context, rep repo.RepositoryWriter, progress *Progress, snapshotID, dest string, log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, error) {
+func Restore(ctx context.Context, rep repo.RepositoryWriter, progress *Progress, snapshotID, dest string, volMode uploader.PersistentVolumeMode,
+	log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, error) {
 	log.Info("Start to restore...")
 
 	kopiaCtx := kopia.SetupKopiaLog(ctx, log)
@@ -360,7 +377,7 @@ func Restore(ctx context.Context, rep repo.RepositoryWriter, progress *Progress,
 		return 0, 0, errors.Wrapf(err, "Unable to resolve path %v", dest)
 	}
 
-	output := &restore.FilesystemOutput{
+	fsOutput := &restore.FilesystemOutput{
 		TargetPath:             path,
 		OverwriteDirectories:   true,
 		OverwriteFiles:         true,
@@ -368,11 +385,18 @@ func Restore(ctx context.Context, rep repo.RepositoryWriter, progress *Progress,
 		IgnorePermissionErrors: true,
 	}
 
-	err = output.Init(ctx)
+	err = fsOutput.Init(ctx)
 	if err != nil {
 		return 0, 0, errors.Wrap(err, "error to init output")
 	}
 
+	var output restore.Output = fsOutput
+	if volMode == uploader.PersistentVolumeBlock {
+		output = &BlockOutput{
+			FilesystemOutput: fsOutput,
+		}
+	}
+
 	stat, err := restoreEntryFunc(kopiaCtx, rep, output, rootEntry, restore.Options{
 		Parallel:               runtime.NumCPU(),
 		RestoreDirEntryAtDepth: math.MaxInt32,

pkg/uploader/kopia/snapshot_test.go

@@ -23,6 +23,7 @@ import (
 	"time"
 
 	"github.com/kopia/kopia/fs"
+	"github.com/kopia/kopia/fs/virtualfs"
 	"github.com/kopia/kopia/repo"
 	"github.com/kopia/kopia/repo/manifest"
 	"github.com/kopia/kopia/snapshot"
@@ -594,11 +595,11 @@ func TestBackup(t *testing.T) {
 			expectedError: errors.New("Unable to read dir"),
 		},
 		{
-			name:          "Unable to handle block mode",
+			name:          "Source path is not a block device",
 			sourcePath:    "/",
 			tags:          nil,
 			volMode:       uploader.PersistentVolumeBlock,
-			expectedError: errors.New("unable to handle block storage"),
+			expectedError: errors.New("source path / is not a block device"),
 		},
 	}
 
@@ -660,6 +661,7 @@ func TestRestore(t *testing.T) {
 		expectedBytes       int64
 		expectedCount       int32
 		expectedError       error
+		volMode             uploader.PersistentVolumeMode
 	}
 
 	// Define test cases
@@ -697,6 +699,46 @@ func TestRestore(t *testing.T) {
 			snapshotID:    "snapshot-123",
 			expectedError: nil,
 		},
+		{
+			name: "Expect block volume successful",
+			filesystemEntryFunc: func(ctx context.Context, rep repo.Repository, rootID string, consistentAttributes bool) (fs.Entry, error) {
+				return snapshotfs.EntryFromDirEntry(rep, &snapshot.DirEntry{Type: snapshot.EntryTypeFile}), nil
+			},
+			restoreEntryFunc: func(ctx context.Context, rep repo.Repository, output restore.Output, rootEntry fs.Entry, options restore.Options) (restore.Stats, error) {
+				return restore.Stats{}, nil
+			},
+			snapshotID:    "snapshot-123",
+			expectedError: nil,
+			volMode:       uploader.PersistentVolumeBlock,
+		},
+		{
+			name: "Unable to evaluate symlinks for block volume",
+			filesystemEntryFunc: func(ctx context.Context, rep repo.Repository, rootID string, consistentAttributes bool) (fs.Entry, error) {
+				return snapshotfs.EntryFromDirEntry(rep, &snapshot.DirEntry{Type: snapshot.EntryTypeFile}), nil
+			},
+			restoreEntryFunc: func(ctx context.Context, rep repo.Repository, output restore.Output, rootEntry fs.Entry, options restore.Options) (restore.Stats, error) {
+				err := output.BeginDirectory(ctx, "fake-dir", virtualfs.NewStaticDirectory("fake-dir", nil))
+				return restore.Stats{}, err
+			},
+			snapshotID:    "snapshot-123",
+			expectedError: errors.New("unable to evaluate symlinks for"),
+			volMode:       uploader.PersistentVolumeBlock,
+			dest:          "/wrong-dest",
+		},
+		{
+			name: "Target file is not a block device",
+			filesystemEntryFunc: func(ctx context.Context, rep repo.Repository, rootID string, consistentAttributes bool) (fs.Entry, error) {
+				return snapshotfs.EntryFromDirEntry(rep, &snapshot.DirEntry{Type: snapshot.EntryTypeFile}), nil
+			},
+			restoreEntryFunc: func(ctx context.Context, rep repo.Repository, output restore.Output, rootEntry fs.Entry, options restore.Options) (restore.Stats, error) {
+				err := output.BeginDirectory(ctx, "fake-dir", virtualfs.NewStaticDirectory("fake-dir", nil))
+				return restore.Stats{}, err
+			},
+			snapshotID:    "snapshot-123",
+			expectedError: errors.New("target file /tmp is not a block device"),
+			volMode:       uploader.PersistentVolumeBlock,
+			dest:          "/tmp",
+		},
 	}
 
 	em := &manifest.EntryMetadata{
@@ -706,6 +748,10 @@ func TestRestore(t *testing.T) {
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
+			if tc.volMode == "" {
+				tc.volMode = uploader.PersistentVolumeFilesystem
+			}
+
 			if tc.invalidManifestType {
 				em.Labels[manifest.TypeLabelKey] = ""
 			} else {
@@ -725,7 +771,7 @@ func TestRestore(t *testing.T) {
 			repoWriterMock.On("OpenObject", mock.Anything, mock.Anything).Return(em, nil)
 
 			progress := new(Progress)
-			bytesRestored, fileCount, err := Restore(context.Background(), repoWriterMock, progress, tc.snapshotID, tc.dest, logrus.New(), nil)
+			bytesRestored, fileCount, err := Restore(context.Background(), repoWriterMock, progress, tc.snapshotID, tc.dest, tc.volMode, logrus.New(), nil)
 
 			// Check if the returned error matches the expected error
 			if tc.expectedError != nil {

pkg/uploader/provider/kopia.go

@@ -128,11 +128,6 @@ func (kp *kopiaProvider) RunBackup(
 		return "", false, errors.New("path is empty")
 	}
 
-	// For now, error on block mode
-	if volMode == uploader.PersistentVolumeBlock {
-		return "", false, errors.New("unable to currently support block mode")
-	}
-
 	log := kp.log.WithFields(logrus.Fields{
 		"path":           path,
 		"realSource":     realSource,
@@ -214,10 +209,6 @@ func (kp *kopiaProvider) RunRestore(
 		"volumePath": volumePath,
 	})
 
-	if volMode == uploader.PersistentVolumeBlock {
-		return errors.New("unable to currently support block mode")
-	}
-
 	repoWriter := kopia.NewShimRepo(kp.bkRepo)
 	progress := new(kopia.Progress)
 	progress.InitThrottle(restoreProgressCheckInterval)
@@ -235,7 +226,7 @@ func (kp *kopiaProvider) RunRestore(
 	// We use the cancel channel to control the restore cancel, so don't pass a context with cancel to Kopia restore.
 	// Otherwise, Kopia restore will not response to the cancel control but return an arbitrary error.
 	// Kopia restore cancel is not designed as well as Kopia backup which uses the context to control backup cancel all the way.
-	size, fileCount, err := RestoreFunc(context.Background(), repoWriter, progress, snapshotID, volumePath, log, restoreCancel)
+	size, fileCount, err := RestoreFunc(context.Background(), repoWriter, progress, snapshotID, volumePath, volMode, log, restoreCancel)
 
 	if err != nil {
 		return errors.Wrapf(err, "Failed to run kopia restore")

pkg/uploader/provider/kopia_test.go

@@ -94,12 +94,12 @@ func TestRunBackup(t *testing.T) {
 			notError: false,
 		},
 		{
-			name: "error on vol mode",
+			name: "success to backup block mode volume",
 			hookBackupFunc: func(ctx context.Context, fsUploader kopia.SnapshotUploader, repoWriter repo.RepositoryWriter, sourcePath string, realSource string, forceFull bool, parentSnapshot string, volMode uploader.PersistentVolumeMode, tags map[string]string, log logrus.FieldLogger) (*uploader.SnapshotInfo, bool, error) {
-				return nil, true, nil
+				return &uploader.SnapshotInfo{}, false, nil
 			},
 			volMode:  uploader.PersistentVolumeBlock,
-			notError: false,
+			notError: true,
 		},
 	}
 	for _, tc := range testCases {
@@ -125,31 +125,31 @@ func TestRunRestore(t *testing.T) {
 
 	testCases := []struct {
 		name            string
-		hookRestoreFunc func(ctx context.Context, rep repo.RepositoryWriter, progress *kopia.Progress, snapshotID, dest string, log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, error)
+		hookRestoreFunc func(ctx context.Context, rep repo.RepositoryWriter, progress *kopia.Progress, snapshotID, dest string, volMode uploader.PersistentVolumeMode, log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, error)
 		notError        bool
 		volMode         uploader.PersistentVolumeMode
 	}{
 		{
 			name: "normal restore",
-			hookRestoreFunc: func(ctx context.Context, rep repo.RepositoryWriter, progress *kopia.Progress, snapshotID, dest string, log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, error) {
+			hookRestoreFunc: func(ctx context.Context, rep repo.RepositoryWriter, progress *kopia.Progress, snapshotID, dest string, volMode uploader.PersistentVolumeMode, log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, error) {
 				return 0, 0, nil
 			},
 			notError: true,
 		},
 		{
 			name: "failed to restore",
-			hookRestoreFunc: func(ctx context.Context, rep repo.RepositoryWriter, progress *kopia.Progress, snapshotID, dest string, log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, error) {
+			hookRestoreFunc: func(ctx context.Context, rep repo.RepositoryWriter, progress *kopia.Progress, snapshotID, dest string, volMode uploader.PersistentVolumeMode, log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, error) {
 				return 0, 0, errors.New("failed to restore")
 			},
 			notError: false,
 		},
 		{
-			name: "failed to restore block mode",
-			hookRestoreFunc: func(ctx context.Context, rep repo.RepositoryWriter, progress *kopia.Progress, snapshotID, dest string, log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, error) {
-				return 0, 0, errors.New("failed to restore")
+			name: "normal block mode restore",
+			hookRestoreFunc: func(ctx context.Context, rep repo.RepositoryWriter, progress *kopia.Progress, snapshotID, dest string, volMode uploader.PersistentVolumeMode, log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, error) {
+				return 0, 0, nil
 			},
 			volMode:  uploader.PersistentVolumeBlock,
-			notError: false,
+			notError: true,
 		},
 	}
 

pkg/uploader/provider/restic_test.go

@@ -212,6 +212,9 @@ func TestResticRunRestore(t *testing.T) {
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
+			if tc.volMode == "" {
+				tc.volMode = uploader.PersistentVolumeFilesystem
+			}
 			resticRestoreCMDFunc = tc.hookResticRestoreFunc
 			if tc.volMode == "" {
 				tc.volMode = uploader.PersistentVolumeFilesystem

pkg/util/kube/pvc_pv.go

@@ -316,3 +316,23 @@ func WaitPVBound(ctx context.Context, pvGetter corev1client.CoreV1Interface, pvN
 func IsPVCBound(pvc *corev1api.PersistentVolumeClaim) bool {
 	return pvc.Spec.VolumeName != ""
 }
+
+// MakePodPVCAttachment returns the volume mounts and devices for a pod needed to attach a PVC
+func MakePodPVCAttachment(volumeName string, volumeMode *corev1api.PersistentVolumeMode) ([]corev1api.VolumeMount, []corev1api.VolumeDevice) {
+	var volumeMounts []corev1api.VolumeMount = nil
+	var volumeDevices []corev1api.VolumeDevice = nil
+
+	if volumeMode != nil && *volumeMode == corev1api.PersistentVolumeBlock {
+		volumeDevices = []corev1api.VolumeDevice{{
+			Name:       volumeName,
+			DevicePath: "/" + volumeName,
+		}}
+	} else {
+		volumeMounts = []corev1api.VolumeMount{{
+			Name:      volumeName,
+			MountPath: "/" + volumeName,
+		}}
+	}
+
+	return volumeMounts, volumeDevices
+}

pkg/util/kube/utils.go

@@ -35,6 +35,7 @@ import (
 	corev1client "k8s.io/client-go/kubernetes/typed/core/v1"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
+	"github.com/vmware-tanzu/velero/pkg/uploader"
 	"github.com/vmware-tanzu/velero/pkg/util/filesystem"
 )
 
@@ -50,6 +51,8 @@ const (
 	KubeAnnSelectedNode           = "volume.kubernetes.io/selected-node"
 )
 
+var ErrorPodVolumeIsNotPVC = errors.New("pod volume is not a PVC")
+
 // NamespaceAndName returns a string in the format <namespace>/<name>
 func NamespaceAndName(objMeta metav1.Object) string {
 	if objMeta.GetNamespace() == "" {
@@ -122,6 +125,57 @@ func EnsureNamespaceExistsAndIsReady(namespace *corev1api.Namespace, client core
 // where the specified volume lives.
 // For volumes with a CSIVolumeSource, append "/mount" to the directory name.
 func GetVolumeDirectory(ctx context.Context, log logrus.FieldLogger, pod *corev1api.Pod, volumeName string, cli client.Client) (string, error) {
+	pvc, pv, volume, err := GetPodPVCVolume(ctx, log, pod, volumeName, cli)
+	if err != nil {
+		// This case implies the administrator created the PV and attached it directly, without PVC.
+		// Note that only one VolumeSource can be populated per Volume on a pod
+		if err == ErrorPodVolumeIsNotPVC {
+			if volume.VolumeSource.CSI != nil {
+				return volume.Name + "/mount", nil
+			}
+			return volume.Name, nil
+		}
+		return "", errors.WithStack(err)
+	}
+
+	// Most common case is that we have a PVC VolumeSource, and we need to check the PV it points to for a CSI source.
+	// PV's been created with a CSI source.
+	isProvisionedByCSI, err := isProvisionedByCSI(log, pv, cli)
+	if err != nil {
+		return "", errors.WithStack(err)
+	}
+	if isProvisionedByCSI {
+		if pv.Spec.VolumeMode != nil && *pv.Spec.VolumeMode == corev1api.PersistentVolumeBlock {
+			return pvc.Spec.VolumeName, nil
+		}
+		return pvc.Spec.VolumeName + "/mount", nil
+	}
+
+	return pvc.Spec.VolumeName, nil
+}
+
+// GetVolumeMode gets the uploader.PersistentVolumeMode of the volume.
+func GetVolumeMode(ctx context.Context, log logrus.FieldLogger, pod *corev1api.Pod, volumeName string, cli client.Client) (
+	uploader.PersistentVolumeMode, error) {
+	_, pv, _, err := GetPodPVCVolume(ctx, log, pod, volumeName, cli)
+
+	if err != nil {
+		if err == ErrorPodVolumeIsNotPVC {
+			return uploader.PersistentVolumeFilesystem, nil
+		}
+		return "", errors.WithStack(err)
+	}
+
+	if pv.Spec.VolumeMode != nil && *pv.Spec.VolumeMode == corev1api.PersistentVolumeBlock {
+		return uploader.PersistentVolumeBlock, nil
+	}
+	return uploader.PersistentVolumeFilesystem, nil
+}
+
+// GetPodPVCVolume gets the PVC, PV and volume for a pod volume name.
+// Returns pod volume in case of ErrorPodVolumeIsNotPVC error
+func GetPodPVCVolume(ctx context.Context, log logrus.FieldLogger, pod *corev1api.Pod, volumeName string, cli client.Client) (
+	*corev1api.PersistentVolumeClaim, *corev1api.PersistentVolume, *corev1api.Volume, error) {
 	var volume *corev1api.Volume
 
 	for i := range pod.Spec.Volumes {
@@ -132,41 +186,26 @@ func GetVolumeDirectory(ctx context.Context, log logrus.FieldLogger, pod *corev1
 	}
 
 	if volume == nil {
-		return "", errors.New("volume not found in pod")
+		return nil, nil, nil, errors.New("volume not found in pod")
 	}
 
-	// This case implies the administrator created the PV and attached it directly, without PVC.
-	// Note that only one VolumeSource can be populated per Volume on a pod
 	if volume.VolumeSource.PersistentVolumeClaim == nil {
-		if volume.VolumeSource.CSI != nil {
-			return volume.Name + "/mount", nil
-		}
-		return volume.Name, nil
+		return nil, nil, volume, ErrorPodVolumeIsNotPVC // There is a pod volume but it is not a PVC
 	}
 
-	// Most common case is that we have a PVC VolumeSource, and we need to check the PV it points to for a CSI source.
 	pvc := &corev1api.PersistentVolumeClaim{}
 	err := cli.Get(ctx, client.ObjectKey{Namespace: pod.Namespace, Name: volume.VolumeSource.PersistentVolumeClaim.ClaimName}, pvc)
 	if err != nil {
-		return "", errors.WithStack(err)
+		return nil, nil, nil, errors.WithStack(err)
 	}
 
 	pv := &corev1api.PersistentVolume{}
 	err = cli.Get(ctx, client.ObjectKey{Name: pvc.Spec.VolumeName}, pv)
 	if err != nil {
-		return "", errors.WithStack(err)
+		return nil, nil, nil, errors.WithStack(err)
 	}
 
-	// PV's been created with a CSI source.
-	isProvisionedByCSI, err := isProvisionedByCSI(log, pv, cli)
-	if err != nil {
-		return "", errors.WithStack(err)
-	}
-	if isProvisionedByCSI {
-		return pvc.Spec.VolumeName + "/mount", nil
-	}
-
-	return pvc.Spec.VolumeName, nil
+	return pvc, pv, volume, nil
 }
 
 // isProvisionedByCSI function checks whether this is a CSI PV by annotation.

pkg/util/kube/utils_test.go

@@ -38,6 +38,7 @@ import (
 
 	"github.com/vmware-tanzu/velero/pkg/builder"
 	velerotest "github.com/vmware-tanzu/velero/pkg/test"
+	"github.com/vmware-tanzu/velero/pkg/uploader"
 )
 
 func TestNamespaceAndName(t *testing.T) {
@@ -164,6 +165,13 @@ func TestGetVolumeDirectorySuccess(t *testing.T) {
 			pv:   builder.ForPersistentVolume("a-pv").CSI("csi.test.com", "provider-volume-id").Result(),
 			want: "a-pv/mount",
 		},
+		{
+			name: "Block CSI volume with a PVC/PV does not append '/mount' to the volume name",
+			pod:  builder.ForPod("ns-1", "my-pod").Volumes(builder.ForVolume("my-vol").PersistentVolumeClaimSource("my-pvc").Result()).Result(),
+			pvc:  builder.ForPersistentVolumeClaim("ns-1", "my-pvc").VolumeName("a-pv").Result(),
+			pv:   builder.ForPersistentVolume("a-pv").CSI("csi.test.com", "provider-volume-id").VolumeMode(corev1.PersistentVolumeBlock).Result(),
+			want: "a-pv",
+		},
 		{
 			name: "CSI volume mounted without a PVC appends '/mount' to the volume name",
 			pod:  builder.ForPod("ns-1", "my-pod").Volumes(builder.ForVolume("my-vol").CSISource("csi.test.com").Result()).Result(),
@@ -211,6 +219,54 @@ func TestGetVolumeDirectorySuccess(t *testing.T) {
 	}
 }
 
+// TestGetVolumeModeSuccess tests the GetVolumeMode function
+func TestGetVolumeModeSuccess(t *testing.T) {
+	tests := []struct {
+		name string
+		pod  *corev1.Pod
+		pvc  *corev1.PersistentVolumeClaim
+		pv   *corev1.PersistentVolume
+		want uploader.PersistentVolumeMode
+	}{
+		{
+			name: "Filesystem PVC volume",
+			pod:  builder.ForPod("ns-1", "my-pod").Volumes(builder.ForVolume("my-vol").PersistentVolumeClaimSource("my-pvc").Result()).Result(),
+			pvc:  builder.ForPersistentVolumeClaim("ns-1", "my-pvc").VolumeName("a-pv").Result(),
+			pv:   builder.ForPersistentVolume("a-pv").VolumeMode(corev1.PersistentVolumeFilesystem).Result(),
+			want: uploader.PersistentVolumeFilesystem,
+		},
+		{
+			name: "Block PVC volume",
+			pod:  builder.ForPod("ns-1", "my-pod").Volumes(builder.ForVolume("my-vol").PersistentVolumeClaimSource("my-pvc").Result()).Result(),
+			pvc:  builder.ForPersistentVolumeClaim("ns-1", "my-pvc").VolumeName("a-pv").Result(),
+			pv:   builder.ForPersistentVolume("a-pv").VolumeMode(corev1.PersistentVolumeBlock).Result(),
+			want: uploader.PersistentVolumeBlock,
+		},
+		{
+			name: "Pod volume without a PVC",
+			pod:  builder.ForPod("ns-1", "my-pod").Volumes(builder.ForVolume("my-vol").Result()).Result(),
+			want: uploader.PersistentVolumeFilesystem,
+		},
+	}
+
+	for _, tc := range tests {
+		clientBuilder := fake.NewClientBuilder()
+
+		if tc.pvc != nil {
+			clientBuilder = clientBuilder.WithObjects(tc.pvc)
+		}
+		if tc.pv != nil {
+			clientBuilder = clientBuilder.WithObjects(tc.pv)
+		}
+
+		// Function under test
+		mode, err := GetVolumeMode(context.Background(), logrus.StandardLogger(), tc.pod, tc.pod.Spec.Volumes[0].Name, clientBuilder.Build())
+
+		require.NoError(t, err)
+		assert.Equal(t, tc.want, mode)
+	}
+}
+
 func TestIsV1Beta1CRDReady(t *testing.T) {
 	tests := []struct {
 		name string

pkg/util/podvolume/pod_volume.go

@@ -0,0 +1,108 @@
+/*
+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 podvolume
+
+import (
+	"strings"
+
+	corev1api "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+)
+
+// GetVolumesByPod returns a list of volume names to backup for the provided pod.
+func GetVolumesByPod(pod *corev1api.Pod, defaultVolumesToFsBackup bool) ([]string, []string) {
+	// tracks the volumes that have been explicitly opted out of backup via the annotation in the pod
+	optedOutVolumes := make([]string, 0)
+
+	if !defaultVolumesToFsBackup {
+		return GetVolumesToBackup(pod), optedOutVolumes
+	}
+
+	volsToExclude := getVolumesToExclude(pod)
+	podVolumes := []string{}
+	for _, pv := range pod.Spec.Volumes {
+		// cannot backup hostpath volumes as they are not mounted into /var/lib/kubelet/pods
+		// and therefore not accessible to the node agent daemon set.
+		if pv.HostPath != nil {
+			continue
+		}
+		// don't backup volumes mounting secrets. Secrets will be backed up separately.
+		if pv.Secret != nil {
+			continue
+		}
+		// don't backup volumes mounting config maps. Config maps will be backed up separately.
+		if pv.ConfigMap != nil {
+			continue
+		}
+		// don't backup volumes mounted as projected volumes, all data in those come from kube state.
+		if pv.Projected != nil {
+			continue
+		}
+		// don't backup DownwardAPI volumes, all data in those come from kube state.
+		if pv.DownwardAPI != nil {
+			continue
+		}
+		// don't backup volumes that are included in the exclude list.
+		if contains(volsToExclude, pv.Name) {
+			optedOutVolumes = append(optedOutVolumes, pv.Name)
+			continue
+		}
+		// don't include volumes that mount the default service account token.
+		if strings.HasPrefix(pv.Name, "default-token") {
+			continue
+		}
+		podVolumes = append(podVolumes, pv.Name)
+	}
+	return podVolumes, optedOutVolumes
+}
+
+// GetVolumesToBackup returns a list of volume names to backup for
+// the provided pod.
+// Deprecated: Use GetVolumesByPod instead.
+func GetVolumesToBackup(obj metav1.Object) []string {
+	annotations := obj.GetAnnotations()
+	if annotations == nil {
+		return nil
+	}
+
+	backupsValue := annotations[velerov1api.VolumesToBackupAnnotation]
+	if backupsValue == "" {
+		return nil
+	}
+
+	return strings.Split(backupsValue, ",")
+}
+
+func getVolumesToExclude(obj metav1.Object) []string {
+	annotations := obj.GetAnnotations()
+	if annotations == nil {
+		return nil
+	}
+
+	return strings.Split(annotations[velerov1api.VolumesToExcludeAnnotation], ",")
+}
+
+func contains(list []string, k string) bool {
+	for _, i := range list {
+		if i == k {
+			return true
+		}
+	}
+	return false
+}

pkg/util/podvolume/pod_volume_test.go

@@ -0,0 +1,347 @@
+/*
+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 podvolume
+
+import (
+	"sort"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	corev1api "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+)
+
+func TestGetVolumesToBackup(t *testing.T) {
+	tests := []struct {
+		name        string
+		annotations map[string]string
+		expected    []string
+	}{
+		{
+			name:        "nil annotations",
+			annotations: nil,
+			expected:    nil,
+		},
+		{
+			name:        "no volumes to backup",
+			annotations: map[string]string{"foo": "bar"},
+			expected:    nil,
+		},
+		{
+			name:        "one volume to backup",
+			annotations: map[string]string{"foo": "bar", velerov1api.VolumesToBackupAnnotation: "volume-1"},
+			expected:    []string{"volume-1"},
+		},
+		{
+			name:        "multiple volumes to backup",
+			annotations: map[string]string{"foo": "bar", velerov1api.VolumesToBackupAnnotation: "volume-1,volume-2,volume-3"},
+			expected:    []string{"volume-1", "volume-2", "volume-3"},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			pod := &corev1api.Pod{}
+			pod.Annotations = test.annotations
+
+			res := GetVolumesToBackup(pod)
+
+			// sort to ensure good compare of slices
+			sort.Strings(test.expected)
+			sort.Strings(res)
+
+			assert.Equal(t, test.expected, res)
+		})
+	}
+}
+
+func TestGetVolumesByPod(t *testing.T) {
+	testCases := []struct {
+		name     string
+		pod      *corev1api.Pod
+		expected struct {
+			included []string
+			optedOut []string
+		}
+		defaultVolumesToFsBackup bool
+	}{
+		{
+			name:                     "should get PVs from VolumesToBackupAnnotation when defaultVolumesToFsBackup is false",
+			defaultVolumesToFsBackup: false,
+			pod: &corev1api.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{
+						velerov1api.VolumesToBackupAnnotation: "pvbPV1,pvbPV2,pvbPV3",
+					},
+				},
+			},
+			expected: struct {
+				included []string
+				optedOut []string
+			}{
+				included: []string{"pvbPV1", "pvbPV2", "pvbPV3"},
+				optedOut: []string{},
+			},
+		},
+		{
+			name:                     "should get all pod volumes when defaultVolumesToFsBackup is true and no PVs are excluded",
+			defaultVolumesToFsBackup: true,
+			pod: &corev1api.Pod{
+				Spec: corev1api.PodSpec{
+					Volumes: []corev1api.Volume{
+						// PVB Volumes
+						{Name: "pvbPV1"}, {Name: "pvbPV2"}, {Name: "pvbPV3"},
+					},
+				},
+			},
+			expected: struct {
+				included []string
+				optedOut []string
+			}{
+				included: []string{"pvbPV1", "pvbPV2", "pvbPV3"},
+				optedOut: []string{},
+			},
+		},
+		{
+			name:                     "should get all pod volumes except ones excluded when defaultVolumesToFsBackup is true",
+			defaultVolumesToFsBackup: true,
+			pod: &corev1api.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{
+						velerov1api.VolumesToExcludeAnnotation: "nonPvbPV1,nonPvbPV2,nonPvbPV3",
+					},
+				},
+				Spec: corev1api.PodSpec{
+					Volumes: []corev1api.Volume{
+						// PVB Volumes
+						{Name: "pvbPV1"}, {Name: "pvbPV2"}, {Name: "pvbPV3"},
+						/// Excluded from PVB through annotation
+						{Name: "nonPvbPV1"}, {Name: "nonPvbPV2"}, {Name: "nonPvbPV3"},
+					},
+				},
+			},
+			expected: struct {
+				included []string
+				optedOut []string
+			}{
+				included: []string{"pvbPV1", "pvbPV2", "pvbPV3"},
+				optedOut: []string{"nonPvbPV1", "nonPvbPV2", "nonPvbPV3"},
+			},
+		},
+		{
+			name:                     "should exclude default service account token from pod volume backup",
+			defaultVolumesToFsBackup: true,
+			pod: &corev1api.Pod{
+				Spec: corev1api.PodSpec{
+					Volumes: []corev1api.Volume{
+						// PVB Volumes
+						{Name: "pvbPV1"}, {Name: "pvbPV2"}, {Name: "pvbPV3"},
+						/// Excluded from PVB because colume mounting default service account token
+						{Name: "default-token-5xq45"},
+					},
+				},
+			},
+			expected: struct {
+				included []string
+				optedOut []string
+			}{
+				included: []string{"pvbPV1", "pvbPV2", "pvbPV3"},
+				optedOut: []string{},
+			},
+		},
+		{
+			name:                     "should exclude host path volumes from pod volume backups",
+			defaultVolumesToFsBackup: true,
+			pod: &corev1api.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{
+						velerov1api.VolumesToExcludeAnnotation: "nonPvbPV1,nonPvbPV2,nonPvbPV3",
+					},
+				},
+				Spec: corev1api.PodSpec{
+					Volumes: []corev1api.Volume{
+						// PVB Volumes
+						{Name: "pvbPV1"}, {Name: "pvbPV2"}, {Name: "pvbPV3"},
+						/// Excluded from pod volume backup through annotation
+						{Name: "nonPvbPV1"}, {Name: "nonPvbPV2"}, {Name: "nonPvbPV3"},
+						// Excluded from pod volume backup because hostpath
+						{Name: "hostPath1", VolumeSource: corev1api.VolumeSource{HostPath: &corev1api.HostPathVolumeSource{Path: "/hostpathVol"}}},
+					},
+				},
+			},
+			expected: struct {
+				included []string
+				optedOut []string
+			}{
+				included: []string{"pvbPV1", "pvbPV2", "pvbPV3"},
+				optedOut: []string{"nonPvbPV1", "nonPvbPV2", "nonPvbPV3"},
+			},
+		},
+		{
+			name:                     "should exclude volumes mounting secrets",
+			defaultVolumesToFsBackup: true,
+			pod: &corev1api.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{
+						velerov1api.VolumesToExcludeAnnotation: "nonPvbPV1,nonPvbPV2,nonPvbPV3",
+					},
+				},
+				Spec: corev1api.PodSpec{
+					Volumes: []corev1api.Volume{
+						// PVB Volumes
+						{Name: "pvbPV1"}, {Name: "pvbPV2"}, {Name: "pvbPV3"},
+						/// Excluded from pod volume backup through annotation
+						{Name: "nonPvbPV1"}, {Name: "nonPvbPV2"}, {Name: "nonPvbPV3"},
+						// Excluded from pod volume backup because hostpath
+						{Name: "superSecret", VolumeSource: corev1api.VolumeSource{Secret: &corev1api.SecretVolumeSource{SecretName: "super-secret"}}},
+					},
+				},
+			},
+			expected: struct {
+				included []string
+				optedOut []string
+			}{
+				included: []string{"pvbPV1", "pvbPV2", "pvbPV3"},
+				optedOut: []string{"nonPvbPV1", "nonPvbPV2", "nonPvbPV3"},
+			},
+		},
+		{
+			name:                     "should exclude volumes mounting config maps",
+			defaultVolumesToFsBackup: true,
+			pod: &corev1api.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{
+						velerov1api.VolumesToExcludeAnnotation: "nonPvbPV1,nonPvbPV2,nonPvbPV3",
+					},
+				},
+				Spec: corev1api.PodSpec{
+					Volumes: []corev1api.Volume{
+						// PVB Volumes
+						{Name: "pvbPV1"}, {Name: "pvbPV2"}, {Name: "pvbPV3"},
+						/// Excluded from pod volume backup through annotation
+						{Name: "nonPvbPV1"}, {Name: "nonPvbPV2"}, {Name: "nonPvbPV3"},
+						// Excluded from pod volume backup because hostpath
+						{Name: "appCOnfig", VolumeSource: corev1api.VolumeSource{ConfigMap: &corev1api.ConfigMapVolumeSource{LocalObjectReference: corev1api.LocalObjectReference{Name: "app-config"}}}},
+					},
+				},
+			},
+			expected: struct {
+				included []string
+				optedOut []string
+			}{
+				included: []string{"pvbPV1", "pvbPV2", "pvbPV3"},
+				optedOut: []string{"nonPvbPV1", "nonPvbPV2", "nonPvbPV3"},
+			},
+		},
+		{
+			name:                     "should exclude projected volumes",
+			defaultVolumesToFsBackup: true,
+			pod: &corev1api.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{
+						velerov1api.VolumesToExcludeAnnotation: "nonPvbPV1,nonPvbPV2,nonPvbPV3",
+					},
+				},
+				Spec: corev1api.PodSpec{
+					Volumes: []corev1api.Volume{
+						{Name: "pvbPV1"}, {Name: "pvbPV2"}, {Name: "pvbPV3"},
+						{
+							Name: "projected",
+							VolumeSource: corev1api.VolumeSource{
+								Projected: &corev1api.ProjectedVolumeSource{
+									Sources: []corev1api.VolumeProjection{{
+										Secret: &corev1api.SecretProjection{
+											LocalObjectReference: corev1api.LocalObjectReference{},
+											Items:                nil,
+											Optional:             nil,
+										},
+										DownwardAPI:         nil,
+										ConfigMap:           nil,
+										ServiceAccountToken: nil,
+									}},
+									DefaultMode: nil,
+								},
+							},
+						},
+					},
+				},
+			},
+			expected: struct {
+				included []string
+				optedOut []string
+			}{
+				included: []string{"pvbPV1", "pvbPV2", "pvbPV3"},
+				optedOut: []string{},
+			},
+		},
+		{
+			name:                     "should exclude DownwardAPI volumes",
+			defaultVolumesToFsBackup: true,
+			pod: &corev1api.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{
+						velerov1api.VolumesToExcludeAnnotation: "nonPvbPV1,nonPvbPV2,nonPvbPV3",
+					},
+				},
+				Spec: corev1api.PodSpec{
+					Volumes: []corev1api.Volume{
+						{Name: "pvbPV1"}, {Name: "pvbPV2"}, {Name: "pvbPV3"},
+						{
+							Name: "downwardAPI",
+							VolumeSource: corev1api.VolumeSource{
+								DownwardAPI: &corev1api.DownwardAPIVolumeSource{
+									Items: []corev1api.DownwardAPIVolumeFile{
+										{
+											Path: "labels",
+											FieldRef: &corev1api.ObjectFieldSelector{
+												APIVersion: "v1",
+												FieldPath:  "metadata.labels",
+											},
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			expected: struct {
+				included []string
+				optedOut []string
+			}{
+				included: []string{"pvbPV1", "pvbPV2", "pvbPV3"},
+				optedOut: []string{},
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			actualIncluded, actualOptedOut := GetVolumesByPod(tc.pod, tc.defaultVolumesToFsBackup)
+
+			sort.Strings(tc.expected.included)
+			sort.Strings(actualIncluded)
+			assert.Equal(t, tc.expected.included, actualIncluded)
+
+			sort.Strings(tc.expected.optedOut)
+			sort.Strings(actualOptedOut)
+			assert.Equal(t, tc.expected.optedOut, actualOptedOut)
+		})
+	}
+}