Merge pull request #9621 from Lyndon-Li/release-1.18

[1.18] Fix compile error for Windows

Dockerfile

@@ -13,7 +13,7 @@
 # limitations under the License.
 
 # Velero binary build section
-FROM --platform=$BUILDPLATFORM golang:1.25-bookworm AS velero-builder
+FROM --platform=$BUILDPLATFORM golang:1.25.7-bookworm AS velero-builder
 
 ARG GOPROXY
 ARG BIN
@@ -49,7 +49,7 @@ RUN mkdir -p /output/usr/bin && \
     go clean -modcache -cache
 
 # Restic binary build section
-FROM --platform=$BUILDPLATFORM golang:1.25-bookworm AS restic-builder
+FROM --platform=$BUILDPLATFORM golang:1.25.7-bookworm AS restic-builder
 
 ARG GOPROXY
 ARG BIN
@@ -73,7 +73,7 @@ RUN mkdir -p /output/usr/bin && \
     go clean -modcache -cache
 
 # Velero image packing section
-FROM paketobuildpacks/run-jammy-tiny:latest
+FROM paketobuildpacks/run-jammy-tiny:0.2.104
 
 LABEL maintainer="Xun Jiang <jxun@vmware.com>"
 

Dockerfile-Windows

@@ -15,7 +15,7 @@
 ARG OS_VERSION=1809
 
 # Velero binary build section
-FROM --platform=$BUILDPLATFORM golang:1.25-bookworm AS velero-builder
+FROM --platform=$BUILDPLATFORM golang:1.25.7-bookworm AS velero-builder
 
 ARG GOPROXY
 ARG BIN

Tiltfile

@@ -52,7 +52,7 @@ git_sha = str(local("git rev-parse HEAD", quiet = True, echo_off = True)).strip(
 
 tilt_helper_dockerfile_header = """
 # Tilt image
-FROM golang:1.25 as tilt-helper
+FROM golang:1.25.7 as tilt-helper
 
 # Support live reloading with Tilt
 RUN wget --output-document /restart.sh --quiet https://raw.githubusercontent.com/windmilleng/rerun-process-wrapper/master/restart.sh  && \

changelogs/CHANGELOG-1.18.md

@@ -16,7 +16,7 @@ https://velero.io/docs/v1.18/upgrade-to-1.18/
 #### Concurrent backup
 In v1.18, Velero is capable to process multiple backups concurrently. This is a significant usability improvement, especially for multiple tenants or multiple users case, backups submitted from different users could run their backups simultaneously without interfering with each other.  
 
-Check design https://github.com/vmware-tanzu/velero/blob/main/design/concurrent-backup-processing.md for more details.
+Check design https://github.com/vmware-tanzu/velero/blob/main/design/Implemented/concurrent-backup-processing.md for more details.
 
 #### Cache volume for data movers
 In v1.18, Velero allows users to configure cache volumes for data mover pods during restore for CSI snapshot data movement and fs-backup. This brings below benefits:
@@ -24,7 +24,7 @@ In v1.18, Velero allows users to configure cache volumes for data mover pods dur
 - Solve the problem that multiple data mover pods fail to run concurrently in one node when the node's ephemeral disk is limited
 - Working together with backup repository's cache limit configuration, cache volume with appropriate size helps to improve the restore throughput
 
-Check design https://github.com/vmware-tanzu/velero/blob/main/design/backup-repo-cache-volume.md for more details.
+Check design https://github.com/vmware-tanzu/velero/blob/main/design/Implemented/backup-repo-cache-volume.md for more details.
 
 #### Incremental size for data movers
 In v1.18, Velero allows users to observe the incremental size of data movers backups for CSI snapshot data movement and fs-backup, so that users could visually see the data reduction due to incremental backup.

changelogs/unreleased/9594-Lyndon-Li‎

@@ -0,0 +1 @@
+Fix issue #9343, include PV topology to data mover pod affinities

changelogs/unreleased/9596-blackpiglet

@@ -0,0 +1 @@
+Add ephemeral storage limit and request support for data mover and maintenance job

changelogs/unreleased/9597-blackpiglet

@@ -0,0 +1 @@
+If BIA return updateObj with SkipFromBackupAnnotation, treat it as skip the resource from backup.

changelogs/unreleased/9604-shubham-pampattiwar

@@ -0,0 +1 @@
+Fix DBR stuck when CSI snapshot no longer exists in cloud provider

changelogs/unreleased/9606-Lyndon-Li‎‎

@@ -0,0 +1 @@
+Fix issue #9496, support customized host os

changelogs/unreleased/9609-Lyndon-Li‎‎

@@ -0,0 +1 @@
+Fix issue #9475, use node-selector instead of nodName for generic restore

changelogs/unreleased/9610-Lyndon-Li‎‎

@@ -0,0 +1 @@
+Fix issue #9460, flush buffer before data mover completes

go.mod

@@ -1,6 +1,6 @@
 module github.com/vmware-tanzu/velero
 
-go 1.25.0
+go 1.25.7
 
 require (
 	cloud.google.com/go/storage v1.57.2
@@ -43,6 +43,7 @@ require (
 	go.uber.org/zap v1.27.1
 	golang.org/x/mod v0.30.0
 	golang.org/x/oauth2 v0.33.0
+	golang.org/x/sys v0.40.0
 	golang.org/x/text v0.31.0
 	google.golang.org/api v0.256.0
 	google.golang.org/grpc v1.77.0
@@ -171,11 +172,11 @@ require (
 	go.opentelemetry.io/contrib/detectors/gcp v1.38.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.61.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.61.0 // indirect
-	go.opentelemetry.io/otel v1.38.0 // indirect
-	go.opentelemetry.io/otel/metric v1.38.0 // indirect
-	go.opentelemetry.io/otel/sdk v1.38.0 // indirect
-	go.opentelemetry.io/otel/sdk/metric v1.38.0 // indirect
-	go.opentelemetry.io/otel/trace v1.38.0 // indirect
+	go.opentelemetry.io/otel v1.40.0 // indirect
+	go.opentelemetry.io/otel/metric v1.40.0 // indirect
+	go.opentelemetry.io/otel/sdk v1.40.0 // indirect
+	go.opentelemetry.io/otel/sdk/metric v1.40.0 // indirect
+	go.opentelemetry.io/otel/trace v1.40.0 // indirect
 	go.starlark.net v0.0.0-20230525235612-a134d8f9ddca // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	go.yaml.in/yaml/v2 v2.4.3 // indirect
@@ -183,7 +184,6 @@ require (
 	golang.org/x/exp v0.0.0-20240719175910-8a7402abbf56 // indirect
 	golang.org/x/net v0.47.0 // indirect
 	golang.org/x/sync v0.18.0 // indirect
-	golang.org/x/sys v0.38.0 // indirect
 	golang.org/x/term v0.37.0 // indirect
 	golang.org/x/time v0.14.0 // indirect
 	golang.org/x/tools v0.38.0 // indirect

go.sum

@@ -748,18 +748,18 @@ go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.6
 go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.61.0/go.mod h1:snMWehoOh2wsEwnvvwtDyFCxVeDAODenXHtn5vzrKjo=
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.61.0 h1:F7Jx+6hwnZ41NSFTO5q4LYDtJRXBf2PD0rNBkeB/lus=
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.61.0/go.mod h1:UHB22Z8QsdRDrnAtX4PntOl36ajSxcdUMt1sF7Y6E7Q=
-go.opentelemetry.io/otel v1.38.0 h1:RkfdswUDRimDg0m2Az18RKOsnI8UDzppJAtj01/Ymk8=
-go.opentelemetry.io/otel v1.38.0/go.mod h1:zcmtmQ1+YmQM9wrNsTGV/q/uyusom3P8RxwExxkZhjM=
+go.opentelemetry.io/otel v1.40.0 h1:oA5YeOcpRTXq6NN7frwmwFR0Cn3RhTVZvXsP4duvCms=
+go.opentelemetry.io/otel v1.40.0/go.mod h1:IMb+uXZUKkMXdPddhwAHm6UfOwJyh4ct1ybIlV14J0g=
 go.opentelemetry.io/otel/exporters/stdout/stdoutmetric v1.36.0 h1:rixTyDGXFxRy1xzhKrotaHy3/KXdPhlWARrCgK+eqUY=
 go.opentelemetry.io/otel/exporters/stdout/stdoutmetric v1.36.0/go.mod h1:dowW6UsM9MKbJq5JTz2AMVp3/5iW5I/TStsk8S+CfHw=
-go.opentelemetry.io/otel/metric v1.38.0 h1:Kl6lzIYGAh5M159u9NgiRkmoMKjvbsKtYRwgfrA6WpA=
-go.opentelemetry.io/otel/metric v1.38.0/go.mod h1:kB5n/QoRM8YwmUahxvI3bO34eVtQf2i4utNVLr9gEmI=
-go.opentelemetry.io/otel/sdk v1.38.0 h1:l48sr5YbNf2hpCUj/FoGhW9yDkl+Ma+LrVl8qaM5b+E=
-go.opentelemetry.io/otel/sdk v1.38.0/go.mod h1:ghmNdGlVemJI3+ZB5iDEuk4bWA3GkTpW+DOoZMYBVVg=
-go.opentelemetry.io/otel/sdk/metric v1.38.0 h1:aSH66iL0aZqo//xXzQLYozmWrXxyFkBJ6qT5wthqPoM=
-go.opentelemetry.io/otel/sdk/metric v1.38.0/go.mod h1:dg9PBnW9XdQ1Hd6ZnRz689CbtrUp0wMMs9iPcgT9EZA=
-go.opentelemetry.io/otel/trace v1.38.0 h1:Fxk5bKrDZJUH+AMyyIXGcFAPah0oRcT+LuNtJrmcNLE=
-go.opentelemetry.io/otel/trace v1.38.0/go.mod h1:j1P9ivuFsTceSWe1oY+EeW3sc+Pp42sO++GHkg4wwhs=
+go.opentelemetry.io/otel/metric v1.40.0 h1:rcZe317KPftE2rstWIBitCdVp89A2HqjkxR3c11+p9g=
+go.opentelemetry.io/otel/metric v1.40.0/go.mod h1:ib/crwQH7N3r5kfiBZQbwrTge743UDc7DTFVZrrXnqc=
+go.opentelemetry.io/otel/sdk v1.40.0 h1:KHW/jUzgo6wsPh9At46+h4upjtccTmuZCFAc9OJ71f8=
+go.opentelemetry.io/otel/sdk v1.40.0/go.mod h1:Ph7EFdYvxq72Y8Li9q8KebuYUr2KoeyHx0DRMKrYBUE=
+go.opentelemetry.io/otel/sdk/metric v1.40.0 h1:mtmdVqgQkeRxHgRv4qhyJduP3fYJRMX4AtAlbuWdCYw=
+go.opentelemetry.io/otel/sdk/metric v1.40.0/go.mod h1:4Z2bGMf0KSK3uRjlczMOeMhKU2rhUqdWNoKcYrtcBPg=
+go.opentelemetry.io/otel/trace v1.40.0 h1:WA4etStDttCSYuhwvEa8OP8I5EWu24lkOzp+ZYblVjw=
+go.opentelemetry.io/otel/trace v1.40.0/go.mod h1:zeAhriXecNGP/s2SEG3+Y8X9ujcJOTqQ5RgdEJcawiA=
 go.starlark.net v0.0.0-20200306205701-8dd3e2ee1dd5/go.mod h1:nmDLcffg48OtT/PSW0Hg7FvpRQsQh5OSqIylirxKC7o=
 go.starlark.net v0.0.0-20201006213952-227f4aabceb5/go.mod h1:f0znQkUKRrkk36XxWbGjMqQM8wGv/xHBVE2qc3B5oFU=
 go.starlark.net v0.0.0-20230525235612-a134d8f9ddca h1:VdD38733bfYv5tUZwEIskMM93VanwNIi5bIKnDrJdEY=
@@ -969,8 +969,8 @@ golang.org/x/sys v0.0.0-20211019181941-9d821ace8654/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.38.0 h1:3yZWxaJjBmCWXqhN1qh02AkOnCQ1poK6oF+a7xWL6Gc=
-golang.org/x/sys v0.38.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
+golang.org/x/sys v0.40.0 h1:DBZZqJ2Rkml6QMQsZywtnjnnGvHza6BTfYFWY9kjEWQ=
+golang.org/x/sys v0.40.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
 golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210220032956-6a3ed077a48d/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=

hack/build-image/Dockerfile

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-FROM --platform=$TARGETPLATFORM golang:1.25-bookworm
+FROM --platform=$TARGETPLATFORM golang:1.25.7-bookworm
 
 ARG GOPROXY
 
@@ -21,9 +21,11 @@ ENV GO111MODULE=on
 ENV GOPROXY=${GOPROXY}
 
 # kubebuilder test bundle is separated from kubebuilder. Need to setup it for CI test.
-RUN curl -sSLo envtest-bins.tar.gz https://go.kubebuilder.io/test-tools/1.22.1/linux/$(go env GOARCH) && \
-    mkdir /usr/local/kubebuilder && \
-    tar -C /usr/local/kubebuilder --strip-components=1 -zvxf envtest-bins.tar.gz
+# Using setup-envtest to download envtest binaries
+RUN go install sigs.k8s.io/controller-runtime/tools/setup-envtest@latest && \
+    mkdir -p /usr/local/kubebuilder/bin && \
+    ENVTEST_ASSETS_DIR=$(setup-envtest use 1.33.0 --bin-dir /usr/local/kubebuilder/bin -p path) && \
+    cp -r ${ENVTEST_ASSETS_DIR}/* /usr/local/kubebuilder/bin/
 
 RUN wget --quiet https://github.com/kubernetes-sigs/kubebuilder/releases/download/v3.2.0/kubebuilder_linux_$(go env GOARCH) && \
     mv kubebuilder_linux_$(go env GOARCH) /usr/local/kubebuilder/bin/kubebuilder && \

internal/delete/actions/csi/volumesnapshotcontent_action.go

@@ -137,6 +137,10 @@ func (p *volumeSnapshotContentDeleteItemAction) Execute(
 			return checkVSCReadiness(ctx, &snapCont, p.crClient)
 		},
 	); err != nil {
+		// Clean up the VSC we created since it can't become ready
+		if deleteErr := p.crClient.Delete(context.TODO(), &snapCont); deleteErr != nil && !apierrors.IsNotFound(deleteErr) {
+			p.log.WithError(deleteErr).Errorf("Failed to clean up VolumeSnapshotContent %s", snapCont.Name)
+		}
 		return errors.Wrapf(err, "fail to wait VolumeSnapshotContent %s becomes ready.", snapCont.Name)
 	}
 
@@ -167,6 +171,13 @@ var checkVSCReadiness = func(
 		return true, nil
 	}
 
+	// Fail fast on permanent CSI driver errors (e.g., InvalidSnapshot.NotFound)
+	if tmpVSC.Status != nil && tmpVSC.Status.Error != nil && tmpVSC.Status.Error.Message != nil {
+		return false, errors.Errorf(
+			"VolumeSnapshotContent %s has error: %s", vsc.Name, *tmpVSC.Status.Error.Message,
+		)
+	}
+
 	return false, nil
 }
 

internal/delete/actions/csi/volumesnapshotcontent_action_test.go

@@ -94,6 +94,19 @@ func TestVSCExecute(t *testing.T) {
 				return false, errors.Errorf("test error case")
 			},
 		},
+		{
+			name:      "Error case with CSI error, dangling VSC should be cleaned up",
+			vsc:       builder.ForVolumeSnapshotContent("bar").ObjectMeta(builder.WithLabelsMap(map[string]string{velerov1api.BackupNameLabel: "backup"})).Status(&snapshotv1api.VolumeSnapshotContentStatus{SnapshotHandle: &snapshotHandleStr}).Result(),
+			backup:    builder.ForBackup("velero", "backup").ObjectMeta(builder.WithAnnotationsMap(map[string]string{velerov1api.ResourceTimeoutAnnotation: "5s"})).Result(),
+			expectErr: true,
+			function: func(
+				ctx context.Context,
+				vsc *snapshotv1api.VolumeSnapshotContent,
+				client crclient.Client,
+			) (bool, error) {
+				return false, errors.Errorf("VolumeSnapshotContent %s has error: InvalidSnapshot.NotFound", vsc.Name)
+			},
+		},
 	}
 
 	for _, test := range tests {
@@ -190,6 +203,24 @@ func TestCheckVSCReadiness(t *testing.T) {
 			expectErr: false,
 			ready:     false,
 		},
+		{
+			name: "VSC has error from CSI driver",
+			vsc: &snapshotv1api.VolumeSnapshotContent{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "vsc-1",
+					Namespace: "velero",
+				},
+				Status: &snapshotv1api.VolumeSnapshotContentStatus{
+					ReadyToUse: boolPtr(false),
+					Error: &snapshotv1api.VolumeSnapshotError{
+						Message: stringPtr("InvalidSnapshot.NotFound: The snapshot 'snap-0abc123' does not exist."),
+					},
+				},
+			},
+			createVSC: true,
+			expectErr: true,
+			ready:     false,
+		},
 	}
 
 	for _, test := range tests {
@@ -207,3 +238,11 @@ func TestCheckVSCReadiness(t *testing.T) {
 		})
 	}
 }
+
+func boolPtr(b bool) *bool {
+	return &b
+}
+
+func stringPtr(s string) *string {
+	return &s
+}

internal/volumehelper/volume_policy_helper.go

@@ -134,6 +134,7 @@ func (v *volumeHelperImpl) ShouldPerformSnapshot(obj runtime.Unstructured, group
 	pv := new(corev1api.PersistentVolume)
 	var err error
 
+	var pvNotFoundErr error
 	if groupResource == kuberesource.PersistentVolumeClaims {
 		if err = runtime.DefaultUnstructuredConverter.FromUnstructured(obj.UnstructuredContent(), &pvc); err != nil {
 			v.logger.WithError(err).Error("fail to convert unstructured into PVC")
@@ -142,8 +143,10 @@ func (v *volumeHelperImpl) ShouldPerformSnapshot(obj runtime.Unstructured, group
 
 		pv, err = kubeutil.GetPVForPVC(pvc, v.client)
 		if err != nil {
-			v.logger.WithError(err).Errorf("fail to get PV for PVC %s", pvc.Namespace+"/"+pvc.Name)
-			return false, err
+			// Any error means PV not available - save to return later if no policy matches
+			v.logger.Debugf("PV not found for PVC %s: %v", pvc.Namespace+"/"+pvc.Name, err)
+			pvNotFoundErr = err
+			pv = nil
 		}
 	}
 
@@ -158,7 +161,7 @@ func (v *volumeHelperImpl) ShouldPerformSnapshot(obj runtime.Unstructured, group
 		vfd := resourcepolicies.NewVolumeFilterData(pv, nil, pvc)
 		action, err := v.volumePolicy.GetMatchAction(vfd)
 		if err != nil {
-			v.logger.WithError(err).Errorf("fail to get VolumePolicy match action for PV %s", pv.Name)
+			v.logger.WithError(err).Errorf("fail to get VolumePolicy match action for %+v", vfd)
 			return false, err
 		}
 
@@ -167,15 +170,21 @@ func (v *volumeHelperImpl) ShouldPerformSnapshot(obj runtime.Unstructured, group
 		// If there is no match action, go on to the next check.
 		if action != nil {
 			if action.Type == resourcepolicies.Snapshot {
-				v.logger.Infof(fmt.Sprintf("performing snapshot action for pv %s", pv.Name))
+				v.logger.Infof("performing snapshot action for %+v", vfd)
 				return true, nil
 			} else {
-				v.logger.Infof("Skip snapshot action for pv %s as the action type is %s", pv.Name, action.Type)
+				v.logger.Infof("Skip snapshot action for %+v as the action type is %s", vfd, action.Type)
 				return false, nil
 			}
 		}
 	}
 
+	// If resource is PVC, and PV is nil (e.g., Pending/Lost PVC with no matching policy), return the original error
+	if groupResource == kuberesource.PersistentVolumeClaims && pv == nil && pvNotFoundErr != nil {
+		v.logger.WithError(pvNotFoundErr).Errorf("fail to get PV for PVC %s", pvc.Namespace+"/"+pvc.Name)
+		return false, pvNotFoundErr
+	}
+
 	// If this PV is claimed, see if we've already taken a (pod volume backup)
 	// snapshot of the contents of this PV. If so, don't take a snapshot.
 	if pv.Spec.ClaimRef != nil {
@@ -209,7 +218,7 @@ func (v *volumeHelperImpl) ShouldPerformSnapshot(obj runtime.Unstructured, group
 		return true, nil
 	}
 
-	v.logger.Infof(fmt.Sprintf("skipping snapshot action for pv %s possibly due to no volume policy setting or snapshotVolumes is false", pv.Name))
+	v.logger.Infof("skipping snapshot action for pv %s possibly due to no volume policy setting or snapshotVolumes is false", pv.Name)
 	return false, nil
 }
 
@@ -219,6 +228,7 @@ func (v volumeHelperImpl) ShouldPerformFSBackup(volume corev1api.Volume, pod cor
 		return false, nil
 	}
 
+	var pvNotFoundErr error
 	if v.volumePolicy != nil {
 		var resource any
 		var err error
@@ -230,10 +240,13 @@ func (v volumeHelperImpl) ShouldPerformFSBackup(volume corev1api.Volume, pod cor
 				v.logger.WithError(err).Errorf("fail to get PVC for pod %s", pod.Namespace+"/"+pod.Name)
 				return false, err
 			}
-			resource, err = kubeutil.GetPVForPVC(pvc, v.client)
+			pvResource, err := kubeutil.GetPVForPVC(pvc, v.client)
 			if err != nil {
-				v.logger.WithError(err).Errorf("fail to get PV for PVC %s", pvc.Namespace+"/"+pvc.Name)
-				return false, err
+				// Any error means PV not available - save to return later if no policy matches
+				v.logger.Debugf("PV not found for PVC %s: %v", pvc.Namespace+"/"+pvc.Name, err)
+				pvNotFoundErr = err
+			} else {
+				resource = pvResource
 			}
 		}
 
@@ -260,6 +273,12 @@ func (v volumeHelperImpl) ShouldPerformFSBackup(volume corev1api.Volume, pod cor
 				return false, nil
 			}
 		}
+
+		// If no policy matched and PV was not found, return the original error
+		if pvNotFoundErr != nil {
+			v.logger.WithError(pvNotFoundErr).Errorf("fail to get PV for PVC %s", pvc.Namespace+"/"+pvc.Name)
+			return false, pvNotFoundErr
+		}
 	}
 
 	if v.shouldPerformFSBackupLegacy(volume, pod) {

internal/volumehelper/volume_policy_helper_test.go

@@ -286,7 +286,7 @@ func TestVolumeHelperImpl_ShouldPerformSnapshot(t *testing.T) {
 			expectedErr:         false,
 		},
 		{
-			name:          "PVC not having PV, return false and error case PV not found",
+			name:          "PVC not having PV, return false and error when no matching policy",
 			inputObj:      builder.ForPersistentVolumeClaim("default", "example-pvc").StorageClass("gp2-csi").Result(),
 			groupResource: kuberesource.PersistentVolumeClaims,
 			resourcePolicies: &resourcepolicies.ResourcePolicies{
@@ -1234,3 +1234,312 @@ func TestNewVolumeHelperImplWithCache_UsesCache(t *testing.T) {
 	require.NoError(t, err)
 	require.False(t, shouldSnapshot, "Expected snapshot to be skipped due to fs-backup selection via cache")
 }
+
+// TestVolumeHelperImpl_ShouldPerformSnapshot_UnboundPVC tests that Pending and Lost PVCs with
+// phase-based skip policies don't cause errors when GetPVForPVC would fail.
+func TestVolumeHelperImpl_ShouldPerformSnapshot_UnboundPVC(t *testing.T) {
+	testCases := []struct {
+		name             string
+		inputPVC         *corev1api.PersistentVolumeClaim
+		resourcePolicies *resourcepolicies.ResourcePolicies
+		shouldSnapshot   bool
+		expectedErr      bool
+	}{
+		{
+			name: "Pending PVC with phase-based skip policy should not error and return false",
+			inputPVC: builder.ForPersistentVolumeClaim("ns", "pvc-pending").
+				StorageClass("non-existent-class").
+				Phase(corev1api.ClaimPending).
+				Result(),
+			resourcePolicies: &resourcepolicies.ResourcePolicies{
+				Version: "v1",
+				VolumePolicies: []resourcepolicies.VolumePolicy{
+					{
+						Conditions: map[string]any{
+							"pvcPhase": []string{"Pending"},
+						},
+						Action: resourcepolicies.Action{
+							Type: resourcepolicies.Skip,
+						},
+					},
+				},
+			},
+			shouldSnapshot: false,
+			expectedErr:    false,
+		},
+		{
+			name: "Pending PVC without matching skip policy should error (no PV)",
+			inputPVC: builder.ForPersistentVolumeClaim("ns", "pvc-pending-no-policy").
+				StorageClass("non-existent-class").
+				Phase(corev1api.ClaimPending).
+				Result(),
+			resourcePolicies: &resourcepolicies.ResourcePolicies{
+				Version: "v1",
+				VolumePolicies: []resourcepolicies.VolumePolicy{
+					{
+						Conditions: map[string]any{
+							"storageClass": []string{"gp2-csi"},
+						},
+						Action: resourcepolicies.Action{
+							Type: resourcepolicies.Skip,
+						},
+					},
+				},
+			},
+			shouldSnapshot: false,
+			expectedErr:    true,
+		},
+		{
+			name: "Lost PVC with phase-based skip policy should not error and return false",
+			inputPVC: builder.ForPersistentVolumeClaim("ns", "pvc-lost").
+				StorageClass("some-class").
+				Phase(corev1api.ClaimLost).
+				Result(),
+			resourcePolicies: &resourcepolicies.ResourcePolicies{
+				Version: "v1",
+				VolumePolicies: []resourcepolicies.VolumePolicy{
+					{
+						Conditions: map[string]any{
+							"pvcPhase": []string{"Lost"},
+						},
+						Action: resourcepolicies.Action{
+							Type: resourcepolicies.Skip,
+						},
+					},
+				},
+			},
+			shouldSnapshot: false,
+			expectedErr:    false,
+		},
+		{
+			name: "Lost PVC with policy for Pending and Lost should not error and return false",
+			inputPVC: builder.ForPersistentVolumeClaim("ns", "pvc-lost").
+				StorageClass("some-class").
+				Phase(corev1api.ClaimLost).
+				Result(),
+			resourcePolicies: &resourcepolicies.ResourcePolicies{
+				Version: "v1",
+				VolumePolicies: []resourcepolicies.VolumePolicy{
+					{
+						Conditions: map[string]any{
+							"pvcPhase": []string{"Pending", "Lost"},
+						},
+						Action: resourcepolicies.Action{
+							Type: resourcepolicies.Skip,
+						},
+					},
+				},
+			},
+			shouldSnapshot: false,
+			expectedErr:    false,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			fakeClient := velerotest.NewFakeControllerRuntimeClient(t)
+
+			var p *resourcepolicies.Policies
+			if tc.resourcePolicies != nil {
+				p = &resourcepolicies.Policies{}
+				err := p.BuildPolicy(tc.resourcePolicies)
+				require.NoError(t, err)
+			}
+
+			vh := NewVolumeHelperImpl(
+				p,
+				ptr.To(true),
+				logrus.StandardLogger(),
+				fakeClient,
+				false,
+				false,
+			)
+
+			obj, err := runtime.DefaultUnstructuredConverter.ToUnstructured(tc.inputPVC)
+			require.NoError(t, err)
+
+			actualShouldSnapshot, actualError := vh.ShouldPerformSnapshot(&unstructured.Unstructured{Object: obj}, kuberesource.PersistentVolumeClaims)
+			if tc.expectedErr {
+				require.Error(t, actualError, "Want error; Got nil error")
+				return
+			}
+
+			require.NoError(t, actualError)
+			require.Equalf(t, tc.shouldSnapshot, actualShouldSnapshot, "Want shouldSnapshot as %t; Got shouldSnapshot as %t", tc.shouldSnapshot, actualShouldSnapshot)
+		})
+	}
+}
+
+// TestVolumeHelperImpl_ShouldPerformFSBackup_UnboundPVC tests that Pending and Lost PVCs with
+// phase-based skip policies don't cause errors when GetPVForPVC would fail.
+func TestVolumeHelperImpl_ShouldPerformFSBackup_UnboundPVC(t *testing.T) {
+	testCases := []struct {
+		name             string
+		pod              *corev1api.Pod
+		pvc              *corev1api.PersistentVolumeClaim
+		resourcePolicies *resourcepolicies.ResourcePolicies
+		shouldFSBackup   bool
+		expectedErr      bool
+	}{
+		{
+			name: "Pending PVC with phase-based skip policy should not error and return false",
+			pod: builder.ForPod("ns", "pod-1").
+				Volumes(
+					&corev1api.Volume{
+						Name: "vol-pending",
+						VolumeSource: corev1api.VolumeSource{
+							PersistentVolumeClaim: &corev1api.PersistentVolumeClaimVolumeSource{
+								ClaimName: "pvc-pending",
+							},
+						},
+					}).Result(),
+			pvc: builder.ForPersistentVolumeClaim("ns", "pvc-pending").
+				StorageClass("non-existent-class").
+				Phase(corev1api.ClaimPending).
+				Result(),
+			resourcePolicies: &resourcepolicies.ResourcePolicies{
+				Version: "v1",
+				VolumePolicies: []resourcepolicies.VolumePolicy{
+					{
+						Conditions: map[string]any{
+							"pvcPhase": []string{"Pending"},
+						},
+						Action: resourcepolicies.Action{
+							Type: resourcepolicies.Skip,
+						},
+					},
+				},
+			},
+			shouldFSBackup: false,
+			expectedErr:    false,
+		},
+		{
+			name: "Pending PVC without matching skip policy should error (no PV)",
+			pod: builder.ForPod("ns", "pod-1").
+				Volumes(
+					&corev1api.Volume{
+						Name: "vol-pending",
+						VolumeSource: corev1api.VolumeSource{
+							PersistentVolumeClaim: &corev1api.PersistentVolumeClaimVolumeSource{
+								ClaimName: "pvc-pending-no-policy",
+							},
+						},
+					}).Result(),
+			pvc: builder.ForPersistentVolumeClaim("ns", "pvc-pending-no-policy").
+				StorageClass("non-existent-class").
+				Phase(corev1api.ClaimPending).
+				Result(),
+			resourcePolicies: &resourcepolicies.ResourcePolicies{
+				Version: "v1",
+				VolumePolicies: []resourcepolicies.VolumePolicy{
+					{
+						Conditions: map[string]any{
+							"storageClass": []string{"gp2-csi"},
+						},
+						Action: resourcepolicies.Action{
+							Type: resourcepolicies.Skip,
+						},
+					},
+				},
+			},
+			shouldFSBackup: false,
+			expectedErr:    true,
+		},
+		{
+			name: "Lost PVC with phase-based skip policy should not error and return false",
+			pod: builder.ForPod("ns", "pod-1").
+				Volumes(
+					&corev1api.Volume{
+						Name: "vol-lost",
+						VolumeSource: corev1api.VolumeSource{
+							PersistentVolumeClaim: &corev1api.PersistentVolumeClaimVolumeSource{
+								ClaimName: "pvc-lost",
+							},
+						},
+					}).Result(),
+			pvc: builder.ForPersistentVolumeClaim("ns", "pvc-lost").
+				StorageClass("some-class").
+				Phase(corev1api.ClaimLost).
+				Result(),
+			resourcePolicies: &resourcepolicies.ResourcePolicies{
+				Version: "v1",
+				VolumePolicies: []resourcepolicies.VolumePolicy{
+					{
+						Conditions: map[string]any{
+							"pvcPhase": []string{"Lost"},
+						},
+						Action: resourcepolicies.Action{
+							Type: resourcepolicies.Skip,
+						},
+					},
+				},
+			},
+			shouldFSBackup: false,
+			expectedErr:    false,
+		},
+		{
+			name: "Lost PVC with policy for Pending and Lost should not error and return false",
+			pod: builder.ForPod("ns", "pod-1").
+				Volumes(
+					&corev1api.Volume{
+						Name: "vol-lost",
+						VolumeSource: corev1api.VolumeSource{
+							PersistentVolumeClaim: &corev1api.PersistentVolumeClaimVolumeSource{
+								ClaimName: "pvc-lost",
+							},
+						},
+					}).Result(),
+			pvc: builder.ForPersistentVolumeClaim("ns", "pvc-lost").
+				StorageClass("some-class").
+				Phase(corev1api.ClaimLost).
+				Result(),
+			resourcePolicies: &resourcepolicies.ResourcePolicies{
+				Version: "v1",
+				VolumePolicies: []resourcepolicies.VolumePolicy{
+					{
+						Conditions: map[string]any{
+							"pvcPhase": []string{"Pending", "Lost"},
+						},
+						Action: resourcepolicies.Action{
+							Type: resourcepolicies.Skip,
+						},
+					},
+				},
+			},
+			shouldFSBackup: false,
+			expectedErr:    false,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			fakeClient := velerotest.NewFakeControllerRuntimeClient(t, tc.pvc)
+			require.NoError(t, fakeClient.Create(t.Context(), tc.pod))
+
+			var p *resourcepolicies.Policies
+			if tc.resourcePolicies != nil {
+				p = &resourcepolicies.Policies{}
+				err := p.BuildPolicy(tc.resourcePolicies)
+				require.NoError(t, err)
+			}
+
+			vh := NewVolumeHelperImpl(
+				p,
+				ptr.To(true),
+				logrus.StandardLogger(),
+				fakeClient,
+				false,
+				false,
+			)
+
+			actualShouldFSBackup, actualError := vh.ShouldPerformFSBackup(tc.pod.Spec.Volumes[0], *tc.pod)
+			if tc.expectedErr {
+				require.Error(t, actualError, "Want error; Got nil error")
+				return
+			}
+
+			require.NoError(t, actualError)
+			require.Equalf(t, tc.shouldFSBackup, actualShouldFSBackup, "Want shouldFSBackup as %t; Got shouldFSBackup as %t", tc.shouldFSBackup, actualShouldFSBackup)
+		})
+	}
+}

pkg/apis/velero/v1/labels_annotations.go

@@ -102,6 +102,15 @@ const (
 	// even if the resource contains a matching selector label.
 	ExcludeFromBackupLabel = "velero.io/exclude-from-backup"
 
+	// SkipFromBackupAnnotation is the annotation used by internal BackupItemActions
+	// to indicate that a resource should be skipped from backup,
+	// even if it doesn't have the ExcludeFromBackupLabel.
+	// This is used in cases where we want to skip backup of a resource based on some logic in a plugin.
+	//
+	// Notice: SkipFromBackupAnnotation's priority is higher than MustIncludeAdditionalItemAnnotation.
+	// If SkipFromBackupAnnotation is set, the resource will be skipped even if MustIncludeAdditionalItemAnnotation is set.
+	SkipFromBackupAnnotation = "velero.io/skip-from-backup"
+
 	// defaultVGSLabelKey is the default label key used to group PVCs under a VolumeGroupSnapshot
 	DefaultVGSLabelKey = "velero.io/volume-group"
 

pkg/backup/backed_up_items_map.go

@@ -98,6 +98,14 @@ func (m *backedUpItemsMap) AddItem(key itemKey) {
 	m.totalItems[key] = struct{}{}
 }
 
+func (m *backedUpItemsMap) DeleteItem(key itemKey) {
+	m.Lock()
+	defer m.Unlock()
+
+	delete(m.backedUpItems, key)
+	delete(m.totalItems, key)
+}
+
 func (m *backedUpItemsMap) AddItemToTotal(key itemKey) {
 	m.Lock()
 	defer m.Unlock()

pkg/backup/item_backupper.go

@@ -244,6 +244,14 @@ func (ib *itemBackupper) backupItemInternal(logger logrus.FieldLogger, obj runti
 		return false, itemFiles, kubeerrs.NewAggregate(backupErrs)
 	}
 
+	// If err is nil and updatedObj is nil, it means the item is skipped by plugin action,
+	// we should return here to avoid backing up the item, and avoid potential NPE in the following code.
+	if updatedObj == nil {
+		log.Infof("Remove item from the backup's backupItems list and totalItems list because it's skipped by plugin action.")
+		ib.backupRequest.BackedUpItems.DeleteItem(key)
+		return false, itemFiles, nil
+	}
+
 	itemFiles = append(itemFiles, additionalItemFiles...)
 	obj = updatedObj
 	if metadata, err = meta.Accessor(obj); err != nil {
@@ -398,6 +406,13 @@ func (ib *itemBackupper) executeActions(
 		}
 
 		u := &unstructured.Unstructured{Object: updatedItem.UnstructuredContent()}
+
+		if _, ok := u.GetAnnotations()[velerov1api.SkipFromBackupAnnotation]; ok {
+			log.Infof("Resource (groupResource=%s, namespace=%s, name=%s) is skipped from backup by action %s.",
+				groupResource.String(), namespace, name, actionName)
+			return nil, itemFiles, nil
+		}
+
 		if actionName == csiBIAPluginName {
 			if additionalItemIdentifiers == nil && u.GetAnnotations()[velerov1api.SkippedNoCSIPVAnnotation] == "true" {
 				// snapshot was skipped by CSI plugin
@@ -687,15 +702,14 @@ func (ib *itemBackupper) getMatchAction(obj runtime.Unstructured, groupResource
 			return nil, errors.WithStack(err)
 		}
 
-		pvName := pvc.Spec.VolumeName
-		if pvName == "" {
-			return nil, errors.Errorf("PVC has no volume backing this claim")
-		}
-
-		pv := &corev1api.PersistentVolume{}
-		if err := ib.kbClient.Get(context.Background(), kbClient.ObjectKey{Name: pvName}, pv); err != nil {
-			return nil, errors.WithStack(err)
+		var pv *corev1api.PersistentVolume
+		if pvName := pvc.Spec.VolumeName; pvName != "" {
+			pv = &corev1api.PersistentVolume{}
+			if err := ib.kbClient.Get(context.Background(), kbClient.ObjectKey{Name: pvName}, pv); err != nil {
+				return nil, errors.WithStack(err)
+			}
 		}
+		// If pv is nil for unbound PVCs - policy matching will use PVC-only conditions
 		vfd := resourcepolicies.NewVolumeFilterData(pv, nil, pvc)
 		return ib.backupRequest.ResPolicies.GetMatchAction(vfd)
 	}
@@ -709,7 +723,10 @@ func (ib *itemBackupper) trackSkippedPV(obj runtime.Unstructured, groupResource
 	if name, err := getPVName(obj, groupResource); len(name) > 0 && err == nil {
 		ib.backupRequest.SkippedPVTracker.Track(name, approach, reason)
 	} else if err != nil {
-		log.WithError(err).Warnf("unable to get PV name, skip tracking.")
+		// Log at info level for tracking purposes. This is not an error because
+		// it's expected for some resources (e.g., PVCs in Pending or Lost phase)
+		// to not have a PV name. This occurs when volume policy skips unbound PVCs.
+		log.WithError(err).Infof("unable to get PV name, skip tracking.")
 	}
 }
 
@@ -719,6 +736,17 @@ func (ib *itemBackupper) unTrackSkippedPV(obj runtime.Unstructured, groupResourc
 	if name, err := getPVName(obj, groupResource); len(name) > 0 && err == nil {
 		ib.backupRequest.SkippedPVTracker.Untrack(name)
 	} else if err != nil {
+		// For PVCs in Pending or Lost phase, it's expected that there's no PV name.
+		// Log at debug level instead of warning to reduce noise.
+		if groupResource == kuberesource.PersistentVolumeClaims {
+			pvc := new(corev1api.PersistentVolumeClaim)
+			if convErr := runtime.DefaultUnstructuredConverter.FromUnstructured(obj.UnstructuredContent(), pvc); convErr == nil {
+				if pvc.Status.Phase == corev1api.ClaimPending || pvc.Status.Phase == corev1api.ClaimLost {
+					log.WithError(err).Debugf("unable to get PV name for %s PVC, skip untracking.", pvc.Status.Phase)
+					return
+				}
+			}
+		}
 		log.WithError(err).Warnf("unable to get PV name, skip untracking.")
 	}
 }

pkg/backup/item_backupper_test.go

@@ -17,12 +17,15 @@ limitations under the License.
 package backup
 
 import (
+	"bytes"
 	"testing"
 
 	"github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/require"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	ctrlfake "sigs.k8s.io/controller-runtime/pkg/client/fake"
 
+	"github.com/vmware-tanzu/velero/internal/resourcepolicies"
 	"github.com/vmware-tanzu/velero/pkg/kuberesource"
 
 	"github.com/stretchr/testify/assert"
@@ -269,3 +272,225 @@ func TestAddVolumeInfo(t *testing.T) {
 		})
 	}
 }
+
+func TestGetMatchAction_PendingLostPVC(t *testing.T) {
+	scheme := runtime.NewScheme()
+	require.NoError(t, corev1api.AddToScheme(scheme))
+
+	// Create resource policies that skip Pending/Lost PVCs
+	resPolicies := &resourcepolicies.ResourcePolicies{
+		Version: "v1",
+		VolumePolicies: []resourcepolicies.VolumePolicy{
+			{
+				Conditions: map[string]any{
+					"pvcPhase": []string{"Pending", "Lost"},
+				},
+				Action: resourcepolicies.Action{
+					Type: resourcepolicies.Skip,
+				},
+			},
+		},
+	}
+	policies := &resourcepolicies.Policies{}
+	err := policies.BuildPolicy(resPolicies)
+	require.NoError(t, err)
+
+	testCases := []struct {
+		name           string
+		pvc            *corev1api.PersistentVolumeClaim
+		pv             *corev1api.PersistentVolume
+		expectedAction *resourcepolicies.Action
+		expectError    bool
+	}{
+		{
+			name: "Pending PVC with no VolumeName should match pvcPhase policy",
+			pvc: builder.ForPersistentVolumeClaim("ns", "pending-pvc").
+				StorageClass("test-sc").
+				Phase(corev1api.ClaimPending).
+				Result(),
+			pv:             nil,
+			expectedAction: &resourcepolicies.Action{Type: resourcepolicies.Skip},
+			expectError:    false,
+		},
+		{
+			name: "Lost PVC with no VolumeName should match pvcPhase policy",
+			pvc: builder.ForPersistentVolumeClaim("ns", "lost-pvc").
+				StorageClass("test-sc").
+				Phase(corev1api.ClaimLost).
+				Result(),
+			pv:             nil,
+			expectedAction: &resourcepolicies.Action{Type: resourcepolicies.Skip},
+			expectError:    false,
+		},
+		{
+			name: "Bound PVC with VolumeName and matching PV should not match pvcPhase policy",
+			pvc: builder.ForPersistentVolumeClaim("ns", "bound-pvc").
+				StorageClass("test-sc").
+				VolumeName("test-pv").
+				Phase(corev1api.ClaimBound).
+				Result(),
+			pv:             builder.ForPersistentVolume("test-pv").StorageClass("test-sc").Result(),
+			expectedAction: nil,
+			expectError:    false,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			// Build fake client with PV if present
+			clientBuilder := ctrlfake.NewClientBuilder().WithScheme(scheme)
+			if tc.pv != nil {
+				clientBuilder = clientBuilder.WithObjects(tc.pv)
+			}
+			fakeClient := clientBuilder.Build()
+
+			ib := &itemBackupper{
+				kbClient: fakeClient,
+				backupRequest: &Request{
+					ResPolicies: policies,
+				},
+			}
+
+			// Convert PVC to unstructured
+			pvcData, err := runtime.DefaultUnstructuredConverter.ToUnstructured(tc.pvc)
+			require.NoError(t, err)
+			obj := &unstructured.Unstructured{Object: pvcData}
+
+			action, err := ib.getMatchAction(obj, kuberesource.PersistentVolumeClaims, csiBIAPluginName)
+			if tc.expectError {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+			}
+
+			if tc.expectedAction == nil {
+				assert.Nil(t, action)
+			} else {
+				require.NotNil(t, action)
+				assert.Equal(t, tc.expectedAction.Type, action.Type)
+			}
+		})
+	}
+}
+
+func TestTrackSkippedPV_PendingLostPVC(t *testing.T) {
+	testCases := []struct {
+		name string
+		pvc  *corev1api.PersistentVolumeClaim
+	}{
+		{
+			name: "Pending PVC should log at info level",
+			pvc: builder.ForPersistentVolumeClaim("ns", "pending-pvc").
+				Phase(corev1api.ClaimPending).
+				Result(),
+		},
+		{
+			name: "Lost PVC should log at info level",
+			pvc: builder.ForPersistentVolumeClaim("ns", "lost-pvc").
+				Phase(corev1api.ClaimLost).
+				Result(),
+		},
+		{
+			name: "Bound PVC without VolumeName should log at info level",
+			pvc: builder.ForPersistentVolumeClaim("ns", "bound-pvc").
+				Phase(corev1api.ClaimBound).
+				Result(),
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			ib := &itemBackupper{
+				backupRequest: &Request{
+					SkippedPVTracker: NewSkipPVTracker(),
+				},
+			}
+
+			// Set up log capture
+			logOutput := &bytes.Buffer{}
+			logger := logrus.New()
+			logger.SetOutput(logOutput)
+			logger.SetLevel(logrus.DebugLevel)
+
+			// Convert PVC to unstructured
+			pvcData, err := runtime.DefaultUnstructuredConverter.ToUnstructured(tc.pvc)
+			require.NoError(t, err)
+			obj := &unstructured.Unstructured{Object: pvcData}
+
+			ib.trackSkippedPV(obj, kuberesource.PersistentVolumeClaims, "", "test reason", logger)
+
+			logStr := logOutput.String()
+			assert.Contains(t, logStr, "level=info")
+			assert.Contains(t, logStr, "unable to get PV name, skip tracking.")
+		})
+	}
+}
+
+func TestUnTrackSkippedPV_PendingLostPVC(t *testing.T) {
+	testCases := []struct {
+		name               string
+		pvc                *corev1api.PersistentVolumeClaim
+		expectWarningLog   bool
+		expectDebugMessage string
+	}{
+		{
+			name: "Pending PVC should log at debug level, not warning",
+			pvc: builder.ForPersistentVolumeClaim("ns", "pending-pvc").
+				Phase(corev1api.ClaimPending).
+				Result(),
+			expectWarningLog:   false,
+			expectDebugMessage: "unable to get PV name for Pending PVC, skip untracking.",
+		},
+		{
+			name: "Lost PVC should log at debug level, not warning",
+			pvc: builder.ForPersistentVolumeClaim("ns", "lost-pvc").
+				Phase(corev1api.ClaimLost).
+				Result(),
+			expectWarningLog:   false,
+			expectDebugMessage: "unable to get PV name for Lost PVC, skip untracking.",
+		},
+		{
+			name: "Bound PVC without VolumeName should log warning",
+			pvc: builder.ForPersistentVolumeClaim("ns", "bound-pvc").
+				Phase(corev1api.ClaimBound).
+				Result(),
+			expectWarningLog:   true,
+			expectDebugMessage: "",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			ib := &itemBackupper{
+				backupRequest: &Request{
+					SkippedPVTracker: NewSkipPVTracker(),
+				},
+			}
+
+			// Set up log capture
+			logOutput := &bytes.Buffer{}
+			logger := logrus.New()
+			logger.SetOutput(logOutput)
+			logger.SetLevel(logrus.DebugLevel)
+
+			// Convert PVC to unstructured
+			pvcData, err := runtime.DefaultUnstructuredConverter.ToUnstructured(tc.pvc)
+			require.NoError(t, err)
+			obj := &unstructured.Unstructured{Object: pvcData}
+
+			ib.unTrackSkippedPV(obj, kuberesource.PersistentVolumeClaims, logger)
+
+			logStr := logOutput.String()
+			if tc.expectWarningLog {
+				assert.Contains(t, logStr, "level=warning")
+				assert.Contains(t, logStr, "unable to get PV name, skip untracking.")
+			} else {
+				assert.NotContains(t, logStr, "level=warning")
+				if tc.expectDebugMessage != "" {
+					assert.Contains(t, logStr, "level=debug")
+					assert.Contains(t, logStr, tc.expectDebugMessage)
+				}
+			}
+		})
+	}
+}

pkg/cmd/cli/install/install.go

@@ -275,11 +275,21 @@ func (o *Options) AsVeleroOptions() (*install.VeleroOptions, error) {
 			return nil, err
 		}
 	}
-	veleroPodResources, err := kubeutil.ParseResourceRequirements(o.VeleroPodCPURequest, o.VeleroPodMemRequest, o.VeleroPodCPULimit, o.VeleroPodMemLimit)
+	veleroPodResources, err := kubeutil.ParseCPUAndMemoryResources(
+		o.VeleroPodCPURequest,
+		o.VeleroPodMemRequest,
+		o.VeleroPodCPULimit,
+		o.VeleroPodMemLimit,
+	)
 	if err != nil {
 		return nil, err
 	}
-	nodeAgentPodResources, err := kubeutil.ParseResourceRequirements(o.NodeAgentPodCPURequest, o.NodeAgentPodMemRequest, o.NodeAgentPodCPULimit, o.NodeAgentPodMemLimit)
+	nodeAgentPodResources, err := kubeutil.ParseCPUAndMemoryResources(
+		o.NodeAgentPodCPURequest,
+		o.NodeAgentPodMemRequest,
+		o.NodeAgentPodCPULimit,
+		o.NodeAgentPodMemLimit,
+	)
 	if err != nil {
 		return nil, err
 	}

pkg/cmd/cli/nodeagent/server.go

@@ -323,7 +323,25 @@ func (s *nodeAgentServer) run() {
 
 	podResources := corev1api.ResourceRequirements{}
 	if s.dataPathConfigs != nil && s.dataPathConfigs.PodResources != nil {
-		if res, err := kube.ParseResourceRequirements(s.dataPathConfigs.PodResources.CPURequest, s.dataPathConfigs.PodResources.MemoryRequest, s.dataPathConfigs.PodResources.CPULimit, s.dataPathConfigs.PodResources.MemoryLimit); err != nil {
+		// To make the PodResources ConfigMap without ephemeral storage request/limit backward compatible,
+		// need to avoid set value as empty, because empty string will cause parsing error.
+		ephemeralStorageRequest := constant.DefaultEphemeralStorageRequest
+		if s.dataPathConfigs.PodResources.EphemeralStorageRequest != "" {
+			ephemeralStorageRequest = s.dataPathConfigs.PodResources.EphemeralStorageRequest
+		}
+		ephemeralStorageLimit := constant.DefaultEphemeralStorageLimit
+		if s.dataPathConfigs.PodResources.EphemeralStorageLimit != "" {
+			ephemeralStorageLimit = s.dataPathConfigs.PodResources.EphemeralStorageLimit
+		}
+
+		if res, err := kube.ParseResourceRequirements(
+			s.dataPathConfigs.PodResources.CPURequest,
+			s.dataPathConfigs.PodResources.MemoryRequest,
+			ephemeralStorageRequest,
+			s.dataPathConfigs.PodResources.CPULimit,
+			s.dataPathConfigs.PodResources.MemoryLimit,
+			ephemeralStorageLimit,
+		); err != nil {
 			s.logger.WithError(err).Warn("Pod resource requirements are invalid, ignore")
 		} else {
 			podResources = res

pkg/constant/constant.go

@@ -23,4 +23,7 @@ const (
 
 	PluginCSIPVCRestoreRIA            = "velero.io/csi-pvc-restorer"
 	PluginCsiVolumeSnapshotRestoreRIA = "velero.io/csi-volumesnapshot-restorer"
+
+	DefaultEphemeralStorageRequest = "0"
+	DefaultEphemeralStorageLimit   = "0"
 )

pkg/exposer/csi_snapshot.go

@@ -124,6 +124,15 @@ func (e *csiSnapshotExposer) Expose(ctx context.Context, ownerObject corev1api.O
 		"owner": ownerObject.Name,
 	})
 
+	volumeTopology, err := kube.GetVolumeTopology(ctx, e.kubeClient.CoreV1(), e.kubeClient.StorageV1(), csiExposeParam.SourcePVName, csiExposeParam.StorageClass)
+	if err != nil {
+		return errors.Wrapf(err, "error getting volume topology for PV %s, storage class %s", csiExposeParam.SourcePVName, csiExposeParam.StorageClass)
+	}
+
+	if volumeTopology != nil {
+		curLog.Infof("Using volume topology %v", volumeTopology)
+	}
+
 	curLog.Info("Exposing CSI snapshot")
 
 	volumeSnapshot, err := csi.WaitVolumeSnapshotReady(ctx, e.csiSnapshotClient, csiExposeParam.SnapshotName, csiExposeParam.SourceNamespace, csiExposeParam.ExposeTimeout, curLog)
@@ -254,6 +263,7 @@ func (e *csiSnapshotExposer) Expose(ctx context.Context, ownerObject corev1api.O
 		csiExposeParam.NodeOS,
 		csiExposeParam.PriorityClassName,
 		intoleratableNodes,
+		volumeTopology,
 	)
 	if err != nil {
 		return errors.Wrap(err, "error to create backup pod")
@@ -320,7 +330,8 @@ func (e *csiSnapshotExposer) GetExposed(ctx context.Context, ownerObject corev1a
 	curLog.WithField("pod", pod.Name).Infof("Backup volume is found in pod at index %v", i)
 
 	var nodeOS *string
-	if os, found := pod.Spec.NodeSelector[kube.NodeOSLabel]; found {
+	if pod.Spec.OS != nil {
+		os := string(pod.Spec.OS.Name)
 		nodeOS = &os
 	}
 
@@ -588,6 +599,7 @@ func (e *csiSnapshotExposer) createBackupPod(
 	nodeOS string,
 	priorityClassName string,
 	intoleratableNodes []string,
+	volumeTopology *corev1api.NodeSelector,
 ) (*corev1api.Pod, error) {
 	podName := ownerObject.Name
 
@@ -643,6 +655,10 @@ func (e *csiSnapshotExposer) createBackupPod(
 	args = append(args, podInfo.logFormatArgs...)
 	args = append(args, podInfo.logLevelArgs...)
 
+	if affinity == nil {
+		affinity = &kube.LoadAffinity{}
+	}
+
 	var securityCtx *corev1api.PodSecurityContext
 	nodeSelector := map[string]string{}
 	podOS := corev1api.PodOS{}
@@ -654,9 +670,14 @@ func (e *csiSnapshotExposer) createBackupPod(
 			},
 		}
 
-		nodeSelector[kube.NodeOSLabel] = kube.NodeOSWindows
 		podOS.Name = kube.NodeOSWindows
 
+		affinity.NodeSelector.MatchExpressions = append(affinity.NodeSelector.MatchExpressions, metav1.LabelSelectorRequirement{
+			Key:      kube.NodeOSLabel,
+			Values:   []string{kube.NodeOSWindows},
+			Operator: metav1.LabelSelectorOpIn,
+		})
+
 		toleration = append(toleration, []corev1api.Toleration{
 			{
 				Key:      "os",
@@ -683,11 +704,15 @@ func (e *csiSnapshotExposer) createBackupPod(
 			}
 		}
 
-		nodeSelector[kube.NodeOSLabel] = kube.NodeOSLinux
 		podOS.Name = kube.NodeOSLinux
+
+		affinity.NodeSelector.MatchExpressions = append(affinity.NodeSelector.MatchExpressions, metav1.LabelSelectorRequirement{
+			Key:      kube.NodeOSLabel,
+			Values:   []string{kube.NodeOSWindows},
+			Operator: metav1.LabelSelectorOpNotIn,
+		})
 	}
 
-	var podAffinity *corev1api.Affinity
 	if len(intoleratableNodes) > 0 {
 		if affinity == nil {
 			affinity = &kube.LoadAffinity{}
@@ -700,9 +725,7 @@ func (e *csiSnapshotExposer) createBackupPod(
 		})
 	}
 
-	if affinity != nil {
-		podAffinity = kube.ToSystemAffinity([]*kube.LoadAffinity{affinity})
-	}
+	podAffinity := kube.ToSystemAffinity(affinity, volumeTopology)
 
 	pod := &corev1api.Pod{
 		ObjectMeta: metav1.ObjectMeta{

pkg/exposer/csi_snapshot_priority_test.go

@@ -154,6 +154,7 @@ func TestCreateBackupPodWithPriorityClass(t *testing.T) {
 				kube.NodeOSLinux,
 				tc.expectedPriorityClass,
 				nil,
+				nil,
 			)
 
 			require.NoError(t, err, tc.description)
@@ -239,6 +240,7 @@ func TestCreateBackupPodWithMissingConfigMap(t *testing.T) {
 		kube.NodeOSLinux,
 		"", // empty priority class since config map is missing
 		nil,
+		nil,
 	)
 
 	// Should succeed even when config map is missing

pkg/exposer/csi_snapshot_test.go

@@ -68,6 +68,12 @@ func TestExpose(t *testing.T) {
 
 	var restoreSize int64 = 123456
 
+	scObj := &storagev1api.StorageClass{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "fake-sc",
+		},
+	}
+
 	snapshotClass := "fake-snapshot-class"
 	vsObject := &snapshotv1api.VolumeSnapshot{
 		ObjectMeta: metav1.ObjectMeta{
@@ -199,6 +205,18 @@ func TestExpose(t *testing.T) {
 		expectedAffinity              *corev1api.Affinity
 		expectedPVCAnnotation         map[string]string
 	}{
+		{
+			name:        "get volume topology fail",
+			ownerBackup: backup,
+			exposeParam: CSISnapshotExposeParam{
+				SnapshotName:     "fake-vs",
+				OperationTimeout: time.Millisecond,
+				ExposeTimeout:    time.Millisecond,
+				StorageClass:     "fake-sc",
+				SourcePVName:     "fake-pv",
+			},
+			err: "error getting volume topology for PV fake-pv, storage class fake-sc: error getting storage class fake-sc: storageclasses.storage.k8s.io \"fake-sc\" not found",
+		},
 		{
 			name:        "wait vs ready fail",
 			ownerBackup: backup,
@@ -206,6 +224,11 @@ func TestExpose(t *testing.T) {
 				SnapshotName:     "fake-vs",
 				OperationTimeout: time.Millisecond,
 				ExposeTimeout:    time.Millisecond,
+				StorageClass:     "fake-sc",
+				SourcePVName:     "fake-pv",
+			},
+			kubeClientObj: []runtime.Object{
+				scObj,
 			},
 			err: "error wait volume snapshot ready: error to get VolumeSnapshot /fake-vs: volumesnapshots.snapshot.storage.k8s.io \"fake-vs\" not found",
 		},
@@ -217,10 +240,15 @@ func TestExpose(t *testing.T) {
 				SourceNamespace:  "fake-ns",
 				OperationTimeout: time.Millisecond,
 				ExposeTimeout:    time.Millisecond,
+				StorageClass:     "fake-sc",
+				SourcePVName:     "fake-pv",
 			},
 			snapshotClientObj: []runtime.Object{
 				vsObject,
 			},
+			kubeClientObj: []runtime.Object{
+				scObj,
+			},
 			err: "error to get volume snapshot content: error getting volume snapshot content from API: volumesnapshotcontents.snapshot.storage.k8s.io \"fake-vsc\" not found",
 		},
 		{
@@ -231,6 +259,8 @@ func TestExpose(t *testing.T) {
 				SourceNamespace:  "fake-ns",
 				OperationTimeout: time.Millisecond,
 				ExposeTimeout:    time.Millisecond,
+				StorageClass:     "fake-sc",
+				SourcePVName:     "fake-pv",
 			},
 			snapshotClientObj: []runtime.Object{
 				vsObject,
@@ -245,6 +275,9 @@ func TestExpose(t *testing.T) {
 					},
 				},
 			},
+			kubeClientObj: []runtime.Object{
+				scObj,
+			},
 			err: "error to delete volume snapshot: error to delete volume snapshot: fake-delete-error",
 		},
 		{
@@ -255,6 +288,8 @@ func TestExpose(t *testing.T) {
 				SourceNamespace:  "fake-ns",
 				OperationTimeout: time.Millisecond,
 				ExposeTimeout:    time.Millisecond,
+				StorageClass:     "fake-sc",
+				SourcePVName:     "fake-pv",
 			},
 			snapshotClientObj: []runtime.Object{
 				vsObject,
@@ -269,6 +304,9 @@ func TestExpose(t *testing.T) {
 					},
 				},
 			},
+			kubeClientObj: []runtime.Object{
+				scObj,
+			},
 			err: "error to delete volume snapshot content: error to delete volume snapshot content: fake-delete-error",
 		},
 		{
@@ -279,6 +317,8 @@ func TestExpose(t *testing.T) {
 				SourceNamespace:  "fake-ns",
 				OperationTimeout: time.Millisecond,
 				ExposeTimeout:    time.Millisecond,
+				StorageClass:     "fake-sc",
+				SourcePVName:     "fake-pv",
 			},
 			snapshotClientObj: []runtime.Object{
 				vsObject,
@@ -293,6 +333,9 @@ func TestExpose(t *testing.T) {
 					},
 				},
 			},
+			kubeClientObj: []runtime.Object{
+				scObj,
+			},
 			err: "error to create backup volume snapshot: fake-create-error",
 		},
 		{
@@ -303,6 +346,8 @@ func TestExpose(t *testing.T) {
 				SourceNamespace:  "fake-ns",
 				OperationTimeout: time.Millisecond,
 				ExposeTimeout:    time.Millisecond,
+				StorageClass:     "fake-sc",
+				SourcePVName:     "fake-pv",
 			},
 			snapshotClientObj: []runtime.Object{
 				vsObject,
@@ -317,6 +362,9 @@ func TestExpose(t *testing.T) {
 					},
 				},
 			},
+			kubeClientObj: []runtime.Object{
+				scObj,
+			},
 			err: "error to create backup volume snapshot content: fake-create-error",
 		},
 		{
@@ -326,11 +374,16 @@ func TestExpose(t *testing.T) {
 				SnapshotName:    "fake-vs",
 				SourceNamespace: "fake-ns",
 				AccessMode:      "fake-mode",
+				StorageClass:    "fake-sc",
+				SourcePVName:    "fake-pv",
 			},
 			snapshotClientObj: []runtime.Object{
 				vsObject,
 				vscObj,
 			},
+			kubeClientObj: []runtime.Object{
+				scObj,
+			},
 			err: "error to create backup pvc: unsupported access mode fake-mode",
 		},
 		{
@@ -342,6 +395,8 @@ func TestExpose(t *testing.T) {
 				OperationTimeout: time.Millisecond,
 				ExposeTimeout:    time.Millisecond,
 				AccessMode:       AccessModeFileSystem,
+				StorageClass:     "fake-sc",
+				SourcePVName:     "fake-pv",
 			},
 			snapshotClientObj: []runtime.Object{
 				vsObject,
@@ -356,6 +411,9 @@ func TestExpose(t *testing.T) {
 					},
 				},
 			},
+			kubeClientObj: []runtime.Object{
+				scObj,
+			},
 			err: "error to create backup pvc: error to create pvc: fake-create-error",
 		},
 		{
@@ -367,6 +425,8 @@ func TestExpose(t *testing.T) {
 				AccessMode:       AccessModeFileSystem,
 				OperationTimeout: time.Millisecond,
 				ExposeTimeout:    time.Millisecond,
+				StorageClass:     "fake-sc",
+				SourcePVName:     "fake-pv",
 			},
 			snapshotClientObj: []runtime.Object{
 				vsObject,
@@ -374,6 +434,7 @@ func TestExpose(t *testing.T) {
 			},
 			kubeClientObj: []runtime.Object{
 				daemonSet,
+				scObj,
 			},
 			kubeReactors: []reactor{
 				{
@@ -395,6 +456,8 @@ func TestExpose(t *testing.T) {
 				AccessMode:       AccessModeFileSystem,
 				OperationTimeout: time.Millisecond,
 				ExposeTimeout:    time.Millisecond,
+				StorageClass:     "fake-sc",
+				SourcePVName:     "fake-pv",
 			},
 			snapshotClientObj: []runtime.Object{
 				vsObject,
@@ -402,6 +465,24 @@ func TestExpose(t *testing.T) {
 			},
 			kubeClientObj: []runtime.Object{
 				daemonSet,
+				scObj,
+			},
+			expectedAffinity: &corev1api.Affinity{
+				NodeAffinity: &corev1api.NodeAffinity{
+					RequiredDuringSchedulingIgnoredDuringExecution: &corev1api.NodeSelector{
+						NodeSelectorTerms: []corev1api.NodeSelectorTerm{
+							{
+								MatchExpressions: []corev1api.NodeSelectorRequirement{
+									{
+										Key:      "kubernetes.io/os",
+										Operator: corev1api.NodeSelectorOpNotIn,
+										Values:   []string{"windows"},
+									},
+								},
+							},
+						},
+					},
+				},
 			},
 		},
 		{
@@ -413,6 +494,8 @@ func TestExpose(t *testing.T) {
 				AccessMode:       AccessModeFileSystem,
 				OperationTimeout: time.Millisecond,
 				ExposeTimeout:    time.Millisecond,
+				StorageClass:     "fake-sc",
+				SourcePVName:     "fake-pv",
 			},
 			snapshotClientObj: []runtime.Object{
 				vsObject,
@@ -420,6 +503,24 @@ func TestExpose(t *testing.T) {
 			},
 			kubeClientObj: []runtime.Object{
 				daemonSet,
+				scObj,
+			},
+			expectedAffinity: &corev1api.Affinity{
+				NodeAffinity: &corev1api.NodeAffinity{
+					RequiredDuringSchedulingIgnoredDuringExecution: &corev1api.NodeSelector{
+						NodeSelectorTerms: []corev1api.NodeSelectorTerm{
+							{
+								MatchExpressions: []corev1api.NodeSelectorRequirement{
+									{
+										Key:      "kubernetes.io/os",
+										Operator: corev1api.NodeSelectorOpNotIn,
+										Values:   []string{"windows"},
+									},
+								},
+							},
+						},
+					},
+				},
 			},
 		},
 		{
@@ -432,6 +533,8 @@ func TestExpose(t *testing.T) {
 				OperationTimeout: time.Millisecond,
 				ExposeTimeout:    time.Millisecond,
 				VolumeSize:       *resource.NewQuantity(567890, ""),
+				StorageClass:     "fake-sc",
+				SourcePVName:     "fake-pv",
 			},
 			snapshotClientObj: []runtime.Object{
 				vsObjectWithoutRestoreSize,
@@ -439,8 +542,26 @@ func TestExpose(t *testing.T) {
 			},
 			kubeClientObj: []runtime.Object{
 				daemonSet,
+				scObj,
 			},
 			expectedVolumeSize: resource.NewQuantity(567890, ""),
+			expectedAffinity: &corev1api.Affinity{
+				NodeAffinity: &corev1api.NodeAffinity{
+					RequiredDuringSchedulingIgnoredDuringExecution: &corev1api.NodeSelector{
+						NodeSelectorTerms: []corev1api.NodeSelectorTerm{
+							{
+								MatchExpressions: []corev1api.NodeSelectorRequirement{
+									{
+										Key:      "kubernetes.io/os",
+										Operator: corev1api.NodeSelectorOpNotIn,
+										Values:   []string{"windows"},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
 		},
 		{
 			name:        "backupPod mounts read only backupPVC",
@@ -449,6 +570,7 @@ func TestExpose(t *testing.T) {
 				SnapshotName:     "fake-vs",
 				SourceNamespace:  "fake-ns",
 				StorageClass:     "fake-sc",
+				SourcePVName:     "fake-pv",
 				AccessMode:       AccessModeFileSystem,
 				OperationTimeout: time.Millisecond,
 				ExposeTimeout:    time.Millisecond,
@@ -465,8 +587,26 @@ func TestExpose(t *testing.T) {
 			},
 			kubeClientObj: []runtime.Object{
 				daemonSet,
+				scObj,
 			},
 			expectedReadOnlyPVC: true,
+			expectedAffinity: &corev1api.Affinity{
+				NodeAffinity: &corev1api.NodeAffinity{
+					RequiredDuringSchedulingIgnoredDuringExecution: &corev1api.NodeSelector{
+						NodeSelectorTerms: []corev1api.NodeSelectorTerm{
+							{
+								MatchExpressions: []corev1api.NodeSelectorRequirement{
+									{
+										Key:      "kubernetes.io/os",
+										Operator: corev1api.NodeSelectorOpNotIn,
+										Values:   []string{"windows"},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
 		},
 		{
 			name:        "backupPod mounts read only backupPVC and storageClass specified in backupPVC config",
@@ -475,6 +615,7 @@ func TestExpose(t *testing.T) {
 				SnapshotName:     "fake-vs",
 				SourceNamespace:  "fake-ns",
 				StorageClass:     "fake-sc",
+				SourcePVName:     "fake-pv",
 				AccessMode:       AccessModeFileSystem,
 				OperationTimeout: time.Millisecond,
 				ExposeTimeout:    time.Millisecond,
@@ -491,9 +632,27 @@ func TestExpose(t *testing.T) {
 			},
 			kubeClientObj: []runtime.Object{
 				daemonSet,
+				scObj,
 			},
 			expectedReadOnlyPVC:           true,
 			expectedBackupPVCStorageClass: "fake-sc-read-only",
+			expectedAffinity: &corev1api.Affinity{
+				NodeAffinity: &corev1api.NodeAffinity{
+					RequiredDuringSchedulingIgnoredDuringExecution: &corev1api.NodeSelector{
+						NodeSelectorTerms: []corev1api.NodeSelectorTerm{
+							{
+								MatchExpressions: []corev1api.NodeSelectorRequirement{
+									{
+										Key:      "kubernetes.io/os",
+										Operator: corev1api.NodeSelectorOpNotIn,
+										Values:   []string{"windows"},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
 		},
 		{
 			name:        "backupPod mounts backupPVC with storageClass specified in backupPVC config",
@@ -502,6 +661,7 @@ func TestExpose(t *testing.T) {
 				SnapshotName:     "fake-vs",
 				SourceNamespace:  "fake-ns",
 				StorageClass:     "fake-sc",
+				SourcePVName:     "fake-pv",
 				AccessMode:       AccessModeFileSystem,
 				OperationTimeout: time.Millisecond,
 				ExposeTimeout:    time.Millisecond,
@@ -517,8 +677,26 @@ func TestExpose(t *testing.T) {
 			},
 			kubeClientObj: []runtime.Object{
 				daemonSet,
+				scObj,
 			},
 			expectedBackupPVCStorageClass: "fake-sc-read-only",
+			expectedAffinity: &corev1api.Affinity{
+				NodeAffinity: &corev1api.NodeAffinity{
+					RequiredDuringSchedulingIgnoredDuringExecution: &corev1api.NodeSelector{
+						NodeSelectorTerms: []corev1api.NodeSelectorTerm{
+							{
+								MatchExpressions: []corev1api.NodeSelectorRequirement{
+									{
+										Key:      "kubernetes.io/os",
+										Operator: corev1api.NodeSelectorOpNotIn,
+										Values:   []string{"windows"},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
 		},
 		{
 			name:        "Affinity per StorageClass",
@@ -527,6 +705,7 @@ func TestExpose(t *testing.T) {
 				SnapshotName:     "fake-vs",
 				SourceNamespace:  "fake-ns",
 				StorageClass:     "fake-sc",
+				SourcePVName:     "fake-pv",
 				AccessMode:       AccessModeFileSystem,
 				OperationTimeout: time.Millisecond,
 				ExposeTimeout:    time.Millisecond,
@@ -551,6 +730,7 @@ func TestExpose(t *testing.T) {
 			},
 			kubeClientObj: []runtime.Object{
 				daemonSet,
+				scObj,
 			},
 			expectedAffinity: &corev1api.Affinity{
 				NodeAffinity: &corev1api.NodeAffinity{
@@ -563,6 +743,11 @@ func TestExpose(t *testing.T) {
 										Operator: corev1api.NodeSelectorOpIn,
 										Values:   []string{"Linux"},
 									},
+									{
+										Key:      "kubernetes.io/os",
+										Operator: corev1api.NodeSelectorOpNotIn,
+										Values:   []string{"windows"},
+									},
 								},
 							},
 						},
@@ -577,6 +762,7 @@ func TestExpose(t *testing.T) {
 				SnapshotName:     "fake-vs",
 				SourceNamespace:  "fake-ns",
 				StorageClass:     "fake-sc",
+				SourcePVName:     "fake-pv",
 				AccessMode:       AccessModeFileSystem,
 				OperationTimeout: time.Millisecond,
 				ExposeTimeout:    time.Millisecond,
@@ -606,6 +792,7 @@ func TestExpose(t *testing.T) {
 			},
 			kubeClientObj: []runtime.Object{
 				daemonSet,
+				scObj,
 			},
 			expectedBackupPVCStorageClass: "fake-sc-read-only",
 			expectedAffinity: &corev1api.Affinity{
@@ -619,6 +806,11 @@ func TestExpose(t *testing.T) {
 										Operator: corev1api.NodeSelectorOpIn,
 										Values:   []string{"amd64"},
 									},
+									{
+										Key:      "kubernetes.io/os",
+										Operator: corev1api.NodeSelectorOpNotIn,
+										Values:   []string{"windows"},
+									},
 								},
 							},
 						},
@@ -633,6 +825,7 @@ func TestExpose(t *testing.T) {
 				SnapshotName:     "fake-vs",
 				SourceNamespace:  "fake-ns",
 				StorageClass:     "fake-sc",
+				SourcePVName:     "fake-pv",
 				AccessMode:       AccessModeFileSystem,
 				OperationTimeout: time.Millisecond,
 				ExposeTimeout:    time.Millisecond,
@@ -649,9 +842,26 @@ func TestExpose(t *testing.T) {
 			},
 			kubeClientObj: []runtime.Object{
 				daemonSet,
+				scObj,
 			},
 			expectedBackupPVCStorageClass: "fake-sc-read-only",
-			expectedAffinity:              nil,
+			expectedAffinity: &corev1api.Affinity{
+				NodeAffinity: &corev1api.NodeAffinity{
+					RequiredDuringSchedulingIgnoredDuringExecution: &corev1api.NodeSelector{
+						NodeSelectorTerms: []corev1api.NodeSelectorTerm{
+							{
+								MatchExpressions: []corev1api.NodeSelectorRequirement{
+									{
+										Key:      "kubernetes.io/os",
+										Operator: corev1api.NodeSelectorOpNotIn,
+										Values:   []string{"windows"},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
 		},
 		{
 			name:        "IntolerateSourceNode, get source node fail",
@@ -677,6 +887,7 @@ func TestExpose(t *testing.T) {
 			},
 			kubeClientObj: []runtime.Object{
 				daemonSet,
+				scObj,
 			},
 			kubeReactors: []reactor{
 				{
@@ -687,7 +898,23 @@ func TestExpose(t *testing.T) {
 					},
 				},
 			},
-			expectedAffinity:      nil,
+			expectedAffinity: &corev1api.Affinity{
+				NodeAffinity: &corev1api.NodeAffinity{
+					RequiredDuringSchedulingIgnoredDuringExecution: &corev1api.NodeSelector{
+						NodeSelectorTerms: []corev1api.NodeSelectorTerm{
+							{
+								MatchExpressions: []corev1api.NodeSelectorRequirement{
+									{
+										Key:      "kubernetes.io/os",
+										Operator: corev1api.NodeSelectorOpNotIn,
+										Values:   []string{"windows"},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
 			expectedPVCAnnotation: nil,
 		},
 		{
@@ -714,8 +941,25 @@ func TestExpose(t *testing.T) {
 			},
 			kubeClientObj: []runtime.Object{
 				daemonSet,
+				scObj,
+			},
+			expectedAffinity: &corev1api.Affinity{
+				NodeAffinity: &corev1api.NodeAffinity{
+					RequiredDuringSchedulingIgnoredDuringExecution: &corev1api.NodeSelector{
+						NodeSelectorTerms: []corev1api.NodeSelectorTerm{
+							{
+								MatchExpressions: []corev1api.NodeSelectorRequirement{
+									{
+										Key:      "kubernetes.io/os",
+										Operator: corev1api.NodeSelectorOpNotIn,
+										Values:   []string{"windows"},
+									},
+								},
+							},
+						},
+					},
+				},
 			},
-			expectedAffinity:      nil,
 			expectedPVCAnnotation: map[string]string{util.VSphereCNSFastCloneAnno: "true"},
 		},
 		{
@@ -744,6 +988,7 @@ func TestExpose(t *testing.T) {
 				daemonSet,
 				volumeAttachement1,
 				volumeAttachement2,
+				scObj,
 			},
 			expectedAffinity: &corev1api.Affinity{
 				NodeAffinity: &corev1api.NodeAffinity{
@@ -751,6 +996,11 @@ func TestExpose(t *testing.T) {
 						NodeSelectorTerms: []corev1api.NodeSelectorTerm{
 							{
 								MatchExpressions: []corev1api.NodeSelectorRequirement{
+									{
+										Key:      "kubernetes.io/os",
+										Operator: corev1api.NodeSelectorOpNotIn,
+										Values:   []string{"windows"},
+									},
 									{
 										Key:      "kubernetes.io/hostname",
 										Operator: corev1api.NodeSelectorOpNotIn,
@@ -844,6 +1094,8 @@ func TestExpose(t *testing.T) {
 
 				if test.expectedAffinity != nil {
 					assert.Equal(t, test.expectedAffinity, backupPod.Spec.Affinity)
+				} else {
+					assert.Nil(t, backupPod.Spec.Affinity)
 				}
 
 				if test.expectedPVCAnnotation != nil {

pkg/exposer/generic_restore.go

@@ -493,13 +493,15 @@ func (e *genericRestoreExposer) createRestorePod(
 	containerName := string(ownerObject.UID)
 	volumeName := string(ownerObject.UID)
 
-	var podAffinity *corev1api.Affinity
-	if selectedNode == "" {
-		e.log.Infof("No selected node for restore pod. Try to get affinity from the node-agent config.")
+	nodeSelector := map[string]string{}
+	if selectedNode != "" {
+		affinity = nil
+		nodeSelector["kubernetes.io/hostname"] = selectedNode
+		e.log.Infof("Selected node for restore pod. Ignore affinity from the node-agent config.")
+	}
 
-		if affinity != nil {
-			podAffinity = kube.ToSystemAffinity([]*kube.LoadAffinity{affinity})
-		}
+	if affinity == nil {
+		affinity = &kube.LoadAffinity{}
 	}
 
 	podInfo, err := getInheritedPodInfo(ctx, e.kubeClient, ownerObject.Namespace, nodeOS)
@@ -566,7 +568,6 @@ func (e *genericRestoreExposer) createRestorePod(
 	args = append(args, podInfo.logLevelArgs...)
 
 	var securityCtx *corev1api.PodSecurityContext
-	nodeSelector := map[string]string{}
 	podOS := corev1api.PodOS{}
 	if nodeOS == kube.NodeOSWindows {
 		userID := "ContainerAdministrator"
@@ -576,9 +577,14 @@ func (e *genericRestoreExposer) createRestorePod(
 			},
 		}
 
-		nodeSelector[kube.NodeOSLabel] = kube.NodeOSWindows
 		podOS.Name = kube.NodeOSWindows
 
+		affinity.NodeSelector.MatchExpressions = append(affinity.NodeSelector.MatchExpressions, metav1.LabelSelectorRequirement{
+			Key:      kube.NodeOSLabel,
+			Values:   []string{kube.NodeOSWindows},
+			Operator: metav1.LabelSelectorOpIn,
+		})
+
 		toleration = append(toleration, []corev1api.Toleration{
 			{
 				Key:      "os",
@@ -599,10 +605,17 @@ func (e *genericRestoreExposer) createRestorePod(
 			RunAsUser: &userID,
 		}
 
-		nodeSelector[kube.NodeOSLabel] = kube.NodeOSLinux
 		podOS.Name = kube.NodeOSLinux
+
+		affinity.NodeSelector.MatchExpressions = append(affinity.NodeSelector.MatchExpressions, metav1.LabelSelectorRequirement{
+			Key:      kube.NodeOSLabel,
+			Values:   []string{kube.NodeOSWindows},
+			Operator: metav1.LabelSelectorOpNotIn,
+		})
 	}
 
+	podAffinity := kube.ToSystemAffinity(affinity, nil)
+
 	pod := &corev1api.Pod{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      restorePodName,
@@ -656,7 +669,6 @@ func (e *genericRestoreExposer) createRestorePod(
 			ServiceAccountName:            podInfo.serviceAccount,
 			TerminationGracePeriodSeconds: &gracePeriod,
 			Volumes:                       volumes,
-			NodeName:                      selectedNode,
 			RestartPolicy:                 corev1api.RestartPolicyNever,
 			SecurityContext:               securityCtx,
 			Tolerations:                   toleration,

pkg/exposer/pod_volume.go

@@ -434,6 +434,8 @@ func (e *podVolumeExposer) createHostingPod(
 	args = append(args, podInfo.logFormatArgs...)
 	args = append(args, podInfo.logLevelArgs...)
 
+	affinity := &kube.LoadAffinity{}
+
 	var securityCtx *corev1api.PodSecurityContext
 	var containerSecurityCtx *corev1api.SecurityContext
 	nodeSelector := map[string]string{}
@@ -446,9 +448,14 @@ func (e *podVolumeExposer) createHostingPod(
 			},
 		}
 
-		nodeSelector[kube.NodeOSLabel] = kube.NodeOSWindows
 		podOS.Name = kube.NodeOSWindows
 
+		affinity.NodeSelector.MatchExpressions = append(affinity.NodeSelector.MatchExpressions, metav1.LabelSelectorRequirement{
+			Key:      kube.NodeOSLabel,
+			Values:   []string{kube.NodeOSWindows},
+			Operator: metav1.LabelSelectorOpIn,
+		})
+
 		toleration = append(toleration, []corev1api.Toleration{
 			{
 				Key:      "os",
@@ -472,10 +479,17 @@ func (e *podVolumeExposer) createHostingPod(
 			Privileged: &privileged,
 		}
 
-		nodeSelector[kube.NodeOSLabel] = kube.NodeOSLinux
 		podOS.Name = kube.NodeOSLinux
+
+		affinity.NodeSelector.MatchExpressions = append(affinity.NodeSelector.MatchExpressions, metav1.LabelSelectorRequirement{
+			Key:      kube.NodeOSLabel,
+			Values:   []string{kube.NodeOSWindows},
+			Operator: metav1.LabelSelectorOpNotIn,
+		})
 	}
 
+	podAffinity := kube.ToSystemAffinity(affinity, nil)
+
 	pod := &corev1api.Pod{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      hostingPodName,
@@ -495,6 +509,7 @@ func (e *podVolumeExposer) createHostingPod(
 		Spec: corev1api.PodSpec{
 			NodeSelector: nodeSelector,
 			OS:           &podOS,
+			Affinity:     podAffinity,
 			Containers: []corev1api.Container{
 				{
 					Name:            containerName,

pkg/install/daemonset.go

@@ -235,12 +235,28 @@ func DaemonSet(namespace string, opts ...podTemplateOption) *appsv1api.DaemonSet
 	if c.forWindows {
 		daemonSet.Spec.Template.Spec.SecurityContext = nil
 		daemonSet.Spec.Template.Spec.Containers[0].SecurityContext = nil
-		daemonSet.Spec.Template.Spec.NodeSelector = map[string]string{
-			"kubernetes.io/os": "windows",
-		}
 		daemonSet.Spec.Template.Spec.OS = &corev1api.PodOS{
 			Name: "windows",
 		}
+
+		daemonSet.Spec.Template.Spec.Affinity = &corev1api.Affinity{
+			NodeAffinity: &corev1api.NodeAffinity{
+				RequiredDuringSchedulingIgnoredDuringExecution: &corev1api.NodeSelector{
+					NodeSelectorTerms: []corev1api.NodeSelectorTerm{
+						{
+							MatchExpressions: []corev1api.NodeSelectorRequirement{
+								{
+									Key:      "kubernetes.io/os",
+									Values:   []string{"windows"},
+									Operator: corev1api.NodeSelectorOpIn,
+								},
+							},
+						},
+					},
+				},
+			},
+		}
+
 		daemonSet.Spec.Template.Spec.Tolerations = []corev1api.Toleration{
 			{
 				Key:      "os",
@@ -256,11 +272,22 @@ func DaemonSet(namespace string, opts ...podTemplateOption) *appsv1api.DaemonSet
 			},
 		}
 	} else {
-		daemonSet.Spec.Template.Spec.NodeSelector = map[string]string{
-			"kubernetes.io/os": "linux",
-		}
-		daemonSet.Spec.Template.Spec.OS = &corev1api.PodOS{
-			Name: "linux",
+		daemonSet.Spec.Template.Spec.Affinity = &corev1api.Affinity{
+			NodeAffinity: &corev1api.NodeAffinity{
+				RequiredDuringSchedulingIgnoredDuringExecution: &corev1api.NodeSelector{
+					NodeSelectorTerms: []corev1api.NodeSelectorTerm{
+						{
+							MatchExpressions: []corev1api.NodeSelectorRequirement{
+								{
+									Key:      "kubernetes.io/os",
+									Values:   []string{"windows"},
+									Operator: corev1api.NodeSelectorOpNotIn,
+								},
+							},
+						},
+					},
+				},
+			},
 		}
 	}
 

pkg/install/daemonset_test.go

@@ -34,8 +34,23 @@ func TestDaemonSet(t *testing.T) {
 	assert.Equal(t, "velero", ds.ObjectMeta.Namespace)
 	assert.Equal(t, "node-agent", ds.Spec.Template.ObjectMeta.Labels["name"])
 	assert.Equal(t, "node-agent", ds.Spec.Template.ObjectMeta.Labels["role"])
-	assert.Equal(t, "linux", ds.Spec.Template.Spec.NodeSelector["kubernetes.io/os"])
-	assert.Equal(t, "linux", string(ds.Spec.Template.Spec.OS.Name))
+	assert.Equal(t, &corev1api.Affinity{
+		NodeAffinity: &corev1api.NodeAffinity{
+			RequiredDuringSchedulingIgnoredDuringExecution: &corev1api.NodeSelector{
+				NodeSelectorTerms: []corev1api.NodeSelectorTerm{
+					{
+						MatchExpressions: []corev1api.NodeSelectorRequirement{
+							{
+								Key:      "kubernetes.io/os",
+								Values:   []string{"windows"},
+								Operator: corev1api.NodeSelectorOpNotIn,
+							},
+						},
+					},
+				},
+			},
+		},
+	}, ds.Spec.Template.Spec.Affinity)
 	assert.Equal(t, corev1api.PodSecurityContext{RunAsUser: &userID}, *ds.Spec.Template.Spec.SecurityContext)
 	assert.Equal(t, corev1api.SecurityContext{Privileged: &boolFalse}, *ds.Spec.Template.Spec.Containers[0].SecurityContext)
 	assert.Len(t, ds.Spec.Template.Spec.Volumes, 3)
@@ -80,8 +95,24 @@ func TestDaemonSet(t *testing.T) {
 	assert.Equal(t, "velero", ds.ObjectMeta.Namespace)
 	assert.Equal(t, "node-agent-windows", ds.Spec.Template.ObjectMeta.Labels["name"])
 	assert.Equal(t, "node-agent", ds.Spec.Template.ObjectMeta.Labels["role"])
-	assert.Equal(t, "windows", ds.Spec.Template.Spec.NodeSelector["kubernetes.io/os"])
 	assert.Equal(t, "windows", string(ds.Spec.Template.Spec.OS.Name))
+	assert.Equal(t, &corev1api.Affinity{
+		NodeAffinity: &corev1api.NodeAffinity{
+			RequiredDuringSchedulingIgnoredDuringExecution: &corev1api.NodeSelector{
+				NodeSelectorTerms: []corev1api.NodeSelectorTerm{
+					{
+						MatchExpressions: []corev1api.NodeSelectorRequirement{
+							{
+								Key:      "kubernetes.io/os",
+								Values:   []string{"windows"},
+								Operator: corev1api.NodeSelectorOpIn,
+							},
+						},
+					},
+				},
+			},
+		},
+	}, ds.Spec.Template.Spec.Affinity)
 	assert.Equal(t, (*corev1api.PodSecurityContext)(nil), ds.Spec.Template.Spec.SecurityContext)
 	assert.Equal(t, (*corev1api.SecurityContext)(nil), ds.Spec.Template.Spec.Containers[0].SecurityContext)
 }

pkg/install/deployment.go

@@ -364,12 +364,26 @@ func Deployment(namespace string, opts ...podTemplateOption) *appsv1api.Deployme
 				Spec: corev1api.PodSpec{
 					RestartPolicy:      corev1api.RestartPolicyAlways,
 					ServiceAccountName: c.serviceAccountName,
-					NodeSelector: map[string]string{
-						"kubernetes.io/os": "linux",
-					},
 					OS: &corev1api.PodOS{
 						Name: "linux",
 					},
+					Affinity: &corev1api.Affinity{
+						NodeAffinity: &corev1api.NodeAffinity{
+							RequiredDuringSchedulingIgnoredDuringExecution: &corev1api.NodeSelector{
+								NodeSelectorTerms: []corev1api.NodeSelectorTerm{
+									{
+										MatchExpressions: []corev1api.NodeSelectorRequirement{
+											{
+												Key:      "kubernetes.io/os",
+												Values:   []string{"windows"},
+												Operator: corev1api.NodeSelectorOpNotIn,
+											},
+										},
+									},
+								},
+							},
+						},
+					},
 					Containers: []corev1api.Container{
 						{
 							Name:            "velero",

pkg/install/deployment_test.go

@@ -100,8 +100,23 @@ func TestDeployment(t *testing.T) {
 	assert.Len(t, deploy.Spec.Template.Spec.Containers[0].Args, 2)
 	assert.Equal(t, "--repo-maintenance-job-configmap=test-repo-maintenance-config", deploy.Spec.Template.Spec.Containers[0].Args[1])
 
-	assert.Equal(t, "linux", deploy.Spec.Template.Spec.NodeSelector["kubernetes.io/os"])
-	assert.Equal(t, "linux", string(deploy.Spec.Template.Spec.OS.Name))
+	assert.Equal(t, &corev1api.Affinity{
+		NodeAffinity: &corev1api.NodeAffinity{
+			RequiredDuringSchedulingIgnoredDuringExecution: &corev1api.NodeSelector{
+				NodeSelectorTerms: []corev1api.NodeSelectorTerm{
+					{
+						MatchExpressions: []corev1api.NodeSelectorRequirement{
+							{
+								Key:      "kubernetes.io/os",
+								Values:   []string{"windows"},
+								Operator: corev1api.NodeSelectorOpNotIn,
+							},
+						},
+					},
+				},
+			},
+		},
+	}, deploy.Spec.Template.Spec.Affinity)
 }
 
 func TestDeploymentWithPriorityClassName(t *testing.T) {

pkg/podvolume/backupper.go

@@ -210,11 +210,9 @@ func resultsKey(ns, name string) string {
 
 func (b *backupper) getMatchAction(resPolicies *resourcepolicies.Policies, pvc *corev1api.PersistentVolumeClaim, volume *corev1api.Volume) (*resourcepolicies.Action, error) {
 	if pvc != nil {
-		pv := new(corev1api.PersistentVolume)
-		err := b.crClient.Get(context.TODO(), ctrlclient.ObjectKey{Name: pvc.Spec.VolumeName}, pv)
-		if err != nil {
-			return nil, errors.Wrapf(err, "error getting pv for pvc %s", pvc.Spec.VolumeName)
-		}
+		// Ignore err, if the PV is not available (Pending/Lost PVC or PV fetch failed) - try matching with PVC only
+		// GetPVForPVC returns nil for all error cases
+		pv, _ := kube.GetPVForPVC(pvc, b.crClient)
 		vfd := resourcepolicies.NewVolumeFilterData(pv, nil, pvc)
 		return resPolicies.GetMatchAction(vfd)
 	}

pkg/podvolume/backupper_test.go

@@ -309,8 +309,8 @@ func createNodeObj() *corev1api.Node {
 
 func TestBackupPodVolumes(t *testing.T) {
 	scheme := runtime.NewScheme()
-	velerov1api.AddToScheme(scheme)
-	corev1api.AddToScheme(scheme)
+	require.NoError(t, velerov1api.AddToScheme(scheme))
+	require.NoError(t, corev1api.AddToScheme(scheme))
 	log := logrus.New()
 
 	tests := []struct {
@@ -778,7 +778,7 @@ func TestWaitAllPodVolumesProcessed(t *testing.T) {
 
 		backuper := newBackupper(c.ctx, log, nil, nil, informer, nil, "", &velerov1api.Backup{})
 		if c.pvb != nil {
-			backuper.pvbIndexer.Add(c.pvb)
+			require.NoError(t, backuper.pvbIndexer.Add(c.pvb))
 			backuper.wg.Add(1)
 		}
 
@@ -833,3 +833,185 @@ func TestPVCBackupSummary(t *testing.T) {
 	assert.Empty(t, pbs.Skipped)
 	assert.Len(t, pbs.Backedup, 2)
 }
+
+func TestGetMatchAction_PendingPVC(t *testing.T) {
+	// Create resource policies that skip Pending/Lost PVCs
+	resPolicies := &resourcepolicies.ResourcePolicies{
+		Version: "v1",
+		VolumePolicies: []resourcepolicies.VolumePolicy{
+			{
+				Conditions: map[string]any{
+					"pvcPhase": []string{"Pending", "Lost"},
+				},
+				Action: resourcepolicies.Action{
+					Type: resourcepolicies.Skip,
+				},
+			},
+		},
+	}
+	policies := &resourcepolicies.Policies{}
+	err := policies.BuildPolicy(resPolicies)
+	require.NoError(t, err)
+
+	testCases := []struct {
+		name           string
+		pvc            *corev1api.PersistentVolumeClaim
+		volume         *corev1api.Volume
+		pv             *corev1api.PersistentVolume
+		expectedAction *resourcepolicies.Action
+		expectError    bool
+	}{
+		{
+			name: "Pending PVC with pvcPhase skip policy should return skip action",
+			pvc: builder.ForPersistentVolumeClaim("ns", "pending-pvc").
+				StorageClass("test-sc").
+				Phase(corev1api.ClaimPending).
+				Result(),
+			volume: &corev1api.Volume{
+				Name: "test-volume",
+				VolumeSource: corev1api.VolumeSource{
+					PersistentVolumeClaim: &corev1api.PersistentVolumeClaimVolumeSource{
+						ClaimName: "pending-pvc",
+					},
+				},
+			},
+			pv:             nil,
+			expectedAction: &resourcepolicies.Action{Type: resourcepolicies.Skip},
+			expectError:    false,
+		},
+		{
+			name: "Lost PVC with pvcPhase skip policy should return skip action",
+			pvc: builder.ForPersistentVolumeClaim("ns", "lost-pvc").
+				StorageClass("test-sc").
+				Phase(corev1api.ClaimLost).
+				Result(),
+			volume: &corev1api.Volume{
+				Name: "test-volume",
+				VolumeSource: corev1api.VolumeSource{
+					PersistentVolumeClaim: &corev1api.PersistentVolumeClaimVolumeSource{
+						ClaimName: "lost-pvc",
+					},
+				},
+			},
+			pv:             nil,
+			expectedAction: &resourcepolicies.Action{Type: resourcepolicies.Skip},
+			expectError:    false,
+		},
+		{
+			name: "Bound PVC with matching PV should not match pvcPhase policy",
+			pvc: builder.ForPersistentVolumeClaim("ns", "bound-pvc").
+				StorageClass("test-sc").
+				VolumeName("test-pv").
+				Phase(corev1api.ClaimBound).
+				Result(),
+			volume: &corev1api.Volume{
+				Name: "test-volume",
+				VolumeSource: corev1api.VolumeSource{
+					PersistentVolumeClaim: &corev1api.PersistentVolumeClaimVolumeSource{
+						ClaimName: "bound-pvc",
+					},
+				},
+			},
+			pv:             builder.ForPersistentVolume("test-pv").StorageClass("test-sc").Result(),
+			expectedAction: nil,
+			expectError:    false,
+		},
+		{
+			name: "Pending PVC with no matching policy should return nil action",
+			pvc: builder.ForPersistentVolumeClaim("ns", "pending-pvc-no-match").
+				StorageClass("test-sc").
+				Phase(corev1api.ClaimPending).
+				Result(),
+			volume: &corev1api.Volume{
+				Name: "test-volume",
+				VolumeSource: corev1api.VolumeSource{
+					PersistentVolumeClaim: &corev1api.PersistentVolumeClaimVolumeSource{
+						ClaimName: "pending-pvc-no-match",
+					},
+				},
+			},
+			pv:             nil,
+			expectedAction: &resourcepolicies.Action{Type: resourcepolicies.Skip}, // Will match the pvcPhase policy
+			expectError:    false,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			// Build fake client with PV if present
+			var objs []runtime.Object
+			if tc.pv != nil {
+				objs = append(objs, tc.pv)
+			}
+			fakeClient := velerotest.NewFakeControllerRuntimeClient(t, objs...)
+
+			b := &backupper{
+				crClient: fakeClient,
+			}
+
+			action, err := b.getMatchAction(policies, tc.pvc, tc.volume)
+			if tc.expectError {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+			}
+
+			if tc.expectedAction == nil {
+				assert.Nil(t, action)
+			} else {
+				require.NotNil(t, action)
+				assert.Equal(t, tc.expectedAction.Type, action.Type)
+			}
+		})
+	}
+}
+
+func TestGetMatchAction_PVCWithoutPVLookupError(t *testing.T) {
+	// Test that when a PVC has a VolumeName but the PV doesn't exist,
+	// the function ignores the error and tries to match with PVC only
+	resPolicies := &resourcepolicies.ResourcePolicies{
+		Version: "v1",
+		VolumePolicies: []resourcepolicies.VolumePolicy{
+			{
+				Conditions: map[string]any{
+					"pvcPhase": []string{"Pending"},
+				},
+				Action: resourcepolicies.Action{
+					Type: resourcepolicies.Skip,
+				},
+			},
+		},
+	}
+	policies := &resourcepolicies.Policies{}
+	err := policies.BuildPolicy(resPolicies)
+	require.NoError(t, err)
+
+	// Pending PVC without a matching PV in the cluster
+	pvc := builder.ForPersistentVolumeClaim("ns", "pending-pvc").
+		StorageClass("test-sc").
+		Phase(corev1api.ClaimPending).
+		Result()
+
+	volume := &corev1api.Volume{
+		Name: "test-volume",
+		VolumeSource: corev1api.VolumeSource{
+			PersistentVolumeClaim: &corev1api.PersistentVolumeClaimVolumeSource{
+				ClaimName: "pending-pvc",
+			},
+		},
+	}
+
+	// Empty client - no PV exists
+	fakeClient := velerotest.NewFakeControllerRuntimeClient(t)
+
+	b := &backupper{
+		crClient: fakeClient,
+	}
+
+	// Should succeed even though PV lookup would fail
+	// because the function ignores PV lookup errors and uses PVC-only matching
+	action, err := b.getMatchAction(policies, pvc, volume)
+	require.NoError(t, err)
+	require.NotNil(t, action)
+	assert.Equal(t, resourcepolicies.Skip, action.Type)
+}

pkg/repository/maintenance/maintenance.go

@@ -38,6 +38,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	"github.com/vmware-tanzu/velero/pkg/constant"
 	velerolabel "github.com/vmware-tanzu/velero/pkg/label"
 	velerotypes "github.com/vmware-tanzu/velero/pkg/types"
 	"github.com/vmware-tanzu/velero/pkg/util"
@@ -574,15 +575,32 @@ func buildJob(
 	// Set resource limits and requests
 	cpuRequest := DefaultMaintenanceJobCPURequest
 	memRequest := DefaultMaintenanceJobMemRequest
+	ephemeralStorageRequest := constant.DefaultEphemeralStorageRequest
 	cpuLimit := DefaultMaintenanceJobCPULimit
 	memLimit := DefaultMaintenanceJobMemLimit
+	ephemeralStorageLimit := constant.DefaultEphemeralStorageLimit
 	if config != nil && config.PodResources != nil {
 		cpuRequest = config.PodResources.CPURequest
 		memRequest = config.PodResources.MemoryRequest
 		cpuLimit = config.PodResources.CPULimit
 		memLimit = config.PodResources.MemoryLimit
+		// To make the PodResources ConfigMap without ephemeral storage request/limit backward compatible,
+		// need to avoid set value as empty, because empty string will cause parsing error.
+		if config.PodResources.EphemeralStorageRequest != "" {
+			ephemeralStorageRequest = config.PodResources.EphemeralStorageRequest
+		}
+		if config.PodResources.EphemeralStorageLimit != "" {
+			ephemeralStorageLimit = config.PodResources.EphemeralStorageLimit
+		}
 	}
-	resources, err := kube.ParseResourceRequirements(cpuRequest, memRequest, cpuLimit, memLimit)
+	resources, err := kube.ParseResourceRequirements(
+		cpuRequest,
+		memRequest,
+		ephemeralStorageRequest,
+		cpuLimit,
+		memLimit,
+		ephemeralStorageLimit,
+	)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to parse resource requirements for maintenance job")
 	}
@@ -671,8 +689,7 @@ func buildJob(
 	}
 
 	if config != nil && len(config.LoadAffinities) > 0 {
-		// Maintenance job only takes the first loadAffinity.
-		affinity := kube.ToSystemAffinity([]*kube.LoadAffinity{config.LoadAffinities[0]})
+		affinity := kube.ToSystemAffinity(config.LoadAffinities[0], nil)
 		job.Spec.Template.Spec.Affinity = affinity
 	}
 

pkg/restore/actions/pod_volume_restore_action.go

@@ -163,12 +163,19 @@ func (a *PodVolumeRestoreAction) Execute(input *velero.RestoreItemActionExecuteI
 		memLimit = defaultMemRequestLimit
 	}
 
-	resourceReqs, err := kube.ParseResourceRequirements(cpuRequest, memRequest, cpuLimit, memLimit)
+	resourceReqs, err := kube.ParseCPUAndMemoryResources(
+		cpuRequest,
+		memRequest,
+		cpuLimit,
+		memLimit,
+	)
 	if err != nil {
 		log.Errorf("couldn't parse resource requirements: %s.", err)
-		resourceReqs, _ = kube.ParseResourceRequirements(
-			defaultCPURequestLimit, defaultMemRequestLimit, // requests
-			defaultCPURequestLimit, defaultMemRequestLimit, // limits
+		resourceReqs, _ = kube.ParseCPUAndMemoryResources(
+			defaultCPURequestLimit,
+			defaultMemRequestLimit,
+			defaultCPURequestLimit,
+			defaultMemRequestLimit,
 		)
 	}
 

pkg/restore/actions/pod_volume_restore_action_test.go

@@ -117,9 +117,11 @@ func TestGetImage(t *testing.T) {
 
 // TestPodVolumeRestoreActionExecute tests the pod volume restore item action plugin's Execute method.
 func TestPodVolumeRestoreActionExecute(t *testing.T) {
-	resourceReqs, _ := kube.ParseResourceRequirements(
-		defaultCPURequestLimit, defaultMemRequestLimit, // requests
-		defaultCPURequestLimit, defaultMemRequestLimit, // limits
+	resourceReqs, _ := kube.ParseCPUAndMemoryResources(
+		defaultCPURequestLimit,
+		defaultMemRequestLimit,
+		defaultCPURequestLimit,
+		defaultMemRequestLimit,
 	)
 	id := int64(1000)
 	securityContext := corev1api.SecurityContext{

pkg/uploader/kopia/block_restore.go

@@ -35,6 +35,7 @@ type BlockOutput struct {
 	*restore.FilesystemOutput
 
 	targetFileName string
+	targetFile     *os.File
 }
 
 var _ restore.Output = &BlockOutput{}
@@ -52,7 +53,7 @@ func (o *BlockOutput) WriteFile(ctx context.Context, relativePath string, remote
 	if err != nil {
 		return errors.Wrapf(err, "failed to open file %s", o.targetFileName)
 	}
-	defer targetFile.Close()
+	o.targetFile = targetFile
 
 	buffer := make([]byte, bufferSize)
 
@@ -101,3 +102,23 @@ func (o *BlockOutput) BeginDirectory(ctx context.Context, relativePath string, e
 
 	return nil
 }
+
+func (o *BlockOutput) Flush() error {
+	if o.targetFile != nil {
+		if err := o.targetFile.Sync(); err != nil {
+			return errors.Wrapf(err, "error syncing block dev %v", o.targetFileName)
+		}
+	}
+
+	return nil
+}
+
+func (o *BlockOutput) Terminate() error {
+	if o.targetFile != nil {
+		if err := o.targetFile.Close(); err != nil {
+			return errors.Wrapf(err, "error closing block dev %v", o.targetFileName)
+		}
+	}
+
+	return nil
+}

pkg/uploader/kopia/block_restore_windows.go

@@ -40,3 +40,11 @@ func (o *BlockOutput) WriteFile(ctx context.Context, relativePath string, remote
 func (o *BlockOutput) BeginDirectory(ctx context.Context, relativePath string, e fs.Directory) error {
 	return fmt.Errorf("block mode is not supported for Windows")
 }
+
+func (o *BlockOutput) Flush() error {
+	return flushVolume(o.targetFileName)
+}
+
+func (o *BlockOutput) Terminate() error {
+	return nil
+}

pkg/uploader/kopia/flush_volume_linux.go

@@ -0,0 +1,50 @@
+//go:build linux
+// +build linux
+
+/*
+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"
+
+	"github.com/pkg/errors"
+	"golang.org/x/sys/unix"
+)
+
+func flushVolume(dirPath string) error {
+	dir, err := os.Open(dirPath)
+	if err != nil {
+		return errors.Wrapf(err, "error opening dir %v", dirPath)
+	}
+
+	raw, err := dir.SyscallConn()
+	if err != nil {
+		return errors.Wrapf(err, "error getting handle of dir %v", dirPath)
+	}
+
+	var syncErr error
+	if err := raw.Control(func(fd uintptr) {
+		if e := unix.Syncfs(int(fd)); e != nil {
+			syncErr = e
+		}
+	}); err != nil {
+		return errors.Wrapf(err, "error calling fs sync from %v", dirPath)
+	}
+
+	return errors.Wrapf(syncErr, "error syncing fs from %v", dirPath)
+}

pkg/uploader/kopia/flush_volume_other.go

@@ -0,0 +1,24 @@
+//go:build !linux
+// +build !linux
+
+/*
+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
+
+func flushVolume(_ string) error {
+	return errFlushUnsupported
+}

pkg/uploader/kopia/restore_output.go

@@ -0,0 +1,30 @@
+/*
+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 (
+	"github.com/kopia/kopia/snapshot/restore"
+	"github.com/pkg/errors"
+)
+
+var errFlushUnsupported = errors.New("flush is not supported")
+
+type RestoreOutput interface {
+	restore.Output
+	Flush() error
+	Terminate() error
+}

pkg/uploader/kopia/snapshot.go

@@ -53,6 +53,7 @@ var loadSnapshotFunc = snapshot.LoadSnapshot
 var listSnapshotsFunc = snapshot.ListSnapshots
 var filesystemEntryFunc = snapshotfs.FilesystemEntryFromIDWithPath
 var restoreEntryFunc = restore.Entry
+var flushVolumeFunc = flushVolume
 
 const UploaderConfigMultipartKey = "uploader-multipart"
 const MaxErrorReported = 10
@@ -375,6 +376,18 @@ func findPreviousSnapshotManifest(ctx context.Context, rep repo.Repository, sour
 	return result, nil
 }
 
+type fileSystemRestoreOutput struct {
+	*restore.FilesystemOutput
+}
+
+func (o *fileSystemRestoreOutput) Flush() error {
+	return flushVolumeFunc(o.TargetPath)
+}
+
+func (o *fileSystemRestoreOutput) Terminate() error {
+	return nil
+}
+
 // Restore restore specific sourcePath with given snapshotID and update progress
 func Restore(ctx context.Context, rep repo.RepositoryWriter, progress *Progress, snapshotID, dest string, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string,
 	log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, error) {
@@ -434,13 +447,23 @@ func Restore(ctx context.Context, rep repo.RepositoryWriter, progress *Progress,
 		return 0, 0, errors.Wrap(err, "error to init output")
 	}
 
-	var output restore.Output = fsOutput
+	var output RestoreOutput
 	if volMode == uploader.PersistentVolumeBlock {
 		output = &BlockOutput{
 			FilesystemOutput: fsOutput,
 		}
+	} else {
+		output = &fileSystemRestoreOutput{
+			FilesystemOutput: fsOutput,
+		}
 	}
 
+	defer func() {
+		if err := output.Terminate(); err != nil {
+			log.Warnf("error terminating restore output for %v", path)
+		}
+	}()
+
 	stat, err := restoreEntryFunc(kopiaCtx, rep, output, rootEntry, restore.Options{
 		Parallel:               restoreConcurrency,
 		RestoreDirEntryAtDepth: math.MaxInt32,
@@ -453,5 +476,16 @@ func Restore(ctx context.Context, rep repo.RepositoryWriter, progress *Progress,
 	if err != nil {
 		return 0, 0, errors.Wrapf(err, "Failed to copy snapshot data to the target")
 	}
+
+	if err := output.Flush(); err != nil {
+		if err == errFlushUnsupported {
+			log.Warnf("Skip flushing data for %v under the current OS %v", path, runtime.GOOS)
+		} else {
+			return 0, 0, errors.Wrapf(err, "Failed to flush data to target")
+		}
+	} else {
+		log.Infof("Flush done for volume dir %v", path)
+	}
+
 	return stat.RestoredTotalFileSize, stat.RestoredFileCount, nil
 }

pkg/uploader/kopia/snapshot_test.go

@@ -675,6 +675,7 @@ func TestRestore(t *testing.T) {
 		invalidManifestType bool
 		filesystemEntryFunc func(ctx context.Context, rep repo.Repository, rootID string, consistentAttributes bool) (fs.Entry, error)
 		restoreEntryFunc    func(ctx context.Context, rep repo.Repository, output restore.Output, rootEntry fs.Entry, options restore.Options) (restore.Stats, error)
+		flushVolumeFunc     func(string) error
 		dest                string
 		expectedBytes       int64
 		expectedCount       int32
@@ -757,6 +758,30 @@ func TestRestore(t *testing.T) {
 			volMode:       uploader.PersistentVolumeBlock,
 			dest:          "/tmp",
 		},
+		{
+			name: "Flush is not supported",
+			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
+			},
+			flushVolumeFunc: func(string) error { return errFlushUnsupported },
+			snapshotID:      "snapshot-123",
+			expectedError:   nil,
+		},
+		{
+			name: "Flush fails",
+			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
+			},
+			flushVolumeFunc: func(string) error { return errors.New("fake-flush-error") },
+			snapshotID:      "snapshot-123",
+			expectedError:   errors.New("fake-flush-error"),
+		},
 	}
 
 	em := &manifest.EntryMetadata{
@@ -784,6 +809,10 @@ func TestRestore(t *testing.T) {
 				restoreEntryFunc = tc.restoreEntryFunc
 			}
 
+			if tc.flushVolumeFunc != nil {
+				flushVolumeFunc = tc.flushVolumeFunc
+			}
+
 			repoWriterMock := &repomocks.RepositoryWriter{}
 			repoWriterMock.On("GetManifest", mock.Anything, mock.Anything, mock.Anything).Return(em, nil)
 			repoWriterMock.On("OpenObject", mock.Anything, mock.Anything).Return(em, nil)

pkg/util/collections/includes_excludes.go

@@ -666,10 +666,22 @@ func validateNamespaceName(ns string) []error {
 		return nil
 	}
 
-	// Kubernetes does not allow asterisks in namespaces but Velero uses them as
-	// wildcards. Replace asterisks with an arbitrary letter to pass Kubernetes
-	// validation.
-	tmpNamespace := strings.ReplaceAll(ns, "*", "x")
+	// Validate the namespace name to ensure it is a valid wildcard pattern
+	if err := wildcard.ValidateNamespaceName(ns); err != nil {
+		return []error{err}
+	}
+
+	// Kubernetes does not allow wildcard characters in namespaces but Velero uses them
+	// for glob patterns. Replace wildcard characters with valid characters to pass
+	// Kubernetes validation.
+	tmpNamespace := ns
+
+	// Replace glob wildcard characters with valid alphanumeric characters
+	// Note: Validation of wildcard patterns is handled by the wildcard package.
+	tmpNamespace = strings.ReplaceAll(tmpNamespace, "*", "x") // matches any sequence
+	tmpNamespace = strings.ReplaceAll(tmpNamespace, "?", "x") // matches single character
+	tmpNamespace = strings.ReplaceAll(tmpNamespace, "[", "x") // character class start
+	tmpNamespace = strings.ReplaceAll(tmpNamespace, "]", "x") // character class end
 
 	if errMsgs := validation.ValidateNamespaceName(tmpNamespace, false); errMsgs != nil {
 		for _, msg := range errMsgs {

pkg/util/collections/includes_excludes_test.go

@@ -289,6 +289,54 @@ func TestValidateNamespaceIncludesExcludes(t *testing.T) {
 			excludes: []string{"bar"},
 			wantErr:  true,
 		},
+		{
+			name:     "glob characters in includes should not error",
+			includes: []string{"kube-*", "test-?", "ns-[0-9]"},
+			excludes: []string{},
+			wantErr:  false,
+		},
+		{
+			name:     "glob characters in excludes should not error",
+			includes: []string{"default"},
+			excludes: []string{"test-*", "app-?", "ns-[1-5]"},
+			wantErr:  false,
+		},
+		{
+			name:     "character class in includes should not error",
+			includes: []string{"ns-[abc]", "test-[0-9]"},
+			excludes: []string{},
+			wantErr:  false,
+		},
+		{
+			name:     "mixed glob patterns should not error",
+			includes: []string{"kube-*", "test-?"},
+			excludes: []string{"*-test", "debug-[0-9]"},
+			wantErr:  false,
+		},
+		{
+			name:     "pipe character in includes should error",
+			includes: []string{"namespace|other"},
+			excludes: []string{},
+			wantErr:  true,
+		},
+		{
+			name:     "parentheses in includes should error",
+			includes: []string{"namespace(prod)", "test-(dev)"},
+			excludes: []string{},
+			wantErr:  true,
+		},
+		{
+			name:     "exclamation mark in includes should error",
+			includes: []string{"!namespace", "test!"},
+			excludes: []string{},
+			wantErr:  true,
+		},
+		{
+			name:     "unsupported characters in excludes should error",
+			includes: []string{"default"},
+			excludes: []string{"test|prod", "app(staging)"},
+			wantErr:  true,
+		},
 	}
 
 	for _, tc := range tests {
@@ -1082,16 +1130,6 @@ func TestExpandIncludesExcludes(t *testing.T) {
 			expectedWildcardExpanded: true,
 			expectError:              false,
 		},
-		{
-			name:                     "brace wildcard pattern",
-			includes:                 []string{"app-{prod,dev}"},
-			excludes:                 []string{},
-			activeNamespaces:         []string{"app-prod", "app-dev", "app-test", "default"},
-			expectedIncludes:         []string{"app-prod", "app-dev"},
-			expectedExcludes:         []string{},
-			expectedWildcardExpanded: true,
-			expectError:              false,
-		},
 		{
 			name:                     "empty activeNamespaces with wildcards",
 			includes:                 []string{"kube-*"},
@@ -1233,13 +1271,6 @@ func TestResolveNamespaceList(t *testing.T) {
 			expectedNamespaces: []string{"kube-system", "kube-public"},
 			preExpandWildcards: true,
 		},
-		{
-			name:               "complex wildcard pattern",
-			includes:           []string{"app-{prod,dev}", "kube-*"},
-			excludes:           []string{"*-test"},
-			activeNamespaces:   []string{"app-prod", "app-dev", "app-test", "kube-system", "kube-test", "default"},
-			expectedNamespaces: []string{"app-prod", "app-dev", "kube-system"},
-		},
 		{
 			name:               "question mark wildcard pattern",
 			includes:           []string{"ns-?"},

pkg/util/kube/node.go

@@ -17,7 +17,6 @@ package kube
 
 import (
 	"context"
-	"fmt"
 
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
@@ -34,6 +33,11 @@ const (
 	NodeOSLabel   = "kubernetes.io/os"
 )
 
+var realNodeOSMap = map[string]string{
+	"linux":   NodeOSLinux,
+	"windows": NodeOSWindows,
+}
+
 func IsLinuxNode(ctx context.Context, nodeName string, client client.Client) error {
 	node := &corev1api.Node{}
 	if err := client.Get(ctx, types.NamespacedName{Name: nodeName}, node); err != nil {
@@ -41,12 +45,11 @@ func IsLinuxNode(ctx context.Context, nodeName string, client client.Client) err
 	}
 
 	os, found := node.Labels[NodeOSLabel]
-
 	if !found {
 		return errors.Errorf("no os type label for node %s", nodeName)
 	}
 
-	if os != NodeOSLinux {
+	if getRealOS(os) != NodeOSLinux {
 		return errors.Errorf("os type %s for node %s is not linux", os, nodeName)
 	}
 
@@ -72,7 +75,7 @@ func withOSNode(ctx context.Context, client client.Client, osType string, log lo
 	for _, node := range nodeList.Items {
 		os, found := node.Labels[NodeOSLabel]
 
-		if os == osType {
+		if getRealOS(os) == osType {
 			return true
 		}
 
@@ -98,7 +101,7 @@ func GetNodeOS(ctx context.Context, nodeName string, nodeClient corev1client.Cor
 		return "", nil
 	}
 
-	return node.Labels[NodeOSLabel], nil
+	return getRealOS(node.Labels[NodeOSLabel]), nil
 }
 
 func HasNodeWithOS(ctx context.Context, os string, nodeClient corev1client.CoreV1Interface) error {
@@ -106,14 +109,29 @@ func HasNodeWithOS(ctx context.Context, os string, nodeClient corev1client.CoreV
 		return errors.New("invalid node OS")
 	}
 
-	nodes, err := nodeClient.Nodes().List(ctx, metav1.ListOptions{LabelSelector: fmt.Sprintf("%s=%s", NodeOSLabel, os)})
+	nodes, err := nodeClient.Nodes().List(ctx, metav1.ListOptions{})
 	if err != nil {
 		return errors.Wrapf(err, "error listing nodes with OS %s", os)
 	}
 
-	if len(nodes.Items) == 0 {
-		return errors.Errorf("node with OS %s doesn't exist", os)
+	for _, node := range nodes.Items {
+		osLabel, found := node.Labels[NodeOSLabel]
+		if !found {
+			continue
+		}
+
+		if getRealOS(osLabel) == os {
+			return nil
+		}
 	}
 
-	return nil
+	return errors.Errorf("node with OS %s doesn't exist", os)
+}
+
+func getRealOS(osLabel string) string {
+	if os, found := realNodeOSMap[osLabel]; !found {
+		return NodeOSLinux
+	} else {
+		return os
+	}
 }

pkg/util/kube/pod.go

@@ -40,10 +40,12 @@ type LoadAffinity struct {
 }
 
 type PodResources struct {
-	CPURequest    string `json:"cpuRequest,omitempty"`
-	MemoryRequest string `json:"memoryRequest,omitempty"`
-	CPULimit      string `json:"cpuLimit,omitempty"`
-	MemoryLimit   string `json:"memoryLimit,omitempty"`
+	CPURequest              string `json:"cpuRequest,omitempty"`
+	CPULimit                string `json:"cpuLimit,omitempty"`
+	MemoryRequest           string `json:"memoryRequest,omitempty"`
+	MemoryLimit             string `json:"memoryLimit,omitempty"`
+	EphemeralStorageRequest string `json:"ephemeralStorageRequest,omitempty"`
+	EphemeralStorageLimit   string `json:"ephemeralStorageLimit,omitempty"`
 }
 
 // IsPodRunning does a well-rounded check to make sure the specified pod is running stably.
@@ -230,14 +232,9 @@ func CollectPodLogs(ctx context.Context, podGetter corev1client.CoreV1Interface,
 	return nil
 }
 
-func ToSystemAffinity(loadAffinities []*LoadAffinity) *corev1api.Affinity {
-	if len(loadAffinities) == 0 {
-		return nil
-	}
-	nodeSelectorTermList := make([]corev1api.NodeSelectorTerm, 0)
-
-	for _, loadAffinity := range loadAffinities {
-		requirements := []corev1api.NodeSelectorRequirement{}
+func ToSystemAffinity(loadAffinity *LoadAffinity, volumeTopology *corev1api.NodeSelector) *corev1api.Affinity {
+	requirements := []corev1api.NodeSelectorRequirement{}
+	if loadAffinity != nil {
 		for k, v := range loadAffinity.NodeSelector.MatchLabels {
 			requirements = append(requirements, corev1api.NodeSelectorRequirement{
 				Key:      k,
@@ -253,25 +250,25 @@ func ToSystemAffinity(loadAffinities []*LoadAffinity) *corev1api.Affinity {
 				Operator: corev1api.NodeSelectorOperator(exp.Operator),
 			})
 		}
-
-		nodeSelectorTermList = append(
-			nodeSelectorTermList,
-			corev1api.NodeSelectorTerm{
-				MatchExpressions: requirements,
-			},
-		)
 	}
 
-	if len(nodeSelectorTermList) > 0 {
-		result := new(corev1api.Affinity)
-		result.NodeAffinity = new(corev1api.NodeAffinity)
-		result.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution = new(corev1api.NodeSelector)
-		result.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms = nodeSelectorTermList
+	result := new(corev1api.Affinity)
+	result.NodeAffinity = new(corev1api.NodeAffinity)
+	result.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution = new(corev1api.NodeSelector)
 
-		return result
+	if volumeTopology != nil {
+		result.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms = append(result.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms, volumeTopology.NodeSelectorTerms...)
+	} else if len(requirements) > 0 {
+		result.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms = make([]corev1api.NodeSelectorTerm, 1)
+	} else {
+		return nil
 	}
 
-	return nil
+	for i := range result.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms {
+		result.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms[i].MatchExpressions = append(result.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms[i].MatchExpressions, requirements...)
+	}
+
+	return result
 }
 
 func DiagnosePod(pod *corev1api.Pod, events *corev1api.EventList) string {

pkg/util/kube/pod_test.go

@@ -747,24 +747,23 @@ func TestCollectPodLogs(t *testing.T) {
 func TestToSystemAffinity(t *testing.T) {
 	tests := []struct {
 		name           string
-		loadAffinities []*LoadAffinity
+		loadAffinity   *LoadAffinity
+		volumeTopology *corev1api.NodeSelector
 		expected       *corev1api.Affinity
 	}{
 		{
 			name: "loadAffinity is nil",
 		},
 		{
-			name:           "loadAffinity is empty",
-			loadAffinities: []*LoadAffinity{},
+			name:         "loadAffinity is empty",
+			loadAffinity: &LoadAffinity{},
 		},
 		{
 			name: "with match label",
-			loadAffinities: []*LoadAffinity{
-				{
-					NodeSelector: metav1.LabelSelector{
-						MatchLabels: map[string]string{
-							"key-1": "value-1",
-						},
+			loadAffinity: &LoadAffinity{
+				NodeSelector: metav1.LabelSelector{
+					MatchLabels: map[string]string{
+						"key-1": "value-1",
 					},
 				},
 			},
@@ -788,23 +787,21 @@ func TestToSystemAffinity(t *testing.T) {
 		},
 		{
 			name: "with match expression",
-			loadAffinities: []*LoadAffinity{
-				{
-					NodeSelector: metav1.LabelSelector{
-						MatchLabels: map[string]string{
-							"key-2": "value-2",
+			loadAffinity: &LoadAffinity{
+				NodeSelector: metav1.LabelSelector{
+					MatchLabels: map[string]string{
+						"key-2": "value-2",
+					},
+					MatchExpressions: []metav1.LabelSelectorRequirement{
+						{
+							Key:      "key-3",
+							Values:   []string{"value-3-1", "value-3-2"},
+							Operator: metav1.LabelSelectorOpNotIn,
 						},
-						MatchExpressions: []metav1.LabelSelectorRequirement{
-							{
-								Key:      "key-3",
-								Values:   []string{"value-3-1", "value-3-2"},
-								Operator: metav1.LabelSelectorOpNotIn,
-							},
-							{
-								Key:      "key-4",
-								Values:   []string{"value-4-1", "value-4-2", "value-4-3"},
-								Operator: metav1.LabelSelectorOpDoesNotExist,
-							},
+						{
+							Key:      "key-4",
+							Values:   []string{"value-4-1", "value-4-2", "value-4-3"},
+							Operator: metav1.LabelSelectorOpDoesNotExist,
 						},
 					},
 				},
@@ -838,19 +835,49 @@ func TestToSystemAffinity(t *testing.T) {
 			},
 		},
 		{
-			name: "multiple load affinities",
-			loadAffinities: []*LoadAffinity{
-				{
-					NodeSelector: metav1.LabelSelector{
-						MatchLabels: map[string]string{
-							"key-1": "value-1",
+			name: "with olume topology",
+			volumeTopology: &corev1api.NodeSelector{
+				NodeSelectorTerms: []corev1api.NodeSelectorTerm{
+					{
+						MatchExpressions: []corev1api.NodeSelectorRequirement{
+							{
+								Key:      "key-5",
+								Values:   []string{"value-5-1", "value-5-2", "value-5-3"},
+								Operator: corev1api.NodeSelectorOpGt,
+							},
+							{
+								Key:      "key-6",
+								Values:   []string{"value-5-1", "value-5-2", "value-5-3"},
+								Operator: corev1api.NodeSelectorOpGt,
+							},
 						},
 					},
-				},
-				{
-					NodeSelector: metav1.LabelSelector{
-						MatchLabels: map[string]string{
-							"key-2": "value-2",
+					{
+						MatchExpressions: []corev1api.NodeSelectorRequirement{
+							{
+								Key:      "key-7",
+								Values:   []string{"value-7-1", "value-7-2", "value-7-3"},
+								Operator: corev1api.NodeSelectorOpGt,
+							},
+							{
+								Key:      "key-8",
+								Values:   []string{"value-8-1", "value-8-2", "value-8-3"},
+								Operator: corev1api.NodeSelectorOpGt,
+							},
+						},
+					},
+					{
+						MatchFields: []corev1api.NodeSelectorRequirement{
+							{
+								Key:      "key-9",
+								Values:   []string{"value-9-1", "value-9-2", "value-9-3"},
+								Operator: corev1api.NodeSelectorOpGt,
+							},
+							{
+								Key:      "key-a",
+								Values:   []string{"value-a-1", "value-a-2", "value-a-3"},
+								Operator: corev1api.NodeSelectorOpGt,
+							},
 						},
 					},
 				},
@@ -862,10 +889,177 @@ func TestToSystemAffinity(t *testing.T) {
 							{
 								MatchExpressions: []corev1api.NodeSelectorRequirement{
 									{
-										Key:      "key-1",
-										Values:   []string{"value-1"},
+										Key:      "key-5",
+										Values:   []string{"value-5-1", "value-5-2", "value-5-3"},
+										Operator: corev1api.NodeSelectorOpGt,
+									},
+									{
+										Key:      "key-6",
+										Values:   []string{"value-5-1", "value-5-2", "value-5-3"},
+										Operator: corev1api.NodeSelectorOpGt,
+									},
+								},
+							},
+							{
+								MatchExpressions: []corev1api.NodeSelectorRequirement{
+									{
+										Key:      "key-7",
+										Values:   []string{"value-7-1", "value-7-2", "value-7-3"},
+										Operator: corev1api.NodeSelectorOpGt,
+									},
+									{
+										Key:      "key-8",
+										Values:   []string{"value-8-1", "value-8-2", "value-8-3"},
+										Operator: corev1api.NodeSelectorOpGt,
+									},
+								},
+							},
+							{
+								MatchFields: []corev1api.NodeSelectorRequirement{
+									{
+										Key:      "key-9",
+										Values:   []string{"value-9-1", "value-9-2", "value-9-3"},
+										Operator: corev1api.NodeSelectorOpGt,
+									},
+									{
+										Key:      "key-a",
+										Values:   []string{"value-a-1", "value-a-2", "value-a-3"},
+										Operator: corev1api.NodeSelectorOpGt,
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "with match expression and volume topology",
+			loadAffinity: &LoadAffinity{
+				NodeSelector: metav1.LabelSelector{
+					MatchLabels: map[string]string{
+						"key-2": "value-2",
+					},
+					MatchExpressions: []metav1.LabelSelectorRequirement{
+						{
+							Key:      "key-3",
+							Values:   []string{"value-3-1", "value-3-2"},
+							Operator: metav1.LabelSelectorOpNotIn,
+						},
+						{
+							Key:      "key-4",
+							Values:   []string{"value-4-1", "value-4-2", "value-4-3"},
+							Operator: metav1.LabelSelectorOpDoesNotExist,
+						},
+					},
+				},
+			},
+			volumeTopology: &corev1api.NodeSelector{
+				NodeSelectorTerms: []corev1api.NodeSelectorTerm{
+					{
+						MatchExpressions: []corev1api.NodeSelectorRequirement{
+							{
+								Key:      "key-5",
+								Values:   []string{"value-5-1", "value-5-2", "value-5-3"},
+								Operator: corev1api.NodeSelectorOpGt,
+							},
+							{
+								Key:      "key-6",
+								Values:   []string{"value-5-1", "value-5-2", "value-5-3"},
+								Operator: corev1api.NodeSelectorOpGt,
+							},
+						},
+					},
+					{
+						MatchExpressions: []corev1api.NodeSelectorRequirement{
+							{
+								Key:      "key-7",
+								Values:   []string{"value-7-1", "value-7-2", "value-7-3"},
+								Operator: corev1api.NodeSelectorOpGt,
+							},
+							{
+								Key:      "key-8",
+								Values:   []string{"value-8-1", "value-8-2", "value-8-3"},
+								Operator: corev1api.NodeSelectorOpGt,
+							},
+						},
+					},
+					{
+						MatchFields: []corev1api.NodeSelectorRequirement{
+							{
+								Key:      "key-9",
+								Values:   []string{"value-9-1", "value-9-2", "value-9-3"},
+								Operator: corev1api.NodeSelectorOpGt,
+							},
+							{
+								Key:      "key-a",
+								Values:   []string{"value-a-1", "value-a-2", "value-a-3"},
+								Operator: corev1api.NodeSelectorOpGt,
+							},
+						},
+					},
+				},
+			},
+			expected: &corev1api.Affinity{
+				NodeAffinity: &corev1api.NodeAffinity{
+					RequiredDuringSchedulingIgnoredDuringExecution: &corev1api.NodeSelector{
+						NodeSelectorTerms: []corev1api.NodeSelectorTerm{
+							{
+								MatchExpressions: []corev1api.NodeSelectorRequirement{
+									{
+										Key:      "key-5",
+										Values:   []string{"value-5-1", "value-5-2", "value-5-3"},
+										Operator: corev1api.NodeSelectorOpGt,
+									},
+									{
+										Key:      "key-6",
+										Values:   []string{"value-5-1", "value-5-2", "value-5-3"},
+										Operator: corev1api.NodeSelectorOpGt,
+									},
+									{
+										Key:      "key-2",
+										Values:   []string{"value-2"},
 										Operator: corev1api.NodeSelectorOpIn,
 									},
+									{
+										Key:      "key-3",
+										Values:   []string{"value-3-1", "value-3-2"},
+										Operator: corev1api.NodeSelectorOpNotIn,
+									},
+									{
+										Key:      "key-4",
+										Values:   []string{"value-4-1", "value-4-2", "value-4-3"},
+										Operator: corev1api.NodeSelectorOpDoesNotExist,
+									},
+								},
+							},
+							{
+								MatchExpressions: []corev1api.NodeSelectorRequirement{
+									{
+										Key:      "key-7",
+										Values:   []string{"value-7-1", "value-7-2", "value-7-3"},
+										Operator: corev1api.NodeSelectorOpGt,
+									},
+									{
+										Key:      "key-8",
+										Values:   []string{"value-8-1", "value-8-2", "value-8-3"},
+										Operator: corev1api.NodeSelectorOpGt,
+									},
+									{
+										Key:      "key-2",
+										Values:   []string{"value-2"},
+										Operator: corev1api.NodeSelectorOpIn,
+									},
+									{
+										Key:      "key-3",
+										Values:   []string{"value-3-1", "value-3-2"},
+										Operator: corev1api.NodeSelectorOpNotIn,
+									},
+									{
+										Key:      "key-4",
+										Values:   []string{"value-4-1", "value-4-2", "value-4-3"},
+										Operator: corev1api.NodeSelectorOpDoesNotExist,
+									},
 								},
 							},
 							{
@@ -875,6 +1069,28 @@ func TestToSystemAffinity(t *testing.T) {
 										Values:   []string{"value-2"},
 										Operator: corev1api.NodeSelectorOpIn,
 									},
+									{
+										Key:      "key-3",
+										Values:   []string{"value-3-1", "value-3-2"},
+										Operator: corev1api.NodeSelectorOpNotIn,
+									},
+									{
+										Key:      "key-4",
+										Values:   []string{"value-4-1", "value-4-2", "value-4-3"},
+										Operator: corev1api.NodeSelectorOpDoesNotExist,
+									},
+								},
+								MatchFields: []corev1api.NodeSelectorRequirement{
+									{
+										Key:      "key-9",
+										Values:   []string{"value-9-1", "value-9-2", "value-9-3"},
+										Operator: corev1api.NodeSelectorOpGt,
+									},
+									{
+										Key:      "key-a",
+										Values:   []string{"value-a-1", "value-a-2", "value-a-3"},
+										Operator: corev1api.NodeSelectorOpGt,
+									},
 								},
 							},
 						},
@@ -886,7 +1102,7 @@ func TestToSystemAffinity(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			affinity := ToSystemAffinity(test.loadAffinities)
+			affinity := ToSystemAffinity(test.loadAffinity, test.volumeTopology)
 			assert.True(t, reflect.DeepEqual(affinity, test.expected))
 		})
 	}

pkg/util/kube/pvc_pv.go

@@ -417,19 +417,19 @@ func MakePodPVCAttachment(volumeName string, volumeMode *corev1api.PersistentVol
 	return volumeMounts, volumeDevices, volumePath
 }
 
+// GetPVForPVC returns the PersistentVolume backing a PVC
+// returns PV, error.
+// PV will be nil on error
 func GetPVForPVC(
 	pvc *corev1api.PersistentVolumeClaim,
 	crClient crclient.Client,
 ) (*corev1api.PersistentVolume, error) {
 	if pvc.Spec.VolumeName == "" {
-		return nil, errors.Errorf("PVC %s/%s has no volume backing this claim",
-			pvc.Namespace, pvc.Name)
+		return nil, errors.Errorf("PVC %s/%s has no volume backing this claim", pvc.Namespace, pvc.Name)
 	}
 	if pvc.Status.Phase != corev1api.ClaimBound {
-		// TODO: confirm if this PVC should be snapshotted if it has no PV bound
-		return nil,
-			errors.Errorf("PVC %s/%s is in phase %v and is not bound to a volume",
-				pvc.Namespace, pvc.Name, pvc.Status.Phase)
+		return nil, errors.Errorf("PVC %s/%s is in phase %v and is not bound to a volume",
+			pvc.Namespace, pvc.Name, pvc.Status.Phase)
 	}
 
 	pv := &corev1api.PersistentVolume{}
@@ -580,3 +580,29 @@ func GetPVAttachedNodes(ctx context.Context, pv string, storageClient storagev1.
 
 	return nodes, nil
 }
+
+func GetVolumeTopology(ctx context.Context, volumeClient corev1client.CoreV1Interface, storageClient storagev1.StorageV1Interface, pvName string, scName string) (*corev1api.NodeSelector, error) {
+	if pvName == "" || scName == "" {
+		return nil, errors.Errorf("invalid parameter, pv %s, sc %s", pvName, scName)
+	}
+
+	sc, err := storageClient.StorageClasses().Get(ctx, scName, metav1.GetOptions{})
+	if err != nil {
+		return nil, errors.Wrapf(err, "error getting storage class %s", scName)
+	}
+
+	if sc.VolumeBindingMode == nil || *sc.VolumeBindingMode != storagev1api.VolumeBindingWaitForFirstConsumer {
+		return nil, nil
+	}
+
+	pv, err := volumeClient.PersistentVolumes().Get(ctx, pvName, metav1.GetOptions{})
+	if err != nil {
+		return nil, errors.Wrapf(err, "error getting PV %s", pvName)
+	}
+
+	if pv.Spec.NodeAffinity == nil {
+		return nil, nil
+	}
+
+	return pv.Spec.NodeAffinity.Required, nil
+}

pkg/util/kube/pvc_pv_test.go

@@ -1909,3 +1909,143 @@ func TestGetPVCAttachingNodeOS(t *testing.T) {
 		})
 	}
 }
+
+func TestGetVolumeTopology(t *testing.T) {
+	pvWithoutNodeAffinity := &corev1api.PersistentVolume{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "fake-pv",
+		},
+	}
+
+	pvWithNodeAffinity := &corev1api.PersistentVolume{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "fake-pv",
+		},
+		Spec: corev1api.PersistentVolumeSpec{
+			NodeAffinity: &corev1api.VolumeNodeAffinity{
+				Required: &corev1api.NodeSelector{
+					NodeSelectorTerms: []corev1api.NodeSelectorTerm{
+						{
+							MatchExpressions: []corev1api.NodeSelectorRequirement{
+								{
+									Key: "fake-key",
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+
+	scObjWithoutVolumeBind := &storagev1api.StorageClass{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "fake-storage-class",
+		},
+	}
+
+	volumeBindImmediate := storagev1api.VolumeBindingImmediate
+	scObjWithImeediateBind := &storagev1api.StorageClass{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "fake-storage-class",
+		},
+		VolumeBindingMode: &volumeBindImmediate,
+	}
+
+	volumeBindWffc := storagev1api.VolumeBindingWaitForFirstConsumer
+	scObjWithWffcBind := &storagev1api.StorageClass{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "fake-storage-class",
+		},
+		VolumeBindingMode: &volumeBindWffc,
+	}
+
+	tests := []struct {
+		name          string
+		pvName        string
+		scName        string
+		kubeClientObj []runtime.Object
+		expectedErr   string
+		expected      *corev1api.NodeSelector
+	}{
+		{
+			name:        "invalid pvName",
+			scName:      "fake-storage-class",
+			expectedErr: "invalid parameter, pv , sc fake-storage-class",
+		},
+		{
+			name:        "invalid scName",
+			pvName:      "fake-pv",
+			expectedErr: "invalid parameter, pv fake-pv, sc ",
+		},
+		{
+			name:        "no sc",
+			pvName:      "fake-pv",
+			scName:      "fake-storage-class",
+			expectedErr: "error getting storage class fake-storage-class: storageclasses.storage.k8s.io \"fake-storage-class\" not found",
+		},
+		{
+			name:          "sc without binding mode",
+			pvName:        "fake-pv",
+			scName:        "fake-storage-class",
+			kubeClientObj: []runtime.Object{scObjWithoutVolumeBind},
+		},
+		{
+			name:          "sc without immediate binding mode",
+			pvName:        "fake-pv",
+			scName:        "fake-storage-class",
+			kubeClientObj: []runtime.Object{scObjWithImeediateBind},
+		},
+		{
+			name:          "get pv fail",
+			pvName:        "fake-pv",
+			scName:        "fake-storage-class",
+			kubeClientObj: []runtime.Object{scObjWithWffcBind},
+			expectedErr:   "error getting PV fake-pv: persistentvolumes \"fake-pv\" not found",
+		},
+		{
+			name:   "pv with no affinity",
+			pvName: "fake-pv",
+			scName: "fake-storage-class",
+			kubeClientObj: []runtime.Object{
+				scObjWithWffcBind,
+				pvWithoutNodeAffinity,
+			},
+		},
+		{
+			name:   "pv with affinity",
+			pvName: "fake-pv",
+			scName: "fake-storage-class",
+			kubeClientObj: []runtime.Object{
+				scObjWithWffcBind,
+				pvWithNodeAffinity,
+			},
+			expected: &corev1api.NodeSelector{
+				NodeSelectorTerms: []corev1api.NodeSelectorTerm{
+					{
+						MatchExpressions: []corev1api.NodeSelectorRequirement{
+							{
+								Key: "fake-key",
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			fakeKubeClient := fake.NewSimpleClientset(test.kubeClientObj...)
+
+			var kubeClient kubernetes.Interface = fakeKubeClient
+
+			affinity, err := GetVolumeTopology(t.Context(), kubeClient.CoreV1(), kubeClient.StorageV1(), test.pvName, test.scName)
+
+			if test.expectedErr != "" {
+				assert.EqualError(t, err, test.expectedErr)
+			} else {
+				assert.Equal(t, test.expected, affinity)
+			}
+		})
+	}
+}

pkg/util/kube/resource_requirements.go

@@ -20,12 +20,34 @@ import (
 	"github.com/pkg/errors"
 	corev1api "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
+
+	"github.com/vmware-tanzu/velero/pkg/constant"
 )
 
-// ParseResourceRequirements takes a set of CPU and memory requests and limit string
+// ParseCPUAndMemoryResources is a helper function that parses CPU and memory requests and limits,
+// using default values for ephemeral storage.
+func ParseCPUAndMemoryResources(cpuRequest, memRequest, cpuLimit, memLimit string) (corev1api.ResourceRequirements, error) {
+	return ParseResourceRequirements(
+		cpuRequest,
+		memRequest,
+		constant.DefaultEphemeralStorageRequest,
+		cpuLimit,
+		memLimit,
+		constant.DefaultEphemeralStorageLimit,
+	)
+}
+
+// ParseResourceRequirements takes a set of CPU, memory, ephemeral storage requests and limit string
 // values and returns a ResourceRequirements struct to be used in a Container.
 // An error is returned if we cannot parse the request/limit.
-func ParseResourceRequirements(cpuRequest, memRequest, cpuLimit, memLimit string) (corev1api.ResourceRequirements, error) {
+func ParseResourceRequirements(
+	cpuRequest,
+	memRequest,
+	ephemeralStorageRequest,
+	cpuLimit,
+	memLimit,
+	ephemeralStorageLimit string,
+) (corev1api.ResourceRequirements, error) {
 	resources := corev1api.ResourceRequirements{
 		Requests: corev1api.ResourceList{},
 		Limits:   corev1api.ResourceList{},
@@ -41,6 +63,11 @@ func ParseResourceRequirements(cpuRequest, memRequest, cpuLimit, memLimit string
 		return resources, errors.Wrapf(err, `couldn't parse memory request "%s"`, memRequest)
 	}
 
+	parsedEphemeralStorageRequest, err := resource.ParseQuantity(ephemeralStorageRequest)
+	if err != nil {
+		return resources, errors.Wrapf(err, `couldn't parse ephemeral storage request "%s"`, ephemeralStorageRequest)
+	}
+
 	parsedCPULimit, err := resource.ParseQuantity(cpuLimit)
 	if err != nil {
 		return resources, errors.Wrapf(err, `couldn't parse CPU limit "%s"`, cpuLimit)
@@ -51,6 +78,11 @@ func ParseResourceRequirements(cpuRequest, memRequest, cpuLimit, memLimit string
 		return resources, errors.Wrapf(err, `couldn't parse memory limit "%s"`, memLimit)
 	}
 
+	parsedEphemeralStorageLimit, err := resource.ParseQuantity(ephemeralStorageLimit)
+	if err != nil {
+		return resources, errors.Wrapf(err, `couldn't parse ephemeral storage limit "%s"`, ephemeralStorageLimit)
+	}
+
 	// A quantity of 0 is treated as unbounded
 	unbounded := resource.MustParse("0")
 
@@ -62,6 +94,10 @@ func ParseResourceRequirements(cpuRequest, memRequest, cpuLimit, memLimit string
 		return resources, errors.WithStack(errors.Errorf(`Memory request "%s" must be less than or equal to Memory limit "%s"`, memRequest, memLimit))
 	}
 
+	if parsedEphemeralStorageLimit != unbounded && parsedEphemeralStorageRequest.Cmp(parsedEphemeralStorageLimit) > 0 {
+		return resources, errors.WithStack(errors.Errorf(`Ephemeral storage request "%s" must be less than or equal to Ephemeral storage limit "%s"`, ephemeralStorageRequest, ephemeralStorageLimit))
+	}
+
 	// Only set resources if they are not unbounded
 	if parsedCPURequest != unbounded {
 		resources.Requests[corev1api.ResourceCPU] = parsedCPURequest
@@ -69,12 +105,18 @@ func ParseResourceRequirements(cpuRequest, memRequest, cpuLimit, memLimit string
 	if parsedMemRequest != unbounded {
 		resources.Requests[corev1api.ResourceMemory] = parsedMemRequest
 	}
+	if parsedEphemeralStorageRequest != unbounded {
+		resources.Requests[corev1api.ResourceEphemeralStorage] = parsedEphemeralStorageRequest
+	}
 	if parsedCPULimit != unbounded {
 		resources.Limits[corev1api.ResourceCPU] = parsedCPULimit
 	}
 	if parsedMemLimit != unbounded {
 		resources.Limits[corev1api.ResourceMemory] = parsedMemLimit
 	}
+	if parsedEphemeralStorageLimit != unbounded {
+		resources.Limits[corev1api.ResourceEphemeralStorage] = parsedEphemeralStorageLimit
+	}
 
 	return resources, nil
 }

pkg/util/kube/resource_requirements_test.go

@@ -27,10 +27,12 @@ import (
 
 func TestParseResourceRequirements(t *testing.T) {
 	type args struct {
-		cpuRequest string
-		memRequest string
-		cpuLimit   string
-		memLimit   string
+		cpuRequest              string
+		memRequest              string
+		ephemeralStorageRequest string
+		cpuLimit                string
+		memLimit                string
+		ephemeralStorageLimit   string
 	}
 	tests := []struct {
 		name     string
@@ -38,43 +40,61 @@ func TestParseResourceRequirements(t *testing.T) {
 		wantErr  bool
 		expected *corev1api.ResourceRequirements
 	}{
-		{"unbounded quantities", args{"0", "0", "0", "0"}, false, &corev1api.ResourceRequirements{
+		{"unbounded quantities", args{"0", "0", "0", "0", "0", "0"}, false, &corev1api.ResourceRequirements{
 			Requests: corev1api.ResourceList{},
 			Limits:   corev1api.ResourceList{},
 		}},
-		{"valid quantities", args{"100m", "128Mi", "200m", "256Mi"}, false, nil},
-		{"CPU request with unbounded limit", args{"100m", "128Mi", "0", "256Mi"}, false, &corev1api.ResourceRequirements{
+		{"valid quantities", args{"100m", "128Mi", "5Gi", "200m", "256Mi", "10Gi"}, false, nil},
+		{"CPU request with unbounded limit", args{"100m", "128Mi", "5Gi", "0", "256Mi", "10Gi"}, false, &corev1api.ResourceRequirements{
 			Requests: corev1api.ResourceList{
-				corev1api.ResourceCPU:    resource.MustParse("100m"),
-				corev1api.ResourceMemory: resource.MustParse("128Mi"),
+				corev1api.ResourceCPU:              resource.MustParse("100m"),
+				corev1api.ResourceMemory:           resource.MustParse("128Mi"),
+				corev1api.ResourceEphemeralStorage: resource.MustParse("5Gi"),
 			},
 			Limits: corev1api.ResourceList{
+				corev1api.ResourceMemory:           resource.MustParse("256Mi"),
+				corev1api.ResourceEphemeralStorage: resource.MustParse("10Gi"),
+			},
+		}},
+		{"Mem request with unbounded limit", args{"100m", "128Mi", "5Gi", "200m", "0", "10Gi"}, false, &corev1api.ResourceRequirements{
+			Requests: corev1api.ResourceList{
+				corev1api.ResourceCPU:              resource.MustParse("100m"),
+				corev1api.ResourceMemory:           resource.MustParse("128Mi"),
+				corev1api.ResourceEphemeralStorage: resource.MustParse("5Gi"),
+			},
+			Limits: corev1api.ResourceList{
+				corev1api.ResourceCPU:              resource.MustParse("200m"),
+				corev1api.ResourceEphemeralStorage: resource.MustParse("10Gi"),
+			},
+		}},
+		{"Ephemeral storage request with unbounded limit", args{"100m", "128Mi", "5Gi", "200m", "256Mi", "0"}, false, &corev1api.ResourceRequirements{
+			Requests: corev1api.ResourceList{
+				corev1api.ResourceCPU:              resource.MustParse("100m"),
+				corev1api.ResourceMemory:           resource.MustParse("128Mi"),
+				corev1api.ResourceEphemeralStorage: resource.MustParse("5Gi"),
+			},
+			Limits: corev1api.ResourceList{
+				corev1api.ResourceCPU:    resource.MustParse("200m"),
 				corev1api.ResourceMemory: resource.MustParse("256Mi"),
 			},
 		}},
-		{"Mem request with unbounded limit", args{"100m", "128Mi", "200m", "0"}, false, &corev1api.ResourceRequirements{
+
+		{"CPU/Mem/EphemeralStorage requests with unbounded limits", args{"100m", "128Mi", "5Gi", "0", "0", "0"}, false, &corev1api.ResourceRequirements{
 			Requests: corev1api.ResourceList{
-				corev1api.ResourceCPU:    resource.MustParse("100m"),
-				corev1api.ResourceMemory: resource.MustParse("128Mi"),
-			},
-			Limits: corev1api.ResourceList{
-				corev1api.ResourceCPU: resource.MustParse("200m"),
-			},
-		}},
-		{"CPU/Mem requests with unbounded limits", args{"100m", "128Mi", "0", "0"}, false, &corev1api.ResourceRequirements{
-			Requests: corev1api.ResourceList{
-				corev1api.ResourceCPU:    resource.MustParse("100m"),
-				corev1api.ResourceMemory: resource.MustParse("128Mi"),
+				corev1api.ResourceCPU:              resource.MustParse("100m"),
+				corev1api.ResourceMemory:           resource.MustParse("128Mi"),
+				corev1api.ResourceEphemeralStorage: resource.MustParse("5Gi"),
 			},
 			Limits: corev1api.ResourceList{},
 		}},
-		{"invalid quantity", args{"100m", "invalid", "200m", "256Mi"}, true, nil},
-		{"CPU request greater than limit", args{"300m", "128Mi", "200m", "256Mi"}, true, nil},
-		{"memory request greater than limit", args{"100m", "512Mi", "200m", "256Mi"}, true, nil},
+		{"invalid quantity", args{"100m", "invalid", "1Gi", "200m", "256Mi", "valid"}, true, nil},
+		{"CPU request greater than limit", args{"300m", "128Mi", "5Gi", "200m", "256Mi", "10Gi"}, true, nil},
+		{"memory request greater than limit", args{"100m", "512Mi", "5Gi", "200m", "256Mi", "10Gi"}, true, nil},
+		{"ephemeral storage request greater than limit", args{"100m", "128Mi", "10Gi", "200m", "256Mi", "5Gi"}, true, nil},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got, err := ParseResourceRequirements(tt.args.cpuRequest, tt.args.memRequest, tt.args.cpuLimit, tt.args.memLimit)
+			got, err := ParseResourceRequirements(tt.args.cpuRequest, tt.args.memRequest, tt.args.ephemeralStorageRequest, tt.args.cpuLimit, tt.args.memLimit, tt.args.ephemeralStorageLimit)
 			if tt.wantErr {
 				assert.Error(t, err)
 				return
@@ -85,12 +105,14 @@ func TestParseResourceRequirements(t *testing.T) {
 			if tt.expected == nil {
 				expected = corev1api.ResourceRequirements{
 					Requests: corev1api.ResourceList{
-						corev1api.ResourceCPU:    resource.MustParse(tt.args.cpuRequest),
-						corev1api.ResourceMemory: resource.MustParse(tt.args.memRequest),
+						corev1api.ResourceCPU:              resource.MustParse(tt.args.cpuRequest),
+						corev1api.ResourceMemory:           resource.MustParse(tt.args.memRequest),
+						corev1api.ResourceEphemeralStorage: resource.MustParse(tt.args.ephemeralStorageRequest),
 					},
 					Limits: corev1api.ResourceList{
-						corev1api.ResourceCPU:    resource.MustParse(tt.args.cpuLimit),
-						corev1api.ResourceMemory: resource.MustParse(tt.args.memLimit),
+						corev1api.ResourceCPU:              resource.MustParse(tt.args.cpuLimit),
+						corev1api.ResourceMemory:           resource.MustParse(tt.args.memLimit),
+						corev1api.ResourceEphemeralStorage: resource.MustParse(tt.args.ephemeralStorageLimit),
 					},
 				}
 			} else {

pkg/util/podvolume/pod_volume_test.go

@@ -156,7 +156,7 @@ func TestGetVolumesByPod(t *testing.T) {
 					Volumes: []corev1api.Volume{
 						// PVB Volumes
 						{Name: "pvbPV1"}, {Name: "pvbPV2"}, {Name: "pvbPV3"},
-						/// Excluded from PVB because colume mounting default service account token
+						/// Excluded from PVB because volume mounting default service account token
 						{Name: "default-token-5xq45"},
 					},
 				},

pkg/util/wildcard/expand.go

@@ -31,70 +31,77 @@ func ShouldExpandWildcards(includes []string, excludes []string) bool {
 }
 
 // containsWildcardPattern checks if a pattern contains any wildcard symbols
-// Supported patterns: *, ?, [abc], {a,b,c}
+// Supported patterns: *, ?, [abc]
 // Note: . and + are treated as literal characters (not wildcards)
 // Note: ** and consecutive asterisks are NOT supported (will cause validation error)
 func containsWildcardPattern(pattern string) bool {
-	return strings.ContainsAny(pattern, "*?[{")
+	return strings.ContainsAny(pattern, "*?[")
 }
 
 func validateWildcardPatterns(patterns []string) error {
 	for _, pattern := range patterns {
-		// Check for invalid regex-only patterns that we don't support
-		if strings.ContainsAny(pattern, "|()") {
-			return errors.New("wildcard pattern contains unsupported regex symbols: |, (, )")
-		}
-
-		// Check for consecutive asterisks (2 or more)
-		if strings.Contains(pattern, "**") {
-			return errors.New("wildcard pattern contains consecutive asterisks (only single * allowed)")
-		}
-
-		// Check for malformed brace patterns
-		if err := validateBracePatterns(pattern); err != nil {
+		if err := ValidateNamespaceName(pattern); err != nil {
 			return err
 		}
 	}
 	return nil
 }
 
+func ValidateNamespaceName(pattern string) error {
+	// Check for invalid characters that are not supported in glob patterns
+	if strings.ContainsAny(pattern, "|()!{},") {
+		return errors.New("wildcard pattern contains unsupported characters: |, (, ), !, {, }, ,")
+	}
+
+	// Check for consecutive asterisks (2 or more)
+	if strings.Contains(pattern, "**") {
+		return errors.New("wildcard pattern contains consecutive asterisks (only single * allowed)")
+	}
+
+	// Check for malformed brace patterns
+	if err := validateBracePatterns(pattern); err != nil {
+		return err
+	}
+
+	return nil
+}
+
 // validateBracePatterns checks for malformed brace patterns like unclosed braces or empty braces
+// Also validates bracket patterns [] for character classes
 func validateBracePatterns(pattern string) error {
-	depth := 0
+	bracketDepth := 0
 
 	for i := 0; i < len(pattern); i++ {
-		if pattern[i] == '{' {
-			braceStart := i
-			depth++
+		if pattern[i] == '[' {
+			bracketStart := i
+			bracketDepth++
 
-			// Scan ahead to find the matching closing brace and validate content
-			for j := i + 1; j < len(pattern) && depth > 0; j++ {
-				if pattern[j] == '{' {
-					depth++
-				} else if pattern[j] == '}' {
-					depth--
-					if depth == 0 {
-						// Found matching closing brace - validate content
-						content := pattern[braceStart+1 : j]
-						if strings.Trim(content, ", \t") == "" {
-							return errors.New("wildcard pattern contains empty brace pattern '{}'")
+			// Scan ahead to find the matching closing bracket and validate content
+			for j := i + 1; j < len(pattern) && bracketDepth > 0; j++ {
+				if pattern[j] == ']' {
+					bracketDepth--
+					if bracketDepth == 0 {
+						// Found matching closing bracket - validate content
+						content := pattern[bracketStart+1 : j]
+						if content == "" {
+							return errors.New("wildcard pattern contains empty bracket pattern '[]'")
 						}
-						// Skip to the closing brace
+						// Skip to the closing bracket
 						i = j
 						break
 					}
 				}
 			}
 
-			// If we exited the loop without finding a match (depth > 0), brace is unclosed
-			if depth > 0 {
-				return errors.New("wildcard pattern contains unclosed brace '{'")
+			// If we exited the loop without finding a match (bracketDepth > 0), bracket is unclosed
+			if bracketDepth > 0 {
+				return errors.New("wildcard pattern contains unclosed bracket '['")
 			}
 
-			// i is now positioned at the closing brace; the outer loop will increment it
-		} else if pattern[i] == '}' {
-			// Found a closing brace without a matching opening brace
-			return errors.New("wildcard pattern contains unmatched closing brace '}'")
+			// i is now positioned at the closing bracket; the outer loop will increment it
+		} else if pattern[i] == ']' {
+			// Found a closing bracket without a matching opening bracket
+			return errors.New("wildcard pattern contains unmatched closing bracket ']'")
 		}
 	}
 

pkg/util/wildcard/expand_test.go

@@ -90,7 +90,7 @@ func TestShouldExpandWildcards(t *testing.T) {
 			name:     "brace alternatives wildcard",
 			includes: []string{"ns{prod,staging}"},
 			excludes: []string{},
-			expected: true, // brace alternatives are considered wildcard
+			expected: false, // brace alternatives are not supported
 		},
 		{
 			name:     "dot is literal - not wildcard",
@@ -237,9 +237,9 @@ func TestExpandWildcards(t *testing.T) {
 			activeNamespaces: []string{"app-prod", "app-staging", "app-dev", "db-prod"},
 			includes:         []string{"app-{prod,staging}"},
 			excludes:         []string{},
-			expectedIncludes: []string{"app-prod", "app-staging"}, // {prod,staging} matches either
+			expectedIncludes: nil,
 			expectedExcludes: nil,
-			expectError:      false,
+			expectError:      true,
 		},
 		{
 			name:             "literal dot and plus patterns",
@@ -259,33 +259,6 @@ func TestExpandWildcards(t *testing.T) {
 			expectedExcludes: nil,
 			expectError:      true, // |, (, ) are not supported
 		},
-		{
-			name:             "unclosed brace patterns should error",
-			activeNamespaces: []string{"app-prod"},
-			includes:         []string{"app-{prod,staging"},
-			excludes:         []string{},
-			expectedIncludes: nil,
-			expectedExcludes: nil,
-			expectError:      true, // unclosed brace
-		},
-		{
-			name:             "empty brace patterns should error",
-			activeNamespaces: []string{"app-prod"},
-			includes:         []string{"app-{}"},
-			excludes:         []string{},
-			expectedIncludes: nil,
-			expectedExcludes: nil,
-			expectError:      true, // empty braces
-		},
-		{
-			name:             "unmatched closing brace should error",
-			activeNamespaces: []string{"app-prod"},
-			includes:         []string{"app-prod}"},
-			excludes:         []string{},
-			expectedIncludes: nil,
-			expectedExcludes: nil,
-			expectError:      true, // unmatched closing brace
-		},
 	}
 
 	for _, tt := range tests {
@@ -354,13 +327,6 @@ func TestExpandWildcardsPrivate(t *testing.T) {
 			expected:         []string{}, // returns empty slice, not nil
 			expectError:      false,
 		},
-		{
-			name:             "brace patterns work correctly",
-			patterns:         []string{"app-{prod,staging}"},
-			activeNamespaces: []string{"app-prod", "app-staging", "app-dev", "app-{prod,staging}"},
-			expected:         []string{"app-prod", "app-staging"}, // brace patterns do expand
-			expectError:      false,
-		},
 		{
 			name:             "duplicate matches from multiple patterns",
 			patterns:         []string{"app-*", "*-prod"},
@@ -389,20 +355,6 @@ func TestExpandWildcardsPrivate(t *testing.T) {
 			expected:         []string{"nsa", "nsb", "nsc"}, // [a-c] matches a to c
 			expectError:      false,
 		},
-		{
-			name:             "negated character class",
-			patterns:         []string{"ns[!abc]"},
-			activeNamespaces: []string{"nsa", "nsb", "nsc", "nsd", "ns1"},
-			expected:         []string{"nsd", "ns1"}, // [!abc] matches anything except a, b, c
-			expectError:      false,
-		},
-		{
-			name:             "brace alternatives",
-			patterns:         []string{"app-{prod,test}"},
-			activeNamespaces: []string{"app-prod", "app-test", "app-staging", "db-prod"},
-			expected:         []string{"app-prod", "app-test"}, // {prod,test} matches either
-			expectError:      false,
-		},
 		{
 			name:             "double asterisk should error",
 			patterns:         []string{"**"},
@@ -410,13 +362,6 @@ func TestExpandWildcardsPrivate(t *testing.T) {
 			expected:         nil,
 			expectError:      true, // ** is not allowed
 		},
-		{
-			name:             "literal dot and plus",
-			patterns:         []string{"app.prod", "service+"},
-			activeNamespaces: []string{"app.prod", "appXprod", "service+", "service"},
-			expected:         []string{"app.prod", "service+"}, // . and + are literal
-			expectError:      false,
-		},
 		{
 			name:             "unsupported regex symbols should error",
 			patterns:         []string{"ns(1|2)"},
@@ -468,153 +413,101 @@ func TestValidateBracePatterns(t *testing.T) {
 		expectError bool
 		errorMsg    string
 	}{
-		// Valid patterns
+		// Valid square bracket patterns
 		{
-			name:        "valid single brace pattern",
-			pattern:     "app-{prod,staging}",
+			name:        "valid square bracket pattern",
+			pattern:     "ns[abc]",
 			expectError: false,
 		},
 		{
-			name:        "valid brace with single option",
-			pattern:     "app-{prod}",
+			name:        "valid square bracket pattern with range",
+			pattern:     "ns[a-z]",
 			expectError: false,
 		},
 		{
-			name:        "valid brace with three options",
-			pattern:     "app-{prod,staging,dev}",
+			name:        "valid square bracket pattern with numbers",
+			pattern:     "ns[0-9]",
 			expectError: false,
 		},
 		{
-			name:        "valid pattern with text before and after brace",
-			pattern:     "prefix-{a,b}-suffix",
+			name:        "valid square bracket pattern with mixed",
+			pattern:     "ns[a-z0-9]",
 			expectError: false,
 		},
 		{
-			name:        "valid pattern with no braces",
-			pattern:     "app-prod",
+			name:        "valid square bracket pattern with single character",
+			pattern:     "ns[a]",
 			expectError: false,
 		},
 		{
-			name:        "valid pattern with asterisk",
-			pattern:     "app-*",
+			name:        "valid square bracket pattern with text before and after",
+			pattern:     "prefix-[abc]-suffix",
 			expectError: false,
 		},
+		// Unclosed opening brackets
 		{
-			name:        "valid brace with spaces around content",
-			pattern:     "app-{ prod , staging }",
-			expectError: false,
+			name:        "unclosed opening bracket at end",
+			pattern:     "ns[abc",
+			expectError: true,
+			errorMsg:    "unclosed bracket",
 		},
 		{
-			name:        "valid brace with numbers",
-			pattern:     "ns-{1,2,3}",
-			expectError: false,
+			name:        "unclosed opening bracket at start",
+			pattern:     "[abc",
+			expectError: true,
+			errorMsg:    "unclosed bracket",
 		},
 		{
-			name:        "valid brace with hyphens in options",
-			pattern:     "{app-prod,db-staging}",
-			expectError: false,
+			name:        "unclosed opening bracket in middle",
+			pattern:     "ns[abc-test",
+			expectError: true,
+			errorMsg:    "unclosed bracket",
 		},
 
-		// Unclosed opening braces
+		// Unmatched closing brackets
 		{
-			name:        "unclosed opening brace at end",
-			pattern:     "app-{prod,staging",
+			name:        "unmatched closing bracket at end",
+			pattern:     "ns-abc]",
 			expectError: true,
-			errorMsg:    "unclosed brace",
+			errorMsg:    "unmatched closing bracket",
 		},
 		{
-			name:        "unclosed opening brace at start",
-			pattern:     "{prod,staging",
+			name:        "unmatched closing bracket at start",
+			pattern:     "]ns-abc",
 			expectError: true,
-			errorMsg:    "unclosed brace",
+			errorMsg:    "unmatched closing bracket",
 		},
 		{
-			name:        "unclosed opening brace in middle",
-			pattern:     "app-{prod-test",
+			name:        "unmatched closing bracket in middle",
+			pattern:     "ns-]abc",
 			expectError: true,
-			errorMsg:    "unclosed brace",
+			errorMsg:    "unmatched closing bracket",
 		},
 		{
-			name:        "multiple unclosed braces",
-			pattern:     "app-{prod-{staging",
+			name:        "extra closing bracket after valid pair",
+			pattern:     "ns[abc]]",
 			expectError: true,
-			errorMsg:    "unclosed brace",
+			errorMsg:    "unmatched closing bracket",
 		},
 
-		// Unmatched closing braces
+		// Empty bracket patterns
 		{
-			name:        "unmatched closing brace at end",
-			pattern:     "app-prod}",
+			name:        "completely empty brackets",
+			pattern:     "ns[]",
 			expectError: true,
-			errorMsg:    "unmatched closing brace",
+			errorMsg:    "empty bracket pattern",
 		},
 		{
-			name:        "unmatched closing brace at start",
-			pattern:     "}app-prod",
+			name:        "empty brackets at start",
+			pattern:     "[]ns",
 			expectError: true,
-			errorMsg:    "unmatched closing brace",
+			errorMsg:    "empty bracket pattern",
 		},
 		{
-			name:        "unmatched closing brace in middle",
-			pattern:     "app-}prod",
+			name:        "empty brackets standalone",
+			pattern:     "[]",
 			expectError: true,
-			errorMsg:    "unmatched closing brace",
-		},
-		{
-			name:        "extra closing brace after valid pair",
-			pattern:     "app-{prod,staging}}",
-			expectError: true,
-			errorMsg:    "unmatched closing brace",
-		},
-
-		// Empty brace patterns
-		{
-			name:        "completely empty braces",
-			pattern:     "app-{}",
-			expectError: true,
-			errorMsg:    "empty brace pattern",
-		},
-		{
-			name:        "braces with only spaces",
-			pattern:     "app-{   }",
-			expectError: true,
-			errorMsg:    "empty brace pattern",
-		},
-		{
-			name:        "braces with only comma",
-			pattern:     "app-{,}",
-			expectError: true,
-			errorMsg:    "empty brace pattern",
-		},
-		{
-			name:        "braces with only commas",
-			pattern:     "app-{,,,}",
-			expectError: true,
-			errorMsg:    "empty brace pattern",
-		},
-		{
-			name:        "braces with commas and spaces",
-			pattern:     "app-{ , , }",
-			expectError: true,
-			errorMsg:    "empty brace pattern",
-		},
-		{
-			name:        "braces with tabs and commas",
-			pattern:     "app-{\t,\t}",
-			expectError: true,
-			errorMsg:    "empty brace pattern",
-		},
-		{
-			name:        "empty braces at start",
-			pattern:     "{}app-prod",
-			expectError: true,
-			errorMsg:    "empty brace pattern",
-		},
-		{
-			name:        "empty braces standalone",
-			pattern:     "{}",
-			expectError: true,
-			errorMsg:    "empty brace pattern",
+			errorMsg:    "empty bracket pattern",
 		},
 
 		// Edge cases
@@ -623,58 +516,6 @@ func TestValidateBracePatterns(t *testing.T) {
 			pattern:     "",
 			expectError: false,
 		},
-		{
-			name:        "pattern with only opening brace",
-			pattern:     "{",
-			expectError: true,
-			errorMsg:    "unclosed brace",
-		},
-		{
-			name:        "pattern with only closing brace",
-			pattern:     "}",
-			expectError: true,
-			errorMsg:    "unmatched closing brace",
-		},
-		{
-			name:        "valid brace with special characters inside",
-			pattern:     "app-{prod-1,staging_2,dev.3}",
-			expectError: false,
-		},
-		{
-			name:        "brace with asterisk inside option",
-			pattern:     "app-{prod*,staging}",
-			expectError: false,
-		},
-		{
-			name:        "multiple valid brace patterns",
-			pattern:     "{app,db}-{prod,staging}",
-			expectError: false,
-		},
-		{
-			name:        "brace with single character",
-			pattern:     "app-{a}",
-			expectError: false,
-		},
-		{
-			name:        "brace with trailing comma but has content",
-			pattern:     "app-{prod,staging,}",
-			expectError: false, // Has content, so it's valid
-		},
-		{
-			name:        "brace with leading comma but has content",
-			pattern:     "app-{,prod,staging}",
-			expectError: false, // Has content, so it's valid
-		},
-		{
-			name:        "brace with leading comma but has content",
-			pattern:     "app-{{,prod,staging}",
-			expectError: true, // unclosed brace
-		},
-		{
-			name:        "brace with leading comma but has content",
-			pattern:     "app-{,prod,staging}}",
-			expectError: true, // unmatched closing brace
-		},
 	}
 
 	for _, tt := range tests {
@@ -723,20 +564,6 @@ func TestExpandWildcardsEdgeCases(t *testing.T) {
 		assert.ElementsMatch(t, []string{"ns-1", "ns_2", "ns.3", "ns@4"}, result)
 	})
 
-	t.Run("complex glob combinations", func(t *testing.T) {
-		activeNamespaces := []string{"app1-prod", "app2-prod", "app1-test", "db-prod", "service"}
-		result, err := expandWildcards([]string{"app?-{prod,test}"}, activeNamespaces)
-		require.NoError(t, err)
-		assert.ElementsMatch(t, []string{"app1-prod", "app2-prod", "app1-test"}, result)
-	})
-
-	t.Run("escaped characters", func(t *testing.T) {
-		activeNamespaces := []string{"app*", "app-prod", "app?test", "app-test"}
-		result, err := expandWildcards([]string{"app\\*"}, activeNamespaces)
-		require.NoError(t, err)
-		assert.ElementsMatch(t, []string{"app*"}, result)
-	})
-
 	t.Run("mixed literal and wildcard patterns", func(t *testing.T) {
 		activeNamespaces := []string{"app.prod", "app-prod", "app_prod", "test.ns"}
 		result, err := expandWildcards([]string{"app.prod", "app?prod"}, activeNamespaces)
@@ -777,12 +604,8 @@ func TestExpandWildcardsEdgeCases(t *testing.T) {
 			shouldError bool
 		}{
 			{"unclosed bracket", "ns[abc", true},
-			{"unclosed brace", "app-{prod,staging", true},
-			{"nested unclosed", "ns[a{bc", true},
 			{"valid bracket", "ns[abc]", false},
-			{"valid brace", "app-{prod,staging}", false},
 			{"empty bracket", "ns[]", true}, // empty brackets are invalid
-			{"empty brace", "app-{}", true}, // empty braces are invalid
 		}
 
 		for _, tt := range tests {

site/content/docs/main/api-types/backup.md

@@ -16,6 +16,8 @@ Backup belongs to the API group version `velero.io/v1`.
 
 Here is a sample `Backup` object with each of the fields documented:
 
+**Note:** Namespace includes/excludes support glob patterns (`*`, `?`, `[abc]`). See [Namespace Glob Patterns](../namespace-glob-patterns) for more details.
+
 ```yaml
 # Standard Kubernetes API Version declaration. Required.
 apiVersion: velero.io/v1
@@ -42,11 +44,12 @@ spec:
   resourcePolicy:
     kind: configmap
     name: resource-policy-configmap
-  # Array of namespaces to include in the backup. If unspecified, all namespaces are included.
-  # Optional.
+  # Array of namespaces to include in the backup. Accepts glob patterns (*, ?, [abc]).
+  # Note: '*' alone is reserved for empty fields, which means all namespaces.
+  # If unspecified, all namespaces are included. Optional.
   includedNamespaces:
   - '*'
-  # Array of namespaces to exclude from the backup. Optional.
+  # Array of namespaces to exclude from the backup. Accepts glob patterns (*, ?, [abc]). Optional.
   excludedNamespaces:
   - some-namespace
   # Array of resources to include in the backup. Resources may be shortcuts (for example 'po' for 'pods')

site/content/docs/main/api-types/restore.md

@@ -16,6 +16,8 @@ Restore belongs to the API group version `velero.io/v1`.
 
 Here is a sample `Restore` object with each of the fields documented:
 
+**Note:** Namespace includes/excludes support glob patterns (`*`, `?`, `[abc]`). See [Namespace Glob Patterns](../namespace-glob-patterns) for more details.
+
 ```yaml
 # Standard Kubernetes API Version declaration. Required.
 apiVersion: velero.io/v1
@@ -45,11 +47,11 @@ spec:
     writeSparseFiles: true
     # ParallelFilesDownload is the concurrency number setting for restore
     parallelFilesDownload: 10
-  # Array of namespaces to include in the restore. If unspecified, all namespaces are included.
-  # Optional.
+  # Array of namespaces to include in the restore. Accepts glob patterns (*, ?, [abc]).
+  # If unspecified, all namespaces are included. Optional.
   includedNamespaces:
   - '*'
-  # Array of namespaces to exclude from the restore. Optional.
+  # Array of namespaces to exclude from the restore. Accepts glob patterns (*, ?, [abc]). Optional.
   excludedNamespaces:
   - some-namespace
   # Array of resources to include in the restore. Resources may be shortcuts (for example 'po' for 'pods')

site/content/docs/main/data-movement-pod-resource-configuration.md

@@ -7,7 +7,7 @@ During [CSI Snapshot Data Movement][1], Velero built-in data mover launches data
 During [fs-backup][2], Velero also launches data mover pods to run the data transfer.  
 The data transfer is a time and resource consuming activity.  
 
-Velero by default uses the [BestEffort QoS][2] for the data mover pods, which guarantees the best performance of the data movement activities. On the other hand, it may take lots of cluster resource, i.e., CPU, memory, and how many resources are taken is decided by the concurrency and the scale of data to be moved.  
+Velero by default uses the [BestEffort QoS][2] for the data mover pods, which guarantees the best performance of the data movement activities. On the other hand, it may take lots of cluster resource, i.e., CPU, memory, ephemeral storage, and how many resources are taken is decided by the concurrency and the scale of data to be moved.  
 
 If the cluster nodes don't have sufficient resource, Velero also allows you to customize the resources for the data mover pods.    
 Note: If less resources are assigned to data mover pods, the data movement activities may take longer time; or the data mover pods may be OOM killed if the assigned memory resource doesn't meet the requirements. Consequently, the dataUpload/dataDownload may run longer or fail.  
@@ -25,6 +25,8 @@ Here is a sample of the configMap with ```podResources```:
     "podResources": {
         "cpuRequest": "1000m",
         "cpuLimit": "1000m",
+        "ephemeralStorageRequest": "5Gi",
+        "ephemeralStorageLimit": "10Gi",
         "memoryRequest": "512Mi",
         "memoryLimit": "1Gi"        
     }

site/content/docs/main/namespace-glob-patterns.md

@@ -0,0 +1,71 @@
+---
+title: "Namespace Glob Patterns"
+layout: docs
+---
+
+When using `--include-namespaces` and `--exclude-namespaces` flags with backup and restore commands, you can use glob patterns to match multiple namespaces.
+
+## Supported Patterns
+
+Velero supports the following glob pattern characters:
+
+- `*` - Matches any sequence of characters
+  ```bash
+  velero backup create my-backup --include-namespaces "app-*"
+  # Matches: app-prod, app-staging, app-dev, etc.
+  ```
+
+- `?` - Matches exactly one character
+  ```bash
+  velero backup create my-backup --include-namespaces "ns?"
+  # Matches: ns1, ns2, nsa, but NOT ns10
+  ```
+
+- `[abc]` - Matches any single character in the brackets
+  ```bash
+  velero backup create my-backup --include-namespaces "ns[123]"
+  # Matches: ns1, ns2, ns3
+  ```
+
+- `[a-z]` - Matches any single character in the range
+  ```bash
+  velero backup create my-backup --include-namespaces "ns[a-c]"
+  # Matches: nsa, nsb, nsc
+  ```
+
+## Unsupported Patterns
+
+The following patterns are **not supported** and will cause validation errors:
+
+- `**` - Consecutive asterisks
+- `|` - Alternation (regex operator)
+- `()` - Grouping (regex operators)
+- `!` - Negation
+- `{}` - Brace expansion
+- `,` - Comma (used in brace expansion) 
+
+## Special Cases
+
+- `*` alone means "all namespaces" and is not expanded
+- Empty brackets `[]` are invalid
+- Unmatched or unclosed brackets will cause validation errors
+
+## Examples
+
+Combine patterns with include and exclude flags:
+
+```bash
+# Backup all production namespaces except test
+velero backup create prod-backup \
+  --include-namespaces "*-prod" \
+  --exclude-namespaces "test-*"
+
+# Backup specific numbered namespaces
+velero backup create numbered-backup \
+  --include-namespaces "app-[0-9]"
+
+# Restore namespaces matching multiple patterns
+velero restore create my-restore \
+  --from-backup my-backup \
+  --include-namespaces "frontend-*,backend-*"
+```

site/content/docs/main/repository-maintenance.md

@@ -72,6 +72,8 @@ data:
       "podResources": {
         "cpuRequest": "100m",
         "cpuLimit": "200m",
+        "ephemeralStorageRequest": "5Gi",
+        "ephemeralStorageLimit": "10Gi",
         "memoryRequest": "100Mi",
         "memoryLimit": "200Mi"
       },
@@ -99,6 +101,8 @@ data:
       "podResources": {
         "cpuRequest": "200m",
         "cpuLimit": "400m",
+        "ephemeralStorageRequest": "5Gi",
+        "ephemeralStorageLimit": "10Gi",
         "memoryRequest": "200Mi",
         "memoryLimit": "400Mi"
       },

site/content/docs/main/resource-filtering.md

@@ -17,7 +17,11 @@ Wildcard takes precedence when both a wildcard and specific resource are include
 
 ### --include-namespaces
 
-Namespaces to include. Default is `*`, all namespaces.
+Namespaces to include. Accepts glob patterns (`*`, `?`, `[abc]`). Default is `*`, all namespaces.
+
+See [Namespace Glob Patterns](namespace-glob-patterns) for more details on supported patterns.
+
+Note: `*` alone is reserved for empty fields, which means all namespaces.
 
 * Backup a namespace and it's objects.
 
@@ -158,7 +162,9 @@ Wildcard excludes are ignored.
 
 ### --exclude-namespaces
 
-Namespaces to exclude.
+Namespaces to exclude. Accepts glob patterns (`*`, `?`, `[abc]`).
+
+See [Namespace Glob Patterns](namespace-glob-patterns.md) for more details on supported patterns.
 
 * Exclude kube-system from the cluster backup.
 

site/content/docs/main/supported-configmaps/node-agent-configmap.md

@@ -224,7 +224,7 @@ Configure different node selection rules for specific storage classes:
 ```
 
 ### Pod Resources (`podResources`)
-Configure CPU and memory resources for Data Mover Pods to optimize performance and prevent resource conflict.
+Configure CPU, memory and ephemeral storage resources for Data Mover Pods to optimize performance and prevent resource conflict.
 
 The configurations work for PodVolumeBackup, PodVolumeRestore, DataUpload, and DataDownload pods.
 
@@ -233,6 +233,8 @@ The configurations work for PodVolumeBackup, PodVolumeRestore, DataUpload, and D
   "podResources": {
     "cpuRequest": "1000m",
     "cpuLimit": "2000m",
+    "ephemeralStorageRequest": "5Gi",
+    "ephemeralStorageLimit": "10Gi",
     "memoryRequest": "1Gi",
     "memoryLimit": "4Gi"
   }
@@ -535,6 +537,8 @@ Here's a comprehensive example showing how all configuration sections work toget
   "podResources": {
     "cpuRequest": "500m",
     "cpuLimit": "1000m",
+    "ephemeralStorageRequest": "5Gi",
+    "ephemeralStorageLimit": "10Gi",
     "memoryRequest": "1Gi",
     "memoryLimit": "2Gi"
   },

site/content/docs/v1.18/api-types/backup.md

@@ -16,6 +16,8 @@ Backup belongs to the API group version `velero.io/v1`.
 
 Here is a sample `Backup` object with each of the fields documented:
 
+**Note:** Namespace includes/excludes support glob patterns (`*`, `?`, `[abc]`). See [Namespace Glob Patterns](../namespace-glob-patterns) for more details.
+
 ```yaml
 # Standard Kubernetes API Version declaration. Required.
 apiVersion: velero.io/v1
@@ -42,11 +44,11 @@ spec:
   resourcePolicy:
     kind: configmap
     name: resource-policy-configmap
-  # Array of namespaces to include in the backup. If unspecified, all namespaces are included.
-  # Optional.
+  # Array of namespaces to include in the backup. Accepts glob patterns (*, ?, [abc]).
+  # If unspecified, all namespaces are included. Optional.
   includedNamespaces:
   - '*'
-  # Array of namespaces to exclude from the backup. Optional.
+  # Array of namespaces to exclude from the backup. Accepts glob patterns (*, ?, [abc]). Optional.
   excludedNamespaces:
   - some-namespace
   # Array of resources to include in the backup. Resources may be shortcuts (for example 'po' for 'pods')

site/content/docs/v1.18/api-types/restore.md

@@ -16,6 +16,8 @@ Restore belongs to the API group version `velero.io/v1`.
 
 Here is a sample `Restore` object with each of the fields documented:
 
+**Note:** Namespace includes/excludes support glob patterns (`*`, `?`, `[abc]`). See [Namespace Glob Patterns](../namespace-glob-patterns) for more details.
+
 ```yaml
 # Standard Kubernetes API Version declaration. Required.
 apiVersion: velero.io/v1
@@ -45,11 +47,11 @@ spec:
     writeSparseFiles: true
     # ParallelFilesDownload is the concurrency number setting for restore
     parallelFilesDownload: 10
-  # Array of namespaces to include in the restore. If unspecified, all namespaces are included.
-  # Optional.
+  # Array of namespaces to include in the restore. Accepts glob patterns (*, ?, [abc]).
+  # If unspecified, all namespaces are included. Optional.
   includedNamespaces:
   - '*'
-  # Array of namespaces to exclude from the restore. Optional.
+  # Array of namespaces to exclude from the restore. Accepts glob patterns (*, ?, [abc]). Optional.
   excludedNamespaces:
   - some-namespace
   # Array of resources to include in the restore. Resources may be shortcuts (for example 'po' for 'pods')

site/content/docs/v1.18/namespace-glob-patterns.md

@@ -0,0 +1,71 @@
+---
+title: "Namespace Glob Patterns"
+layout: docs
+---
+
+When using `--include-namespaces` and `--exclude-namespaces` flags with backup and restore commands, you can use glob patterns to match multiple namespaces.
+
+## Supported Patterns
+
+Velero supports the following glob pattern characters:
+
+- `*` - Matches any sequence of characters
+  ```bash
+  velero backup create my-backup --include-namespaces "app-*"
+  # Matches: app-prod, app-staging, app-dev, etc.
+  ```
+
+- `?` - Matches exactly one character
+  ```bash
+  velero backup create my-backup --include-namespaces "ns?"
+  # Matches: ns1, ns2, nsa, but NOT ns10
+  ```
+
+- `[abc]` - Matches any single character in the brackets
+  ```bash
+  velero backup create my-backup --include-namespaces "ns[123]"
+  # Matches: ns1, ns2, ns3
+  ```
+
+- `[a-z]` - Matches any single character in the range
+  ```bash
+  velero backup create my-backup --include-namespaces "ns[a-c]"
+  # Matches: nsa, nsb, nsc
+  ```
+
+## Unsupported Patterns
+
+The following patterns are **not supported** and will cause validation errors:
+
+- `**` - Consecutive asterisks
+- `|` - Alternation (regex operator)
+- `()` - Grouping (regex operators)
+- `!` - Negation
+- `{}` - Brace expansion
+- `,` - Comma (used in brace expansion)
+
+## Special Cases
+
+- `*` alone means "all namespaces" and is not expanded
+- Empty brackets `[]` are invalid
+- Unmatched or unclosed brackets will cause validation errors
+
+## Examples
+
+Combine patterns with include and exclude flags:
+
+```bash
+# Backup all production namespaces except test
+velero backup create prod-backup \
+  --include-namespaces "*-prod" \
+  --exclude-namespaces "test-*"
+
+# Backup specific numbered namespaces
+velero backup create numbered-backup \
+  --include-namespaces "app-[0-9]"
+
+# Restore namespaces matching multiple patterns
+velero restore create my-restore \
+  --from-backup my-backup \
+  --include-namespaces "frontend-*,backend-*"
+```

site/content/docs/v1.18/resource-filtering.md

@@ -17,7 +17,11 @@ Wildcard takes precedence when both a wildcard and specific resource are include
 
 ### --include-namespaces
 
-Namespaces to include. Default is `*`, all namespaces.
+Namespaces to include. Accepts glob patterns (`*`, `?`, `[abc]`). Default is `*`, all namespaces.
+
+See [Namespace Glob Patterns](namespace-glob-patterns) for more details on supported patterns.
+
+Note: `*` alone is reserved for empty fields, which means all namespaces.
 
 * Backup a namespace and it's objects.
 
@@ -158,7 +162,9 @@ Wildcard excludes are ignored.
 
 ### --exclude-namespaces
 
-Namespaces to exclude.
+Namespaces to exclude. Accepts glob patterns (`*`, `?`, `[abc]`).
+
+See [Namespace Glob Patterns](namespace-glob-patterns) for more details on supported patterns.
 
 * Exclude kube-system from the cluster backup.
 

site/data/docs/main-toc.yml

@@ -33,6 +33,8 @@ toc:
         url: /enable-api-group-versions-feature
       - page: Resource filtering
         url: /resource-filtering
+      - page: Namespace glob patterns
+        url: /namespace-glob-patterns
       - page: Backup reference
         url: /backup-reference
       - page: Backup hooks

site/data/docs/v1-18-toc.yml

@@ -33,6 +33,8 @@ toc:
         url: /enable-api-group-versions-feature
       - page: Resource filtering
         url: /resource-filtering
+      - page: Namespace glob patterns
+        url: /namespace-glob-patterns
       - page: Backup reference
         url: /backup-reference
       - page: Backup hooks

test/Makefile

@@ -76,7 +76,7 @@ HAS_VSPHERE_PLUGIN ?= false
 RESTORE_HELPER_IMAGE ?=
 
 #Released version only
-UPGRADE_FROM_VELERO_VERSION ?= v1.15.2,v1.16.2
+UPGRADE_FROM_VELERO_VERSION ?= v1.16.2,v1.17.2
 
 # UPGRADE_FROM_VELERO_CLI can has the same format(a list divided by comma) with UPGRADE_FROM_VELERO_VERSION
 # Upgrade tests will be executed sequently according to the list by UPGRADE_FROM_VELERO_VERSION
@@ -85,7 +85,7 @@ UPGRADE_FROM_VELERO_VERSION ?= v1.15.2,v1.16.2
 # to the end, nil string will be set if UPGRADE_FROM_VELERO_CLI is shorter than UPGRADE_FROM_VELERO_VERSION
 UPGRADE_FROM_VELERO_CLI ?=
 
-MIGRATE_FROM_VELERO_VERSION ?= v1.16.2,$(VERSION)
+MIGRATE_FROM_VELERO_VERSION ?= v1.17.2,$(VERSION)
 MIGRATE_FROM_VELERO_CLI ?=
 
 VELERO_NAMESPACE ?= velero

test/e2e/nodeagentconfig/node-agent-config.go

@@ -36,7 +36,6 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/builder"
 	velerotypes "github.com/vmware-tanzu/velero/pkg/types"
 	"github.com/vmware-tanzu/velero/pkg/util/kube"
-	velerokubeutil "github.com/vmware-tanzu/velero/pkg/util/kube"
 	"github.com/vmware-tanzu/velero/test"
 	. "github.com/vmware-tanzu/velero/test/e2e/test"
 	k8sutil "github.com/vmware-tanzu/velero/test/util/k8s"
@@ -240,9 +239,13 @@ func (n *NodeAgentConfigTestCase) Backup() error {
 	Expect(backupPodList.Items[0].Spec.PriorityClassName).To(Equal(n.nodeAgentConfigs.PriorityClassName))
 
 	// In backup, only the second element of LoadAffinity array should be used.
-	expectedAffinity := velerokubeutil.ToSystemAffinity(n.nodeAgentConfigs.LoadAffinity[1:])
+	expectedLabelKey, _, ok := popFromMap(n.nodeAgentConfigs.LoadAffinity[1].NodeSelector.MatchLabels)
+	Expect(ok).To(BeTrue(), "Expected LoadAffinity's MatchLabels should at least have one key-value pair")
 
-	Expect(backupPodList.Items[0].Spec.Affinity).To(Equal(expectedAffinity))
+	// From 1.18.1, Velero adds some default affinity in the backup/restore pod,
+	// so we can't directly compare the whole affinity,
+	// but we can verify if the expected affinity is contained in the pod affinity.
+	Expect(backupPodList.Items[0].Spec.Affinity.String()).To(ContainSubstring(expectedLabelKey))
 
 	fmt.Println("backupPod content verification completed successfully.")
 
@@ -317,9 +320,13 @@ func (n *NodeAgentConfigTestCase) Restore() error {
 	Expect(restorePodList.Items[0].Spec.PriorityClassName).To(Equal(n.nodeAgentConfigs.PriorityClassName))
 
 	// In restore, only the first element of LoadAffinity array should be used.
-	expectedAffinity := velerokubeutil.ToSystemAffinity(n.nodeAgentConfigs.LoadAffinity[:1])
+	expectedLabelKey, _, ok := popFromMap(n.nodeAgentConfigs.LoadAffinity[0].NodeSelector.MatchLabels)
+	Expect(ok).To(BeTrue(), "Expected LoadAffinity's MatchLabels should at least have one key-value pair")
 
-	Expect(restorePodList.Items[0].Spec.Affinity).To(Equal(expectedAffinity))
+	// From 1.18.1, Velero adds some default affinity in the backup/restore pod,
+	// so we can't directly compare the whole affinity,
+	// but we can verify if the expected affinity is contained in the pod affinity.
+	Expect(restorePodList.Items[0].Spec.Affinity.String()).To(ContainSubstring(expectedLabelKey))
 
 	fmt.Println("restorePod content verification completed successfully.")
 
@@ -345,3 +352,12 @@ func (n *NodeAgentConfigTestCase) Restore() error {
 
 	return nil
 }
+
+func popFromMap[K comparable, V any](m map[K]V) (k K, v V, ok bool) {
+	for key, val := range m {
+		delete(m, key)
+		return key, val, true
+	}
+
+	return
+}

test/e2e/repomaintenance/repo_maintenance_config.go

@@ -70,10 +70,12 @@ var SpecificRepoMaintenanceTest func() = TestFunc(&RepoMaintenanceTestCase{
 	jobConfigs: velerotypes.JobConfigs{
 		KeepLatestMaintenanceJobs: &keepJobNum,
 		PodResources: &velerokubeutil.PodResources{
-			CPURequest:    "100m",
-			MemoryRequest: "100Mi",
-			CPULimit:      "200m",
-			MemoryLimit:   "200Mi",
+			CPURequest:              "100m",
+			MemoryRequest:           "100Mi",
+			EphemeralStorageRequest: "5Gi",
+			CPULimit:                "200m",
+			MemoryLimit:             "200Mi",
+			EphemeralStorageLimit:   "10Gi",
 		},
 		PriorityClassName: test.PriorityClassNameForRepoMaintenance,
 	},
@@ -230,8 +232,10 @@ func (r *RepoMaintenanceTestCase) Verify() error {
 	resources, err := kube.ParseResourceRequirements(
 		r.jobConfigs.PodResources.CPURequest,
 		r.jobConfigs.PodResources.MemoryRequest,
+		r.jobConfigs.PodResources.EphemeralStorageRequest,
 		r.jobConfigs.PodResources.CPULimit,
 		r.jobConfigs.PodResources.MemoryLimit,
+		r.jobConfigs.PodResources.EphemeralStorageLimit,
 	)
 	if err != nil {
 		return errors.Wrap(err, "failed to parse resource requirements for maintenance job")

test/util/velero/install.go

@@ -365,7 +365,7 @@ func VersionNoOlderThan(version string, targetVersion string) (bool, error) {
 			matches := tagRe.FindStringSubmatch(targetVersion)
 			targetMajor := matches[1]
 			targetMinor := matches[2]
-			if major > targetMajor && minor >= targetMinor {
+			if major >= targetMajor && minor >= targetMinor {
 				return true, nil
 			} else {
 				return false, nil

test/util/velero/install_test.go

@@ -0,0 +1,65 @@
+/*
+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 velero
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func Test_VersionNoOlderThan(t *testing.T) {
+	type versionTest struct {
+		caseName      string
+		version       string
+		targetVersion string
+		result        bool
+		err           error
+	}
+	tests := []versionTest{
+		{
+			caseName:      "branch version compare",
+			version:       "release-1.18",
+			targetVersion: "v1.16",
+			result:        true,
+			err:           nil,
+		},
+		{
+			caseName:      "tag version compare",
+			version:       "v1.18.0",
+			targetVersion: "v1.16",
+			result:        true,
+			err:           nil,
+		},
+		{
+			caseName:      "main version compare",
+			version:       "main",
+			targetVersion: "v1.15",
+			result:        true,
+			err:           nil,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.caseName, func(t *testing.T) {
+			res, err := VersionNoOlderThan(test.version, test.targetVersion)
+
+			require.Equal(t, test.err, err)
+			require.Equal(t, test.result, res)
+		})
+	}
+}

test/util/velero/velero_utils.go

@@ -99,6 +99,15 @@ var ImagesMatrix = map[string]map[string][]string{
 		"velero":                {"velero/velero:v1.16.2"},
 		"velero-restore-helper": {"velero/velero:v1.16.2"},
 	},
+	"v1.17": {
+		"aws":                   {"velero/velero-plugin-for-aws:v1.13.2"},
+		"azure":                 {"velero/velero-plugin-for-microsoft-azure:v1.13.2"},
+		"vsphere":               {"vsphereveleroplugin/velero-plugin-for-vsphere:v1.5.2"},
+		"gcp":                   {"velero/velero-plugin-for-gcp:v1.13.2"},
+		"datamover":             {"velero/velero-plugin-for-aws:v1.13.2"},
+		"velero":                {"velero/velero:v1.17.2"},
+		"velero-restore-helper": {"velero/velero:v1.17.2"},
+	},
 	"main": {
 		"aws":                   {"velero/velero-plugin-for-aws:main"},
 		"azure":                 {"velero/velero-plugin-for-microsoft-azure:main"},
@@ -128,16 +137,13 @@ func SetImagesToDefaultValues(config VeleroConfig, version string) (VeleroConfig
 
 	ret.Plugins = ""
 
-	versionWithoutPatch := "main"
-	if version != "main" {
-		versionWithoutPatch = semver.MajorMinor(version)
-	}
+	versionWithoutPatch := getVersionWithoutPatch(version)
 
 	// Read migration case needs images from the PluginsMatrix map.
 	images, ok := ImagesMatrix[versionWithoutPatch]
 	if !ok {
-		return config, fmt.Errorf("fail to read the images for version %s from the ImagesMatrix",
-			versionWithoutPatch)
+		fmt.Printf("Cannot read the images for version %s from the ImagesMatrix. Use the original values.\n", versionWithoutPatch)
+		return config, nil
 	}
 
 	ret.VeleroImage = images[Velero][0]
@@ -164,6 +170,27 @@ func SetImagesToDefaultValues(config VeleroConfig, version string) (VeleroConfig
 	return ret, nil
 }
 
+func getVersionWithoutPatch(version string) string {
+	versionWithoutPatch := ""
+
+	mainRe := regexp.MustCompile(`^main$`)
+	releaseRe := regexp.MustCompile(`^release-(\d+)\.(\d+)(-dev)?$`)
+
+	switch {
+	case mainRe.MatchString(version):
+		versionWithoutPatch = "main"
+	case releaseRe.MatchString(version):
+		matches := releaseRe.FindStringSubmatch(version)
+		versionWithoutPatch = fmt.Sprintf("v%s.%s", matches[1], matches[2])
+	default:
+		versionWithoutPatch = semver.MajorMinor(version)
+	}
+
+	fmt.Println("The version is ", versionWithoutPatch)
+
+	return versionWithoutPatch
+}
+
 func getPluginsByVersion(version string, cloudProvider string, needDataMoverPlugin bool) ([]string, error) {
 	var cloudMap map[string][]string
 	arr := strings.Split(version, ".")

test/util/velero/velero_utils_test.go

@@ -0,0 +1,54 @@
+/*
+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 velero
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func Test_getVersionWithoutPatch(t *testing.T) {
+	versionTests := []struct {
+		caseName string
+		version  string
+		result   string
+	}{
+		{
+			caseName: "main version",
+			version:  "main",
+			result:   "main",
+		},
+		{
+			caseName: "release version",
+			version:  "release-1.18-dev",
+			result:   "v1.18",
+		},
+		{
+			caseName: "tag version",
+			version:  "v1.17.2",
+			result:   "v1.17",
+		},
+	}
+
+	for _, test := range versionTests {
+		t.Run(test.caseName, func(t *testing.T) {
+			res := getVersionWithoutPatch(test.version)
+			require.Equal(t, test.result, res)
+		})
+	}
+}