Merge branch 'main' into fix/backup-name-validation

changelogs/unreleased/10070-shubham-pampattiwar

@@ -0,0 +1 @@
+Add snapshotClass parameter to volume policy snapshot action

changelogs/unreleased/9697-Joeavaikath

@@ -0,0 +1 @@
+Fail backup validation when built-in data mover is requested but no node-agent pods are running

changelogs/unreleased/9720-Joeavaikath

@@ -0,0 +1 @@
+Add dynamic resource autocompletion to Velero CLI

internal/resourcepolicies/resource_policies.go

@@ -54,6 +54,10 @@ const (
 	// DataMoverParameter is the key of the action parameter that selects the data
 	// mover to be used for the matched volumes when the action type is snapshot.
 	DataMoverParameter = "dataMover"
+
+	// SnapshotClassParameter is the key of the action parameter that selects the
+	// VolumeSnapshotClass to use for CSI snapshots when the action type is snapshot.
+	SnapshotClassParameter = "snapshotClass"
 )
 
 // validDataMovers is the set of data mover values accepted in the snapshot
@@ -109,6 +113,30 @@ func (a *Action) GetDataMover() (string, error) {
 	return dataMover, nil
 }
 
+// GetSnapshotClass returns the VolumeSnapshotClass name configured in the
+// snapshot action's snapshotClass parameter. The snapshotClass parameter is
+// only meaningful for the snapshot action, so it returns an error when the
+// action is nil or its type is not snapshot. When the parameter is absent,
+// it returns an empty string, meaning the caller should fall back to the
+// existing VolumeSnapshotClass selection logic.
+func (a *Action) GetSnapshotClass() (string, error) {
+	if a == nil || a.Type != Snapshot {
+		return "", fmt.Errorf("the %q parameter is only supported for the %q action", SnapshotClassParameter, Snapshot)
+	}
+	if len(a.Parameters) == 0 {
+		return "", nil
+	}
+	raw, ok := a.Parameters[SnapshotClassParameter]
+	if !ok {
+		return "", nil
+	}
+	snapshotClass, ok := raw.(string)
+	if !ok {
+		return "", fmt.Errorf("parameter %q must be a string, got %T", SnapshotClassParameter, raw)
+	}
+	return snapshotClass, nil
+}
+
 // PolicyLabelSelector mirrors metav1.LabelSelector with yaml tags for ConfigMap decode.
 // metav1.LabelSelector only has json tags, which do not populate under go.yaml.in/yaml/v3.
 type PolicyLabelSelector struct {

internal/resourcepolicies/resource_policies_test.go

@@ -3064,3 +3064,60 @@ func TestActionGetDataMover(t *testing.T) {
 		})
 	}
 }
+
+func TestActionGetSnapshotClass(t *testing.T) {
+	testCases := []struct {
+		name          string
+		action        *Action
+		expectedClass string
+		expectErr     bool
+	}{
+		{
+			name:      "nil action",
+			action:    nil,
+			expectErr: true,
+		},
+		{
+			name:          "snapshot action without parameters",
+			action:        &Action{Type: Snapshot},
+			expectedClass: "",
+		},
+		{
+			name:          "snapshot action without snapshotClass parameter",
+			action:        &Action{Type: Snapshot, Parameters: map[string]any{"other": "value"}},
+			expectedClass: "",
+		},
+		{
+			name:          "snapshot action with snapshotClass",
+			action:        &Action{Type: Snapshot, Parameters: map[string]any{"snapshotClass": "my-vsc"}},
+			expectedClass: "my-vsc",
+		},
+		{
+			name:      "non-snapshot action returns error",
+			action:    &Action{Type: FSBackup, Parameters: map[string]any{"snapshotClass": "my-vsc"}},
+			expectErr: true,
+		},
+		{
+			name:      "snapshot action with non-string snapshotClass returns error",
+			action:    &Action{Type: Snapshot, Parameters: map[string]any{"snapshotClass": 123}},
+			expectErr: true,
+		},
+		{
+			name:          "snapshot action with both snapshotClass and dataMover",
+			action:        &Action{Type: Snapshot, Parameters: map[string]any{"snapshotClass": "my-vsc", "dataMover": "velero-fs"}},
+			expectedClass: "my-vsc",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			snapshotClass, err := tc.action.GetSnapshotClass()
+			if tc.expectErr {
+				require.Error(t, err)
+				return
+			}
+			require.NoError(t, err)
+			assert.Equal(t, tc.expectedClass, snapshotClass)
+		})
+	}
+}

internal/resourcepolicies/volume_resources_validator.go

@@ -118,5 +118,19 @@ func (a *Action) validate() error {
 		}
 	}
 
+	if raw, ok := a.Parameters[SnapshotClassParameter]; ok {
+		if a.Type != Snapshot {
+			return fmt.Errorf("parameter %q is only supported for the %q action, but the action type is %q",
+				SnapshotClassParameter, Snapshot, a.Type)
+		}
+		snapshotClass, ok := raw.(string)
+		if !ok {
+			return fmt.Errorf("parameter %q must be a string, got %T", SnapshotClassParameter, raw)
+		}
+		if snapshotClass == "" {
+			return fmt.Errorf("parameter %q must not be empty", SnapshotClassParameter)
+		}
+	}
+
 	return nil
 }

internal/resourcepolicies/volume_resources_validator_test.go

@@ -658,6 +658,86 @@ func TestValidate(t *testing.T) {
 			},
 			wantErr: false,
 		},
+		{
+			name: "snapshot action with valid snapshotClass",
+			res: &ResourcePolicies{
+				Version: "v1",
+				VolumePolicies: []VolumePolicy{
+					{
+						Action: Action{
+							Type:       Snapshot,
+							Parameters: map[string]any{"snapshotClass": "my-vsc"},
+						},
+						Conditions: map[string]any{"storageClass": []string{"gp2"}},
+					},
+				},
+			},
+			wantErr: false,
+		},
+		{
+			name: "snapshot action with both snapshotClass and dataMover",
+			res: &ResourcePolicies{
+				Version: "v1",
+				VolumePolicies: []VolumePolicy{
+					{
+						Action: Action{
+							Type:       Snapshot,
+							Parameters: map[string]any{"snapshotClass": "my-vsc", "dataMover": "velero-fs"},
+						},
+						Conditions: map[string]any{"storageClass": []string{"gp2"}},
+					},
+				},
+			},
+			wantErr: false,
+		},
+		{
+			name: "snapshotClass parameter on non-snapshot action is rejected",
+			res: &ResourcePolicies{
+				Version: "v1",
+				VolumePolicies: []VolumePolicy{
+					{
+						Action: Action{
+							Type:       FSBackup,
+							Parameters: map[string]any{"snapshotClass": "my-vsc"},
+						},
+						Conditions: map[string]any{"storageClass": []string{"gp2"}},
+					},
+				},
+			},
+			wantErr: true,
+		},
+		{
+			name: "snapshot action with non-string snapshotClass is rejected",
+			res: &ResourcePolicies{
+				Version: "v1",
+				VolumePolicies: []VolumePolicy{
+					{
+						Action: Action{
+							Type:       Snapshot,
+							Parameters: map[string]any{"snapshotClass": 123},
+						},
+						Conditions: map[string]any{"storageClass": []string{"gp2"}},
+					},
+				},
+			},
+			wantErr: true,
+		},
+		{
+			name: "snapshot action with empty snapshotClass is rejected",
+			res: &ResourcePolicies{
+				Version: "v1",
+				VolumePolicies: []VolumePolicy{
+					{
+						Action: Action{
+							Type:       Snapshot,
+							Parameters: map[string]any{"snapshotClass": ""},
+						},
+						Conditions: map[string]any{"storageClass": []string{"gp2"}},
+					},
+				},
+			},
+			wantErr: true,
+		},
 	}
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {

internal/volumehelper/volume_policy_helper.go

@@ -430,6 +430,21 @@ func (v *volumeHelperImpl) GetActionParameters(obj runtime.Unstructured, groupRe
 	return false, "", nil, nil
 }
 
+func (v *volumeHelperImpl) GetSnapshotClass(obj runtime.Unstructured, groupResource schema.GroupResource) (string, error) {
+	matched, actionType, params, err := v.GetActionParameters(obj, groupResource)
+	if err != nil {
+		return "", err
+	}
+	if !matched {
+		return "", nil
+	}
+	action := &resourcepolicies.Action{
+		Type:       resourcepolicies.VolumeActionType(actionType),
+		Parameters: params,
+	}
+	return action.GetSnapshotClass()
+}
+
 func (v *volumeHelperImpl) shouldIncludeVolumeInBackup(vol corev1api.Volume) bool {
 	includeVolumeInBackup := true
 	// cannot backup hostpath volumes as they are not mounted into /var/lib/kubelet/pods

pkg/backup/actions/csi/pvc_action.go

@@ -48,6 +48,7 @@ import (
 	veleroclient "github.com/vmware-tanzu/velero/pkg/client"
 	"github.com/vmware-tanzu/velero/pkg/kuberesource"
 	"github.com/vmware-tanzu/velero/pkg/label"
+	"github.com/vmware-tanzu/velero/pkg/nodeagent"
 	plugincommon "github.com/vmware-tanzu/velero/pkg/plugin/framework/common"
 	"github.com/vmware-tanzu/velero/pkg/plugin/utils/volumehelper"
 	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
@@ -55,6 +56,7 @@ import (
 	uploaderUtil "github.com/vmware-tanzu/velero/pkg/uploader/util"
 	"github.com/vmware-tanzu/velero/pkg/util/boolptr"
 	"github.com/vmware-tanzu/velero/pkg/util/csi"
+	datamover "github.com/vmware-tanzu/velero/pkg/util/datamover"
 	kubeutil "github.com/vmware-tanzu/velero/pkg/util/kube"
 	podvolumeutil "github.com/vmware-tanzu/velero/pkg/util/podvolume"
 	vhutil "github.com/vmware-tanzu/velero/pkg/util/volumehelper"
@@ -212,6 +214,7 @@ func (p *pvcBackupItemAction) validatePVCAndPV(
 func (p *pvcBackupItemAction) createVolumeSnapshot(
 	pvc corev1api.PersistentVolumeClaim,
 	backup *velerov1api.Backup,
+	policySnapshotClass string,
 ) (
 	vs *snapshotv1api.VolumeSnapshot,
 	err error,
@@ -232,6 +235,7 @@ func (p *pvcBackupItemAction) createVolumeSnapshot(
 		&pvc,
 		p.log,
 		p.crClient,
+		policySnapshotClass,
 	)
 	if err != nil {
 		return nil, errors.Wrapf(
@@ -338,7 +342,25 @@ func (p *pvcBackupItemAction) Execute(
 		return nil, nil, "", nil, err
 	}
 
-	vs, err := p.getVolumeSnapshotReference(context.TODO(), pvc, backup)
+	// validate that the node-agent daemonset is ready when snapshot data movement with
+	// the built-in data mover is requested. Without this, the DataUpload CR will be
+	// created but never processed (the DataUpload controller runs inside node-agent),
+	// causing the backup to hang until itemOperationTimeout expires.
+	if boolptr.IsSetToTrue(backup.Spec.SnapshotMoveData) && datamover.IsBuiltInDataMover(backup.Spec.DataMover) {
+		if err := nodeagent.IsReady(context.TODO(), backup.Namespace, p.crClient, p.log); err != nil {
+			p.log.WithError(err).Error("cannot perform snapshot data movement without running node-agent pods")
+			return nil, nil, "", nil, errors.Wrap(err, "CSI PVC BIA cannot proceed: node-agent is not ready for snapshot data movement")
+		}
+	}
+
+	policySnapshotClass, scErr := vh.GetSnapshotClass(item, kuberesource.PersistentVolumeClaims)
+	if scErr != nil {
+		p.log.WithError(scErr).Warn("failed to get snapshotClass from volume policy, proceeding without it")
+	} else if policySnapshotClass != "" {
+		p.log.Infof("Volume policy specifies snapshotClass=%s for PVC %s/%s", policySnapshotClass, pvc.Namespace, pvc.Name)
+	}
+
+	vs, err := p.getVolumeSnapshotReference(context.TODO(), pvc, backup, policySnapshotClass)
 	if err != nil {
 		return nil, nil, "", nil, err
 	}
@@ -678,6 +700,7 @@ func (p *pvcBackupItemAction) getVolumeSnapshotReference(
 	ctx context.Context,
 	pvc corev1api.PersistentVolumeClaim,
 	backup *velerov1api.Backup,
+	policySnapshotClass string,
 ) (*snapshotv1api.VolumeSnapshot, error) {
 	vgsLabelKey := backup.Spec.VolumeGroupSnapshotLabelKey
 	group, hasLabel := pvc.Labels[vgsLabelKey]
@@ -808,7 +831,7 @@ func (p *pvcBackupItemAction) getVolumeSnapshotReference(
 	}
 
 	// Legacy fallback: create individual VS
-	return p.createVolumeSnapshot(pvc, backup)
+	return p.createVolumeSnapshot(pvc, backup, policySnapshotClass)
 }
 
 func (p *pvcBackupItemAction) findExistingVSForBackup(

pkg/backup/actions/csi/pvc_action_test.go

@@ -31,6 +31,7 @@ import (
 	"github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	appsv1api "k8s.io/api/apps/v1"
 	corev1api "k8s.io/api/core/v1"
 	storagev1api "k8s.io/api/storage/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
@@ -79,21 +80,23 @@ func (c *errorInjectingClient) Create(ctx context.Context, obj crclient.Object,
 func TestExecute(t *testing.T) {
 	boolTrue := true
 	tests := []struct {
-		name               string
-		backup             *velerov1api.Backup
-		pvc                *corev1api.PersistentVolumeClaim
-		pv                 *corev1api.PersistentVolume
-		sc                 *storagev1api.StorageClass
-		vsClass            *snapshotv1api.VolumeSnapshotClass
-		operationID        string
-		expectedErr        error
-		expectErr          bool // Use bool for cases where we just need to check for any error
-		expectedBackup     *velerov1api.Backup
-		expectedDataUpload *velerov2alpha1.DataUpload
-		expectedPVC        *corev1api.PersistentVolumeClaim
-		resourcePolicy     *corev1api.ConfigMap
-		failVSCreate       bool
-		skipVSReadyUpdate  bool // New flag to control VS readiness
+		name                string
+		backup              *velerov1api.Backup
+		pvc                 *corev1api.PersistentVolumeClaim
+		pv                  *corev1api.PersistentVolume
+		sc                  *storagev1api.StorageClass
+		vsClass             *snapshotv1api.VolumeSnapshotClass
+		operationID         string
+		expectedErr         error
+		expectErr           bool // Use bool for cases where we just need to check for any error
+		expectedBackup      *velerov1api.Backup
+		expectedDataUpload  *velerov2alpha1.DataUpload
+		expectedPVC         *corev1api.PersistentVolumeClaim
+		resourcePolicy      *corev1api.ConfigMap
+		extraObjects        []runtime.Object
+		failVSCreate        bool
+		skipVSReadyUpdate   bool // New flag to control VS readiness
+		expectedVSClassName string
 	}{
 		{
 			name:   "Skip PVC BIA when backup is in finalizing phase",
@@ -120,12 +123,21 @@ func TestExecute(t *testing.T) {
 			expectErr:         true, // Expect an error, but the exact message can vary
 		},
 		{
-			name:        "Test SnapshotMoveData",
-			backup:      builder.ForBackup("velero", "test").SnapshotMoveData(true).CSISnapshotTimeout(1 * time.Minute).Result(),
-			pvc:         builder.ForPersistentVolumeClaim("velero", "testPVC").VolumeName("testPV").StorageClass("testSC").Phase(corev1api.ClaimBound).Result(),
-			pv:          builder.ForPersistentVolume("testPV").CSI("hostpath", "testVolume").Result(),
-			sc:          builder.ForStorageClass("testSC").Provisioner("hostpath").Result(),
-			vsClass:     builder.ForVolumeSnapshotClass("testVSClass").Driver("hostpath").ObjectMeta(builder.WithLabels(velerov1api.VolumeSnapshotClassSelectorLabel, "")).Result(),
+			name:    "Test SnapshotMoveData",
+			backup:  builder.ForBackup("velero", "test").SnapshotMoveData(true).CSISnapshotTimeout(1 * time.Minute).Result(),
+			pvc:     builder.ForPersistentVolumeClaim("velero", "testPVC").VolumeName("testPV").StorageClass("testSC").Phase(corev1api.ClaimBound).Result(),
+			pv:      builder.ForPersistentVolume("testPV").CSI("hostpath", "testVolume").Result(),
+			sc:      builder.ForStorageClass("testSC").Provisioner("hostpath").Result(),
+			vsClass: builder.ForVolumeSnapshotClass("testVSClass").Driver("hostpath").ObjectMeta(builder.WithLabels(velerov1api.VolumeSnapshotClassSelectorLabel, "")).Result(),
+			extraObjects: []runtime.Object{
+				&corev1api.Node{
+					ObjectMeta: metav1.ObjectMeta{Name: "linux-node", Labels: map[string]string{"kubernetes.io/os": "linux"}},
+				},
+				&appsv1api.DaemonSet{
+					ObjectMeta: metav1.ObjectMeta{Namespace: "velero", Name: "node-agent"},
+					Status:     appsv1api.DaemonSetStatus{NumberReady: 3},
+				},
+			},
 			operationID: ".",
 			expectedDataUpload: &velerov2alpha1.DataUpload{
 				TypeMeta: metav1.TypeMeta{
@@ -166,18 +178,37 @@ func TestExecute(t *testing.T) {
 			},
 		},
 		{
-			name:        "Verify PVC is modified as expected",
-			backup:      builder.ForBackup("velero", "test").SnapshotMoveData(true).CSISnapshotTimeout(1 * time.Minute).Result(),
-			pvc:         builder.ForPersistentVolumeClaim("velero", "testPVC").VolumeName("testPV").StorageClass("testSC").Phase(corev1api.ClaimBound).Result(),
-			pv:          builder.ForPersistentVolume("testPV").CSI("hostpath", "testVolume").Result(),
-			sc:          builder.ForStorageClass("testSC").Provisioner("hostpath").Result(),
-			vsClass:     builder.ForVolumeSnapshotClass("tescVSClass").Driver("hostpath").ObjectMeta(builder.WithLabels(velerov1api.VolumeSnapshotClassSelectorLabel, "")).Result(),
+			name:    "Verify PVC is modified as expected",
+			backup:  builder.ForBackup("velero", "test").SnapshotMoveData(true).CSISnapshotTimeout(1 * time.Minute).Result(),
+			pvc:     builder.ForPersistentVolumeClaim("velero", "testPVC").VolumeName("testPV").StorageClass("testSC").Phase(corev1api.ClaimBound).Result(),
+			pv:      builder.ForPersistentVolume("testPV").CSI("hostpath", "testVolume").Result(),
+			sc:      builder.ForStorageClass("testSC").Provisioner("hostpath").Result(),
+			vsClass: builder.ForVolumeSnapshotClass("tescVSClass").Driver("hostpath").ObjectMeta(builder.WithLabels(velerov1api.VolumeSnapshotClassSelectorLabel, "")).Result(),
+			extraObjects: []runtime.Object{
+				&corev1api.Node{
+					ObjectMeta: metav1.ObjectMeta{Name: "linux-node", Labels: map[string]string{"kubernetes.io/os": "linux"}},
+				},
+				&appsv1api.DaemonSet{
+					ObjectMeta: metav1.ObjectMeta{Namespace: "velero", Name: "node-agent"},
+					Status:     appsv1api.DaemonSetStatus{NumberReady: 3},
+				},
+			},
 			operationID: ".",
 			expectedPVC: builder.ForPersistentVolumeClaim("velero", "testPVC").
 				ObjectMeta(builder.WithAnnotations(velerov1api.MustIncludeAdditionalItemAnnotation, "true", velerov1api.DataUploadNameAnnotation, "velero/"),
 					builder.WithLabels(velerov1api.BackupNameLabel, "test")).
 				VolumeName("testPV").StorageClass("testSC").Phase(corev1api.ClaimBound).Result(),
 		},
+		{
+			name:              "Test SnapshotMoveData without node-agent",
+			backup:            builder.ForBackup("velero", "test").SnapshotMoveData(true).CSISnapshotTimeout(1 * time.Minute).Result(),
+			pvc:               builder.ForPersistentVolumeClaim("velero", "testPVC").VolumeName("testPV").StorageClass("testSC").Phase(corev1api.ClaimBound).Result(),
+			pv:                builder.ForPersistentVolume("testPV").CSI("hostpath", "testVolume").Result(),
+			sc:                builder.ForStorageClass("testSC").Provisioner("hostpath").Result(),
+			vsClass:           builder.ForVolumeSnapshotClass("testVSClass").Driver("hostpath").ObjectMeta(builder.WithLabels(velerov1api.VolumeSnapshotClassSelectorLabel, "")).Result(),
+			expectErr:         true,
+			skipVSReadyUpdate: true,
+		},
 		{
 			name:           "Test ResourcePolicy",
 			backup:         builder.ForBackup("velero", "test").ResourcePolicies("resourcePolicy").SnapshotVolumes(false).CSISnapshotTimeout(time.Duration(3600) * time.Second).Result(),
@@ -187,6 +218,16 @@ func TestExecute(t *testing.T) {
 			sc:             builder.ForStorageClass("testSC").Provisioner("hostpath").Result(),
 			vsClass:        builder.ForVolumeSnapshotClass("tescVSClass").Driver("hostpath").ObjectMeta(builder.WithLabels(velerov1api.VolumeSnapshotClassSelectorLabel, "")).Result(),
 		},
+		{
+			name:                "Volume policy with snapshotClass selects correct VolumeSnapshotClass",
+			backup:              builder.ForBackup("velero", "test").ResourcePolicies("resourcePolicy").CSISnapshotTimeout(time.Duration(3600) * time.Second).Result(),
+			resourcePolicy:      builder.ForConfigMap("velero", "resourcePolicy").Data("policy", `{"version":"v1","volumePolicies":[{"conditions":{"csi":{}},"action":{"type":"snapshot","parameters":{"snapshotClass":"policy-selected-vsclass"}}}]}`).Result(),
+			pvc:                 builder.ForPersistentVolumeClaim("velero", "testPVC").VolumeName("testPV").StorageClass("testSC").Phase(corev1api.ClaimBound).Result(),
+			pv:                  builder.ForPersistentVolume("testPV").CSI("hostpath", "testVolume").Result(),
+			sc:                  builder.ForStorageClass("testSC").Provisioner("hostpath").Result(),
+			vsClass:             builder.ForVolumeSnapshotClass("policy-selected-vsclass").Driver("hostpath").Result(),
+			expectedVSClassName: "policy-selected-vsclass",
+		},
 	}
 
 	for _, tc := range tests {
@@ -209,6 +250,7 @@ func TestExecute(t *testing.T) {
 			if tc.resourcePolicy != nil {
 				objects = append(objects, tc.resourcePolicy)
 			}
+			objects = append(objects, tc.extraObjects...)
 
 			var crClient crclient.Client
 			if tc.failVSCreate {
@@ -300,6 +342,15 @@ func TestExecute(t *testing.T) {
 				runtime.DefaultUnstructuredConverter.FromUnstructured(resultUnstructed.UnstructuredContent(), resultPVC)
 				require.True(t, cmp.Equal(tc.expectedPVC, resultPVC, cmpopts.IgnoreFields(corev1api.PersistentVolumeClaim{}, "ResourceVersion", "Annotations", "Labels")))
 			}
+
+			if tc.expectedVSClassName != "" {
+				vsList := new(snapshotv1api.VolumeSnapshotList)
+				require.NoError(t, crClient.List(t.Context(), vsList, &crclient.ListOptions{Namespace: tc.pvc.Namespace}))
+				require.NotEmpty(t, vsList.Items, "expected VolumeSnapshot to be created")
+				require.NotNil(t, vsList.Items[0].Spec.VolumeSnapshotClassName)
+				assert.Equal(t, tc.expectedVSClassName, *vsList.Items[0].Spec.VolumeSnapshotClassName,
+					"VolumeSnapshot should use the VolumeSnapshotClass specified by volume policy")
+			}
 		})
 	}
 }

pkg/cmd/cli/backup/create.go

@@ -33,6 +33,7 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/builder"
 	"github.com/vmware-tanzu/velero/pkg/client"
 	"github.com/vmware-tanzu/velero/pkg/cmd"
+	"github.com/vmware-tanzu/velero/pkg/cmd/cli"
 	"github.com/vmware-tanzu/velero/pkg/cmd/util/flag"
 	"github.com/vmware-tanzu/velero/pkg/cmd/util/output"
 	"github.com/vmware-tanzu/velero/pkg/util/collections"
@@ -76,6 +77,10 @@ func NewCreateCommand(f client.Factory, use string) *cobra.Command {
 	output.BindFlags(c.Flags())
 	output.ClearOutputFlagDefault(c)
 
+	_ = c.RegisterFlagCompletionFunc("from-schedule", cli.CompleteScheduleNames(f))
+	_ = c.RegisterFlagCompletionFunc("storage-location", cli.CompleteBackupStorageLocationNames(f))
+	_ = c.RegisterFlagCompletionFunc("volume-snapshot-locations", cli.CompleteVolumeSnapshotLocationNames(f))
+
 	return c
 }
 

pkg/cmd/cli/backup/delete.go

@@ -64,6 +64,7 @@ func NewDeleteCommand(f client.Factory, use string) *cobra.Command {
 		},
 	}
 
+	c.ValidArgsFunction = cli.CompleteBackupNames(f)
 	o.BindFlags(c.Flags())
 
 	return c

pkg/cmd/cli/backup/describe.go

@@ -29,6 +29,7 @@ import (
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/client"
 	"github.com/vmware-tanzu/velero/pkg/cmd"
+	"github.com/vmware-tanzu/velero/pkg/cmd/cli"
 	"github.com/vmware-tanzu/velero/pkg/cmd/util/output"
 	"github.com/vmware-tanzu/velero/pkg/label"
 )
@@ -112,6 +113,7 @@ func NewDescribeCommand(f client.Factory, use string) *cobra.Command {
 		},
 	}
 
+	c.ValidArgsFunction = cli.CompleteBackupNames(f)
 	c.Flags().StringVarP(&listOptions.LabelSelector, "selector", "l", listOptions.LabelSelector, "Only show items matching this label selector.")
 	c.Flags().BoolVar(&details, "details", details, "Display additional detail in the command output.")
 	c.Flags().BoolVar(&insecureSkipTLSVerify, "insecure-skip-tls-verify", insecureSkipTLSVerify, "If true, the object store's TLS certificate will not be checked for validity. This is insecure and susceptible to man-in-the-middle attacks. Not recommended for production.")

pkg/cmd/cli/backup/download.go

@@ -31,6 +31,7 @@ import (
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/client"
 	"github.com/vmware-tanzu/velero/pkg/cmd"
+	"github.com/vmware-tanzu/velero/pkg/cmd/cli"
 	"github.com/vmware-tanzu/velero/pkg/cmd/util/cacert"
 	"github.com/vmware-tanzu/velero/pkg/cmd/util/downloadrequest"
 )
@@ -55,6 +56,7 @@ func NewDownloadCommand(f client.Factory) *cobra.Command {
 		},
 	}
 
+	c.ValidArgsFunction = cli.CompleteBackupNames(f)
 	o.BindFlags(c.Flags())
 
 	return c

pkg/cmd/cli/backup/get.go

@@ -27,6 +27,7 @@ import (
 	api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/client"
 	"github.com/vmware-tanzu/velero/pkg/cmd"
+	"github.com/vmware-tanzu/velero/pkg/cmd/cli"
 	"github.com/vmware-tanzu/velero/pkg/cmd/util/output"
 )
 
@@ -66,6 +67,7 @@ func NewGetCommand(f client.Factory, use string) *cobra.Command {
 		},
 	}
 
+	c.ValidArgsFunction = cli.CompleteBackupNames(f)
 	c.Flags().StringVarP(&listOptions.LabelSelector, "selector", "l", listOptions.LabelSelector, "Only show items matching this label selector")
 
 	output.BindFlags(c.Flags())

pkg/cmd/cli/backup/logs.go

@@ -30,6 +30,7 @@ import (
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/client"
 	"github.com/vmware-tanzu/velero/pkg/cmd"
+	"github.com/vmware-tanzu/velero/pkg/cmd/cli"
 	"github.com/vmware-tanzu/velero/pkg/cmd/util/cacert"
 	"github.com/vmware-tanzu/velero/pkg/cmd/util/downloadrequest"
 )
@@ -119,6 +120,7 @@ func NewLogsCommand(f client.Factory) *cobra.Command {
 		},
 	}
 
+	c.ValidArgsFunction = cli.CompleteBackupNames(f)
 	l.BindFlags(c.Flags())
 
 	return c

pkg/cmd/cli/backuplocation/delete.go

@@ -62,6 +62,7 @@ func NewDeleteCommand(f client.Factory, use string) *cobra.Command {
 		},
 	}
 
+	c.ValidArgsFunction = cli.CompleteBackupStorageLocationNames(f)
 	o.BindFlags(c.Flags())
 	return c
 }

pkg/cmd/cli/backuplocation/get.go

@@ -27,6 +27,7 @@ import (
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/client"
 	"github.com/vmware-tanzu/velero/pkg/cmd"
+	"github.com/vmware-tanzu/velero/pkg/cmd/cli"
 	"github.com/vmware-tanzu/velero/pkg/cmd/util/output"
 )
 
@@ -89,6 +90,7 @@ func NewGetCommand(f client.Factory, use string) *cobra.Command {
 		},
 	}
 
+	c.ValidArgsFunction = cli.CompleteBackupStorageLocationNames(f)
 	c.Flags().BoolVar(&showDefaultOnly, "default", false, "Displays the current default backup storage location.")
 	c.Flags().StringVarP(&listOptions.LabelSelector, "selector", "l", listOptions.LabelSelector, "Only show items matching this label selector.")
 

pkg/cmd/cli/backuplocation/set.go

@@ -33,6 +33,7 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/builder"
 	"github.com/vmware-tanzu/velero/pkg/client"
 	"github.com/vmware-tanzu/velero/pkg/cmd"
+	"github.com/vmware-tanzu/velero/pkg/cmd/cli"
 	"github.com/vmware-tanzu/velero/pkg/cmd/util/flag"
 	"github.com/vmware-tanzu/velero/pkg/util/boolptr"
 )
@@ -51,6 +52,7 @@ func NewSetCommand(f client.Factory, use string) *cobra.Command {
 		},
 	}
 
+	c.ValidArgsFunction = cli.CompleteBackupStorageLocationNames(f)
 	o.BindFlags(c.Flags())
 
 	return c

pkg/cmd/cli/completion_functions.go

@@ -0,0 +1,105 @@
+/*
+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 cli
+
+import (
+	"context"
+	"strings"
+	"time"
+
+	"github.com/spf13/cobra"
+	"k8s.io/apimachinery/pkg/api/meta"
+	kbclient "sigs.k8s.io/controller-runtime/pkg/client"
+
+	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	"github.com/vmware-tanzu/velero/pkg/client"
+)
+
+// completionFunc is the function signature for cobra's ValidArgsFunction.
+type completionFunc = func(cmd *cobra.Command, args []string, toComplete string) ([]string, cobra.ShellCompDirective)
+
+// completeNames builds a completion function for any Velero list type.
+// It extracts resource names via apimachinery's meta helpers.
+func completeNames(f client.Factory, list kbclient.ObjectList) completionFunc {
+	return func(cmd *cobra.Command, args []string, toComplete string) ([]string, cobra.ShellCompDirective) {
+		kbClient, err := f.KubebuilderClient()
+		if err != nil {
+			return nil, cobra.ShellCompDirectiveNoFileComp
+		}
+		parentCtx := context.Background()
+		if cmd != nil && cmd.Context() != nil {
+			parentCtx = cmd.Context()
+		}
+		ctx, cancel := context.WithTimeout(parentCtx, 3*time.Second)
+		defer cancel()
+		freshObject := list.DeepCopyObject()
+		freshList, ok := freshObject.(kbclient.ObjectList)
+		if !ok {
+			return nil, cobra.ShellCompDirectiveNoFileComp
+		}
+		if err := kbClient.List(ctx, freshList, &kbclient.ListOptions{Namespace: f.Namespace()}); err != nil {
+			return nil, cobra.ShellCompDirectiveNoFileComp
+		}
+		items, err := meta.ExtractList(freshList)
+		if err != nil {
+			return nil, cobra.ShellCompDirectiveNoFileComp
+		}
+		seen := make(map[string]bool, len(args))
+		for _, a := range args {
+			seen[a] = true
+		}
+		var filtered []string
+		for _, item := range items {
+			accessor, err := meta.Accessor(item)
+			if err != nil {
+				continue
+			}
+			name := accessor.GetName()
+			if seen[name] {
+				continue
+			}
+			if strings.HasPrefix(name, toComplete) {
+				filtered = append(filtered, name)
+			}
+		}
+		return filtered, cobra.ShellCompDirectiveNoFileComp
+	}
+}
+
+func CompleteBackupNames(f client.Factory) completionFunc {
+	return completeNames(f, &velerov1api.BackupList{})
+}
+
+func CompleteRestoreNames(f client.Factory) completionFunc {
+	return completeNames(f, &velerov1api.RestoreList{})
+}
+
+func CompleteScheduleNames(f client.Factory) completionFunc {
+	return completeNames(f, &velerov1api.ScheduleList{})
+}
+
+func CompleteBackupStorageLocationNames(f client.Factory) completionFunc {
+	return completeNames(f, &velerov1api.BackupStorageLocationList{})
+}
+
+func CompleteVolumeSnapshotLocationNames(f client.Factory) completionFunc {
+	return completeNames(f, &velerov1api.VolumeSnapshotLocationList{})
+}
+
+func CompleteBackupRepositoryNames(f client.Factory) completionFunc {
+	return completeNames(f, &velerov1api.BackupRepositoryList{})
+}

pkg/cmd/cli/completion_functions_test.go

@@ -0,0 +1,212 @@
+/*
+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 cli
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/spf13/cobra"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	kbclient "sigs.k8s.io/controller-runtime/pkg/client"
+
+	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	factorymocks "github.com/vmware-tanzu/velero/pkg/client/mocks"
+	velerotest "github.com/vmware-tanzu/velero/pkg/test"
+)
+
+// TestCompleteNames exercises the core completeNames helper with various list
+// types, prefix filters, and edge cases (empty cluster, no match).
+func TestCompleteNames(t *testing.T) {
+	tests := []struct {
+		name       string
+		objects    []runtime.Object
+		list       kbclient.ObjectList
+		args       []string
+		toComplete string
+		want       []string
+	}{
+		{
+			name:       "no resources returns nil",
+			objects:    nil,
+			list:       &velerov1api.BackupList{},
+			toComplete: "",
+			want:       nil,
+		},
+		{
+			name: "returns all matching names",
+			objects: []runtime.Object{
+				&velerov1api.Backup{ObjectMeta: metav1.ObjectMeta{Name: "daily", Namespace: "velero"}},
+				&velerov1api.Backup{ObjectMeta: metav1.ObjectMeta{Name: "weekly", Namespace: "velero"}},
+			},
+			list:       &velerov1api.BackupList{},
+			toComplete: "",
+			want:       []string{"daily", "weekly"},
+		},
+		{
+			name: "filters by prefix",
+			objects: []runtime.Object{
+				&velerov1api.Backup{ObjectMeta: metav1.ObjectMeta{Name: "daily", Namespace: "velero"}},
+				&velerov1api.Backup{ObjectMeta: metav1.ObjectMeta{Name: "weekly", Namespace: "velero"}},
+				&velerov1api.Backup{ObjectMeta: metav1.ObjectMeta{Name: "daily-full", Namespace: "velero"}},
+			},
+			list:       &velerov1api.BackupList{},
+			toComplete: "dai",
+			want:       []string{"daily", "daily-full"},
+		},
+		{
+			name: "no prefix match returns nil",
+			objects: []runtime.Object{
+				&velerov1api.Backup{ObjectMeta: metav1.ObjectMeta{Name: "daily", Namespace: "velero"}},
+			},
+			list:       &velerov1api.BackupList{},
+			toComplete: "xyz",
+			want:       nil,
+		},
+		{
+			name: "works with RestoreList",
+			objects: []runtime.Object{
+				&velerov1api.Restore{ObjectMeta: metav1.ObjectMeta{Name: "restore-1", Namespace: "velero"}},
+				&velerov1api.Restore{ObjectMeta: metav1.ObjectMeta{Name: "restore-2", Namespace: "velero"}},
+			},
+			list:       &velerov1api.RestoreList{},
+			toComplete: "restore-",
+			want:       []string{"restore-1", "restore-2"},
+		},
+		{
+			name: "works with ScheduleList",
+			objects: []runtime.Object{
+				&velerov1api.Schedule{ObjectMeta: metav1.ObjectMeta{Name: "nightly", Namespace: "velero"}},
+			},
+			list:       &velerov1api.ScheduleList{},
+			toComplete: "",
+			want:       []string{"nightly"},
+		},
+		{
+			name: "works with BackupStorageLocationList",
+			objects: []runtime.Object{
+				&velerov1api.BackupStorageLocation{ObjectMeta: metav1.ObjectMeta{Name: "default", Namespace: "velero"}},
+				&velerov1api.BackupStorageLocation{ObjectMeta: metav1.ObjectMeta{Name: "secondary", Namespace: "velero"}},
+			},
+			list:       &velerov1api.BackupStorageLocationList{},
+			toComplete: "s",
+			want:       []string{"secondary"},
+		},
+		{
+			name: "works with VolumeSnapshotLocationList",
+			objects: []runtime.Object{
+				&velerov1api.VolumeSnapshotLocation{ObjectMeta: metav1.ObjectMeta{Name: "aws-snap", Namespace: "velero"}},
+			},
+			list:       &velerov1api.VolumeSnapshotLocationList{},
+			toComplete: "",
+			want:       []string{"aws-snap"},
+		},
+		{
+			name: "works with BackupRepositoryList",
+			objects: []runtime.Object{
+				&velerov1api.BackupRepository{ObjectMeta: metav1.ObjectMeta{Name: "repo-1", Namespace: "velero"}},
+			},
+			list:       &velerov1api.BackupRepositoryList{},
+			toComplete: "",
+			want:       []string{"repo-1"},
+		},
+		{
+			name: "excludes already-typed args",
+			objects: []runtime.Object{
+				&velerov1api.Backup{ObjectMeta: metav1.ObjectMeta{Name: "daily", Namespace: "velero"}},
+				&velerov1api.Backup{ObjectMeta: metav1.ObjectMeta{Name: "weekly", Namespace: "velero"}},
+				&velerov1api.Backup{ObjectMeta: metav1.ObjectMeta{Name: "monthly", Namespace: "velero"}},
+			},
+			list:       &velerov1api.BackupList{},
+			args:       []string{"daily", "monthly"},
+			toComplete: "",
+			want:       []string{"weekly"},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			kbClient := velerotest.NewFakeControllerRuntimeClient(t, tc.objects...)
+
+			f := new(factorymocks.Factory)
+			f.On("KubebuilderClient").Return(kbClient, nil)
+			f.On("Namespace").Return("velero")
+
+			completionFn := completeNames(f, tc.list)
+			got, directive := completionFn(&cobra.Command{}, tc.args, tc.toComplete)
+
+			assert.Equal(t, cobra.ShellCompDirectiveNoFileComp, directive)
+			assert.ElementsMatch(t, tc.want, got)
+		})
+	}
+}
+
+// TestCompleteNames_KubebuilderClientError verifies that a factory error
+// (e.g. no kubeconfig) returns nil completions instead of panicking.
+func TestCompleteNames_KubebuilderClientError(t *testing.T) {
+	f := new(factorymocks.Factory)
+	f.On("KubebuilderClient").Return(nil, fmt.Errorf("connection refused"))
+
+	completionFn := completeNames(f, &velerov1api.BackupList{})
+	got, directive := completionFn(&cobra.Command{}, nil, "")
+
+	assert.Equal(t, cobra.ShellCompDirectiveNoFileComp, directive)
+	assert.Nil(t, got)
+}
+
+// TestCompleteWrappers verifies each exported Complete*Names wrapper returns
+// only its own resource type. A single fake client holds one object of every
+// type, so each wrapper must filter correctly and not leak other kinds.
+func TestCompleteWrappers(t *testing.T) {
+	objects := []runtime.Object{
+		&velerov1api.Backup{ObjectMeta: metav1.ObjectMeta{Name: "b1", Namespace: "velero"}},
+		&velerov1api.Restore{ObjectMeta: metav1.ObjectMeta{Name: "r1", Namespace: "velero"}},
+		&velerov1api.Schedule{ObjectMeta: metav1.ObjectMeta{Name: "s1", Namespace: "velero"}},
+		&velerov1api.BackupStorageLocation{ObjectMeta: metav1.ObjectMeta{Name: "bsl1", Namespace: "velero"}},
+		&velerov1api.VolumeSnapshotLocation{ObjectMeta: metav1.ObjectMeta{Name: "vsl1", Namespace: "velero"}},
+		&velerov1api.BackupRepository{ObjectMeta: metav1.ObjectMeta{Name: "br1", Namespace: "velero"}},
+	}
+	kbClient := velerotest.NewFakeControllerRuntimeClient(t, objects...)
+
+	f := new(factorymocks.Factory)
+	f.On("KubebuilderClient").Return(kbClient, nil)
+	f.On("Namespace").Return("velero")
+
+	tests := []struct {
+		name     string
+		fn       completionFunc
+		expected []string
+	}{
+		{"CompleteBackupNames", CompleteBackupNames(f), []string{"b1"}},
+		{"CompleteRestoreNames", CompleteRestoreNames(f), []string{"r1"}},
+		{"CompleteScheduleNames", CompleteScheduleNames(f), []string{"s1"}},
+		{"CompleteBackupStorageLocationNames", CompleteBackupStorageLocationNames(f), []string{"bsl1"}},
+		{"CompleteVolumeSnapshotLocationNames", CompleteVolumeSnapshotLocationNames(f), []string{"vsl1"}},
+		{"CompleteBackupRepositoryNames", CompleteBackupRepositoryNames(f), []string{"br1"}},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			got, directive := tc.fn(&cobra.Command{}, nil, "")
+			require.Equal(t, cobra.ShellCompDirectiveNoFileComp, directive)
+			assert.Equal(t, tc.expected, got)
+		})
+	}
+}

pkg/cmd/cli/debug/debug.go

@@ -38,6 +38,7 @@ import (
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/client"
 	"github.com/vmware-tanzu/velero/pkg/cmd"
+	"github.com/vmware-tanzu/velero/pkg/cmd/cli"
 )
 
 //go:embed cshd-scripts/velero.cshd
@@ -171,6 +172,10 @@ specs of resources created by velero server, and optionally the logs of backup a
 		},
 	}
 	o.bindFlags(c.Flags())
+
+	_ = c.RegisterFlagCompletionFunc("backup", cli.CompleteBackupNames(f))
+	_ = c.RegisterFlagCompletionFunc("restore", cli.CompleteRestoreNames(f))
+
 	return c
 }
 

pkg/cmd/cli/repo/get.go

@@ -27,6 +27,7 @@ import (
 	api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/client"
 	"github.com/vmware-tanzu/velero/pkg/cmd"
+	"github.com/vmware-tanzu/velero/pkg/cmd/cli"
 	"github.com/vmware-tanzu/velero/pkg/cmd/util/output"
 )
 
@@ -66,6 +67,7 @@ func NewGetCommand(f client.Factory, use string) *cobra.Command {
 		},
 	}
 
+	c.ValidArgsFunction = cli.CompleteBackupRepositoryNames(f)
 	c.Flags().StringVarP(&listOptions.LabelSelector, "selector", "l", listOptions.LabelSelector, "Only show items matching this label selector.")
 
 	output.BindFlags(c.Flags())

pkg/cmd/cli/restore/create.go

@@ -36,6 +36,7 @@ import (
 	api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/client"
 	"github.com/vmware-tanzu/velero/pkg/cmd"
+	"github.com/vmware-tanzu/velero/pkg/cmd/cli"
 	"github.com/vmware-tanzu/velero/pkg/cmd/util/flag"
 	"github.com/vmware-tanzu/velero/pkg/cmd/util/output"
 	"github.com/vmware-tanzu/velero/pkg/util/boolptr"
@@ -81,6 +82,9 @@ Notes:
 	output.BindFlags(c.Flags())
 	output.ClearOutputFlagDefault(c)
 
+	_ = c.RegisterFlagCompletionFunc("from-backup", cli.CompleteBackupNames(f))
+	_ = c.RegisterFlagCompletionFunc("from-schedule", cli.CompleteScheduleNames(f))
+
 	return c
 }
 

pkg/cmd/cli/restore/delete.go

@@ -61,6 +61,7 @@ func NewDeleteCommand(f client.Factory, use string) *cobra.Command {
 			cmd.CheckError(Run(o))
 		},
 	}
+	c.ValidArgsFunction = cli.CompleteRestoreNames(f)
 	o.BindFlags(c.Flags())
 	return c
 }

pkg/cmd/cli/restore/describe.go

@@ -29,6 +29,7 @@ import (
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/client"
 	"github.com/vmware-tanzu/velero/pkg/cmd"
+	"github.com/vmware-tanzu/velero/pkg/cmd/cli"
 	"github.com/vmware-tanzu/velero/pkg/cmd/util/output"
 	"github.com/vmware-tanzu/velero/pkg/label"
 )
@@ -92,6 +93,7 @@ func NewDescribeCommand(f client.Factory, use string) *cobra.Command {
 		},
 	}
 
+	c.ValidArgsFunction = cli.CompleteRestoreNames(f)
 	c.Flags().StringVarP(&listOptions.LabelSelector, "selector", "l", listOptions.LabelSelector, "Only show items matching this label selector.")
 	c.Flags().BoolVar(&details, "details", details, "Display additional detail in the command output.")
 	c.Flags().BoolVar(&insecureSkipTLSVerify, "insecure-skip-tls-verify", insecureSkipTLSVerify, "If true, the object store's TLS certificate will not be checked for validity. This is insecure and susceptible to man-in-the-middle attacks. Not recommended for production.")

pkg/cmd/cli/restore/get.go

@@ -27,6 +27,7 @@ import (
 	api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/client"
 	"github.com/vmware-tanzu/velero/pkg/cmd"
+	"github.com/vmware-tanzu/velero/pkg/cmd/cli"
 	"github.com/vmware-tanzu/velero/pkg/cmd/util/output"
 )
 
@@ -76,6 +77,7 @@ func NewGetCommand(f client.Factory, use string) *cobra.Command {
 		},
 	}
 
+	c.ValidArgsFunction = cli.CompleteRestoreNames(f)
 	c.Flags().StringVarP(&listOptions.LabelSelector, "selector", "l", listOptions.LabelSelector, "Only show items matching this label selector.")
 
 	output.BindFlags(c.Flags())

pkg/cmd/cli/restore/logs.go

@@ -29,6 +29,7 @@ import (
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/client"
 	"github.com/vmware-tanzu/velero/pkg/cmd"
+	"github.com/vmware-tanzu/velero/pkg/cmd/cli"
 	"github.com/vmware-tanzu/velero/pkg/cmd/util/cacert"
 	"github.com/vmware-tanzu/velero/pkg/cmd/util/downloadrequest"
 )
@@ -82,6 +83,7 @@ func NewLogsCommand(f client.Factory) *cobra.Command {
 		},
 	}
 
+	c.ValidArgsFunction = cli.CompleteRestoreNames(f)
 	c.Flags().DurationVar(&timeout, "timeout", timeout, "How long to wait to receive logs.")
 	c.Flags().BoolVar(&insecureSkipTLSVerify, "insecure-skip-tls-verify", insecureSkipTLSVerify, "If true, the object store's TLS certificate will not be checked for validity. This is insecure and susceptible to man-in-the-middle attacks. Not recommended for production.")
 	c.Flags().StringVar(&caCertFile, "cacert", caCertFile, "Path to a certificate bundle to use when verifying TLS connections. If not specified, the CA certificate from the BackupStorageLocation will be used if available.")

pkg/cmd/cli/schedule/create.go

@@ -30,6 +30,7 @@ import (
 	api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/client"
 	"github.com/vmware-tanzu/velero/pkg/cmd"
+	"github.com/vmware-tanzu/velero/pkg/cmd/cli"
 	"github.com/vmware-tanzu/velero/pkg/cmd/cli/backup"
 	"github.com/vmware-tanzu/velero/pkg/cmd/util/output"
 )
@@ -77,6 +78,9 @@ example: "@every 2h30m".`,
 	output.BindFlags(c.Flags())
 	output.ClearOutputFlagDefault(c)
 
+	_ = c.RegisterFlagCompletionFunc("storage-location", cli.CompleteBackupStorageLocationNames(f))
+	_ = c.RegisterFlagCompletionFunc("volume-snapshot-locations", cli.CompleteVolumeSnapshotLocationNames(f))
+
 	return c
 }
 

pkg/cmd/cli/schedule/delete.go

@@ -62,6 +62,7 @@ func NewDeleteCommand(f client.Factory, use string) *cobra.Command {
 		},
 	}
 
+	c.ValidArgsFunction = cli.CompleteScheduleNames(f)
 	o.BindFlags(c.Flags())
 	return c
 }

pkg/cmd/cli/schedule/describe.go

@@ -28,6 +28,7 @@ import (
 	v1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/client"
 	"github.com/vmware-tanzu/velero/pkg/cmd"
+	"github.com/vmware-tanzu/velero/pkg/cmd/cli"
 	"github.com/vmware-tanzu/velero/pkg/cmd/util/output"
 )
 
@@ -73,6 +74,7 @@ func NewDescribeCommand(f client.Factory, use string) *cobra.Command {
 		},
 	}
 
+	c.ValidArgsFunction = cli.CompleteScheduleNames(f)
 	c.Flags().StringVarP(&listOptions.LabelSelector, "selector", "l", listOptions.LabelSelector, "Only show items matching this label selector.")
 
 	return c

pkg/cmd/cli/schedule/get.go

@@ -27,6 +27,7 @@ import (
 	api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/client"
 	"github.com/vmware-tanzu/velero/pkg/cmd"
+	"github.com/vmware-tanzu/velero/pkg/cmd/cli"
 	"github.com/vmware-tanzu/velero/pkg/cmd/util/output"
 )
 
@@ -71,6 +72,7 @@ func NewGetCommand(f client.Factory, use string) *cobra.Command {
 		},
 	}
 
+	c.ValidArgsFunction = cli.CompleteScheduleNames(f)
 	c.Flags().StringVarP(&listOptions.LabelSelector, "selector", "l", listOptions.LabelSelector, "Only show items matching this label selector.")
 
 	output.BindFlags(c.Flags())

pkg/cmd/cli/schedule/pause.go

@@ -60,6 +60,7 @@ func NewPauseCommand(f client.Factory, use string) *cobra.Command {
 		},
 	}
 
+	c.ValidArgsFunction = cli.CompleteScheduleNames(f)
 	o.BindFlags(c.Flags())
 	pauseOpts.BindFlags(c.Flags())
 

pkg/cmd/cli/schedule/unpause.go

@@ -49,6 +49,7 @@ func NewUnpauseCommand(f client.Factory, use string) *cobra.Command {
 		},
 	}
 
+	c.ValidArgsFunction = cli.CompleteScheduleNames(f)
 	o.BindFlags(c.Flags())
 	pauseOpts.BindFlags(c.Flags())
 

pkg/cmd/cli/snapshotlocation/get.go

@@ -26,6 +26,7 @@ import (
 	api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/client"
 	"github.com/vmware-tanzu/velero/pkg/cmd"
+	"github.com/vmware-tanzu/velero/pkg/cmd/cli"
 	"github.com/vmware-tanzu/velero/pkg/cmd/util/output"
 )
 
@@ -56,6 +57,7 @@ func NewGetCommand(f client.Factory, use string) *cobra.Command {
 			cmd.CheckError(err)
 		},
 	}
+	c.ValidArgsFunction = cli.CompleteVolumeSnapshotLocationNames(f)
 	c.Flags().StringVarP(&listOptions.LabelSelector, "selector", "l", listOptions.LabelSelector, "Only show items matching this label selector")
 	output.BindFlags(c.Flags())
 	return c

pkg/cmd/cli/snapshotlocation/set.go

@@ -30,6 +30,7 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/builder"
 	"github.com/vmware-tanzu/velero/pkg/client"
 	"github.com/vmware-tanzu/velero/pkg/cmd"
+	"github.com/vmware-tanzu/velero/pkg/cmd/cli"
 	"github.com/vmware-tanzu/velero/pkg/cmd/util/flag"
 	"github.com/vmware-tanzu/velero/pkg/cmd/util/output"
 )
@@ -48,6 +49,7 @@ func NewSetCommand(f client.Factory, use string) *cobra.Command {
 		},
 	}
 
+	c.ValidArgsFunction = cli.CompleteVolumeSnapshotLocationNames(f)
 	o.BindFlags(c.Flags())
 	return c
 }

pkg/nodeagent/node_agent.go

@@ -22,6 +22,8 @@ import (
 	"fmt"
 
 	"github.com/cockroachdb/errors"
+	"github.com/sirupsen/logrus"
+	appsv1api "k8s.io/api/apps/v1"
 	corev1api "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -80,6 +82,34 @@ func KbClientIsRunningInNode(ctx context.Context, namespace string, nodeName str
 	return isRunningInNode(ctx, namespace, nodeName, nil, kubeClient)
 }
 
+// IsReady checks whether the node-agent daemonset has at least one ready pod
+// by inspecting the DaemonSet status. It only checks the daemonset for node
+// OS types that are present in the cluster, following the same pattern as
+// server.checkNodeAgent.
+func IsReady(ctx context.Context, namespace string, crClient ctrlclient.Client, log logrus.FieldLogger) error {
+	if kube.WithLinuxNode(ctx, crClient, log) {
+		ds := new(appsv1api.DaemonSet)
+		if err := crClient.Get(ctx, ctrlclient.ObjectKey{Namespace: namespace, Name: daemonSet}, ds); err != nil {
+			return errors.Wrap(err, "failed to get linux node-agent daemonset")
+		}
+		if ds.Status.NumberReady > 0 {
+			return nil
+		}
+	}
+
+	if kube.WithWindowsNode(ctx, crClient, log) {
+		ds := new(appsv1api.DaemonSet)
+		if err := crClient.Get(ctx, ctrlclient.ObjectKey{Namespace: namespace, Name: daemonsetWindows}, ds); err != nil {
+			return errors.Wrap(err, "failed to get windows node-agent daemonset")
+		}
+		if ds.Status.NumberReady > 0 {
+			return nil
+		}
+	}
+
+	return errors.New("node-agent is not ready: no ready pods found")
+}
+
 // IsRunningInNode checks if the node agent pod is running properly in a specified node through controller client. If not, return the error found
 func IsRunningInNode(ctx context.Context, namespace string, nodeName string, crClient ctrlclient.Client) error {
 	return isRunningInNode(ctx, namespace, nodeName, crClient, nil)

pkg/nodeagent/node_agent_test.go

@@ -20,6 +20,7 @@ import (
 	"testing"
 
 	"github.com/cockroachdb/errors"
+	"github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	appsv1api "k8s.io/api/apps/v1"
@@ -213,6 +214,146 @@ func TestIsRunningInNode(t *testing.T) {
 	}
 }
 
+func TestIsReady(t *testing.T) {
+	scheme := runtime.NewScheme()
+	appsv1api.AddToScheme(scheme)
+	corev1api.AddToScheme(scheme)
+
+	log := logrus.New()
+
+	linuxNode := &corev1api.Node{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:   "linux-node",
+			Labels: map[string]string{kube.NodeOSLabel: kube.NodeOSLinux},
+		},
+	}
+	windowsNode := &corev1api.Node{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:   "windows-node",
+			Labels: map[string]string{kube.NodeOSLabel: kube.NodeOSWindows},
+		},
+	}
+
+	dsLinuxNotReady := &appsv1api.DaemonSet{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "fake-ns", Name: "node-agent"},
+		Status:     appsv1api.DaemonSetStatus{NumberReady: 0},
+	}
+	dsLinuxReady := &appsv1api.DaemonSet{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "fake-ns", Name: "node-agent"},
+		Status:     appsv1api.DaemonSetStatus{NumberReady: 3},
+	}
+	dsWindowsNotReady := &appsv1api.DaemonSet{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "fake-ns", Name: "node-agent-windows"},
+		Status:     appsv1api.DaemonSetStatus{NumberReady: 0},
+	}
+	dsWindowsReady := &appsv1api.DaemonSet{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "fake-ns", Name: "node-agent-windows"},
+		Status:     appsv1api.DaemonSetStatus{NumberReady: 2},
+	}
+
+	tests := []struct {
+		name          string
+		kubeClientObj []runtime.Object
+		namespace     string
+		expectErr     string
+	}{
+		{
+			name:      "no nodes in cluster",
+			namespace: "fake-ns",
+			expectErr: "node-agent is not ready: no ready pods found",
+		},
+		{
+			name:      "linux node exists but daemonset not found",
+			namespace: "fake-ns",
+			kubeClientObj: []runtime.Object{
+				linuxNode,
+			},
+			expectErr: "failed to get linux node-agent daemonset",
+		},
+		{
+			name:      "linux node and daemonset exist but no ready pods",
+			namespace: "fake-ns",
+			kubeClientObj: []runtime.Object{
+				linuxNode,
+				dsLinuxNotReady,
+			},
+			expectErr: "node-agent is not ready: no ready pods found",
+		},
+		{
+			name:      "linux node and daemonset with ready pods",
+			namespace: "fake-ns",
+			kubeClientObj: []runtime.Object{
+				linuxNode,
+				dsLinuxReady,
+			},
+		},
+		{
+			name:      "windows node and daemonset with ready pods",
+			namespace: "fake-ns",
+			kubeClientObj: []runtime.Object{
+				windowsNode,
+				dsWindowsReady,
+			},
+		},
+		{
+			name:      "windows node and daemonset with no ready pods",
+			namespace: "fake-ns",
+			kubeClientObj: []runtime.Object{
+				windowsNode,
+				dsWindowsNotReady,
+			},
+			expectErr: "node-agent is not ready: no ready pods found",
+		},
+		{
+			name:      "both node types with both daemonsets ready",
+			namespace: "fake-ns",
+			kubeClientObj: []runtime.Object{
+				linuxNode,
+				windowsNode,
+				dsLinuxReady,
+				dsWindowsReady,
+			},
+		},
+		{
+			name:      "both node types but neither daemonset has ready pods",
+			namespace: "fake-ns",
+			kubeClientObj: []runtime.Object{
+				linuxNode,
+				windowsNode,
+				dsLinuxNotReady,
+				dsWindowsNotReady,
+			},
+			expectErr: "node-agent is not ready: no ready pods found",
+		},
+		{
+			name:      "linux not ready but windows ready",
+			namespace: "fake-ns",
+			kubeClientObj: []runtime.Object{
+				linuxNode,
+				windowsNode,
+				dsLinuxNotReady,
+				dsWindowsReady,
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			fakeClient := clientFake.NewClientBuilder().
+				WithScheme(scheme).
+				WithRuntimeObjects(test.kubeClientObj...).
+				Build()
+
+			err := IsReady(t.Context(), test.namespace, fakeClient, log)
+			if test.expectErr == "" {
+				assert.NoError(t, err)
+			} else {
+				assert.ErrorContains(t, err, test.expectErr)
+			}
+		})
+	}
+}
+
 func TestGetPodSpec(t *testing.T) {
 	podSpec := corev1api.PodSpec{
 		NodeName: "fake-node",

pkg/util/csi/volume_snapshot.go

@@ -314,6 +314,7 @@ func GetVolumeSnapshotClass(
 	pvc *corev1api.PersistentVolumeClaim,
 	log logrus.FieldLogger,
 	crClient crclient.Client,
+	policySnapshotClass string,
 ) (*snapshotv1api.VolumeSnapshotClass, error) {
 	snapshotClasses := new(snapshotv1api.VolumeSnapshotClassList)
 	err := crClient.List(context.TODO(), snapshotClasses)
@@ -331,6 +332,16 @@ func GetVolumeSnapshotClass(
 		return snapshotClass, nil
 	}
 
+	// If a snapshot class is specified by volume policy, use that
+	snapshotClass, err = GetVolumeSnapshotClassFromVolumePolicy(
+		policySnapshotClass, provisioner, snapshotClasses)
+	if err != nil {
+		log.Debugf("Didn't find VolumeSnapshotClass from volume policy: %v", err)
+	}
+	if snapshotClass != nil {
+		return snapshotClass, nil
+	}
+
 	// If there is no annotation in PVC, attempt to fetch it from backup annotations
 	snapshotClass, err = GetVolumeSnapshotClassFromBackupAnnotationsForDriver(
 		backup, provisioner, snapshotClasses)
@@ -412,6 +423,34 @@ func GetVolumeSnapshotClassFromBackupAnnotationsForDriver(
 	)
 }
 
+// GetVolumeSnapshotClassFromVolumePolicy returns a VolumeSnapshotClass
+// specified by a volume policy's snapshotClass parameter. If
+// policySnapshotClass is empty, it returns nil (no match).
+func GetVolumeSnapshotClassFromVolumePolicy(
+	policySnapshotClass string,
+	provisioner string,
+	snapshotClasses *snapshotv1api.VolumeSnapshotClassList,
+) (*snapshotv1api.VolumeSnapshotClass, error) {
+	if policySnapshotClass == "" {
+		return nil, nil
+	}
+	for _, sc := range snapshotClasses.Items {
+		if strings.EqualFold(policySnapshotClass, sc.ObjectMeta.Name) {
+			if !strings.EqualFold(sc.Driver, provisioner) {
+				return nil, errors.Errorf(
+					"VolumeSnapshotClass %s specified by volume policy is not for driver %s",
+					sc.ObjectMeta.Name, provisioner,
+				)
+			}
+			return &sc, nil
+		}
+	}
+	return nil, errors.Errorf(
+		"No CSI VolumeSnapshotClass found with name %s specified by volume policy for driver %s",
+		policySnapshotClass, provisioner,
+	)
+}
+
 // GetVolumeSnapshotClassForStorageClass returns a VolumeSnapshotClass
 // for the supplied volume provisioner/ driver name.
 func GetVolumeSnapshotClassForStorageClass(

pkg/util/csi/volume_snapshot_test.go

@@ -1032,7 +1032,7 @@ func TestGetVolumeSnapshotClass(t *testing.T) {
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
 			actualSnapshotClass, actualError := GetVolumeSnapshotClass(
-				tc.driverName, tc.backup, tc.pvc, logrus.New(), fakeClient)
+				tc.driverName, tc.backup, tc.pvc, logrus.New(), fakeClient, "")
 			if tc.expectError {
 				require.Error(t, actualError)
 				assert.Nil(t, actualSnapshotClass)
@@ -1043,6 +1043,93 @@ func TestGetVolumeSnapshotClass(t *testing.T) {
 	}
 }
 
+func TestGetVolumeSnapshotClassFromVolumePolicy(t *testing.T) {
+	vscArray1 := &snapshotv1api.VolumeSnapshotClass{
+		ObjectMeta: metav1.ObjectMeta{Name: "vsc-array-1"},
+		Driver:     "infinibox-csi-driver",
+	}
+	vscArray2 := &snapshotv1api.VolumeSnapshotClass{
+		ObjectMeta: metav1.ObjectMeta{Name: "vsc-array-2"},
+		Driver:     "infinibox-csi-driver",
+	}
+	vscOther := &snapshotv1api.VolumeSnapshotClass{
+		ObjectMeta: metav1.ObjectMeta{Name: "vsc-other"},
+		Driver:     "other-csi-driver",
+	}
+
+	snapshotClasses := &snapshotv1api.VolumeSnapshotClassList{
+		Items: []snapshotv1api.VolumeSnapshotClass{*vscArray1, *vscArray2, *vscOther},
+	}
+
+	testCases := []struct {
+		name                string
+		policySnapshotClass string
+		provisioner         string
+		expectedVSC         *snapshotv1api.VolumeSnapshotClass
+		expectError         bool
+	}{
+		{
+			name:                "empty policy returns nil",
+			policySnapshotClass: "",
+			provisioner:         "infinibox-csi-driver",
+			expectedVSC:         nil,
+			expectError:         false,
+		},
+		{
+			name:                "matching VSC with correct driver",
+			policySnapshotClass: "vsc-array-1",
+			provisioner:         "infinibox-csi-driver",
+			expectedVSC:         vscArray1,
+			expectError:         false,
+		},
+		{
+			name:                "matching VSC with correct driver second array",
+			policySnapshotClass: "vsc-array-2",
+			provisioner:         "infinibox-csi-driver",
+			expectedVSC:         vscArray2,
+			expectError:         false,
+		},
+		{
+			name:                "VSC exists but wrong driver",
+			policySnapshotClass: "vsc-other",
+			provisioner:         "infinibox-csi-driver",
+			expectError:         true,
+		},
+		{
+			name:                "VSC does not exist",
+			policySnapshotClass: "non-existent",
+			provisioner:         "infinibox-csi-driver",
+			expectError:         true,
+		},
+		{
+			name:                "case-insensitive name matching",
+			policySnapshotClass: "VSC-ARRAY-1",
+			provisioner:         "infinibox-csi-driver",
+			expectedVSC:         vscArray1,
+			expectError:         false,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			actualVSC, actualError := GetVolumeSnapshotClassFromVolumePolicy(
+				tc.policySnapshotClass, tc.provisioner, snapshotClasses)
+			if tc.expectError {
+				require.Error(t, actualError)
+				assert.Nil(t, actualVSC)
+				return
+			}
+			if tc.expectedVSC == nil {
+				assert.Nil(t, actualVSC)
+			} else {
+				require.NotNil(t, actualVSC)
+				assert.Equal(t, tc.expectedVSC.Name, actualVSC.Name)
+				assert.Equal(t, tc.expectedVSC.Driver, actualVSC.Driver)
+			}
+		})
+	}
+}
+
 func TestGetVolumeSnapshotClassForStorageClass(t *testing.T) {
 	hostpathClass := &snapshotv1api.VolumeSnapshotClass{
 		ObjectMeta: metav1.ObjectMeta{

pkg/util/volumehelper/volume_policy_helper.go

@@ -27,4 +27,5 @@ type VolumeHelper interface {
 	ShouldPerformFSBackup(volume corev1api.Volume, pod corev1api.Pod) (bool, error)
 	ShouldPerformCustomAction(obj runtime.Unstructured, groupResource schema.GroupResource, matchParams map[string]any) (bool, error)
 	GetActionParameters(obj runtime.Unstructured, groupResource schema.GroupResource) (bool, string, map[string]any, error)
+	GetSnapshotClass(obj runtime.Unstructured, groupResource schema.GroupResource) (string, error)
 }

site/content/docs/main/csi.md

@@ -86,8 +86,23 @@ This section documents some of the choices made during implementing the CSI snap
         ```
         Note: Please ensure all your annotations are in lowercase. And follow the following format: `velero.io/csi-volumesnapshot-class_<driver name> = <VolumeSnapshotClass Name>`
 
-    3. **Choosing VolumeSnapshotClass for a particular PVC:**
-    If you want to use a particular VolumeSnapshotClass for a particular PVC, you can add a annotation to the PVC to indicate which VolumeSnapshotClass to use. This overrides any annotation added to backup or schedule. For example, if you want to use the VolumeSnapshotClass `test-snapclass` for a particular PVC, you can create a PVC like this:
+    3. **Choosing VolumeSnapshotClass via Volume Policy:**
+    If you want to use a particular VolumeSnapshotClass based on conditions like StorageClass, you can specify the `snapshotClass` parameter in a volume policy's `snapshot` action. This is useful when multiple storage arrays share the same CSI driver but require different VolumeSnapshotClasses. For example:
+        ```yaml
+        version: v1
+        volumePolicies:
+        - conditions:
+            storageClass:
+            - nutanix-files
+          action:
+            type: snapshot
+            parameters:
+              snapshotClass: nutanix-files-snapclass
+        ```
+        This overrides backup/schedule annotations and VolumeSnapshotClass labels, but is overridden by PVC-level annotations. See the [resource filtering documentation](resource-filtering.md) for more volume policy examples.
+
+    4. **Choosing VolumeSnapshotClass for a particular PVC:**
+    If you want to use a particular VolumeSnapshotClass for a particular PVC, you can add a annotation to the PVC to indicate which VolumeSnapshotClass to use. This overrides any other method of selecting a VolumeSnapshotClass. For example, if you want to use the VolumeSnapshotClass `test-snapclass` for a particular PVC, you can create a PVC like this:
         ```yaml
         apiVersion: v1
         kind: PersistentVolumeClaim

site/content/docs/main/customize-installation.md

@@ -356,7 +356,7 @@ Run `velero install --help` or see the [Helm chart documentation](https://vmware
 
 ### Enabling shell autocompletion
 
-**Velero CLI** provides autocompletion support for `Bash` and `Zsh`, which can save you a lot of typing.
+**Velero CLI** provides autocompletion support for `Bash`, `Zsh`, and `Fish`, which can save you a lot of typing. In addition to command and flag names, the CLI dynamically completes resource names (backups, restores, schedules, etc.) by querying the cluster.
 
 Below are the procedures to set up autocompletion for `Bash` (including the difference between `Linux` and `macOS`) and `Zsh`.
 

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

@@ -652,6 +652,7 @@ a volume policy but for a particular volume included in the backup there are no
 in such a scenario the legacy approach will be used for backing up the particular volume. Considering everything, the recommendation would be to use only one
 of the approaches to backup volumes - volume policy approach or the opt-in/opt-out legacy approach, and not mix them for clarity.
 - Snapshot action can either be a native snapshot or a csi snapshot or csi snapshot datamover, as is the case with the current flow where velero itself makes the decision based on the backup CR's existing options.
+- The `snapshot` action supports an optional `snapshotClass` parameter that specifies which VolumeSnapshotClass to use for CSI snapshots. This is useful when multiple storage arrays share the same CSI driver but require different VolumeSnapshotClasses. When specified, this takes priority over backup annotations and VolumeSnapshotClass labels, but is overridden by PVC-level annotations. See the [CSI documentation](csi.md) for the full VolumeSnapshotClass selection priority order.
 - The `snapshot` action via Volume Policy has higher priority if there is a `snapshot` action matching for a particular volume, this volume would be backed up via snapshot irrespective of the value of `backup.Spec.SnapshotVolumes`.
 - If for a particular volume there is no `snapshot` matching action then the volume will be backed up via snapshot given that `backup.Spec.SnapshotVolumes` is not explicitly set to false.
 - Let's see some examples on how to use the volume policy feature for `fs-backup` and `snapshot` action purposes:
@@ -740,6 +741,29 @@ volumePolicies:
    - `fs-backup` on `Volume 1` because `Volume 1` satisfies the criteria for `fs-backup` action. 
    - Also, for Volume 2 as no matching action was found so legacy approach will be used as a fallback option for this volume (`fs-backup` operation will be done as `defaultVolumesToFSBackup: true` is specified by the user).
 
+***Example 6: User has two storage arrays using the same CSI driver and needs different VolumeSnapshotClasses for each***
+1. User specifies the volume policy as follows:
+```yaml
+version: v1
+volumePolicies:
+- conditions:
+    storageClass:
+    - array-1-sc
+  action:
+    type: snapshot
+    parameters:
+      snapshotClass: vsc-array-1
+- conditions:
+    storageClass:
+    - array-2-sc
+  action:
+    type: snapshot
+    parameters:
+      snapshotClass: vsc-array-2
+```
+2. User creates a backup using this volume policy
+3. The outcome would be that velero would use `vsc-array-1` VolumeSnapshotClass for volumes on storage class `array-1-sc` and `vsc-array-2` VolumeSnapshotClass for volumes on storage class `array-2-sc`, even though both storage classes use the same CSI driver.
+
 ### Global backup volume policies
 
 Resource policies (volume policies) are normally opt-in per backup via `--resource-policies-configmap`. An administrator can instead configure a cluster-wide baseline that applies to **every** backup by starting the Velero server with the `--global-backup-volume-policies-configmap` flag, pointing at a ConfigMap in the Velero install namespace: