Merge pull request kubernetes#55039 from msau42/local-binding-4

Automatic merge from submit-queue (batch tested with PRs 51321, 55969, 55039, 56183, 55976). 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>.

Topology aware volume scheduler and PV controller changes

**What this PR does / why we need it**:
Scheduler and PV controller changes to support volume topology aware scheduling, as specified in kubernetes/community#1168

**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 kubernetes#54435

**Special notes for your reviewer**:
* I've split the PR into logical commits to make it easier to review
* The remaining TODOs I plan to address next release unless you think it needs to be done now

**Release note**:
```release-note
Adds alpha support for volume scheduling, which allows the scheduler to make PersistentVolume binding decisions while respecting the Pod's scheduling requirements.  Dynamic provisioning is not supported with this feature yet.

Action required for existing users of the LocalPersistentVolumes alpha feature:
* The VolumeScheduling feature gate also has to be enabled on kube-scheduler and kube-controller-manager.
* The NoVolumeNodeConflict predicate has been removed.  For non-default schedulers, update your scheduler policy.
* The CheckVolumeBinding predicate has to be enabled in non-default schedulers.
```

@kubernetes/sig-storage-pr-reviews @kubernetes/sig-scheduling-pr-reviews

pkg/controller/volume/events/event.go

@@ -29,4 +29,5 @@ const (
 	ProvisioningFailed        = "ProvisioningFailed"
 	ProvisioningCleanupFailed = "ProvisioningCleanupFailed"
 	ProvisioningSucceeded     = "ProvisioningSucceeded"
+	WaitForFirstConsumer      = "WaitForFirstConsumer"
 )

pkg/controller/volume/persistentvolume/BUILD

@@ -12,6 +12,10 @@ go_library(
         "index.go",
         "pv_controller.go",
         "pv_controller_base.go",
+        "scheduler_assume_cache.go",
+        "scheduler_binder.go",
+        "scheduler_binder_cache.go",
+        "scheduler_binder_fake.go",
         "volume_host.go",
     ],
     importpath = "k8s.io/kubernetes/pkg/controller/volume/persistentvolume",
@@ -63,12 +67,16 @@ go_test(
         "provision_test.go",
         "pv_controller_test.go",
         "recycle_test.go",
+        "scheduler_assume_cache_test.go",
+        "scheduler_binder_cache_test.go",
+        "scheduler_binder_test.go",
     ],
     importpath = "k8s.io/kubernetes/pkg/controller/volume/persistentvolume",
     library = ":go_default_library",
     deps = [
         "//pkg/api/testapi:go_default_library",
         "//pkg/apis/core:go_default_library",
+        "//pkg/apis/core/v1/helper:go_default_library",
         "//pkg/controller:go_default_library",
         "//pkg/volume:go_default_library",
         "//vendor/github.com/golang/glog:go_default_library",

pkg/controller/volume/persistentvolume/binder_test.go

@@ -21,6 +21,7 @@ import (
 
 	"k8s.io/api/core/v1"
 	storage "k8s.io/api/storage/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
 )
 
@@ -178,6 +179,25 @@ func TestSync(t *testing.T) {
 			[]string{"Normal FailedBinding"},
 			noerrors, testSyncClaim,
 		},
+		{
+			// syncClaim does not do anything when binding is delayed
+			"1-13 - delayed binding",
+			newVolumeArray("volume1-1", "1Gi", "", "", v1.VolumePending, v1.PersistentVolumeReclaimRetain, classWait),
+			newVolumeArray("volume1-1", "1Gi", "", "", v1.VolumePending, v1.PersistentVolumeReclaimRetain, classWait),
+			newClaimArray("claim1-1", "uid1-1", "1Gi", "", v1.ClaimPending, &classWait),
+			newClaimArray("claim1-1", "uid1-1", "1Gi", "", v1.ClaimPending, &classWait),
+			[]string{"Normal WaitForFirstConsumer"},
+			noerrors, testSyncClaim,
+		},
+		{
+			// syncClaim binds when binding is delayed but PV is prebound to PVC
+			"1-14 - successful prebound PV",
+			newVolumeArray("volume1-1", "1Gi", "", "claim1-1", v1.VolumePending, v1.PersistentVolumeReclaimRetain, classWait),
+			newVolumeArray("volume1-1", "1Gi", "uid1-1", "claim1-1", v1.VolumeBound, v1.PersistentVolumeReclaimRetain, classWait),
+			newClaimArray("claim1-1", "uid1-1", "1Gi", "", v1.ClaimPending, &classWait),
+			newClaimArray("claim1-1", "uid1-1", "1Gi", "volume1-1", v1.ClaimBound, &classWait, annBoundByController, annBindCompleted),
+			noevents, noerrors, testSyncClaim,
+		},
 
 		// [Unit test set 2] User asked for a specific PV.
 		// Test the binding when pv.ClaimRef is already set by controller or
@@ -570,7 +590,15 @@ func TestSync(t *testing.T) {
 		},
 	}
 
-	runSyncTests(t, tests, []*storage.StorageClass{})
+	utilfeature.DefaultFeatureGate.Set("VolumeScheduling=true")
+	defer utilfeature.DefaultFeatureGate.Set("VolumeScheduling=false")
+
+	runSyncTests(t, tests, []*storage.StorageClass{
+		{
+			ObjectMeta:        metav1.ObjectMeta{Name: classWait},
+			VolumeBindingMode: &modeWait,
+		},
+	})
 }
 
 func TestSyncAlphaBlockVolume(t *testing.T) {

pkg/controller/volume/persistentvolume/framework_test.go

@@ -196,6 +196,8 @@ func (r *volumeReactor) React(action core.Action) (handled bool, ret runtime.Obj
 			if storedVer != requestedVer {
 				return true, obj, versionConflictError
 			}
+			// Don't modify the existing object
+			volume = volume.DeepCopy()
 			volume.ResourceVersion = strconv.Itoa(storedVer + 1)
 		} else {
 			return true, nil, fmt.Errorf("Cannot update volume %s: volume not found", volume.Name)
@@ -220,6 +222,8 @@ func (r *volumeReactor) React(action core.Action) (handled bool, ret runtime.Obj
 			if storedVer != requestedVer {
 				return true, obj, versionConflictError
 			}
+			// Don't modify the existing object
+			claim = claim.DeepCopy()
 			claim.ResourceVersion = strconv.Itoa(storedVer + 1)
 		} else {
 			return true, nil, fmt.Errorf("Cannot update claim %s: claim not found", claim.Name)
@@ -301,7 +305,12 @@ func (r *volumeReactor) checkVolumes(expectedVolumes []*v1.PersistentVolume) err
 	gotMap := make(map[string]*v1.PersistentVolume)
 	// Clear any ResourceVersion from both sets
 	for _, v := range expectedVolumes {
+		// Don't modify the existing object
+		v := v.DeepCopy()
 		v.ResourceVersion = ""
+		if v.Spec.ClaimRef != nil {
+			v.Spec.ClaimRef.ResourceVersion = ""
+		}
 		expectedMap[v.Name] = v
 	}
 	for _, v := range r.volumes {
@@ -331,6 +340,8 @@ func (r *volumeReactor) checkClaims(expectedClaims []*v1.PersistentVolumeClaim)
 	expectedMap := make(map[string]*v1.PersistentVolumeClaim)
 	gotMap := make(map[string]*v1.PersistentVolumeClaim)
 	for _, c := range expectedClaims {
+		// Don't modify the existing object
+		c = c.DeepCopy()
 		c.ResourceVersion = ""
 		expectedMap[c.Name] = c
 	}
@@ -822,6 +833,9 @@ var (
 	classUnknownInternal         string = "unknown-internal"
 	classUnsupportedMountOptions string = "unsupported-mountoptions"
 	classLarge                   string = "large"
+	classWait                    string = "wait"
+
+	modeWait = storage.VolumeBindingWaitForFirstConsumer
 )
 
 // wrapTestWithPluginCalls returns a testCall that:

pkg/controller/volume/persistentvolume/index.go

@@ -29,6 +29,7 @@ import (
 	v1helper "k8s.io/kubernetes/pkg/apis/core/v1/helper"
 	"k8s.io/kubernetes/pkg/features"
 	"k8s.io/kubernetes/pkg/volume"
+	volumeutil "k8s.io/kubernetes/pkg/volume/util"
 )
 
 // persistentVolumeOrderedIndex is a cache.Store that keeps persistent volumes
@@ -74,7 +75,7 @@ func (pvIndex *persistentVolumeOrderedIndex) listByAccessModes(modes []v1.Persis
 }
 
 // find returns the nearest PV from the ordered list or nil if a match is not found
-func (pvIndex *persistentVolumeOrderedIndex) findByClaim(claim *v1.PersistentVolumeClaim) (*v1.PersistentVolume, error) {
+func (pvIndex *persistentVolumeOrderedIndex) findByClaim(claim *v1.PersistentVolumeClaim, delayBinding bool) (*v1.PersistentVolume, error) {
 	// PVs are indexed by their access modes to allow easier searching.  Each
 	// index is the string representation of a set of access modes. There is a
 	// finite number of possible sets and PVs will only be indexed in one of
@@ -90,6 +91,45 @@ func (pvIndex *persistentVolumeOrderedIndex) findByClaim(claim *v1.PersistentVol
 	// example above).
 	allPossibleModes := pvIndex.allPossibleMatchingAccessModes(claim.Spec.AccessModes)
 
+	for _, modes := range allPossibleModes {
+		volumes, err := pvIndex.listByAccessModes(modes)
+		if err != nil {
+			return nil, err
+		}
+
+		bestVol, err := findMatchingVolume(claim, volumes, nil /* node for topology binding*/, nil /* exclusion map */, delayBinding)
+		if err != nil {
+			return nil, err
+		}
+
+		if bestVol != nil {
+			return bestVol, nil
+		}
+	}
+	return nil, nil
+}
+
+// findMatchingVolume goes through the list of volumes to find the best matching volume
+// for the claim.
+//
+// This function is used by both the PV controller and scheduler.
+//
+// delayBinding is true only in the PV controller path.  When set, prebound PVs are still returned
+// as a match for the claim, but unbound PVs are skipped.
+//
+// node is set only in the scheduler path. When set, the PV node affinity is checked against
+// the node's labels.
+//
+// excludedVolumes is only used in the scheduler path, and is needed for evaluating multiple
+// unbound PVCs for a single Pod at one time.  As each PVC finds a matching PV, the chosen
+// PV needs to be excluded from future matching.
+func findMatchingVolume(
+	claim *v1.PersistentVolumeClaim,
+	volumes []*v1.PersistentVolume,
+	node *v1.Node,
+	excludedVolumes map[string]*v1.PersistentVolume,
+	delayBinding bool) (*v1.PersistentVolume, error) {
+
 	var smallestVolume *v1.PersistentVolume
 	var smallestVolumeQty resource.Quantity
 	requestedQty := claim.Spec.Resources.Requests[v1.ResourceName(v1.ResourceStorage)]
@@ -105,67 +145,90 @@ func (pvIndex *persistentVolumeOrderedIndex) findByClaim(claim *v1.PersistentVol
 		selector = internalSelector
 	}
 
-	for _, modes := range allPossibleModes {
-		volumes, err := pvIndex.listByAccessModes(modes)
+	// Go through all available volumes with two goals:
+	// - find a volume that is either pre-bound by user or dynamically
+	//   provisioned for this claim. Because of this we need to loop through
+	//   all volumes.
+	// - find the smallest matching one if there is no volume pre-bound to
+	//   the claim.
+	for _, volume := range volumes {
+		if _, ok := excludedVolumes[volume.Name]; ok {
+			// Skip volumes in the excluded list
+			continue
+		}
+
+		volumeQty := volume.Spec.Capacity[v1.ResourceStorage]
+
+		// check if volumeModes do not match (Alpha and feature gate protected)
+		isMisMatch, err := checkVolumeModeMisMatches(&claim.Spec, &volume.Spec)
 		if err != nil {
-			return nil, err
+			return nil, fmt.Errorf("error checking if volumeMode was a mismatch: %v", err)
+		}
+		// filter out mismatching volumeModes
+		if isMisMatch {
+			continue
 		}
 
-		// Go through all available volumes with two goals:
-		// - find a volume that is either pre-bound by user or dynamically
-		//   provisioned for this claim. Because of this we need to loop through
-		//   all volumes.
-		// - find the smallest matching one if there is no volume pre-bound to
-		//   the claim.
-		for _, volume := range volumes {
-			// check if volumeModes do not match (Alpha and feature gate protected)
-			isMisMatch, err := checkVolumeModeMisMatches(&claim.Spec, &volume.Spec)
+		if node != nil {
+			// Scheduler path, check that the PV NodeAffinity
+			// is satisfied by the node
+			err := volumeutil.CheckNodeAffinity(volume, node.Labels)
 			if err != nil {
-				return nil, fmt.Errorf("error checking if volumeMode was a mismatch: %v", err)
-			}
-			// filter out mismatching volumeModes
-			if isMisMatch {
 				continue
 			}
+		}
 
-			if isVolumeBoundToClaim(volume, claim) {
-				// this claim and volume are pre-bound; return
-				// the volume if the size request is satisfied,
-				// otherwise continue searching for a match
-				volumeQty := volume.Spec.Capacity[v1.ResourceStorage]
-				if volumeQty.Cmp(requestedQty) < 0 {
-					continue
-				}
-				return volume, nil
-			}
-
-			// filter out:
-			// - volumes bound to another claim
-			// - volumes whose labels don't match the claim's selector, if specified
-			// - volumes in Class that is not requested
-			if volume.Spec.ClaimRef != nil {
-				continue
-			} else if selector != nil && !selector.Matches(labels.Set(volume.Labels)) {
+		if isVolumeBoundToClaim(volume, claim) {
+			// this claim and volume are pre-bound; return
+			// the volume if the size request is satisfied,
+			// otherwise continue searching for a match
+			if volumeQty.Cmp(requestedQty) < 0 {
 				continue
 			}
-			if v1helper.GetPersistentVolumeClass(volume) != requestedClass {
+			return volume, nil
+		}
+
+		if node == nil && delayBinding {
+			// PV controller does not bind this claim.
+			// Scheduler will handle binding unbound volumes
+			// Scheduler path will have node != nil
+			continue
+		}
+
+		// filter out:
+		// - volumes bound to another claim
+		// - volumes whose labels don't match the claim's selector, if specified
+		// - volumes in Class that is not requested
+		if volume.Spec.ClaimRef != nil {
+			continue
+		} else if selector != nil && !selector.Matches(labels.Set(volume.Labels)) {
+			continue
+		}
+		if v1helper.GetPersistentVolumeClass(volume) != requestedClass {
+			continue
+		}
+
+		if node != nil {
+			// Scheduler path
+			// Check that the access modes match
+			if !checkAccessModes(claim, volume) {
 				continue
 			}
+		}
 
-			volumeQty := volume.Spec.Capacity[v1.ResourceStorage]
-			if volumeQty.Cmp(requestedQty) >= 0 {
-				if smallestVolume == nil || smallestVolumeQty.Cmp(volumeQty) > 0 {
-					smallestVolume = volume
-					smallestVolumeQty = volumeQty
-				}
+		if volumeQty.Cmp(requestedQty) >= 0 {
+			if smallestVolume == nil || smallestVolumeQty.Cmp(volumeQty) > 0 {
+				smallestVolume = volume
+				smallestVolumeQty = volumeQty
 			}
 		}
+	}
 
-		if smallestVolume != nil {
-			// Found a matching volume
-			return smallestVolume, nil
-		}
+	if smallestVolume != nil {
+		// Found a matching volume
+		return smallestVolume, nil
 	}
+
 	return nil, nil
 }
 
@@ -191,8 +254,8 @@ func checkVolumeModeMisMatches(pvcSpec *v1.PersistentVolumeClaimSpec, pvSpec *v1
 }
 
 // findBestMatchForClaim is a convenience method that finds a volume by the claim's AccessModes and requests for Storage
-func (pvIndex *persistentVolumeOrderedIndex) findBestMatchForClaim(claim *v1.PersistentVolumeClaim) (*v1.PersistentVolume, error) {
-	return pvIndex.findByClaim(claim)
+func (pvIndex *persistentVolumeOrderedIndex) findBestMatchForClaim(claim *v1.PersistentVolumeClaim, delayBinding bool) (*v1.PersistentVolume, error) {
+	return pvIndex.findByClaim(claim, delayBinding)
 }
 
 // allPossibleMatchingAccessModes returns an array of AccessMode arrays that
@@ -274,3 +337,19 @@ func claimToClaimKey(claim *v1.PersistentVolumeClaim) string {
 func claimrefToClaimKey(claimref *v1.ObjectReference) string {
 	return fmt.Sprintf("%s/%s", claimref.Namespace, claimref.Name)
 }
+
+// Returns true if PV satisfies all the PVC's requested AccessModes
+func checkAccessModes(claim *v1.PersistentVolumeClaim, volume *v1.PersistentVolume) bool {
+	pvModesMap := map[v1.PersistentVolumeAccessMode]bool{}
+	for _, mode := range volume.Spec.AccessModes {
+		pvModesMap[mode] = true
+	}
+
+	for _, mode := range claim.Spec.AccessModes {
+		_, ok := pvModesMap[mode]
+		if !ok {
+			return false
+		}
+	}
+	return true
+}