volumeattributesclass and core api changes

pkg/api/persistentvolume/util.go

@@ -39,6 +39,11 @@ func DropDisabledSpecFields(pvSpec *api.PersistentVolumeSpec, oldPVSpec *api.Per
 			pvSpec.CSI.NodeExpandSecretRef = nil
 		}
 	}
+	if !utilfeature.DefaultFeatureGate.Enabled(features.VolumeAttributesClass) {
+		if oldPVSpec == nil || oldPVSpec.VolumeAttributesClassName == nil {
+			pvSpec.VolumeAttributesClassName = nil
+		}
+	}
 }
 
 // DropDisabledStatusFields removes disabled fields from the pv status.

pkg/api/persistentvolume/util_test.go

@@ -28,6 +28,7 @@ import (
 	featuregatetesting "k8s.io/component-base/featuregate/testing"
 	api "k8s.io/kubernetes/pkg/apis/core"
 	"k8s.io/kubernetes/pkg/features"
+	"k8s.io/utils/ptr"
 )
 
 func TestDropDisabledFields(t *testing.T) {
@@ -35,6 +36,7 @@ func TestDropDisabledFields(t *testing.T) {
 		Name:      "expansion-secret",
 		Namespace: "default",
 	}
+	vacName := ptr.To("vac")
 
 	tests := map[string]struct {
 		oldSpec             *api.PersistentVolumeSpec
@@ -42,6 +44,7 @@ func TestDropDisabledFields(t *testing.T) {
 		expectOldSpec       *api.PersistentVolumeSpec
 		expectNewSpec       *api.PersistentVolumeSpec
 		csiExpansionEnabled bool
+		vacEnabled          bool
 	}{
 		"disabled csi expansion clears secrets": {
 			csiExpansionEnabled: false,
@@ -85,11 +88,54 @@ func TestDropDisabledFields(t *testing.T) {
 			oldSpec:             specWithCSISecrets(nil),
 			expectOldSpec:       specWithCSISecrets(nil),
 		},
+		"disabled vac clears volume attributes class name": {
+			vacEnabled:    false,
+			newSpec:       specWithVACName(vacName),
+			expectNewSpec: specWithVACName(nil),
+			oldSpec:       nil,
+			expectOldSpec: nil,
+		},
+		"enabled vac preserve volume attributes class name": {
+			vacEnabled:    true,
+			newSpec:       specWithVACName(vacName),
+			expectNewSpec: specWithVACName(vacName),
+			oldSpec:       nil,
+			expectOldSpec: nil,
+		},
+		"enabled vac preserve volume attributes class name when both old and new have it": {
+			vacEnabled:    true,
+			newSpec:       specWithVACName(vacName),
+			expectNewSpec: specWithVACName(vacName),
+			oldSpec:       specWithVACName(vacName),
+			expectOldSpec: specWithVACName(vacName),
+		},
+		"disabled vac old pv had volume attributes class name": {
+			vacEnabled:    false,
+			newSpec:       specWithVACName(vacName),
+			expectNewSpec: specWithVACName(vacName),
+			oldSpec:       specWithVACName(vacName),
+			expectOldSpec: specWithVACName(vacName),
+		},
+		"enabled vac preserves volume attributes class name when old pv did not had it": {
+			vacEnabled:    true,
+			newSpec:       specWithVACName(vacName),
+			expectNewSpec: specWithVACName(vacName),
+			oldSpec:       specWithVACName(nil),
+			expectOldSpec: specWithVACName(nil),
+		},
+		"disabled vac neither new pv nor old pv had volume attributes class name": {
+			vacEnabled:    false,
+			newSpec:       specWithVACName(nil),
+			expectNewSpec: specWithVACName(nil),
+			oldSpec:       specWithVACName(nil),
+			expectOldSpec: specWithVACName(nil),
+		},
 	}
 
 	for name, tc := range tests {
 		t.Run(name, func(t *testing.T) {
 			defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.CSINodeExpandSecret, tc.csiExpansionEnabled)()
+			defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.VolumeAttributesClass, tc.vacEnabled)()
 
 			DropDisabledSpecFields(tc.newSpec, tc.oldSpec)
 			if !reflect.DeepEqual(tc.newSpec, tc.expectNewSpec) {
@@ -118,6 +164,22 @@ func specWithCSISecrets(secret *api.SecretReference) *api.PersistentVolumeSpec {
 	return pvSpec
 }
 
+func specWithVACName(vacName *string) *api.PersistentVolumeSpec {
+	pvSpec := &api.PersistentVolumeSpec{
+		PersistentVolumeSource: api.PersistentVolumeSource{
+			CSI: &api.CSIPersistentVolumeSource{
+				Driver:       "com.google.gcepd",
+				VolumeHandle: "foobar",
+			},
+		},
+	}
+
+	if vacName != nil {
+		pvSpec.VolumeAttributesClassName = vacName
+	}
+	return pvSpec
+}
+
 func TestWarnings(t *testing.T) {
 	testcases := []struct {
 		name     string

pkg/api/persistentvolumeclaim/util.go

@@ -35,6 +35,14 @@ const (
 // DropDisabledFields removes disabled fields from the pvc spec.
 // This should be called from PrepareForCreate/PrepareForUpdate for all resources containing a pvc spec.
 func DropDisabledFields(pvcSpec, oldPVCSpec *core.PersistentVolumeClaimSpec) {
+	// Drop the contents of the volumeAttributesClassName if the VolumeAttributesClass
+	// feature gate is disabled.
+	if !utilfeature.DefaultFeatureGate.Enabled(features.VolumeAttributesClass) {
+		if oldPVCSpec == nil || oldPVCSpec.VolumeAttributesClassName == nil {
+			pvcSpec.VolumeAttributesClassName = nil
+		}
+	}
+
 	// Drop the contents of the dataSourceRef field if the AnyVolumeDataSource
 	// feature gate is disabled.
 	if !utilfeature.DefaultFeatureGate.Enabled(features.AnyVolumeDataSource) {
@@ -91,6 +99,15 @@ func EnforceDataSourceBackwardsCompatibility(pvcSpec, oldPVCSpec *core.Persisten
 }
 
 func DropDisabledFieldsFromStatus(pvc, oldPVC *core.PersistentVolumeClaim) {
+	if !utilfeature.DefaultFeatureGate.Enabled(features.VolumeAttributesClass) {
+		if oldPVC == nil || oldPVC.Status.CurrentVolumeAttributesClassName == nil {
+			pvc.Status.CurrentVolumeAttributesClassName = nil
+		}
+		if oldPVC == nil || oldPVC.Status.ModifyVolumeStatus == nil {
+			pvc.Status.ModifyVolumeStatus = nil
+		}
+	}
+
 	if !utilfeature.DefaultFeatureGate.Enabled(features.RecoverVolumeExpansionFailure) {
 		if !helper.ClaimContainsAllocatedResources(oldPVC) {
 			pvc.Status.AllocatedResources = nil

pkg/api/persistentvolumeclaim/util_test.go

@@ -23,11 +23,12 @@ import (
 
 	"github.com/google/go-cmp/cmp"
 	"k8s.io/apimachinery/pkg/api/resource"
-	"k8s.io/apimachinery/pkg/util/sets"
-
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/sets"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	featuregatetesting "k8s.io/component-base/featuregate/testing"
+	"k8s.io/utils/ptr"
+
 	"k8s.io/kubernetes/pkg/apis/core"
 	"k8s.io/kubernetes/pkg/features"
 )
@@ -384,82 +385,217 @@ func TestDataSourceRef(t *testing.T) {
 	}
 }
 
+func TestDropDisabledVolumeAttributesClass(t *testing.T) {
+	vacName := ptr.To("foo")
+
+	var tests = map[string]struct {
+		spec       core.PersistentVolumeClaimSpec
+		oldSpec    core.PersistentVolumeClaimSpec
+		vacEnabled bool
+		wantVAC    *string
+	}{
+		"vac disabled with empty vac": {
+			spec: core.PersistentVolumeClaimSpec{},
+		},
+		"vac disabled with vac": {
+			spec: core.PersistentVolumeClaimSpec{VolumeAttributesClassName: vacName},
+		},
+		"vac enabled with empty vac": {
+			spec:       core.PersistentVolumeClaimSpec{},
+			vacEnabled: true,
+		},
+		"vac enabled with vac": {
+			spec:       core.PersistentVolumeClaimSpec{VolumeAttributesClassName: vacName},
+			vacEnabled: true,
+			wantVAC:    vacName,
+		},
+		"vac disabled with vac when vac doesn't exists in oldSpec": {
+			spec:    core.PersistentVolumeClaimSpec{VolumeAttributesClassName: vacName},
+			oldSpec: core.PersistentVolumeClaimSpec{},
+		},
+		"vac disabled with vac when vac exists in oldSpec": {
+			spec:       core.PersistentVolumeClaimSpec{VolumeAttributesClassName: vacName},
+			oldSpec:    core.PersistentVolumeClaimSpec{VolumeAttributesClassName: vacName},
+			vacEnabled: false,
+			wantVAC:    vacName,
+		},
+		"vac enabled with vac when vac doesn't exists in oldSpec": {
+			spec:       core.PersistentVolumeClaimSpec{VolumeAttributesClassName: vacName},
+			oldSpec:    core.PersistentVolumeClaimSpec{},
+			vacEnabled: true,
+			wantVAC:    vacName,
+		},
+		"vac enable with vac when vac exists in oldSpec": {
+			spec:       core.PersistentVolumeClaimSpec{VolumeAttributesClassName: vacName},
+			oldSpec:    core.PersistentVolumeClaimSpec{VolumeAttributesClassName: vacName},
+			vacEnabled: true,
+			wantVAC:    vacName,
+		},
+	}
+
+	for testName, test := range tests {
+		t.Run(testName, func(t *testing.T) {
+			defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.VolumeAttributesClass, test.vacEnabled)()
+			DropDisabledFields(&test.spec, &test.oldSpec)
+			if test.spec.VolumeAttributesClassName != test.wantVAC {
+				t.Errorf("expected vac was not met, test: %s, vacEnabled: %v, spec: %+v, expected VAC: %+v",
+					testName, test.vacEnabled, test.spec, test.wantVAC)
+			}
+		})
+	}
+}
+
 func TestDropDisabledFieldsFromStatus(t *testing.T) {
 	tests := []struct {
 		name                                string
-		feature  bool
+		enableRecoverVolumeExpansionFailure bool
+		enableVolumeAttributesClass         bool
 		pvc                                 *core.PersistentVolumeClaim
 		oldPVC                              *core.PersistentVolumeClaim
 		expected                            *core.PersistentVolumeClaim
 	}{
 		{
 			name:                                "for:newPVC=hasAllocatedResource,oldPVC=doesnot,featuregate=false; should drop field",
-			feature:  false,
+			enableRecoverVolumeExpansionFailure: false,
+			enableVolumeAttributesClass:         false,
 			pvc:                                 withAllocatedResource("5G"),
 			oldPVC:                              getPVC(),
 			expected:                            getPVC(),
 		},
 		{
-			name:     "for:newPVC=hasAllocatedResource,oldPVC=doesnot,featuregate=true; should keep field",
+			name:                                "for:newPVC=hasAllocatedResource,oldPVC=doesnot,featuregate=RecoverVolumeExpansionFailure=true; should keep field",
-			feature:  true,
+			enableRecoverVolumeExpansionFailure: true,
+			enableVolumeAttributesClass:         false,
 			pvc:                                 withAllocatedResource("5G"),
 			oldPVC:                              getPVC(),
 			expected:                            withAllocatedResource("5G"),
 		},
 		{
-			name:     "for:newPVC=hasAllocatedResource,oldPVC=hasAllocatedResource,featuregate=true; should keep field",
+			name:                                "for:newPVC=hasAllocatedResource,oldPVC=hasAllocatedResource,featuregate=RecoverVolumeExpansionFailure=true; should keep field",
-			feature:  true,
+			enableRecoverVolumeExpansionFailure: true,
+			enableVolumeAttributesClass:         false,
 			pvc:                                 withAllocatedResource("5G"),
 			oldPVC:                              withAllocatedResource("5G"),
 			expected:                            withAllocatedResource("5G"),
 		},
 		{
 			name:                                "for:newPVC=hasAllocatedResource,oldPVC=hasAllocatedResource,featuregate=false; should keep field",
-			feature:  false,
+			enableRecoverVolumeExpansionFailure: false,
+			enableVolumeAttributesClass:         false,
 			pvc:                                 withAllocatedResource("10G"),
 			oldPVC:                              withAllocatedResource("5G"),
 			expected:                            withAllocatedResource("10G"),
 		},
 		{
 			name:                                "for:newPVC=hasAllocatedResource,oldPVC=nil,featuregate=false; should drop field",
-			feature:  false,
+			enableRecoverVolumeExpansionFailure: false,
+			enableVolumeAttributesClass:         false,
 			pvc:                                 withAllocatedResource("5G"),
 			oldPVC:                              nil,
 			expected:                            getPVC(),
 		},
 		{
 			name:                                "for:newPVC=hasResizeStatus,oldPVC=nil, featuregate=false should drop field",
-			feature:  false,
+			enableRecoverVolumeExpansionFailure: false,
+			enableVolumeAttributesClass:         false,
 			pvc:                                 withResizeStatus(core.PersistentVolumeClaimNodeResizeFailed),
 			oldPVC:                              nil,
 			expected:                            getPVC(),
 		},
 		{
-			name:     "for:newPVC=hasResizeStatus,oldPVC=doesnot,featuregate=true; should keep field",
+			name:                                "for:newPVC=hasResizeStatus,oldPVC=doesnot,featuregate=RecoverVolumeExpansionFailure=true; should keep field",
-			feature:  true,
+			enableRecoverVolumeExpansionFailure: true,
+			enableVolumeAttributesClass:         false,
 			pvc:                                 withResizeStatus(core.PersistentVolumeClaimNodeResizeFailed),
 			oldPVC:                              getPVC(),
 			expected:                            withResizeStatus(core.PersistentVolumeClaimNodeResizeFailed),
 		},
 		{
-			name:     "for:newPVC=hasResizeStatus,oldPVC=hasResizeStatus,featuregate=true; should keep field",
+			name:                                "for:newPVC=hasResizeStatus,oldPVC=hasResizeStatus,featuregate=RecoverVolumeExpansionFailure=true; should keep field",
-			feature:  true,
+			enableRecoverVolumeExpansionFailure: true,
+			enableVolumeAttributesClass:         false,
 			pvc:                                 withResizeStatus(core.PersistentVolumeClaimNodeResizeFailed),
 			oldPVC:                              withResizeStatus(core.PersistentVolumeClaimNodeResizeFailed),
 			expected:                            withResizeStatus(core.PersistentVolumeClaimNodeResizeFailed),
 		},
 		{
 			name:                                "for:newPVC=hasResizeStatus,oldPVC=hasResizeStatus,featuregate=false; should keep field",
-			feature:  false,
+			enableRecoverVolumeExpansionFailure: false,
+			enableVolumeAttributesClass:         false,
 			pvc:                                 withResizeStatus(core.PersistentVolumeClaimNodeResizeFailed),
 			oldPVC:                              withResizeStatus(core.PersistentVolumeClaimNodeResizeFailed),
 			expected:                            withResizeStatus(core.PersistentVolumeClaimNodeResizeFailed),
 		},
+		{
+			name:                                "for:newPVC=hasVolumeAttributeClass,oldPVC=nil, featuregate=false should drop field",
+			enableRecoverVolumeExpansionFailure: false,
+			enableVolumeAttributesClass:         false,
+			pvc:                                 withVolumeAttributesClassName("foo"),
+			oldPVC:                              nil,
+			expected:                            getPVC(),
+		},
+		{
+			name:                                "for:newPVC=hasVolumeAttributeClass,oldPVC=doesnot,featuregate=VolumeAttributesClass=true; should keep field",
+			enableRecoverVolumeExpansionFailure: false,
+			enableVolumeAttributesClass:         true,
+			pvc:                                 withVolumeAttributesClassName("foo"),
+			oldPVC:                              getPVC(),
+			expected:                            withVolumeAttributesClassName("foo"),
+		},
+		{
+			name:                                "for:newPVC=hasVolumeAttributeClass,oldPVC=hasVolumeAttributeClass,featuregate=VolumeAttributesClass=true; should keep field",
+			enableRecoverVolumeExpansionFailure: false,
+			enableVolumeAttributesClass:         true,
+			pvc:                                 withVolumeAttributesClassName("foo"),
+			oldPVC:                              withVolumeAttributesClassName("foo"),
+			expected:                            withVolumeAttributesClassName("foo"),
+		},
+		{
+			name:                                "for:newPVC=hasVolumeAttributeClass,oldPVC=hasVolumeAttributeClass,featuregate=false; should keep field",
+			enableRecoverVolumeExpansionFailure: false,
+			enableVolumeAttributesClass:         false,
+			pvc:                                 withVolumeAttributesClassName("foo"),
+			oldPVC:                              withVolumeAttributesClassName("foo"),
+			expected:                            withVolumeAttributesClassName("foo"),
+		},
+		{
+			name:                                "for:newPVC=hasVolumeAttributesModifyStatus,oldPVC=nil, featuregate=false should drop field",
+			enableRecoverVolumeExpansionFailure: false,
+			enableVolumeAttributesClass:         false,
+			pvc:                                 withVolumeAttributesModifyStatus("bar", core.PersistentVolumeClaimModifyVolumePending),
+			oldPVC:                              nil,
+			expected:                            getPVC(),
+		},
+		{
+			name:                                "for:newPVC=hasVolumeAttributesModifyStatus,oldPVC=doesnot,featuregate=VolumeAttributesClass=true; should keep field",
+			enableRecoverVolumeExpansionFailure: false,
+			enableVolumeAttributesClass:         true,
+			pvc:                                 withVolumeAttributesModifyStatus("bar", core.PersistentVolumeClaimModifyVolumePending),
+			oldPVC:                              getPVC(),
+			expected:                            withVolumeAttributesModifyStatus("bar", core.PersistentVolumeClaimModifyVolumePending),
+		},
+		{
+			name:                                "for:newPVC=hasVolumeAttributesModifyStatus,oldPVC=hasVolumeAttributesModifyStatus,featuregate=VolumeAttributesClass=true; should keep field",
+			enableRecoverVolumeExpansionFailure: false,
+			enableVolumeAttributesClass:         true,
+			pvc:                                 withVolumeAttributesModifyStatus("bar", core.PersistentVolumeClaimModifyVolumePending),
+			oldPVC:                              withVolumeAttributesModifyStatus("bar", core.PersistentVolumeClaimModifyVolumePending),
+			expected:                            withVolumeAttributesModifyStatus("bar", core.PersistentVolumeClaimModifyVolumePending),
+		},
+		{
+			name:                                "for:newPVC=hasVolumeAttributesModifyStatus,oldPVC=hasVolumeAttributesModifyStatus,featuregate=false; should keep field",
+			enableRecoverVolumeExpansionFailure: false,
+			enableVolumeAttributesClass:         false,
+			pvc:                                 withVolumeAttributesModifyStatus("bar", core.PersistentVolumeClaimModifyVolumePending),
+			oldPVC:                              withVolumeAttributesModifyStatus("bar", core.PersistentVolumeClaimModifyVolumePending),
+			expected:                            withVolumeAttributesModifyStatus("bar", core.PersistentVolumeClaimModifyVolumePending),
+		},
 	}
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.RecoverVolumeExpansionFailure, test.feature)()
+			defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.RecoverVolumeExpansionFailure, test.enableRecoverVolumeExpansionFailure)()
+			defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.VolumeAttributesClass, test.enableVolumeAttributesClass)()
 
 			DropDisabledFieldsFromStatus(test.pvc, test.oldPVC)
 
@@ -494,6 +630,25 @@ func withResizeStatus(status core.ClaimResourceStatus) *core.PersistentVolumeCla
 	}
 }
 
+func withVolumeAttributesClassName(vacName string) *core.PersistentVolumeClaim {
+	return &core.PersistentVolumeClaim{
+		Status: core.PersistentVolumeClaimStatus{
+			CurrentVolumeAttributesClassName: &vacName,
+		},
+	}
+}
+
+func withVolumeAttributesModifyStatus(target string, status core.PersistentVolumeClaimModifyVolumeStatus) *core.PersistentVolumeClaim {
+	return &core.PersistentVolumeClaim{
+		Status: core.PersistentVolumeClaimStatus{
+			ModifyVolumeStatus: &core.ModifyVolumeStatus{
+				TargetVolumeAttributesClassName: target,
+				Status:                          status,
+			},
+		},
+	}
+}
+
 func TestWarnings(t *testing.T) {
 	testcases := []struct {
 		name     string

pkg/apis/core/types.go

@@ -335,6 +335,16 @@ type PersistentVolumeSpec struct {
 	// This field influences the scheduling of pods that use this volume.
 	// +optional
 	NodeAffinity *VolumeNodeAffinity
+	// Name of VolumeAttributesClass to which this persistent volume belongs. Empty value
+	// is not allowed. When this field is not set, it indicates that this volume does not belong to any
+	// VolumeAttributesClass. This field is mutable and can be changed by the CSI driver
+	// after a volume has been updated successfully to a new class.
+	// For an unbound PersistentVolume, the volumeAttributesClassName will be matched with unbound
+	// PersistentVolumeClaims during the binding process.
+	// This is an alpha field and requires enabling VolumeAttributesClass feature.
+	// +featureGate=VolumeAttributesClass
+	// +optional
+	VolumeAttributesClassName *string
 }
 
 // VolumeNodeAffinity defines constraints that limit what nodes this volume can be accessed from.
@@ -488,6 +498,21 @@ type PersistentVolumeClaimSpec struct {
 	// (Alpha) Using the namespace field of dataSourceRef requires the CrossNamespaceVolumeDataSource feature gate to be enabled.
 	// +optional
 	DataSourceRef *TypedObjectReference
+	// volumeAttributesClassName may be used to set the VolumeAttributesClass used by this claim.
+	// If specified, the CSI driver will create or update the volume with the attributes defined
+	// in the corresponding VolumeAttributesClass. This has a different purpose than storageClassName,
+	// it can be changed after the claim is created. An empty string value means that no VolumeAttributesClass
+	// will be applied to the claim but it's not allowed to reset this field to empty string once it is set.
+	// If unspecified and the PersistentVolumeClaim is unbound, the default VolumeAttributesClass
+	// will be set by the persistentvolume controller if it exists.
+	// If the resource referred to by volumeAttributesClass does not exist, this PersistentVolumeClaim will be
+	// set to a Pending state, as reflected by the modifyVolumeStatus field, until such as a resource
+	// exists.
+	// More info: https://kubernetes.io/docs/concepts/storage/persistent-volumes#volumeattributesclass
+	// (Alpha) Using this field requires the VolumeAttributesClass feature gate to be enabled.
+	// +featureGate=VolumeAttributesClass
+	// +optional
+	VolumeAttributesClassName *string
 }
 
 type TypedObjectReference struct {
@@ -518,6 +543,11 @@ const (
 	PersistentVolumeClaimResizing PersistentVolumeClaimConditionType = "Resizing"
 	// PersistentVolumeClaimFileSystemResizePending - controller resize is finished and a file system resize is pending on node
 	PersistentVolumeClaimFileSystemResizePending PersistentVolumeClaimConditionType = "FileSystemResizePending"
+
+	// Applying the target VolumeAttributesClass encountered an error
+	PersistentVolumeClaimVolumeModifyVolumeError PersistentVolumeClaimConditionType = "ModifyVolumeError"
+	// Volume is being modified
+	PersistentVolumeClaimVolumeModifyingVolume PersistentVolumeClaimConditionType = "ModifyingVolume"
 )
 
 // +enum
@@ -544,6 +574,38 @@ const (
 	PersistentVolumeClaimNodeResizeFailed ClaimResourceStatus = "NodeResizeFailed"
 )
 
+// +enum
+// New statuses can be added in the future. Consumers should check for unknown statuses and fail appropriately
+type PersistentVolumeClaimModifyVolumeStatus string
+
+const (
+	// Pending indicates that the PersistentVolumeClaim cannot be modified due to unmet requirements, such as
+	// the specified VolumeAttributesClass not existing
+	PersistentVolumeClaimModifyVolumePending PersistentVolumeClaimModifyVolumeStatus = "Pending"
+	// InProgress indicates that the volume is being modified
+	PersistentVolumeClaimModifyVolumeInProgress PersistentVolumeClaimModifyVolumeStatus = "InProgress"
+	// Infeasible indicates that the request has been rejected as invalid by the CSI driver. To
+	// resolve the error, a valid VolumeAttributesClass needs to be specified
+	PersistentVolumeClaimModifyVolumeInfeasible PersistentVolumeClaimModifyVolumeStatus = "Infeasible"
+)
+
+// ModifyVolumeStatus represents the status object of ControllerModifyVolume operation
+type ModifyVolumeStatus struct {
+	// targetVolumeAttributesClassName is the name of the VolumeAttributesClass the PVC currently being reconciled
+	TargetVolumeAttributesClassName string
+	// status is the status of the ControllerModifyVolume operation. It can be in any of following states:
+	//  - Pending
+	//    Pending indicates that the PersistentVolumeClaim cannot be modified due to unmet requirements, such as
+	//    the specified VolumeAttributesClass not existing.
+	//  - InProgress
+	//    InProgress indicates that the volume is being modified.
+	//  - Infeasible
+	//   Infeasible indicates that the request has been rejected as invalid by the CSI driver. To
+	// 	  resolve the error, a valid VolumeAttributesClass needs to be specified.
+	// Note: New statuses can be added in the future. Consumers should check for unknown statuses and fail appropriately.
+	Status PersistentVolumeClaimModifyVolumeStatus
+}
+
 // PersistentVolumeClaimCondition represents the current condition of PV claim
 type PersistentVolumeClaimCondition struct {
 	Type   PersistentVolumeClaimConditionType
@@ -635,6 +697,18 @@ type PersistentVolumeClaimStatus struct {
 	// +mapType=granular
 	// +optional
 	AllocatedResourceStatuses map[ResourceName]ClaimResourceStatus
+	// currentVolumeAttributesClassName is the current name of the VolumeAttributesClass the PVC is using.
+	// When unset, there is no VolumeAttributeClass applied to this PersistentVolumeClaim
+	// This is an alpha field and requires enabling VolumeAttributesClass feature.
+	// +featureGate=VolumeAttributesClass
+	// +optional
+	CurrentVolumeAttributesClassName *string
+	// ModifyVolumeStatus represents the status object of ControllerModifyVolume operation.
+	// When this is unset, there is no ModifyVolume operation being attempted.
+	// This is an alpha field and requires enabling VolumeAttributesClass feature.
+	// +featureGate=VolumeAttributesClass
+	// +optional
+	ModifyVolumeStatus *ModifyVolumeStatus
 }
 
 // PersistentVolumeAccessMode defines various access modes for PV.

pkg/apis/core/validation/validation.go

@@ -1654,6 +1654,8 @@ var allowedTemplateObjectMetaFields = map[string]bool{
 
 // PersistentVolumeSpecValidationOptions contains the different settings for PeristentVolume validation
 type PersistentVolumeSpecValidationOptions struct {
+	// Allow users to modify the class of volume attributes
+	EnableVolumeAttributesClass bool
 }
 
 // ValidatePersistentVolumeName checks that a name is appropriate for a
@@ -1667,7 +1669,13 @@ var supportedReclaimPolicy = sets.NewString(string(core.PersistentVolumeReclaimD
 var supportedVolumeModes = sets.NewString(string(core.PersistentVolumeBlock), string(core.PersistentVolumeFilesystem))
 
 func ValidationOptionsForPersistentVolume(pv, oldPv *core.PersistentVolume) PersistentVolumeSpecValidationOptions {
-	return PersistentVolumeSpecValidationOptions{}
+	opts := PersistentVolumeSpecValidationOptions{
+		EnableVolumeAttributesClass: utilfeature.DefaultMutableFeatureGate.Enabled(features.VolumeAttributesClass),
+	}
+	if oldPv != nil && oldPv.Spec.VolumeAttributesClassName != nil {
+		opts.EnableVolumeAttributesClass = true
+	}
+	return opts
 }
 
 func ValidatePersistentVolumeSpec(pvSpec *core.PersistentVolumeSpec, pvName string, validateInlinePersistentVolumeSpec bool, fldPath *field.Path, opts PersistentVolumeSpecValidationOptions) field.ErrorList {
@@ -1952,6 +1960,18 @@ func ValidatePersistentVolumeSpec(pvSpec *core.PersistentVolumeSpec, pvName stri
 			}
 		}
 	}
+	if pvSpec.VolumeAttributesClassName != nil && opts.EnableVolumeAttributesClass {
+		if len(*pvSpec.VolumeAttributesClassName) == 0 {
+			allErrs = append(allErrs, field.Required(fldPath.Child("volumeAttributesClassName"), "an empty string is disallowed"))
+		} else {
+			for _, msg := range ValidateClassName(*pvSpec.VolumeAttributesClassName, false) {
+				allErrs = append(allErrs, field.Invalid(fldPath.Child("volumeAttributesClassName"), *pvSpec.VolumeAttributesClassName, msg))
+			}
+		}
+		if pvSpec.CSI == nil {
+			allErrs = append(allErrs, field.Required(fldPath.Child("csi"), "has to be specified when using volumeAttributesClassName"))
+		}
+	}
 	return allErrs
 }
 
@@ -1986,6 +2006,17 @@ func ValidatePersistentVolumeUpdate(newPv, oldPv *core.PersistentVolume, opts Pe
 		allErrs = append(allErrs, validatePvNodeAffinity(newPv.Spec.NodeAffinity, oldPv.Spec.NodeAffinity, field.NewPath("nodeAffinity"))...)
 	}
 
+	if !apiequality.Semantic.DeepEqual(oldPv.Spec.VolumeAttributesClassName, newPv.Spec.VolumeAttributesClassName) {
+		if !utilfeature.DefaultFeatureGate.Enabled(features.VolumeAttributesClass) {
+			allErrs = append(allErrs, field.Forbidden(field.NewPath("spec", "volumeAttributesClassName"), "update is forbidden when the VolumeAttributesClass feature gate is disabled"))
+		}
+		if opts.EnableVolumeAttributesClass {
+			if oldPv.Spec.VolumeAttributesClassName != nil && newPv.Spec.VolumeAttributesClassName == nil {
+				allErrs = append(allErrs, field.Forbidden(field.NewPath("spec", "volumeAttributesClassName"), "update from non-nil value to nil is forbidden"))
+			}
+		}
+	}
+
 	return allErrs
 }
 
@@ -2005,12 +2036,15 @@ type PersistentVolumeClaimSpecValidationOptions struct {
 	AllowInvalidLabelValueInSelector bool
 	// Allow to validate the API group of the data source and data source reference
 	AllowInvalidAPIGroupInDataSourceOrRef bool
+	// Allow users to modify the class of volume attributes
+	EnableVolumeAttributesClass bool
 }
 
 func ValidationOptionsForPersistentVolumeClaim(pvc, oldPvc *core.PersistentVolumeClaim) PersistentVolumeClaimSpecValidationOptions {
 	opts := PersistentVolumeClaimSpecValidationOptions{
 		EnableRecoverFromExpansionFailure: utilfeature.DefaultFeatureGate.Enabled(features.RecoverVolumeExpansionFailure),
 		AllowInvalidLabelValueInSelector:  false,
+		EnableVolumeAttributesClass:       utilfeature.DefaultFeatureGate.Enabled(features.VolumeAttributesClass),
 	}
 	if oldPvc == nil {
 		// If there's no old PVC, use the options based solely on feature enablement
@@ -2020,6 +2054,11 @@ func ValidationOptionsForPersistentVolumeClaim(pvc, oldPvc *core.PersistentVolum
 	// If the old object had an invalid API group in the data source or data source reference, continue to allow it in the new object
 	opts.AllowInvalidAPIGroupInDataSourceOrRef = allowInvalidAPIGroupInDataSourceOrRef(&oldPvc.Spec)
 
+	if oldPvc.Spec.VolumeAttributesClassName != nil {
+		// If the old object had a volume attributes class, continue to validate it in the new object.
+		opts.EnableVolumeAttributesClass = true
+	}
+
 	labelSelectorValidationOpts := unversionedvalidation.LabelSelectorValidationOptions{
 		AllowInvalidLabelValueInSelector: opts.AllowInvalidLabelValueInSelector,
 	}
@@ -2038,6 +2077,7 @@ func ValidationOptionsForPersistentVolumeClaim(pvc, oldPvc *core.PersistentVolum
 func ValidationOptionsForPersistentVolumeClaimTemplate(claimTemplate, oldClaimTemplate *core.PersistentVolumeClaimTemplate) PersistentVolumeClaimSpecValidationOptions {
 	opts := PersistentVolumeClaimSpecValidationOptions{
 		AllowInvalidLabelValueInSelector: false,
+		EnableVolumeAttributesClass:      utilfeature.DefaultFeatureGate.Enabled(features.VolumeAttributesClass),
 	}
 	if oldClaimTemplate == nil {
 		// If there's no old PVC template, use the options based solely on feature enablement
@@ -2193,6 +2233,11 @@ func ValidatePersistentVolumeClaimSpec(spec *core.PersistentVolumeClaimSpec, fld
 				"must match dataSourceRef"))
 		}
 	}
+	if spec.VolumeAttributesClassName != nil && len(*spec.VolumeAttributesClassName) > 0 && opts.EnableVolumeAttributesClass {
+		for _, msg := range ValidateClassName(*spec.VolumeAttributesClassName, false) {
+			allErrs = append(allErrs, field.Invalid(fldPath.Child("volumeAttributesClassName"), *spec.VolumeAttributesClassName, msg))
+		}
+	}
 
 	return allErrs
 }
@@ -2236,6 +2281,8 @@ func ValidatePersistentVolumeClaimUpdate(newPvc, oldPvc *core.PersistentVolumeCl
 	if newPvc.Status.Phase == core.ClaimBound && newPvcClone.Spec.Resources.Requests != nil {
 		newPvcClone.Spec.Resources.Requests["storage"] = oldPvc.Spec.Resources.Requests["storage"] // +k8s:verify-mutation:reason=clone
 	}
+	// lets make sure volume attributes class name is same.
+	newPvcClone.Spec.VolumeAttributesClassName = oldPvcClone.Spec.VolumeAttributesClassName // +k8s:verify-mutation:reason=clone
 
 	oldSize := oldPvc.Spec.Resources.Requests["storage"]
 	newSize := newPvc.Spec.Resources.Requests["storage"]
@@ -2243,7 +2290,7 @@ func ValidatePersistentVolumeClaimUpdate(newPvc, oldPvc *core.PersistentVolumeCl
 
 	if !apiequality.Semantic.DeepEqual(newPvcClone.Spec, oldPvcClone.Spec) {
 		specDiff := cmp.Diff(oldPvcClone.Spec, newPvcClone.Spec)
-		allErrs = append(allErrs, field.Forbidden(field.NewPath("spec"), fmt.Sprintf("spec is immutable after creation except resources.requests for bound claims\n%v", specDiff)))
+		allErrs = append(allErrs, field.Forbidden(field.NewPath("spec"), fmt.Sprintf("spec is immutable after creation except resources.requests and volumeAttributesClassName for bound claims\n%v", specDiff)))
 	}
 	if newSize.Cmp(oldSize) < 0 {
 		if !opts.EnableRecoverFromExpansionFailure {
@@ -2260,6 +2307,21 @@ func ValidatePersistentVolumeClaimUpdate(newPvc, oldPvc *core.PersistentVolumeCl
 
 	allErrs = append(allErrs, ValidateImmutableField(newPvc.Spec.VolumeMode, oldPvc.Spec.VolumeMode, field.NewPath("volumeMode"))...)
 
+	if !apiequality.Semantic.DeepEqual(oldPvc.Spec.VolumeAttributesClassName, newPvc.Spec.VolumeAttributesClassName) {
+		if !utilfeature.DefaultFeatureGate.Enabled(features.VolumeAttributesClass) {
+			allErrs = append(allErrs, field.Forbidden(field.NewPath("spec", "volumeAttributesClassName"), "update is forbidden when the VolumeAttributesClass feature gate is disabled"))
+		}
+		if opts.EnableVolumeAttributesClass {
+			if oldPvc.Spec.VolumeAttributesClassName != nil {
+				if newPvc.Spec.VolumeAttributesClassName == nil {
+					allErrs = append(allErrs, field.Forbidden(field.NewPath("spec", "volumeAttributesClassName"), "update from non-nil value to nil is forbidden"))
+				} else if len(*newPvc.Spec.VolumeAttributesClassName) == 0 {
+					allErrs = append(allErrs, field.Forbidden(field.NewPath("spec", "volumeAttributesClassName"), "update from non-nil value to an empty string is forbidden"))
+				}
+			}
+		}
+	}
+
 	return allErrs
 }
 

pkg/apis/core/validation/validation_test.go

@@ -46,6 +46,7 @@ import (
 	"k8s.io/kubernetes/pkg/capabilities"
 	"k8s.io/kubernetes/pkg/features"
 	utilpointer "k8s.io/utils/pointer"
+	"k8s.io/utils/ptr"
 )
 
 const (
@@ -110,6 +111,7 @@ func TestValidatePersistentVolumes(t *testing.T) {
 	invalidMode := core.PersistentVolumeMode("fakeVolumeMode")
 	scenarios := map[string]struct {
 		isExpectedFailure           bool
+		enableVolumeAttributesClass bool
 		volume                      *core.PersistentVolume
 	}{
 		"good-volume": {
@@ -478,10 +480,84 @@ func TestValidatePersistentVolumes(t *testing.T) {
 					},
 				}),
 		},
+		"invalid-volume-attributes-class-name": {
+			isExpectedFailure:           true,
+			enableVolumeAttributesClass: true,
+			volume: testVolume("invalid-volume-attributes-class-name", "", core.PersistentVolumeSpec{
+				Capacity: core.ResourceList{
+					core.ResourceName(core.ResourceStorage): resource.MustParse("10G"),
+				},
+				AccessModes: []core.PersistentVolumeAccessMode{core.ReadWriteOnce},
+				PersistentVolumeSource: core.PersistentVolumeSource{
+					HostPath: &core.HostPathVolumeSource{
+						Path: "/foo",
+						Type: newHostPathType(string(core.HostPathDirectory)),
+					},
+				},
+				StorageClassName:          "invalid",
+				VolumeAttributesClassName: ptr.To("-invalid-"),
+			}),
+		},
+		"invalid-empty-volume-attributes-class-name": {
+			isExpectedFailure:           true,
+			enableVolumeAttributesClass: true,
+			volume: testVolume("invalid-empty-volume-attributes-class-name", "", core.PersistentVolumeSpec{
+				Capacity: core.ResourceList{
+					core.ResourceName(core.ResourceStorage): resource.MustParse("10G"),
+				},
+				AccessModes: []core.PersistentVolumeAccessMode{core.ReadWriteOnce},
+				PersistentVolumeSource: core.PersistentVolumeSource{
+					HostPath: &core.HostPathVolumeSource{
+						Path: "/foo",
+						Type: newHostPathType(string(core.HostPathDirectory)),
+					},
+				},
+				StorageClassName:          "invalid",
+				VolumeAttributesClassName: ptr.To(""),
+			}),
+		},
+		"volume-with-good-volume-attributes-class-and-matched-volume-resource-when-feature-gate-is-on": {
+			isExpectedFailure:           false,
+			enableVolumeAttributesClass: true,
+			volume: testVolume("foo", "", core.PersistentVolumeSpec{
+				Capacity: core.ResourceList{
+					core.ResourceName(core.ResourceStorage): resource.MustParse("10G"),
+				},
+				AccessModes: []core.PersistentVolumeAccessMode{core.ReadWriteOnce},
+				PersistentVolumeSource: core.PersistentVolumeSource{
+					CSI: &core.CSIPersistentVolumeSource{
+						Driver:       "test-driver",
+						VolumeHandle: "test-123",
+					},
+				},
+				StorageClassName:          "valid",
+				VolumeAttributesClassName: ptr.To("valid"),
+			}),
+		},
+		"volume-with-good-volume-attributes-class-and-mismatched-volume-resource-when-feature-gate-is-on": {
+			isExpectedFailure:           true,
+			enableVolumeAttributesClass: true,
+			volume: testVolume("foo", "", core.PersistentVolumeSpec{
+				Capacity: core.ResourceList{
+					core.ResourceName(core.ResourceStorage): resource.MustParse("10G"),
+				},
+				AccessModes: []core.PersistentVolumeAccessMode{core.ReadWriteOnce},
+				PersistentVolumeSource: core.PersistentVolumeSource{
+					HostPath: &core.HostPathVolumeSource{
+						Path: "/foo",
+						Type: newHostPathType(string(core.HostPathDirectory)),
+					},
+				},
+				StorageClassName:          "valid",
+				VolumeAttributesClassName: ptr.To("valid"),
+			}),
+		},
 	}
 
 	for name, scenario := range scenarios {
 		t.Run(name, func(t *testing.T) {
+			defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.VolumeAttributesClass, scenario.enableVolumeAttributesClass)()
+
 			opts := ValidationOptionsForPersistentVolume(scenario.volume, nil)
 			errs := ValidatePersistentVolume(scenario.volume, opts)
 			if len(errs) == 0 && scenario.isExpectedFailure {
@@ -883,16 +959,47 @@ func TestValidatePersistentVolumeSourceUpdate(t *testing.T) {
 func TestValidationOptionsForPersistentVolume(t *testing.T) {
 	tests := map[string]struct {
 		oldPv                       *core.PersistentVolume
+		enableVolumeAttributesClass bool
 		expectValidationOpts        PersistentVolumeSpecValidationOptions
 	}{
 		"nil old pv": {
 			oldPv:                nil,
 			expectValidationOpts: PersistentVolumeSpecValidationOptions{},
 		},
+		"nil old pv and feature-gate VolumeAttrributesClass is on": {
+			oldPv:                       nil,
+			enableVolumeAttributesClass: true,
+			expectValidationOpts:        PersistentVolumeSpecValidationOptions{EnableVolumeAttributesClass: true},
+		},
+		"nil old pv and feature-gate VolumeAttrributesClass is off": {
+			oldPv:                       nil,
+			enableVolumeAttributesClass: false,
+			expectValidationOpts:        PersistentVolumeSpecValidationOptions{EnableVolumeAttributesClass: false},
+		},
+		"old pv has volumeAttributesClass and feature-gate VolumeAttrributesClass is on": {
+			oldPv: &core.PersistentVolume{
+				Spec: core.PersistentVolumeSpec{
+					VolumeAttributesClassName: ptr.To("foo"),
+				},
+			},
+			enableVolumeAttributesClass: true,
+			expectValidationOpts:        PersistentVolumeSpecValidationOptions{EnableVolumeAttributesClass: true},
+		},
+		"old pv has volumeAttributesClass and feature-gate VolumeAttrributesClass is off": {
+			oldPv: &core.PersistentVolume{
+				Spec: core.PersistentVolumeSpec{
+					VolumeAttributesClassName: ptr.To("foo"),
+				},
+			},
+			enableVolumeAttributesClass: false,
+			expectValidationOpts:        PersistentVolumeSpecValidationOptions{EnableVolumeAttributesClass: true},
+		},
 	}
 
 	for name, tc := range tests {
 		t.Run(name, func(t *testing.T) {
+			defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.VolumeAttributesClass, tc.enableVolumeAttributesClass)()
+
 			opts := ValidationOptionsForPersistentVolume(nil, tc.oldPv)
 			if opts != tc.expectValidationOpts {
 				t.Errorf("Expected opts: %+v, received: %+v", opts, tc.expectValidationOpts)
@@ -919,6 +1026,14 @@ func getCSIVolumeWithSecret(pv *core.PersistentVolume, secret *core.SecretRefere
 	return pvCopy
 }
 
+func pvcWithVolumeAttributesClassName(vacName *string) *core.PersistentVolumeClaim {
+	return &core.PersistentVolumeClaim{
+		Spec: core.PersistentVolumeClaimSpec{
+			VolumeAttributesClassName: vacName,
+		},
+	}
+}
+
 func pvcWithDataSource(dataSource *core.TypedLocalObjectReference) *core.PersistentVolumeClaim {
 	return &core.PersistentVolumeClaim{
 		Spec: core.PersistentVolumeClaimSpec{
@@ -934,6 +1049,14 @@ func pvcWithDataSourceRef(ref *core.TypedObjectReference) *core.PersistentVolume
 	}
 }
 
+func pvcTemplateWithVolumeAttributesClassName(vacName *string) *core.PersistentVolumeClaimTemplate {
+	return &core.PersistentVolumeClaimTemplate{
+		Spec: core.PersistentVolumeClaimSpec{
+			VolumeAttributesClassName: vacName,
+		},
+	}
+}
+
 func testLocalVolume(path string, affinity *core.VolumeNodeAffinity) core.PersistentVolumeSpec {
 	return core.PersistentVolumeSpec{
 		Capacity: core.ResourceList{
@@ -1001,6 +1124,24 @@ func TestValidateLocalVolumes(t *testing.T) {
 	}
 }
 
+func testVolumeWithVolumeAttributesClass(vacName *string) *core.PersistentVolume {
+	return testVolume("test-volume-with-volume-attributes-class", "",
+		core.PersistentVolumeSpec{
+			Capacity: core.ResourceList{
+				core.ResourceName(core.ResourceStorage): resource.MustParse("10G"),
+			},
+			AccessModes: []core.PersistentVolumeAccessMode{core.ReadWriteOnce},
+			PersistentVolumeSource: core.PersistentVolumeSource{
+				CSI: &core.CSIPersistentVolumeSource{
+					Driver:       "test-driver",
+					VolumeHandle: "test-123",
+				},
+			},
+			StorageClassName:          "test-storage-class",
+			VolumeAttributesClassName: vacName,
+		})
+}
+
 func testVolumeWithNodeAffinity(affinity *core.VolumeNodeAffinity) *core.PersistentVolume {
 	return testVolume("test-affinity-volume", "",
 		core.PersistentVolumeSpec{
@@ -1341,6 +1482,115 @@ func TestValidateVolumeNodeAffinityUpdate(t *testing.T) {
 	}
 }
 
+func TestValidatePeristentVolumeAttributesClassUpdate(t *testing.T) {
+	scenarios := map[string]struct {
+		isExpectedFailure           bool
+		enableVolumeAttributesClass bool
+		oldPV                       *core.PersistentVolume
+		newPV                       *core.PersistentVolume
+	}{
+		"nil-nothing-changed": {
+			isExpectedFailure:           false,
+			enableVolumeAttributesClass: true,
+			oldPV:                       testVolumeWithVolumeAttributesClass(nil),
+			newPV:                       testVolumeWithVolumeAttributesClass(nil),
+		},
+		"vac-nothing-changed": {
+			isExpectedFailure:           false,
+			enableVolumeAttributesClass: true,
+			oldPV:                       testVolumeWithVolumeAttributesClass(ptr.To("foo")),
+			newPV:                       testVolumeWithVolumeAttributesClass(ptr.To("foo")),
+		},
+		"vac-changed": {
+			isExpectedFailure:           false,
+			enableVolumeAttributesClass: true,
+			oldPV:                       testVolumeWithVolumeAttributesClass(ptr.To("foo")),
+			newPV:                       testVolumeWithVolumeAttributesClass(ptr.To("bar")),
+		},
+		"nil-to-string": {
+			isExpectedFailure:           false,
+			enableVolumeAttributesClass: true,
+			oldPV:                       testVolumeWithVolumeAttributesClass(nil),
+			newPV:                       testVolumeWithVolumeAttributesClass(ptr.To("foo")),
+		},
+		"nil-to-empty-string": {
+			isExpectedFailure:           true,
+			enableVolumeAttributesClass: true,
+			oldPV:                       testVolumeWithVolumeAttributesClass(nil),
+			newPV:                       testVolumeWithVolumeAttributesClass(ptr.To("")),
+		},
+		"string-to-nil": {
+			isExpectedFailure:           true,
+			enableVolumeAttributesClass: true,
+			oldPV:                       testVolumeWithVolumeAttributesClass(ptr.To("foo")),
+			newPV:                       testVolumeWithVolumeAttributesClass(nil),
+		},
+		"string-to-empty-string": {
+			isExpectedFailure:           true,
+			enableVolumeAttributesClass: true,
+			oldPV:                       testVolumeWithVolumeAttributesClass(ptr.To("foo")),
+			newPV:                       testVolumeWithVolumeAttributesClass(ptr.To("")),
+		},
+		"vac-nothing-changed-when-feature-gate-is-off": {
+			isExpectedFailure:           false,
+			enableVolumeAttributesClass: false,
+			oldPV:                       testVolumeWithVolumeAttributesClass(ptr.To("foo")),
+			newPV:                       testVolumeWithVolumeAttributesClass(ptr.To("foo")),
+		},
+		"vac-changed-when-feature-gate-is-off": {
+			isExpectedFailure:           true,
+			enableVolumeAttributesClass: false,
+			oldPV:                       testVolumeWithVolumeAttributesClass(ptr.To("foo")),
+			newPV:                       testVolumeWithVolumeAttributesClass(ptr.To("bar")),
+		},
+		"nil-to-string-when-feature-gate-is-off": {
+			isExpectedFailure:           true,
+			enableVolumeAttributesClass: false,
+			oldPV:                       testVolumeWithVolumeAttributesClass(nil),
+			newPV:                       testVolumeWithVolumeAttributesClass(ptr.To("foo")),
+		},
+		"nil-to-empty-string-when-feature-gate-is-off": {
+			isExpectedFailure:           true,
+			enableVolumeAttributesClass: false,
+			oldPV:                       testVolumeWithVolumeAttributesClass(nil),
+			newPV:                       testVolumeWithVolumeAttributesClass(ptr.To("")),
+		},
+		"string-to-nil-when-feature-gate-is-off": {
+			isExpectedFailure:           true,
+			enableVolumeAttributesClass: false,
+			oldPV:                       testVolumeWithVolumeAttributesClass(ptr.To("foo")),
+			newPV:                       testVolumeWithVolumeAttributesClass(nil),
+		},
+		"string-to-empty-string-when-feature-gate-is-off": {
+			isExpectedFailure:           true,
+			enableVolumeAttributesClass: false,
+			oldPV:                       testVolumeWithVolumeAttributesClass(ptr.To("foo")),
+			newPV:                       testVolumeWithVolumeAttributesClass(ptr.To("")),
+		},
+	}
+
+	for name, scenario := range scenarios {
+		defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.VolumeAttributesClass, scenario.enableVolumeAttributesClass)()
+
+		originalNewPV := scenario.newPV.DeepCopy()
+		originalOldPV := scenario.oldPV.DeepCopy()
+		opts := ValidationOptionsForPersistentVolume(scenario.newPV, scenario.oldPV)
+		errs := ValidatePersistentVolumeUpdate(scenario.newPV, scenario.oldPV, opts)
+		if len(errs) == 0 && scenario.isExpectedFailure {
+			t.Errorf("Unexpected success for scenario: %s", name)
+		}
+		if len(errs) > 0 && !scenario.isExpectedFailure {
+			t.Errorf("Unexpected failure for scenario: %s - %+v", name, errs)
+		}
+		if diff := cmp.Diff(originalNewPV, scenario.newPV); len(diff) > 0 {
+			t.Errorf("newPV was modified: %s", diff)
+		}
+		if diff := cmp.Diff(originalOldPV, scenario.oldPV); len(diff) > 0 {
+			t.Errorf("oldPV was modified: %s", diff)
+		}
+	}
+}
+
 func testVolumeClaim(name string, namespace string, spec core.PersistentVolumeClaimSpec) *core.PersistentVolumeClaim {
 	return &core.PersistentVolumeClaim{
 		ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: namespace},
@@ -1517,6 +1767,7 @@ func testValidatePVC(t *testing.T, ephemeral bool) {
 
 	scenarios := map[string]struct {
 		isExpectedFailure           bool
+		enableVolumeAttributesClass bool
 		claim                       *core.PersistentVolumeClaim
 	}{
 		"good-claim": {
@@ -1894,10 +2145,34 @@ func testValidatePVC(t *testing.T, ephemeral bool) {
 				},
 			}),
 		},
+		"invalid-volume-attributes-class-name": {
+			isExpectedFailure:           true,
+			enableVolumeAttributesClass: true,
+			claim: testVolumeClaim(goodName, goodNS, core.PersistentVolumeClaimSpec{
+				Selector: &metav1.LabelSelector{
+					MatchExpressions: []metav1.LabelSelectorRequirement{{
+						Key:      "key2",
+						Operator: "Exists",
+					}},
+				},
+				AccessModes: []core.PersistentVolumeAccessMode{
+					core.ReadWriteOnce,
+					core.ReadOnlyMany,
+				},
+				Resources: core.VolumeResourceRequirements{
+					Requests: core.ResourceList{
+						core.ResourceName(core.ResourceStorage): resource.MustParse("10G"),
+					},
+				},
+				VolumeAttributesClassName: &invalidClassName,
+			}),
+		},
 	}
 
 	for name, scenario := range scenarios {
 		t.Run(name, func(t *testing.T) {
+			defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.VolumeAttributesClass, scenario.enableVolumeAttributesClass)()
+
 			var errs field.ErrorList
 			if ephemeral {
 				volumes := []core.Volume{{
@@ -2422,11 +2697,68 @@ func TestValidatePersistentVolumeClaimUpdate(t *testing.T) {
 		},
 	})
 
+	validClaimNilVolumeAttributesClass := testVolumeClaimWithStatus("foo", "ns", core.PersistentVolumeClaimSpec{
+		AccessModes: []core.PersistentVolumeAccessMode{
+			core.ReadWriteOnce,
+			core.ReadOnlyMany,
+		},
+		Resources: core.VolumeResourceRequirements{
+			Requests: core.ResourceList{
+				core.ResourceName(core.ResourceStorage): resource.MustParse("10G"),
+			},
+		},
+	}, core.PersistentVolumeClaimStatus{
+		Phase: core.ClaimBound,
+	})
+	validClaimEmptyVolumeAttributesClass := testVolumeClaimWithStatus("foo", "ns", core.PersistentVolumeClaimSpec{
+		VolumeAttributesClassName: utilpointer.String(""),
+		AccessModes: []core.PersistentVolumeAccessMode{
+			core.ReadWriteOnce,
+			core.ReadOnlyMany,
+		},
+		Resources: core.VolumeResourceRequirements{
+			Requests: core.ResourceList{
+				core.ResourceName(core.ResourceStorage): resource.MustParse("10G"),
+			},
+		},
+	}, core.PersistentVolumeClaimStatus{
+		Phase: core.ClaimBound,
+	})
+	validClaimVolumeAttributesClass1 := testVolumeClaimWithStatus("foo", "ns", core.PersistentVolumeClaimSpec{
+		VolumeAttributesClassName: utilpointer.String("vac1"),
+		AccessModes: []core.PersistentVolumeAccessMode{
+			core.ReadWriteOnce,
+			core.ReadOnlyMany,
+		},
+		Resources: core.VolumeResourceRequirements{
+			Requests: core.ResourceList{
+				core.ResourceName(core.ResourceStorage): resource.MustParse("10G"),
+			},
+		},
+	}, core.PersistentVolumeClaimStatus{
+		Phase: core.ClaimBound,
+	})
+	validClaimVolumeAttributesClass2 := testVolumeClaimWithStatus("foo", "ns", core.PersistentVolumeClaimSpec{
+		VolumeAttributesClassName: utilpointer.String("vac2"),
+		AccessModes: []core.PersistentVolumeAccessMode{
+			core.ReadWriteOnce,
+			core.ReadOnlyMany,
+		},
+		Resources: core.VolumeResourceRequirements{
+			Requests: core.ResourceList{
+				core.ResourceName(core.ResourceStorage): resource.MustParse("10G"),
+			},
+		},
+	}, core.PersistentVolumeClaimStatus{
+		Phase: core.ClaimBound,
+	})
+
 	scenarios := map[string]struct {
 		isExpectedFailure           bool
 		oldClaim                    *core.PersistentVolumeClaim
 		newClaim                    *core.PersistentVolumeClaim
 		enableRecoverFromExpansion  bool
+		enableVolumeAttributesClass bool
 	}{
 		"valid-update-volumeName-only": {
 			isExpectedFailure: false,
@@ -2636,11 +2968,61 @@ func TestValidatePersistentVolumeClaimUpdate(t *testing.T) {
 			newClaim:          invalidClaimDataSourceRefAPIGroup,
 			isExpectedFailure: false,
 		},
+		"valid-update-volume-attributes-class-from-nil": {
+			oldClaim:                    validClaimNilVolumeAttributesClass,
+			newClaim:                    validClaimVolumeAttributesClass1,
+			enableVolumeAttributesClass: true,
+			isExpectedFailure:           false,
+		},
+		"valid-update-volume-attributes-class-from-empty": {
+			oldClaim:                    validClaimEmptyVolumeAttributesClass,
+			newClaim:                    validClaimVolumeAttributesClass1,
+			enableVolumeAttributesClass: true,
+			isExpectedFailure:           false,
+		},
+		"valid-update-volume-attributes-class": {
+			oldClaim:                    validClaimVolumeAttributesClass1,
+			newClaim:                    validClaimVolumeAttributesClass2,
+			enableVolumeAttributesClass: true,
+			isExpectedFailure:           false,
+		},
+		"invalid-update-volume-attributes-class": {
+			oldClaim:                    validClaimVolumeAttributesClass1,
+			newClaim:                    validClaimNilVolumeAttributesClass,
+			enableVolumeAttributesClass: true,
+			isExpectedFailure:           true,
+		},
+		"invalid-update-volume-attributes-class-to-nil": {
+			oldClaim:                    validClaimVolumeAttributesClass1,
+			newClaim:                    validClaimNilVolumeAttributesClass,
+			enableVolumeAttributesClass: true,
+			isExpectedFailure:           true,
+		},
+		"invalid-update-volume-attributes-class-to-empty": {
+			oldClaim:                    validClaimVolumeAttributesClass1,
+			newClaim:                    validClaimEmptyVolumeAttributesClass,
+			enableVolumeAttributesClass: true,
+			isExpectedFailure:           true,
+		},
+		"invalid-update-volume-attributes-class-to-nil-without-featuregate-enabled": {
+			oldClaim:                    validClaimVolumeAttributesClass1,
+			newClaim:                    validClaimNilVolumeAttributesClass,
+			enableVolumeAttributesClass: false,
+			isExpectedFailure:           true,
+		},
+		"invalid-update-volume-attributes-class-without-featuregate-enabled": {
+			oldClaim:                    validClaimVolumeAttributesClass1,
+			newClaim:                    validClaimVolumeAttributesClass2,
+			enableVolumeAttributesClass: false,
+			isExpectedFailure:           true,
+		},
 	}
 
 	for name, scenario := range scenarios {
 		t.Run(name, func(t *testing.T) {
 			defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.RecoverVolumeExpansionFailure, scenario.enableRecoverFromExpansion)()
+			defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.VolumeAttributesClass, scenario.enableVolumeAttributesClass)()
+
 			scenario.oldClaim.ResourceVersion = "1"
 			scenario.newClaim.ResourceVersion = "1"
 			opts := ValidationOptionsForPersistentVolumeClaim(scenario.newClaim, scenario.oldClaim)
@@ -2660,12 +3042,14 @@ func TestValidationOptionsForPersistentVolumeClaim(t *testing.T) {
 
 	tests := map[string]struct {
 		oldPvc                      *core.PersistentVolumeClaim
+		enableVolumeAttributesClass bool
 		expectValidationOpts        PersistentVolumeClaimSpecValidationOptions
 	}{
 		"nil pv": {
 			oldPvc: nil,
 			expectValidationOpts: PersistentVolumeClaimSpecValidationOptions{
 				EnableRecoverFromExpansionFailure: false,
+				EnableVolumeAttributesClass:       false,
 			},
 		},
 		"invaild apiGroup in dataSource allowed because the old pvc is used": {
@@ -2680,10 +3064,28 @@ func TestValidationOptionsForPersistentVolumeClaim(t *testing.T) {
 				AllowInvalidAPIGroupInDataSourceOrRef: true,
 			},
 		},
+		"volume attributes class allowed because feature enable": {
+			oldPvc:                      pvcWithVolumeAttributesClassName(utilpointer.String("foo")),
+			enableVolumeAttributesClass: true,
+			expectValidationOpts: PersistentVolumeClaimSpecValidationOptions{
+				EnableRecoverFromExpansionFailure: false,
+				EnableVolumeAttributesClass:       true,
+			},
+		},
+		"volume attributes class validated because used and feature disabled": {
+			oldPvc:                      pvcWithVolumeAttributesClassName(utilpointer.String("foo")),
+			enableVolumeAttributesClass: false,
+			expectValidationOpts: PersistentVolumeClaimSpecValidationOptions{
+				EnableRecoverFromExpansionFailure: false,
+				EnableVolumeAttributesClass:       true,
+			},
+		},
 	}
 
 	for name, tc := range tests {
 		t.Run(name, func(t *testing.T) {
+			defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.VolumeAttributesClass, tc.enableVolumeAttributesClass)()
+
 			opts := ValidationOptionsForPersistentVolumeClaim(nil, tc.oldPvc)
 			if opts != tc.expectValidationOpts {
 				t.Errorf("Expected opts: %+v, received: %+v", tc.expectValidationOpts, opts)
@@ -2695,16 +3097,26 @@ func TestValidationOptionsForPersistentVolumeClaim(t *testing.T) {
 func TestValidationOptionsForPersistentVolumeClaimTemplate(t *testing.T) {
 	tests := map[string]struct {
 		oldPvcTemplate              *core.PersistentVolumeClaimTemplate
+		enableVolumeAttributesClass bool
 		expectValidationOpts        PersistentVolumeClaimSpecValidationOptions
 	}{
 		"nil pv": {
 			oldPvcTemplate:       nil,
 			expectValidationOpts: PersistentVolumeClaimSpecValidationOptions{},
 		},
+		"volume attributes class allowed because feature enable": {
+			oldPvcTemplate:              pvcTemplateWithVolumeAttributesClassName(utilpointer.String("foo")),
+			enableVolumeAttributesClass: true,
+			expectValidationOpts: PersistentVolumeClaimSpecValidationOptions{
+				EnableVolumeAttributesClass: true,
+			},
+		},
 	}
 
 	for name, tc := range tests {
 		t.Run(name, func(t *testing.T) {
+			defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.VolumeAttributesClass, tc.enableVolumeAttributesClass)()
+
 			opts := ValidationOptionsForPersistentVolumeClaimTemplate(nil, tc.oldPvcTemplate)
 			if opts != tc.expectValidationOpts {
 				t.Errorf("Expected opts: %+v, received: %+v", opts, tc.expectValidationOpts)
@@ -22180,6 +22592,71 @@ func TestCrossNamespaceSource(t *testing.T) {
 	}
 }
 
+func pvcSpecWithVolumeAttributesClassName(vacName *string) *core.PersistentVolumeClaimSpec {
+	scName := "csi-plugin"
+	spec := core.PersistentVolumeClaimSpec{
+		AccessModes: []core.PersistentVolumeAccessMode{
+			core.ReadOnlyMany,
+		},
+		Resources: core.VolumeResourceRequirements{
+			Requests: core.ResourceList{
+				core.ResourceName(core.ResourceStorage): resource.MustParse("10G"),
+			},
+		},
+		StorageClassName:          &scName,
+		VolumeAttributesClassName: vacName,
+	}
+	return &spec
+}
+
+func TestVolumeAttributesClass(t *testing.T) {
+	testCases := []struct {
+		testName                    string
+		expectedFail                bool
+		enableVolumeAttributesClass bool
+		claimSpec                   *core.PersistentVolumeClaimSpec
+	}{
+		{
+			testName:                    "Feature gate enabled and valid no volumeAttributesClassName specified",
+			expectedFail:                false,
+			enableVolumeAttributesClass: true,
+			claimSpec:                   pvcSpecWithVolumeAttributesClassName(nil),
+		},
+		{
+			testName:                    "Feature gate enabled and an empty volumeAttributesClassName specified",
+			expectedFail:                false,
+			enableVolumeAttributesClass: true,
+			claimSpec:                   pvcSpecWithVolumeAttributesClassName(utilpointer.String("")),
+		},
+		{
+			testName:                    "Feature gate enabled and valid volumeAttributesClassName specified",
+			expectedFail:                false,
+			enableVolumeAttributesClass: true,
+			claimSpec:                   pvcSpecWithVolumeAttributesClassName(utilpointer.String("foo")),
+		},
+		{
+			testName:                    "Feature gate enabled and invalid volumeAttributesClassName specified",
+			expectedFail:                true,
+			enableVolumeAttributesClass: true,
+			claimSpec:                   pvcSpecWithVolumeAttributesClassName(utilpointer.String("-invalid-")),
+		},
+	}
+	for _, tc := range testCases {
+		opts := PersistentVolumeClaimSpecValidationOptions{
+			EnableVolumeAttributesClass: tc.enableVolumeAttributesClass,
+		}
+		if tc.expectedFail {
+			if errs := ValidatePersistentVolumeClaimSpec(tc.claimSpec, field.NewPath("spec"), opts); len(errs) == 0 {
+				t.Errorf("%s: expected failure: %v", tc.testName, errs)
+			}
+		} else {
+			if errs := ValidatePersistentVolumeClaimSpec(tc.claimSpec, field.NewPath("spec"), opts); len(errs) != 0 {
+				t.Errorf("%s: expected success: %v", tc.testName, errs)
+			}
+		}
+	}
+}
+
 func TestValidateTopologySpreadConstraints(t *testing.T) {
 	fieldPath := field.NewPath("field")
 	subFldPath0 := fieldPath.Index(0)

pkg/apis/storage/register.go

@@ -54,6 +54,8 @@ func addKnownTypes(scheme *runtime.Scheme) error {
 		&CSIDriverList{},
 		&CSIStorageCapacity{},
 		&CSIStorageCapacityList{},
+		&VolumeAttributesClass{},
+		&VolumeAttributesClassList{},
 	)
 	return nil
 }

pkg/apis/storage/types.go

@@ -672,3 +672,53 @@ type CSIStorageCapacityList struct {
 	// Items is the list of CSIStorageCapacity objects.
 	Items []CSIStorageCapacity
 }
+
+// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
+
+// VolumeAttributesClass represents a specification of mutable volume attributes
+// defined by the CSI driver. The class can be specified during dynamic provisioning
+// of PersistentVolumeClaims, and changed in the PersistentVolumeClaim spec after provisioning.
+type VolumeAttributesClass struct {
+	metav1.TypeMeta
+
+	// Standard object's metadata.
+	// More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#metadata
+	// +optional
+	metav1.ObjectMeta
+
+	// Name of the CSI driver
+	// This field is immutable.
+	DriverName string
+
+	// parameters hold volume attributes defined by the CSI driver. These values
+	// are opaque to the Kubernetes and are passed directly to the CSI driver.
+	// The underlying storage provider supports changing these attributes on an
+	// existing volume, however the parameters field itself is immutable. To
+	// invoke a volume update, a new VolumeAttributesClass should be created with
+	// new parameters, and the PersistentVolumeClaim should be updated to reference
+	// the new VolumeAttributesClass.
+	//
+	// This field is required and must contain at least one key/value pair.
+	// The keys cannot be empty, and the maximum number of parameters is 512, with
+	// a cumulative max size of 256K. If the CSI driver rejects invalid parameters,
+	// the target PersistentVolumeClaim will be set to an "Infeasible" state in the
+	// modifyVolumeStatus field.
+	Parameters map[string]string
+}
+
+// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
+
+// VolumeAttributesClassList is a collection of VolumeAttributesClass objects.
+type VolumeAttributesClassList struct {
+	metav1.TypeMeta
+
+	// Standard list metadata
+	// More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#metadata
+	// +optional
+	metav1.ListMeta
+
+	// items is the list of VolumeAttributesClass objects.
+	// +listType=map
+	// +listMapKey=name
+	Items []VolumeAttributesClass
+}

pkg/apis/storage/util/helpers.go

@@ -26,6 +26,9 @@ const IsDefaultStorageClassAnnotation = "storageclass.kubernetes.io/is-default-c
 // TODO: remove Beta when no longer used
 const BetaIsDefaultStorageClassAnnotation = "storageclass.beta.kubernetes.io/is-default-class"
 
+// AlphaIsDefaultVolumeAttributesClassAnnotation is the alpha version of IsDefaultVolumeAttributesClassAnnotation.
+const AlphaIsDefaultVolumeAttributesClassAnnotation = "volumeattributesclass.alpha.kubernetes.io/is-default-class"
+
 // IsDefaultAnnotation returns a boolean if
 // the annotation is set
 // TODO: remove Beta when no longer needed
@@ -39,3 +42,9 @@ func IsDefaultAnnotation(obj metav1.ObjectMeta) bool {
 
 	return false
 }
+
+// IsDefaultAnnotationForVolumeAttributesClass returns a boolean if
+// the annotation is set
+func IsDefaultAnnotationForVolumeAttributesClass(obj metav1.ObjectMeta) bool {
+	return obj.Annotations[AlphaIsDefaultVolumeAttributesClassAnnotation] == "true"
+}

pkg/apis/storage/validation/validation.go

@@ -55,7 +55,7 @@ type CSINodeValidationOptions struct {
 func ValidateStorageClass(storageClass *storage.StorageClass) field.ErrorList {
 	allErrs := apivalidation.ValidateObjectMeta(&storageClass.ObjectMeta, false, apivalidation.ValidateClassName, field.NewPath("metadata"))
 	allErrs = append(allErrs, validateProvisioner(storageClass.Provisioner, field.NewPath("provisioner"))...)
-	allErrs = append(allErrs, validateParameters(storageClass.Parameters, field.NewPath("parameters"))...)
+	allErrs = append(allErrs, validateParameters(storageClass.Parameters, true, field.NewPath("parameters"))...)
 	allErrs = append(allErrs, validateReclaimPolicy(storageClass.ReclaimPolicy, field.NewPath("reclaimPolicy"))...)
 	allErrs = append(allErrs, validateVolumeBindingMode(storageClass.VolumeBindingMode, field.NewPath("volumeBindingMode"))...)
 	allErrs = append(allErrs, validateAllowedTopologies(storageClass.AllowedTopologies, field.NewPath("allowedTopologies"))...)
@@ -95,7 +95,7 @@ func validateProvisioner(provisioner string, fldPath *field.Path) field.ErrorLis
 }
 
 // validateParameters tests that keys are qualified names and that provisionerParameter are < 256kB.
-func validateParameters(params map[string]string, fldPath *field.Path) field.ErrorList {
+func validateParameters(params map[string]string, allowEmpty bool, fldPath *field.Path) field.ErrorList {
 	var totalSize int64
 	allErrs := field.ErrorList{}
 
@@ -114,6 +114,10 @@ func validateParameters(params map[string]string, fldPath *field.Path) field.Err
 	if totalSize > maxProvisionerParameterSize {
 		allErrs = append(allErrs, field.TooLong(fldPath, "", maxProvisionerParameterSize))
 	}
+
+	if !allowEmpty && len(params) == 0 {
+		allErrs = append(allErrs, field.Required(fldPath, "must contain at least one key/value pair"))
+	}
 	return allErrs
 }
 
@@ -578,3 +582,23 @@ func ValidateCSIStorageCapacityUpdate(capacity, oldCapacity *storage.CSIStorageC
 
 	return allErrs
 }
+
+// ValidateVolumeAttributesClass validates a VolumeAttributesClass.
+func ValidateVolumeAttributesClass(volumeAttributesClass *storage.VolumeAttributesClass) field.ErrorList {
+	allErrs := apivalidation.ValidateObjectMeta(&volumeAttributesClass.ObjectMeta, false, apivalidation.ValidateClassName, field.NewPath("metadata"))
+	allErrs = append(allErrs, validateProvisioner(volumeAttributesClass.DriverName, field.NewPath("driverName"))...)
+	allErrs = append(allErrs, validateParameters(volumeAttributesClass.Parameters, false, field.NewPath("parameters"))...)
+	return allErrs
+}
+
+// ValidateVolumeAttributesClassUpdate tests if an update to VolumeAttributesClass is valid.
+func ValidateVolumeAttributesClassUpdate(volumeAttributesClass, oldVolumeAttributesClass *storage.VolumeAttributesClass) field.ErrorList {
+	allErrs := apivalidation.ValidateObjectMetaUpdate(&volumeAttributesClass.ObjectMeta, &oldVolumeAttributesClass.ObjectMeta, field.NewPath("metadata"))
+	if volumeAttributesClass.DriverName != oldVolumeAttributesClass.DriverName {
+		allErrs = append(allErrs, field.Forbidden(field.NewPath("driverName"), "updates to driverName are forbidden."))
+	}
+	if !reflect.DeepEqual(oldVolumeAttributesClass.Parameters, volumeAttributesClass.Parameters) {
+		allErrs = append(allErrs, field.Forbidden(field.NewPath("parameters"), "updates to parameters are forbidden."))
+	}
+	return allErrs
+}

pkg/apis/storage/validation/validation_test.go

@@ -2163,3 +2163,178 @@ func TestCSIDriverValidationSELinuxMountEnabledDisabled(t *testing.T) {
 		})
 	}
 }
+
+func TestValidateVolumeAttributesClass(t *testing.T) {
+	successCases := []storage.VolumeAttributesClass{
+		{
+			// driverName without a slash
+			ObjectMeta: metav1.ObjectMeta{Name: "foo"},
+			DriverName: "foo",
+			Parameters: map[string]string{
+				"foo-parameter": "free-form-string",
+			},
+		},
+		{
+			// some parameters
+			ObjectMeta: metav1.ObjectMeta{Name: "foo"},
+			DriverName: "kubernetes.io/foo",
+			Parameters: map[string]string{
+				"kubernetes.io/foo-parameter": "free/form/string",
+				"foo-parameter":               "free-form-string",
+				"foo-parameter2":              "{\"embedded\": \"json\", \"with\": {\"structures\":\"inside\"}}",
+				"foo-parameter3":              "",
+			},
+		}}
+
+	// Success cases are expected to pass validation.
+	for testName, v := range successCases {
+		if errs := ValidateVolumeAttributesClass(&v); len(errs) != 0 {
+			t.Errorf("Expected success for %d, got %v", testName, errs)
+		}
+	}
+
+	// generate a map longer than maxParameterSize
+	longParameters := make(map[string]string)
+	totalSize := 0
+	for totalSize < maxProvisionerParameterSize {
+		k := fmt.Sprintf("param/%d", totalSize)
+		v := fmt.Sprintf("value-%d", totalSize)
+		longParameters[k] = v
+		totalSize = totalSize + len(k) + len(v)
+	}
+
+	errorCases := map[string]storage.VolumeAttributesClass{
+		"namespace is present": {
+			ObjectMeta: metav1.ObjectMeta{Name: "foo", Namespace: "bar"},
+			DriverName: "kubernetes.io/foo",
+			Parameters: map[string]string{
+				"foo-parameter": "free-form-string",
+			},
+		},
+		"invalid driverName": {
+			ObjectMeta: metav1.ObjectMeta{Name: "foo"},
+			DriverName: "kubernetes.io/invalid/foo",
+			Parameters: map[string]string{
+				"foo-parameter": "free-form-string",
+			},
+		},
+		"invalid driverName with invalid chars": {
+			ObjectMeta: metav1.ObjectMeta{Name: "foo"},
+			DriverName: "^/ ",
+			Parameters: map[string]string{
+				"foo-parameter": "free-form-string",
+			},
+		},
+		"empty parameters": {
+			ObjectMeta: metav1.ObjectMeta{Name: "foo"},
+			DriverName: "kubernetes.io/foo",
+			Parameters: map[string]string{},
+		},
+		"nil parameters": {
+			ObjectMeta: metav1.ObjectMeta{Name: "foo"},
+			DriverName: "kubernetes.io/foo",
+		},
+		"invalid empty parameter name": {
+			ObjectMeta: metav1.ObjectMeta{Name: "foo"},
+			DriverName: "kubernetes.io/foo",
+			Parameters: map[string]string{
+				"": "value",
+			},
+		},
+		"driverName: Required value": {
+			ObjectMeta: metav1.ObjectMeta{Name: "foo"},
+			DriverName: "",
+			Parameters: map[string]string{
+				"foo-parameter": "free-form-string",
+			},
+		},
+		"driverName: whitespace": {
+			ObjectMeta: metav1.ObjectMeta{Name: "foo"},
+			DriverName: " ",
+			Parameters: map[string]string{
+				"foo-parameter": "free-form-string",
+			},
+		},
+		"too long parameters": {
+			ObjectMeta: metav1.ObjectMeta{Name: "foo"},
+			DriverName: "kubernetes.io/foo",
+			Parameters: longParameters,
+		},
+	}
+
+	// Error cases are not expected to pass validation.
+	for testName, v := range errorCases {
+		if errs := ValidateVolumeAttributesClass(&v); len(errs) == 0 {
+			t.Errorf("Expected failure for test: %s", testName)
+		}
+	}
+}
+
+func TestValidateVolumeAttributesClassUpdate(t *testing.T) {
+	cases := map[string]struct {
+		oldClass      *storage.VolumeAttributesClass
+		newClass      *storage.VolumeAttributesClass
+		shouldSucceed bool
+	}{
+		"invalid driverName update": {
+			oldClass: &storage.VolumeAttributesClass{
+				DriverName: "kubernetes.io/foo",
+			},
+			newClass: &storage.VolumeAttributesClass{
+				DriverName: "kubernetes.io/bar",
+			},
+			shouldSucceed: false,
+		},
+		"invalid parameter update which changes values": {
+			oldClass: &storage.VolumeAttributesClass{
+				DriverName: "kubernetes.io/foo",
+				Parameters: map[string]string{
+					"foo": "bar1",
+				},
+			},
+			newClass: &storage.VolumeAttributesClass{
+				DriverName: "kubernetes.io/foo",
+				Parameters: map[string]string{
+					"foo": "bar2",
+				},
+			},
+			shouldSucceed: false,
+		},
+		"invalid parameter update which add new item": {
+			oldClass: &storage.VolumeAttributesClass{
+				DriverName: "kubernetes.io/foo",
+				Parameters: map[string]string{},
+			},
+			newClass: &storage.VolumeAttributesClass{
+				DriverName: "kubernetes.io/foo",
+				Parameters: map[string]string{
+					"foo": "bar",
+				},
+			},
+			shouldSucceed: false,
+		},
+		"invalid parameter update which remove a item": {
+			oldClass: &storage.VolumeAttributesClass{
+				DriverName: "kubernetes.io/foo",
+				Parameters: map[string]string{
+					"foo": "bar",
+				},
+			},
+			newClass: &storage.VolumeAttributesClass{
+				DriverName: "kubernetes.io/foo",
+				Parameters: map[string]string{},
+			},
+			shouldSucceed: false,
+		},
+	}
+
+	for testName, testCase := range cases {
+		errs := ValidateVolumeAttributesClassUpdate(testCase.newClass, testCase.oldClass)
+		if testCase.shouldSucceed && len(errs) != 0 {
+			t.Errorf("Expected success for %v, got %v", testName, errs)
+		}
+		if !testCase.shouldSucceed && len(errs) == 0 {
+			t.Errorf("Expected failure for %v, got success", testName)
+		}
+	}
+}

pkg/features/kube_features.go

@@ -862,6 +862,13 @@ const (
 	// Enables user namespace support for stateless pods.
 	UserNamespacesSupport featuregate.Feature = "UserNamespacesSupport"
 
+	// owner: @mattcarry, @sunnylovestiramisu
+	// kep: https://kep.k8s.io/3751
+	// alpha: v1.29
+	//
+	// Enables user specified volume attributes for persistent volumes, like iops and throughput.
+	VolumeAttributesClass featuregate.Feature = "VolumeAttributesClass"
+
 	// owner: @cofyc
 	// alpha: v1.21
 	VolumeCapacityPriority featuregate.Feature = "VolumeCapacityPriority"
@@ -1162,6 +1169,8 @@ var defaultKubernetesFeatureGates = map[featuregate.Feature]featuregate.FeatureS
 
 	UnknownVersionInteroperabilityProxy: {Default: false, PreRelease: featuregate.Alpha},
 
+	VolumeAttributesClass: {Default: false, PreRelease: featuregate.Alpha},
+
 	VolumeCapacityPriority: {Default: false, PreRelease: featuregate.Alpha},
 
 	UserNamespacesSupport: {Default: false, PreRelease: featuregate.Alpha},

pkg/kubeapiserver/default_storage_factory_builder.go

@@ -34,6 +34,7 @@ import (
 	"k8s.io/kubernetes/pkg/apis/extensions"
 	"k8s.io/kubernetes/pkg/apis/networking"
 	"k8s.io/kubernetes/pkg/apis/policy"
+	"k8s.io/kubernetes/pkg/apis/storage"
 )
 
 // SpecialDefaultResourcePrefixes are prefixes compiled into Kubernetes.
@@ -73,6 +74,7 @@ func NewStorageFactoryConfig() *StorageFactoryConfig {
 		admissionregistration.Resource("validatingadmissionpolicybindings").WithVersion("v1beta1"),
 		networking.Resource("ipaddresses").WithVersion("v1alpha1"),
 		certificates.Resource("clustertrustbundles").WithVersion("v1alpha1"),
+		storage.Resource("volumeattributesclasses").WithVersion("v1alpha1"),
 	}
 
 	return &StorageFactoryConfig{

pkg/printers/internalversion/printers.go

@@ -43,6 +43,7 @@ import (
 	resourcev1alpha2 "k8s.io/api/resource/v1alpha2"
 	schedulingv1 "k8s.io/api/scheduling/v1"
 	storagev1 "k8s.io/api/storage/v1"
+	storagev1alpha1 "k8s.io/api/storage/v1alpha1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -305,6 +306,7 @@ func AddHandlers(h printers.PrintHandler) {
 		{Name: "Status", Type: "string", Description: apiv1.PersistentVolumeStatus{}.SwaggerDoc()["phase"]},
 		{Name: "Claim", Type: "string", Description: apiv1.PersistentVolumeSpec{}.SwaggerDoc()["claimRef"]},
 		{Name: "StorageClass", Type: "string", Description: "StorageClass of the pv"},
+		{Name: "VolumeAttributesClass", Type: "string", Description: "VolumeAttributesClass of the pv"},
 		{Name: "Reason", Type: "string", Description: apiv1.PersistentVolumeStatus{}.SwaggerDoc()["reason"]},
 		{Name: "Age", Type: "string", Description: metav1.ObjectMeta{}.SwaggerDoc()["creationTimestamp"]},
 		{Name: "VolumeMode", Type: "string", Priority: 1, Description: apiv1.PersistentVolumeSpec{}.SwaggerDoc()["volumeMode"]},
@@ -319,6 +321,7 @@ func AddHandlers(h printers.PrintHandler) {
 		{Name: "Capacity", Type: "string", Description: apiv1.PersistentVolumeClaimStatus{}.SwaggerDoc()["capacity"]},
 		{Name: "Access Modes", Type: "string", Description: apiv1.PersistentVolumeClaimStatus{}.SwaggerDoc()["accessModes"]},
 		{Name: "StorageClass", Type: "string", Description: "StorageClass of the pvc"},
+		{Name: "VolumeAttributesClass", Type: "string", Description: "VolumeAttributesClass of the pvc"},
 		{Name: "Age", Type: "string", Description: metav1.ObjectMeta{}.SwaggerDoc()["creationTimestamp"]},
 		{Name: "VolumeMode", Type: "string", Priority: 1, Description: apiv1.PersistentVolumeClaimSpec{}.SwaggerDoc()["volumeMode"]},
 	}
@@ -435,6 +438,15 @@ func AddHandlers(h printers.PrintHandler) {
 	_ = h.TableHandler(storageClassColumnDefinitions, printStorageClass)
 	_ = h.TableHandler(storageClassColumnDefinitions, printStorageClassList)
 
+	volumeAttributesClassColumnDefinitions := []metav1.TableColumnDefinition{
+		{Name: "Name", Type: "string", Format: "name", Description: metav1.ObjectMeta{}.SwaggerDoc()["name"]},
+		{Name: "DriverName", Type: "string", Description: storagev1alpha1.VolumeAttributesClass{}.SwaggerDoc()["driverName"]},
+		{Name: "Age", Type: "string", Description: metav1.ObjectMeta{}.SwaggerDoc()["creationTimestamp"]},
+	}
+
+	_ = h.TableHandler(volumeAttributesClassColumnDefinitions, printVolumeAttributesClass)
+	_ = h.TableHandler(volumeAttributesClassColumnDefinitions, printVolumeAttributesClassList)
+
 	statusColumnDefinitions := []metav1.TableColumnDefinition{
 		{Name: "Status", Type: "string", Description: metav1.Status{}.SwaggerDoc()["status"]},
 		{Name: "Reason", Type: "string", Description: metav1.Status{}.SwaggerDoc()["reason"]},
@@ -1882,8 +1894,13 @@ func printPersistentVolume(obj *api.PersistentVolume, options printers.GenerateO
 		volumeMode = string(*obj.Spec.VolumeMode)
 	}
 
+	volumeAttributeClass := "<unset>"
+	if obj.Spec.VolumeAttributesClassName != nil {
+		volumeAttributeClass = *obj.Spec.VolumeAttributesClassName
+	}
+
 	row.Cells = append(row.Cells, obj.Name, aSize, modesStr, reclaimPolicyStr,
-		string(phase), claimRefUID, helper.GetPersistentVolumeClass(obj),
+		string(phase), claimRefUID, helper.GetPersistentVolumeClass(obj), volumeAttributeClass,
 		obj.Status.Reason, translateTimestampSince(obj.CreationTimestamp), volumeMode)
 	return []metav1.TableRow{row}, nil
 }
@@ -1910,10 +1927,16 @@ func printPersistentVolumeClaim(obj *api.PersistentVolumeClaim, options printers
 		phase = "Terminating"
 	}
 
+	volumeAttributeClass := "<unset>"
 	storage := obj.Spec.Resources.Requests[api.ResourceStorage]
 	capacity := ""
 	accessModes := ""
 	volumeMode := "<unset>"
+
+	if obj.Spec.VolumeAttributesClassName != nil {
+		volumeAttributeClass = *obj.Spec.VolumeAttributesClassName
+	}
+
 	if obj.Spec.VolumeName != "" {
 		accessModes = helper.GetAccessModesAsString(obj.Status.AccessModes)
 		storage = obj.Status.Capacity[api.ResourceStorage]
@@ -1925,7 +1948,7 @@ func printPersistentVolumeClaim(obj *api.PersistentVolumeClaim, options printers
 	}
 
 	row.Cells = append(row.Cells, obj.Name, string(phase), obj.Spec.VolumeName, capacity, accessModes,
-		helper.GetPersistentVolumeClaimClass(obj), translateTimestampSince(obj.CreationTimestamp), volumeMode)
+		helper.GetPersistentVolumeClaimClass(obj), volumeAttributeClass, translateTimestampSince(obj.CreationTimestamp), volumeMode)
 	return []metav1.TableRow{row}, nil
 }
 
@@ -2434,6 +2457,33 @@ func printStorageClassList(list *storage.StorageClassList, options printers.Gene
 	return rows, nil
 }
 
+func printVolumeAttributesClass(obj *storage.VolumeAttributesClass, options printers.GenerateOptions) ([]metav1.TableRow, error) {
+	row := metav1.TableRow{
+		Object: runtime.RawExtension{Object: obj},
+	}
+
+	name := obj.Name
+	if storageutil.IsDefaultAnnotationForVolumeAttributesClass(obj.ObjectMeta) {
+		name += " (default)"
+	}
+
+	row.Cells = append(row.Cells, name, obj.DriverName, translateTimestampSince(obj.CreationTimestamp))
+
+	return []metav1.TableRow{row}, nil
+}
+
+func printVolumeAttributesClassList(list *storage.VolumeAttributesClassList, options printers.GenerateOptions) ([]metav1.TableRow, error) {
+	rows := make([]metav1.TableRow, 0, len(list.Items))
+	for i := range list.Items {
+		r, err := printVolumeAttributesClass(&list.Items[i], options)
+		if err != nil {
+			return nil, err
+		}
+		rows = append(rows, r...)
+	}
+	return rows, nil
+}
+
 func printLease(obj *coordination.Lease, options printers.GenerateOptions) ([]metav1.TableRow, error) {
 	row := metav1.TableRow{
 		Object: runtime.RawExtension{Object: obj},

pkg/printers/internalversion/printers_test.go

@@ -4711,6 +4711,7 @@ func TestPrintStatefulSet(t *testing.T) {
 
 func TestPrintPersistentVolume(t *testing.T) {
 	myScn := "my-scn"
+	myVacn := "my-vacn"
 
 	claimRef := api.ObjectReference{
 		Name:      "test",
@@ -4737,7 +4738,7 @@ func TestPrintPersistentVolume(t *testing.T) {
 					Phase: api.VolumeBound,
 				},
 			},
-			expected: []metav1.TableRow{{Cells: []interface{}{"test1", "4Gi", "ROX", "", "Bound", "default/test", "", "", "<unknown>", "<unset>"}}},
+			expected: []metav1.TableRow{{Cells: []interface{}{"test1", "4Gi", "ROX", "", "Bound", "default/test", "", "<unset>", "", "<unknown>", "<unset>"}}},
 		},
 		{
 			// Test failed
@@ -4756,7 +4757,7 @@ func TestPrintPersistentVolume(t *testing.T) {
 					Phase: api.VolumeFailed,
 				},
 			},
-			expected: []metav1.TableRow{{Cells: []interface{}{"test2", "4Gi", "ROX", "", "Failed", "default/test", "", "", "<unknown>", "<unset>"}}},
+			expected: []metav1.TableRow{{Cells: []interface{}{"test2", "4Gi", "ROX", "", "Failed", "default/test", "", "<unset>", "", "<unknown>", "<unset>"}}},
 		},
 		{
 			// Test pending
@@ -4775,7 +4776,7 @@ func TestPrintPersistentVolume(t *testing.T) {
 					Phase: api.VolumePending,
 				},
 			},
-			expected: []metav1.TableRow{{Cells: []interface{}{"test3", "10Gi", "RWX", "", "Pending", "default/test", "", "", "<unknown>", "<unset>"}}},
+			expected: []metav1.TableRow{{Cells: []interface{}{"test3", "10Gi", "RWX", "", "Pending", "default/test", "", "<unset>", "", "<unknown>", "<unset>"}}},
 		},
 		{
 			// Test pending, storageClass
@@ -4795,7 +4796,28 @@ func TestPrintPersistentVolume(t *testing.T) {
 					Phase: api.VolumePending,
 				},
 			},
-			expected: []metav1.TableRow{{Cells: []interface{}{"test4", "10Gi", "RWO", "", "Pending", "default/test", "my-scn", "", "<unknown>", "<unset>"}}},
+			expected: []metav1.TableRow{{Cells: []interface{}{"test4", "10Gi", "RWO", "", "Pending", "default/test", "my-scn", "<unset>", "", "<unknown>", "<unset>"}}},
+		},
+		{
+			// Test pending, storageClass, volumeAttributesClass
+			pv: api.PersistentVolume{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "test4",
+				},
+				Spec: api.PersistentVolumeSpec{
+					ClaimRef:                  &claimRef,
+					StorageClassName:          myScn,
+					VolumeAttributesClassName: &myVacn,
+					AccessModes:               []api.PersistentVolumeAccessMode{api.ReadWriteOnce},
+					Capacity: map[api.ResourceName]resource.Quantity{
+						api.ResourceStorage: resource.MustParse("10Gi"),
+					},
+				},
+				Status: api.PersistentVolumeStatus{
+					Phase: api.VolumePending,
+				},
+			},
+			expected: []metav1.TableRow{{Cells: []interface{}{"test4", "10Gi", "RWO", "", "Pending", "default/test", "my-scn", "my-vacn", "", "<unknown>", "<unset>"}}},
 		},
 		{
 			// Test available
@@ -4815,7 +4837,7 @@ func TestPrintPersistentVolume(t *testing.T) {
 					Phase: api.VolumeAvailable,
 				},
 			},
-			expected: []metav1.TableRow{{Cells: []interface{}{"test5", "10Gi", "RWO", "", "Available", "default/test", "my-scn", "", "<unknown>", "<unset>"}}},
+			expected: []metav1.TableRow{{Cells: []interface{}{"test5", "10Gi", "RWO", "", "Available", "default/test", "my-scn", "<unset>", "", "<unknown>", "<unset>"}}},
 		},
 		{
 			// Test released
@@ -4835,7 +4857,7 @@ func TestPrintPersistentVolume(t *testing.T) {
 					Phase: api.VolumeReleased,
 				},
 			},
-			expected: []metav1.TableRow{{Cells: []interface{}{"test6", "10Gi", "RWO", "", "Released", "default/test", "my-scn", "", "<unknown>", "<unset>"}}},
+			expected: []metav1.TableRow{{Cells: []interface{}{"test6", "10Gi", "RWO", "", "Released", "default/test", "my-scn", "<unset>", "", "<unknown>", "<unset>"}}},
 		},
 	}
 
@@ -4855,6 +4877,7 @@ func TestPrintPersistentVolume(t *testing.T) {
 
 func TestPrintPersistentVolumeClaim(t *testing.T) {
 	volumeMode := api.PersistentVolumeFilesystem
+	myVacn := "my-vacn"
 	myScn := "my-scn"
 	tests := []struct {
 		pvc      api.PersistentVolumeClaim
@@ -4878,7 +4901,7 @@ func TestPrintPersistentVolumeClaim(t *testing.T) {
 					},
 				},
 			},
-			expected: []metav1.TableRow{{Cells: []interface{}{"test1", "Bound", "my-volume", "4Gi", "ROX", "", "<unknown>", "Filesystem"}}},
+			expected: []metav1.TableRow{{Cells: []interface{}{"test1", "Bound", "my-volume", "4Gi", "ROX", "", "<unset>", "<unknown>", "Filesystem"}}},
 		},
 		{
 			// Test name, num of containers, restarts, container ready status
@@ -4897,7 +4920,7 @@ func TestPrintPersistentVolumeClaim(t *testing.T) {
 					},
 				},
 			},
-			expected: []metav1.TableRow{{Cells: []interface{}{"test2", "Lost", "", "", "", "", "<unknown>", "Filesystem"}}},
+			expected: []metav1.TableRow{{Cells: []interface{}{"test2", "Lost", "", "", "", "", "<unset>", "<unknown>", "Filesystem"}}},
 		},
 		{
 			// Test name, num of containers, restarts, container ready status
@@ -4917,7 +4940,7 @@ func TestPrintPersistentVolumeClaim(t *testing.T) {
 					},
 				},
 			},
-			expected: []metav1.TableRow{{Cells: []interface{}{"test3", "Pending", "my-volume", "10Gi", "RWX", "", "<unknown>", "Filesystem"}}},
+			expected: []metav1.TableRow{{Cells: []interface{}{"test3", "Pending", "my-volume", "10Gi", "RWX", "", "<unset>", "<unknown>", "Filesystem"}}},
 		},
 		{
 			// Test name, num of containers, restarts, container ready status
@@ -4938,7 +4961,7 @@ func TestPrintPersistentVolumeClaim(t *testing.T) {
 					},
 				},
 			},
-			expected: []metav1.TableRow{{Cells: []interface{}{"test4", "Pending", "my-volume", "10Gi", "RWO", "my-scn", "<unknown>", "Filesystem"}}},
+			expected: []metav1.TableRow{{Cells: []interface{}{"test4", "Pending", "my-volume", "10Gi", "RWO", "my-scn", "<unset>", "<unknown>", "Filesystem"}}},
 		},
 		{
 			// Test name, num of containers, restarts, container ready status
@@ -4958,7 +4981,28 @@ func TestPrintPersistentVolumeClaim(t *testing.T) {
 					},
 				},
 			},
-			expected: []metav1.TableRow{{Cells: []interface{}{"test5", "Pending", "my-volume", "10Gi", "RWO", "my-scn", "<unknown>", "<unset>"}}},
+			expected: []metav1.TableRow{{Cells: []interface{}{"test5", "Pending", "my-volume", "10Gi", "RWO", "my-scn", "<unset>", "<unknown>", "<unset>"}}},
+		},
+		{
+			// Test name, num of containers, restarts, container ready status
+			pvc: api.PersistentVolumeClaim{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "test5",
+				},
+				Spec: api.PersistentVolumeClaimSpec{
+					VolumeName:                "my-volume",
+					StorageClassName:          &myScn,
+					VolumeAttributesClassName: &myVacn,
+				},
+				Status: api.PersistentVolumeClaimStatus{
+					Phase:       api.ClaimPending,
+					AccessModes: []api.PersistentVolumeAccessMode{api.ReadWriteOnce},
+					Capacity: map[api.ResourceName]resource.Quantity{
+						api.ResourceStorage: resource.MustParse("10Gi"),
+					},
+				},
+			},
+			expected: []metav1.TableRow{{Cells: []interface{}{"test5", "Pending", "my-volume", "10Gi", "RWO", "my-scn", "my-vacn", "<unknown>", "<unset>"}}},
 		},
 	}
 
@@ -5346,6 +5390,51 @@ func TestPrintStorageClass(t *testing.T) {
 	}
 }
 
+func TestPrintVolumeAttributesClass(t *testing.T) {
+	tests := []struct {
+		vac      storage.VolumeAttributesClass
+		expected []metav1.TableRow
+	}{
+		{
+			vac: storage.VolumeAttributesClass{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:              "vac1",
+					CreationTimestamp: metav1.Time{Time: time.Now().Add(1.9e9)},
+				},
+				DriverName: "fake",
+			},
+			expected: []metav1.TableRow{{Cells: []interface{}{"vac1", "fake", "0s"}}},
+		},
+		{
+			vac: storage.VolumeAttributesClass{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:              "vac2",
+					CreationTimestamp: metav1.Time{Time: time.Now().Add(-3e11)},
+				},
+				DriverName: "fake",
+				Parameters: map[string]string{
+					"iops":       "500",
+					"throughput": "50MiB/s",
+				},
+			},
+			expected: []metav1.TableRow{{Cells: []interface{}{"vac2", "fake", "5m"}}},
+		},
+	}
+
+	for i, test := range tests {
+		rows, err := printVolumeAttributesClass(&test.vac, printers.GenerateOptions{})
+		if err != nil {
+			t.Fatal(err)
+		}
+		for i := range rows {
+			rows[i].Object.Object = nil
+		}
+		if !reflect.DeepEqual(test.expected, rows) {
+			t.Errorf("%d mismatch: %s", i, cmp.Diff(test.expected, rows))
+		}
+	}
+}
+
 func TestPrintLease(t *testing.T) {
 	holder1 := "holder1"
 	holder2 := "holder2"

pkg/registry/storage/rest/storage_storage.go

@@ -31,6 +31,7 @@ import (
 	csistoragecapacitystore "k8s.io/kubernetes/pkg/registry/storage/csistoragecapacity/storage"
 	storageclassstore "k8s.io/kubernetes/pkg/registry/storage/storageclass/storage"
 	volumeattachmentstore "k8s.io/kubernetes/pkg/registry/storage/volumeattachment/storage"
+	volumeattributesclassstore "k8s.io/kubernetes/pkg/registry/storage/volumeattributesclass/storage"
 )
 
 type RESTStorageProvider struct {
@@ -72,6 +73,15 @@ func (p RESTStorageProvider) v1alpha1Storage(apiResourceConfigSource serverstora
 		storage[resource] = csiStorageStorage.CSIStorageCapacity
 	}
 
+	// register volumeattributesclasses
+	if resource := "volumeattributesclasses"; apiResourceConfigSource.ResourceEnabled(storageapiv1alpha1.SchemeGroupVersion.WithResource(resource)) {
+		volumeAttributesClassStorage, err := volumeattributesclassstore.NewREST(restOptionsGetter)
+		if err != nil {
+			return storage, err
+		}
+		storage[resource] = volumeAttributesClassStorage
+	}
+
 	return storage, nil
 }
 

pkg/registry/storage/volumeattributesclass/doc.go

@@ -0,0 +1,19 @@
+/*
+Copyright 2023 The Kubernetes Authors.
+
+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 volumeattributesclass provides Registry interface and its REST
+// implementation for storing volumeattributesclass api objects.
+package volumeattributesclass

pkg/registry/storage/volumeattributesclass/storage/storage.go

@@ -0,0 +1,65 @@
+/*
+Copyright 2023 The Kubernetes Authors.
+
+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 storage
+
+import (
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apiserver/pkg/registry/generic"
+	genericregistry "k8s.io/apiserver/pkg/registry/generic/registry"
+	"k8s.io/apiserver/pkg/registry/rest"
+	storageapi "k8s.io/kubernetes/pkg/apis/storage"
+	"k8s.io/kubernetes/pkg/printers"
+	printersinternal "k8s.io/kubernetes/pkg/printers/internalversion"
+	printerstorage "k8s.io/kubernetes/pkg/printers/storage"
+	"k8s.io/kubernetes/pkg/registry/storage/volumeattributesclass"
+)
+
+// REST implements a RESTStorage for volume attributes classes.
+type REST struct {
+	*genericregistry.Store
+}
+
+// NewREST returns a RESTStorage object that will work against storage classes.
+func NewREST(optsGetter generic.RESTOptionsGetter) (*REST, error) {
+	store := &genericregistry.Store{
+		NewFunc:                   func() runtime.Object { return &storageapi.VolumeAttributesClass{} },
+		NewListFunc:               func() runtime.Object { return &storageapi.VolumeAttributesClassList{} },
+		DefaultQualifiedResource:  storageapi.Resource("volumeattributesclasses"),
+		SingularQualifiedResource: storageapi.Resource("volumeattributesclass"),
+
+		CreateStrategy:      volumeattributesclass.Strategy,
+		UpdateStrategy:      volumeattributesclass.Strategy,
+		DeleteStrategy:      volumeattributesclass.Strategy,
+		ReturnDeletedObject: true,
+
+		TableConvertor: printerstorage.TableConvertor{TableGenerator: printers.NewTableGenerator().With(printersinternal.AddHandlers)},
+	}
+	options := &generic.StoreOptions{RESTOptions: optsGetter}
+	if err := store.CompleteWithOptions(options); err != nil {
+		return nil, err
+	}
+
+	return &REST{store}, nil
+}
+
+// Implement ShortNamesProvider
+var _ rest.ShortNamesProvider = &REST{}
+
+// ShortNames implements the ShortNamesProvider interface. Returns a list of short names for a resource.
+func (r *REST) ShortNames() []string {
+	return []string{"vac"}
+}

pkg/registry/storage/volumeattributesclass/storage/storage_test.go

@@ -0,0 +1,148 @@
+/*
+Copyright 2023 The Kubernetes Authors.
+
+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 storage
+
+import (
+	"testing"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/fields"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apiserver/pkg/registry/generic"
+	genericregistrytest "k8s.io/apiserver/pkg/registry/generic/testing"
+	etcd3testing "k8s.io/apiserver/pkg/storage/etcd3/testing"
+	storageapi "k8s.io/kubernetes/pkg/apis/storage"
+	"k8s.io/kubernetes/pkg/registry/registrytest"
+)
+
+func newStorage(t *testing.T) (*REST, *etcd3testing.EtcdTestServer) {
+	etcdStorage, server := registrytest.NewEtcdStorageForResource(t, storageapi.SchemeGroupVersion.WithResource("volumeattributesclasses").GroupResource())
+	restOptions := generic.RESTOptions{
+		StorageConfig:           etcdStorage,
+		Decorator:               generic.UndecoratedStorage,
+		DeleteCollectionWorkers: 1,
+		ResourcePrefix:          "volumeattributesclasses",
+	}
+	volumeAttributesClassStorage, err := NewREST(restOptions)
+	if err != nil {
+		t.Fatalf("unexpected error from REST storage: %v", err)
+	}
+	return volumeAttributesClassStorage, server
+}
+
+func validNewVolumeAttributesClass(name string) *storageapi.VolumeAttributesClass {
+	return &storageapi.VolumeAttributesClass{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: name,
+		},
+		DriverName: "fake",
+		Parameters: map[string]string{
+			"foo": "bar",
+		},
+	}
+}
+
+func TestCreate(t *testing.T) {
+	storage, server := newStorage(t)
+	defer server.Terminate(t)
+	defer storage.Store.DestroyFunc()
+	test := genericregistrytest.New(t, storage.Store).ClusterScope()
+	volumeAttributesClass := validNewVolumeAttributesClass("foo")
+	volumeAttributesClass.ObjectMeta = metav1.ObjectMeta{GenerateName: "foo"}
+	test.TestCreate(
+		// valid
+		volumeAttributesClass,
+		// invalid
+		&storageapi.VolumeAttributesClass{
+			ObjectMeta: metav1.ObjectMeta{Name: "*BadName!"},
+			Parameters: map[string]string{"foo": "bar"},
+		},
+	)
+}
+
+func TestUpdate(t *testing.T) {
+	storage, server := newStorage(t)
+	defer server.Terminate(t)
+	defer storage.Store.DestroyFunc()
+	test := genericregistrytest.New(t, storage.Store).ClusterScope()
+	test.TestUpdate(
+		// valid
+		validNewVolumeAttributesClass("foo"),
+		// updateFunc
+		func(obj runtime.Object) runtime.Object {
+			object := obj.(*storageapi.VolumeAttributesClass)
+			object.Parameters = map[string]string{"foo": "bar"}
+			return object
+		},
+		// invalid update
+		func(obj runtime.Object) runtime.Object {
+			object := obj.(*storageapi.VolumeAttributesClass)
+			object.Parameters = map[string]string{"faz": "bar"}
+			return object
+		},
+	)
+
+}
+
+func TestDelete(t *testing.T) {
+	storage, server := newStorage(t)
+	defer server.Terminate(t)
+	defer storage.Store.DestroyFunc()
+	test := genericregistrytest.New(t, storage.Store).ClusterScope().ReturnDeletedObject()
+	test.TestDelete(validNewVolumeAttributesClass("foo"))
+}
+
+func TestGet(t *testing.T) {
+	storage, server := newStorage(t)
+	defer server.Terminate(t)
+	defer storage.Store.DestroyFunc()
+	test := genericregistrytest.New(t, storage.Store).ClusterScope()
+	test.TestGet(validNewVolumeAttributesClass("foo"))
+}
+
+func TestList(t *testing.T) {
+	storage, server := newStorage(t)
+	defer server.Terminate(t)
+	defer storage.Store.DestroyFunc()
+	test := genericregistrytest.New(t, storage.Store).ClusterScope()
+	test.TestList(validNewVolumeAttributesClass("foo"))
+}
+
+func TestWatch(t *testing.T) {
+	storage, server := newStorage(t)
+	defer server.Terminate(t)
+	defer storage.Store.DestroyFunc()
+	test := genericregistrytest.New(t, storage.Store).ClusterScope()
+	test.TestWatch(
+		validNewVolumeAttributesClass("foo"),
+		// matching labels
+		[]labels.Set{},
+		// not matching labels
+		[]labels.Set{
+			{"foo": "bar"},
+		},
+		// matching fields
+		[]fields.Set{
+			{"metadata.name": "foo"},
+		},
+		// not matching fields
+		[]fields.Set{
+			{"metadata.name": "bar"},
+		},
+	)
+}

pkg/registry/storage/volumeattributesclass/strategy.go

@@ -0,0 +1,82 @@
+/*
+Copyright 2023 The Kubernetes Authors.
+
+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 volumeattributesclass
+
+import (
+	"context"
+
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/util/validation/field"
+	"k8s.io/apiserver/pkg/storage/names"
+	"k8s.io/kubernetes/pkg/api/legacyscheme"
+	"k8s.io/kubernetes/pkg/apis/storage"
+	"k8s.io/kubernetes/pkg/apis/storage/validation"
+)
+
+// volumeAttributesClassStrategy implements behavior for VolumeAttributesClassStrategy objects
+type volumeAttributesClassStrategy struct {
+	runtime.ObjectTyper
+	names.NameGenerator
+}
+
+// Strategy is the default logic that applies when creating and updating
+// VolumeAttributesClass objects via the REST API.
+var Strategy = volumeAttributesClassStrategy{legacyscheme.Scheme, names.SimpleNameGenerator}
+
+func (volumeAttributesClassStrategy) NamespaceScoped() bool {
+	return false
+}
+
+// ResetBeforeCreate clears the Status field which is not allowed to be set by end users on creation.
+func (volumeAttributesClassStrategy) PrepareForCreate(ctx context.Context, obj runtime.Object) {
+}
+
+func (volumeAttributesClassStrategy) Validate(ctx context.Context, obj runtime.Object) field.ErrorList {
+	volumeAttributesClass := obj.(*storage.VolumeAttributesClass)
+	return validation.ValidateVolumeAttributesClass(volumeAttributesClass)
+}
+
+// WarningsOnCreate returns warnings for the creation of the given object.
+func (volumeAttributesClassStrategy) WarningsOnCreate(ctx context.Context, obj runtime.Object) []string {
+	return nil
+}
+
+// Canonicalize normalizes the object after validation.
+func (volumeAttributesClassStrategy) Canonicalize(obj runtime.Object) {
+}
+
+func (volumeAttributesClassStrategy) AllowCreateOnUpdate() bool {
+	return false
+}
+
+// PrepareForUpdate sets the Status fields which is not allowed to be set by an end user updating a PV
+func (volumeAttributesClassStrategy) PrepareForUpdate(ctx context.Context, obj, old runtime.Object) {
+}
+
+func (volumeAttributesClassStrategy) ValidateUpdate(ctx context.Context, obj, old runtime.Object) field.ErrorList {
+	errorList := validation.ValidateVolumeAttributesClass(obj.(*storage.VolumeAttributesClass))
+	return append(errorList, validation.ValidateVolumeAttributesClassUpdate(obj.(*storage.VolumeAttributesClass), old.(*storage.VolumeAttributesClass))...)
+}
+
+// WarningsOnUpdate returns warnings for the given update.
+func (volumeAttributesClassStrategy) WarningsOnUpdate(ctx context.Context, obj, old runtime.Object) []string {
+	return nil
+}
+
+func (volumeAttributesClassStrategy) AllowUnconditionalUpdate() bool {
+	return true
+}

pkg/registry/storage/volumeattributesclass/strategy_test.go

@@ -0,0 +1,70 @@
+/*
+Copyright 2023 The Kubernetes Authors.
+
+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 volumeattributesclass
+
+import (
+	"testing"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	genericapirequest "k8s.io/apiserver/pkg/endpoints/request"
+	"k8s.io/kubernetes/pkg/apis/storage"
+)
+
+func TestVolumeAttributesClassStrategy(t *testing.T) {
+	ctx := genericapirequest.NewDefaultContext()
+	if Strategy.NamespaceScoped() {
+		t.Errorf("VolumeAttributesClassStrategy must not be namespace scoped")
+	}
+	if Strategy.AllowCreateOnUpdate() {
+		t.Errorf("VolumeAttributesClassStrategy should not allow create on update")
+	}
+
+	class := &storage.VolumeAttributesClass{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "valid-class",
+		},
+		DriverName: "fake",
+		Parameters: map[string]string{
+			"foo": "bar",
+		},
+	}
+
+	Strategy.PrepareForCreate(ctx, class)
+
+	errs := Strategy.Validate(ctx, class)
+	if len(errs) != 0 {
+		t.Errorf("unexpected error validating %v", errs)
+	}
+
+	newClass := &storage.VolumeAttributesClass{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:            "valid-class-2",
+			ResourceVersion: "4",
+		},
+		DriverName: "fake",
+		Parameters: map[string]string{
+			"foo": "bar",
+		},
+	}
+
+	Strategy.PrepareForUpdate(ctx, newClass, class)
+
+	errs = Strategy.ValidateUpdate(ctx, newClass, class)
+	if len(errs) == 0 {
+		t.Errorf("Expected a validation error")
+	}
+}

pkg/volume/util/volumeattributesclass.go

@@ -0,0 +1,72 @@
+/*
+Copyright 2023 The Kubernetes Authors.
+
+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 util
+
+import (
+	"sort"
+
+	storagev1alpha1 "k8s.io/api/storage/v1alpha1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	storagev1alpha1listers "k8s.io/client-go/listers/storage/v1alpha1"
+	"k8s.io/klog/v2"
+)
+
+const (
+	// AlphaIsDefaultVolumeAttributesClassAnnotation is the alpha version of IsDefaultVolumeAttributesClassAnnotation.
+	AlphaIsDefaultVolumeAttributesClassAnnotation = "volumeattributesclass.alpha.kubernetes.io/is-default-class"
+)
+
+// GetDefaultVolumeAttributesClass returns the default VolumeAttributesClass from the store, or nil.
+func GetDefaultVolumeAttributesClass(lister storagev1alpha1listers.VolumeAttributesClassLister, driverName string) (*storagev1alpha1.VolumeAttributesClass, error) {
+	list, err := lister.List(labels.Everything())
+	if err != nil {
+		return nil, err
+	}
+
+	defaultClasses := []*storagev1alpha1.VolumeAttributesClass{}
+	for _, class := range list {
+		if IsDefaultVolumeAttributesClassAnnotation(class.ObjectMeta) && class.DriverName == driverName {
+			defaultClasses = append(defaultClasses, class)
+			klog.V(4).Infof("GetDefaultVolumeAttributesClass added: %s", class.Name)
+		}
+	}
+
+	if len(defaultClasses) == 0 {
+		return nil, nil
+	}
+
+	// Primary sort by creation timestamp, newest first
+	// Secondary sort by class name, ascending order
+	sort.Slice(defaultClasses, func(i, j int) bool {
+		if defaultClasses[i].CreationTimestamp.UnixNano() == defaultClasses[j].CreationTimestamp.UnixNano() {
+			return defaultClasses[i].Name < defaultClasses[j].Name
+		}
+		return defaultClasses[i].CreationTimestamp.UnixNano() > defaultClasses[j].CreationTimestamp.UnixNano()
+	})
+	if len(defaultClasses) > 1 {
+		klog.V(4).Infof("%d default VolumeAttributesClass were found, choosing: %s", len(defaultClasses), defaultClasses[0].Name)
+	}
+
+	return defaultClasses[0], nil
+}
+
+// IsDefaultVolumeAttributesClassAnnotation returns a boolean if the default
+// volume attributes class annotation is set
+func IsDefaultVolumeAttributesClassAnnotation(obj metav1.ObjectMeta) bool {
+	return obj.Annotations[AlphaIsDefaultVolumeAttributesClassAnnotation] == "true"
+}

pkg/volume/util/volumeattributesclass_test.go

@@ -0,0 +1,224 @@
+/*
+Copyright 2023 The Kubernetes Authors.
+
+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 util
+
+import (
+	"testing"
+	"time"
+
+	storagev1alpha1 "k8s.io/api/storage/v1alpha1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/informers"
+	"k8s.io/kubernetes/pkg/controller"
+)
+
+func TestGetDefaultVolumeAttributesClass(t *testing.T) {
+	var (
+		t1 = time.Now()
+		t2 = time.Now().Add(1 * time.Hour)
+	)
+
+	dirverName1 := "my-driver1"
+	vac1 := &storagev1alpha1.VolumeAttributesClass{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "my-vac1",
+			Annotations: map[string]string{
+				"a": "b",
+			},
+		},
+		DriverName: dirverName1,
+	}
+	vac2 := &storagev1alpha1.VolumeAttributesClass{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "my-vac2",
+			Annotations: map[string]string{
+				"a": "b",
+			},
+		},
+		DriverName: dirverName1,
+	}
+	vac3 := &storagev1alpha1.VolumeAttributesClass{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "my-vac3",
+			Annotations: map[string]string{
+				AlphaIsDefaultVolumeAttributesClassAnnotation: "true",
+			},
+			CreationTimestamp: metav1.Time{Time: t1},
+		},
+		DriverName: dirverName1,
+	}
+	vac4 := &storagev1alpha1.VolumeAttributesClass{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "my-vac4",
+			Annotations: map[string]string{
+				AlphaIsDefaultVolumeAttributesClassAnnotation: "true",
+			},
+			CreationTimestamp: metav1.Time{Time: t2},
+		},
+		DriverName: dirverName1,
+	}
+	vac5 := &storagev1alpha1.VolumeAttributesClass{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "my-vac5",
+			Annotations: map[string]string{
+				AlphaIsDefaultVolumeAttributesClassAnnotation: "true",
+			},
+			CreationTimestamp: metav1.Time{Time: t2},
+		},
+		DriverName: dirverName1,
+	}
+
+	dirverName2 := "my-driver2"
+	vac6 := &storagev1alpha1.VolumeAttributesClass{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "my-vac6",
+			Annotations: map[string]string{
+				"a": "b",
+			},
+		},
+		DriverName: dirverName2,
+	}
+	vac7 := &storagev1alpha1.VolumeAttributesClass{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "my-vac7",
+			Annotations: map[string]string{
+				AlphaIsDefaultVolumeAttributesClassAnnotation: "true",
+			},
+		},
+		DriverName: dirverName2,
+	}
+
+	testCases := []struct {
+		name       string
+		driverName string
+		classes    []*storagev1alpha1.VolumeAttributesClass
+		expect     *storagev1alpha1.VolumeAttributesClass
+	}{
+		{
+			name:       "no volume attributes class",
+			driverName: dirverName1,
+		},
+		{
+			name:       "no default volume attributes class",
+			driverName: dirverName1,
+			classes:    []*storagev1alpha1.VolumeAttributesClass{vac1, vac2, vac6},
+			expect:     nil,
+		},
+		{
+			name:       "no default volume attributes class for the driverName1",
+			driverName: dirverName1,
+			classes:    []*storagev1alpha1.VolumeAttributesClass{vac1, vac2, vac6, vac7},
+			expect:     nil,
+		},
+		{
+			name:       "one default volume attributes class for the driverName1",
+			driverName: dirverName1,
+			classes:    []*storagev1alpha1.VolumeAttributesClass{vac1, vac2, vac3, vac6, vac7},
+			expect:     vac3,
+		},
+		{
+			name:       "two default volume attributes class with different creation timestamp for the driverName1",
+			driverName: dirverName1,
+			classes:    []*storagev1alpha1.VolumeAttributesClass{vac3, vac4, vac6, vac7},
+			expect:     vac4,
+		},
+		{
+			name:       "two default volume attributes class with same creation timestamp for the driverName1",
+			driverName: dirverName1,
+			classes:    []*storagev1alpha1.VolumeAttributesClass{vac4, vac5, vac6, vac7},
+			expect:     vac4,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			informerFactory := informers.NewSharedInformerFactory(nil, controller.NoResyncPeriodFunc())
+			for _, c := range tc.classes {
+				err := informerFactory.Storage().V1alpha1().VolumeAttributesClasses().Informer().GetStore().Add(c)
+				if err != nil {
+					t.Errorf("Expected no error, got %v", err)
+					return
+				}
+			}
+			lister := informerFactory.Storage().V1alpha1().VolumeAttributesClasses().Lister()
+			actual, err := GetDefaultVolumeAttributesClass(lister, tc.driverName)
+			if err != nil {
+				t.Errorf("Expected no error, got %v", err)
+				return
+			}
+			if tc.expect != actual {
+				t.Errorf("Expected %v, got %v", tc.expect, actual)
+			}
+		})
+	}
+}
+
+func TestIsDefaultVolumeAttributesClassAnnotation(t *testing.T) {
+	testCases := []struct {
+		name   string
+		class  *storagev1alpha1.VolumeAttributesClass
+		expect bool
+	}{
+		{
+			name:   "no annotation",
+			class:  &storagev1alpha1.VolumeAttributesClass{},
+			expect: false,
+		},
+		{
+			name: "annotation is not boolean",
+			class: &storagev1alpha1.VolumeAttributesClass{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{
+						AlphaIsDefaultVolumeAttributesClassAnnotation: "not-boolean",
+					},
+				},
+			},
+			expect: false,
+		},
+		{
+			name: "annotation is false",
+			class: &storagev1alpha1.VolumeAttributesClass{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{
+						AlphaIsDefaultVolumeAttributesClassAnnotation: "false",
+					},
+				},
+			},
+			expect: false,
+		},
+		{
+			name: "annotation is true",
+			class: &storagev1alpha1.VolumeAttributesClass{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{
+						AlphaIsDefaultVolumeAttributesClassAnnotation: "true",
+					},
+				},
+			},
+			expect: true,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			actual := IsDefaultVolumeAttributesClassAnnotation(tc.class.ObjectMeta)
+			if tc.expect != actual {
+				t.Errorf("Expected %v, got %v", tc.expect, actual)
+			}
+		})
+	}
+}

staging/src/k8s.io/api/core/v1/types.go

@@ -363,6 +363,16 @@ type PersistentVolumeSpec struct {
 	// This field influences the scheduling of pods that use this volume.
 	// +optional
 	NodeAffinity *VolumeNodeAffinity `json:"nodeAffinity,omitempty" protobuf:"bytes,9,opt,name=nodeAffinity"`
+	// Name of VolumeAttributesClass to which this persistent volume belongs. Empty value
+	// is not allowed. When this field is not set, it indicates that this volume does not belong to any
+	// VolumeAttributesClass. This field is mutable and can be changed by the CSI driver
+	// after a volume has been updated successfully to a new class.
+	// For an unbound PersistentVolume, the volumeAttributesClassName will be matched with unbound
+	// PersistentVolumeClaims during the binding process.
+	// This is an alpha field and requires enabling VolumeAttributesClass feature.
+	// +featureGate=VolumeAttributesClass
+	// +optional
+	VolumeAttributesClassName *string `json:"volumeAttributesClassName,omitempty" protobuf:"bytes,10,opt,name=volumeAttributesClassName"`
 }
 
 // VolumeNodeAffinity defines constraints that limit what nodes this volume can be accessed from.
@@ -533,6 +543,21 @@ type PersistentVolumeClaimSpec struct {
 	// (Alpha) Using the namespace field of dataSourceRef requires the CrossNamespaceVolumeDataSource feature gate to be enabled.
 	// +optional
 	DataSourceRef *TypedObjectReference `json:"dataSourceRef,omitempty" protobuf:"bytes,8,opt,name=dataSourceRef"`
+	// volumeAttributesClassName may be used to set the VolumeAttributesClass used by this claim.
+	// If specified, the CSI driver will create or update the volume with the attributes defined
+	// in the corresponding VolumeAttributesClass. This has a different purpose than storageClassName,
+	// it can be changed after the claim is created. An empty string value means that no VolumeAttributesClass
+	// will be applied to the claim but it's not allowed to reset this field to empty string once it is set.
+	// If unspecified and the PersistentVolumeClaim is unbound, the default VolumeAttributesClass
+	// will be set by the persistentvolume controller if it exists.
+	// If the resource referred to by volumeAttributesClass does not exist, this PersistentVolumeClaim will be
+	// set to a Pending state, as reflected by the modifyVolumeStatus field, until such as a resource
+	// exists.
+	// More info: https://kubernetes.io/docs/concepts/storage/persistent-volumes#volumeattributesclass
+	// (Alpha) Using this field requires the VolumeAttributesClass feature gate to be enabled.
+	// +featureGate=VolumeAttributesClass
+	// +optional
+	VolumeAttributesClassName *string `json:"volumeAttributesClassName,omitempty" protobuf:"bytes,9,opt,name=volumeAttributesClassName"`
 }
 
 type TypedObjectReference struct {
@@ -561,6 +586,11 @@ const (
 	PersistentVolumeClaimResizing PersistentVolumeClaimConditionType = "Resizing"
 	// PersistentVolumeClaimFileSystemResizePending - controller resize is finished and a file system resize is pending on node
 	PersistentVolumeClaimFileSystemResizePending PersistentVolumeClaimConditionType = "FileSystemResizePending"
+
+	// Applying the target VolumeAttributesClass encountered an error
+	PersistentVolumeClaimVolumeModifyVolumeError PersistentVolumeClaimConditionType = "ModifyVolumeError"
+	// Volume is being modified
+	PersistentVolumeClaimVolumeModifyingVolume PersistentVolumeClaimConditionType = "ModifyingVolume"
 )
 
 // +enum
@@ -587,6 +617,38 @@ const (
 	PersistentVolumeClaimNodeResizeFailed ClaimResourceStatus = "NodeResizeFailed"
 )
 
+// +enum
+// New statuses can be added in the future. Consumers should check for unknown statuses and fail appropriately
+type PersistentVolumeClaimModifyVolumeStatus string
+
+const (
+	// Pending indicates that the PersistentVolumeClaim cannot be modified due to unmet requirements, such as
+	// the specified VolumeAttributesClass not existing
+	PersistentVolumeClaimModifyVolumePending PersistentVolumeClaimModifyVolumeStatus = "Pending"
+	// InProgress indicates that the volume is being modified
+	PersistentVolumeClaimModifyVolumeInProgress PersistentVolumeClaimModifyVolumeStatus = "InProgress"
+	// Infeasible indicates that the request has been rejected as invalid by the CSI driver. To
+	// resolve the error, a valid VolumeAttributesClass needs to be specified
+	PersistentVolumeClaimModifyVolumeInfeasible PersistentVolumeClaimModifyVolumeStatus = "Infeasible"
+)
+
+// ModifyVolumeStatus represents the status object of ControllerModifyVolume operation
+type ModifyVolumeStatus struct {
+	// targetVolumeAttributesClassName is the name of the VolumeAttributesClass the PVC currently being reconciled
+	TargetVolumeAttributesClassName string `json:"targetVolumeAttributesClassName,omitempty" protobuf:"bytes,1,opt,name=targetVolumeAttributesClassName"`
+	// status is the status of the ControllerModifyVolume operation. It can be in any of following states:
+	//  - Pending
+	//    Pending indicates that the PersistentVolumeClaim cannot be modified due to unmet requirements, such as
+	//    the specified VolumeAttributesClass not existing.
+	//  - InProgress
+	//    InProgress indicates that the volume is being modified.
+	//  - Infeasible
+	//   Infeasible indicates that the request has been rejected as invalid by the CSI driver. To
+	// 	  resolve the error, a valid VolumeAttributesClass needs to be specified.
+	// Note: New statuses can be added in the future. Consumers should check for unknown statuses and fail appropriately.
+	Status PersistentVolumeClaimModifyVolumeStatus `json:"status" protobuf:"bytes,2,opt,name=status,casttype=PersistentVolumeClaimModifyVolumeStatus"`
+}
+
 // PersistentVolumeClaimCondition contains details about state of pvc
 type PersistentVolumeClaimCondition struct {
 	Type   PersistentVolumeClaimConditionType `json:"type" protobuf:"bytes,1,opt,name=type,casttype=PersistentVolumeClaimConditionType"`
@@ -693,6 +755,18 @@ type PersistentVolumeClaimStatus struct {
 	// +mapType=granular
 	// +optional
 	AllocatedResourceStatuses map[ResourceName]ClaimResourceStatus `json:"allocatedResourceStatuses,omitempty" protobuf:"bytes,7,rep,name=allocatedResourceStatuses"`
+	// currentVolumeAttributesClassName is the current name of the VolumeAttributesClass the PVC is using.
+	// When unset, there is no VolumeAttributeClass applied to this PersistentVolumeClaim
+	// This is an alpha field and requires enabling VolumeAttributesClass feature.
+	// +featureGate=VolumeAttributesClass
+	// +optional
+	CurrentVolumeAttributesClassName *string `json:"currentVolumeAttributesClassName,omitempty" protobuf:"bytes,8,opt,name=currentVolumeAttributesClassName"`
+	// ModifyVolumeStatus represents the status object of ControllerModifyVolume operation.
+	// When this is unset, there is no ModifyVolume operation being attempted.
+	// This is an alpha field and requires enabling VolumeAttributesClass feature.
+	// +featureGate=VolumeAttributesClass
+	// +optional
+	ModifyVolumeStatus *ModifyVolumeStatus `json:"modifyVolumeStatus,omitempty" protobuf:"bytes,9,opt,name=modifyVolumeStatus"`
 }
 
 // +enum

staging/src/k8s.io/api/storage/v1alpha1/register.go

@@ -45,6 +45,8 @@ func addKnownTypes(scheme *runtime.Scheme) error {
 		&VolumeAttachmentList{},
 		&CSIStorageCapacity{},
 		&CSIStorageCapacityList{},
+		&VolumeAttributesClass{},
+		&VolumeAttributesClassList{},
 	)
 
 	metav1.AddToGroupVersion(scheme, SchemeGroupVersion)

staging/src/k8s.io/api/storage/v1alpha1/types.go

@@ -251,3 +251,55 @@ type CSIStorageCapacityList struct {
 	// +listMapKey=name
 	Items []CSIStorageCapacity `json:"items" protobuf:"bytes,2,rep,name=items"`
 }
+
+// +genclient
+// +genclient:nonNamespaced
+// +k8s:prerelease-lifecycle-gen:introduced=1.29
+// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
+
+// VolumeAttributesClass represents a specification of mutable volume attributes
+// defined by the CSI driver. The class can be specified during dynamic provisioning
+// of PersistentVolumeClaims, and changed in the PersistentVolumeClaim spec after provisioning.
+type VolumeAttributesClass struct {
+	metav1.TypeMeta `json:",inline"`
+
+	// Standard object's metadata.
+	// More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#metadata
+	// +optional
+	metav1.ObjectMeta `json:"metadata,omitempty" protobuf:"bytes,1,opt,name=metadata"`
+
+	// Name of the CSI driver
+	// This field is immutable.
+	DriverName string `json:"driverName" protobuf:"bytes,2,opt,name=driverName"`
+
+	// parameters hold volume attributes defined by the CSI driver. These values
+	// are opaque to the Kubernetes and are passed directly to the CSI driver.
+	// The underlying storage provider supports changing these attributes on an
+	// existing volume, however the parameters field itself is immutable. To
+	// invoke a volume update, a new VolumeAttributesClass should be created with
+	// new parameters, and the PersistentVolumeClaim should be updated to reference
+	// the new VolumeAttributesClass.
+	//
+	// This field is required and must contain at least one key/value pair.
+	// The keys cannot be empty, and the maximum number of parameters is 512, with
+	// a cumulative max size of 256K. If the CSI driver rejects invalid parameters,
+	// the target PersistentVolumeClaim will be set to an "Infeasible" state in the
+	// modifyVolumeStatus field.
+	Parameters map[string]string `json:"parameters,omitempty" protobuf:"bytes,3,rep,name=parameters"`
+}
+
+// +k8s:prerelease-lifecycle-gen:introduced=1.29
+// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
+
+// VolumeAttributesClassList is a collection of VolumeAttributesClass objects.
+type VolumeAttributesClassList struct {
+	metav1.TypeMeta `json:",inline"`
+
+	// Standard list metadata
+	// More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#metadata
+	// +optional
+	metav1.ListMeta `json:"metadata,omitempty" protobuf:"bytes,1,opt,name=metadata"`
+
+	// items is the list of VolumeAttributesClass objects.
+	Items []VolumeAttributesClass `json:"items" protobuf:"bytes,2,rep,name=items"`
+}

test/integration/etcd/data.go

@@ -294,6 +294,13 @@ func GetEtcdStorageDataForNamespace(namespace string) map[schema.GroupVersionRes
 		},
 		// --
 
+		// k8s.io/kubernetes/pkg/apis/storage/v1alpha1
+		gvr("storage.k8s.io", "v1alpha1", "volumeattributesclasses"): {
+			Stub:             `{"metadata": {"name": "vac1"}, "driverName": "example.com/driver", "parameters": {"foo": "bar"}}`,
+			ExpectedEtcdPath: "/registry/volumeattributesclasses/vac1",
+		},
+		// --
+
 		// k8s.io/kubernetes/pkg/apis/storage/v1beta1
 		gvr("storage.k8s.io", "v1beta1", "csistoragecapacities"): {
 			Stub:             `{"metadata": {"name": "csc-12345-2"}, "storageClassName": "sc1"}`,