Merge pull request #59391 from msau42/topology-beta

Automatic merge from submit-queue. If you want to cherry-pick this change to another branch, please follow the instructions <a href="https://github.com/kubernetes/community/blob/master/contributors/devel/cherry-picks.md">here</a>.

Move volume scheduling and local storage to beta

**What this PR does / why we need it**:
* Move the feature gates and APIs for volume scheduling and local storage to beta
* Update tests to use the beta fields
@kubernetes/sig-storage-pr-reviews 

**Which issue(s) this PR fixes** *(optional, in `fixes #<issue number>(, fixes #<issue_number>, ...)` format, will close the issue(s) when PR gets merged)*:
Fixes #59390

**Special notes for your reviewer**:

**Release note**:

```release-note
ACTION REQUIRED: VolumeScheduling and LocalPersistentVolume features are beta and enabled by default.  The PersistentVolume NodeAffinity alpha annotation is deprecated and will be removed in a future release.
```

pkg/volume/util/util.go

@@ -237,6 +237,13 @@ func GetClassForVolume(kubeClient clientset.Interface, pv *v1.PersistentVolume)
 // CheckNodeAffinity looks at the PV node affinity, and checks if the node has the same corresponding labels
 // This ensures that we don't mount a volume that doesn't belong to this node
 func CheckNodeAffinity(pv *v1.PersistentVolume, nodeLabels map[string]string) error {
+	if err := checkAlphaNodeAffinity(pv, nodeLabels); err != nil {
+		return err
+	}
+	return checkVolumeNodeAffinity(pv, nodeLabels)
+}
+
+func checkAlphaNodeAffinity(pv *v1.PersistentVolume, nodeLabels map[string]string) error {
 	affinity, err := v1helper.GetStorageNodeAffinityFromAnnotation(pv.Annotations)
 	if err != nil {
 		return fmt.Errorf("Error getting storage node affinity: %v", err)
@@ -261,6 +268,27 @@ func CheckNodeAffinity(pv *v1.PersistentVolume, nodeLabels map[string]string) er
 	return nil
 }
 
+func checkVolumeNodeAffinity(pv *v1.PersistentVolume, nodeLabels map[string]string) error {
+	if pv.Spec.NodeAffinity == nil {
+		return nil
+	}
+
+	if pv.Spec.NodeAffinity.Required != nil {
+		terms := pv.Spec.NodeAffinity.Required.NodeSelectorTerms
+		glog.V(10).Infof("Match for Required node selector terms %+v", terms)
+		for _, term := range terms {
+			selector, err := v1helper.NodeSelectorRequirementsAsSelector(term.MatchExpressions)
+			if err != nil {
+				return fmt.Errorf("Failed to parse MatchExpressions: %v", err)
+			}
+			if !selector.Matches(labels.Set(nodeLabels)) {
+				return fmt.Errorf("NodeSelectorTerm %+v does not match node labels", term.MatchExpressions)
+			}
+		}
+	}
+	return nil
+}
+
 // LoadPodFromFile will read, decode, and return a Pod from a file.
 func LoadPodFromFile(filePath string) (*v1.Pod, error) {
 	if filePath == "" {