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

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
+}

pkg/controller/volume/persistentvolume/pv_controller.go

@@ -26,6 +26,7 @@ import (
 	storage "k8s.io/api/storage/v1"
 	apierrs "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	clientset "k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/kubernetes/scheme"
 	corelisters "k8s.io/client-go/listers/core/v1"
@@ -37,6 +38,7 @@ import (
 	v1helper "k8s.io/kubernetes/pkg/apis/core/v1/helper"
 	"k8s.io/kubernetes/pkg/cloudprovider"
 	"k8s.io/kubernetes/pkg/controller/volume/events"
+	"k8s.io/kubernetes/pkg/features"
 	"k8s.io/kubernetes/pkg/util/goroutinemap"
 	"k8s.io/kubernetes/pkg/util/goroutinemap/exponentialbackoff"
 	vol "k8s.io/kubernetes/pkg/volume"
@@ -254,6 +256,30 @@ func checkVolumeSatisfyClaim(volume *v1.PersistentVolume, claim *v1.PersistentVo
 	return nil
 }
 
+func (ctrl *PersistentVolumeController) shouldDelayBinding(claim *v1.PersistentVolumeClaim) (bool, error) {
+	if !utilfeature.DefaultFeatureGate.Enabled(features.VolumeScheduling) {
+		return false, nil
+	}
+
+	className := v1helper.GetPersistentVolumeClaimClass(claim)
+	if className == "" {
+		return false, nil
+	}
+
+	class, err := ctrl.classLister.Get(className)
+	if err != nil {
+		return false, nil
+	}
+
+	if class.VolumeBindingMode == nil {
+		return false, fmt.Errorf("VolumeBindingMode not set for StorageClass %q", className)
+	}
+
+	// TODO: add check to handle dynamic provisioning later
+
+	return *class.VolumeBindingMode == storage.VolumeBindingWaitForFirstConsumer, nil
+}
+
 // syncUnboundClaim is the main controller method to decide what to do with an
 // unbound claim.
 func (ctrl *PersistentVolumeController) syncUnboundClaim(claim *v1.PersistentVolumeClaim) error {
@@ -261,9 +287,13 @@ func (ctrl *PersistentVolumeController) syncUnboundClaim(claim *v1.PersistentVol
 	// OBSERVATION: pvc is "Pending"
 	if claim.Spec.VolumeName == "" {
 		// User did not care which PV they get.
+		delayBinding, err := ctrl.shouldDelayBinding(claim)
+		if err != nil {
+			return err
+		}
 
 		// [Unit test set 1]
-		volume, err := ctrl.volumes.findBestMatchForClaim(claim)
+		volume, err := ctrl.volumes.findBestMatchForClaim(claim, delayBinding)
 		if err != nil {
 			glog.V(2).Infof("synchronizing unbound PersistentVolumeClaim[%s]: Error finding PV for claim: %v", claimToClaimKey(claim), err)
 			return fmt.Errorf("Error finding PV for claim %q: %v", claimToClaimKey(claim), err)
@@ -272,15 +302,21 @@ func (ctrl *PersistentVolumeController) syncUnboundClaim(claim *v1.PersistentVol
 			glog.V(4).Infof("synchronizing unbound PersistentVolumeClaim[%s]: no volume found", claimToClaimKey(claim))
 			// No PV could be found
 			// OBSERVATION: pvc is "Pending", will retry
-			if v1helper.GetPersistentVolumeClaimClass(claim) != "" {
+			switch {
+			case delayBinding:
+				// TODO: Skip dynamic provisioning for now
+				ctrl.eventRecorder.Event(claim, v1.EventTypeNormal, events.WaitForFirstConsumer, "waiting for first consumer to be created before binding")
+			case v1helper.GetPersistentVolumeClaimClass(claim) != "":
 				if err = ctrl.provisionClaim(claim); err != nil {
 					return err
 				}
 				return nil
+			default:
+				ctrl.eventRecorder.Event(claim, v1.EventTypeNormal, events.FailedBinding, "no persistent volumes available for this claim and no storage class is set")
 			}
+
 			// Mark the claim as Pending and try to find a match in the next
 			// periodic syncClaim
-			ctrl.eventRecorder.Event(claim, v1.EventTypeNormal, events.FailedBinding, "no persistent volumes available for this claim and no storage class is set")
 			if _, err = ctrl.updateClaimStatus(claim, v1.ClaimPending, nil); err != nil {
 				return err
 			}
@@ -748,6 +784,42 @@ func (ctrl *PersistentVolumeController) updateVolumePhaseWithEvent(volume *v1.Pe
 func (ctrl *PersistentVolumeController) bindVolumeToClaim(volume *v1.PersistentVolume, claim *v1.PersistentVolumeClaim) (*v1.PersistentVolume, error) {
 	glog.V(4).Infof("updating PersistentVolume[%s]: binding to %q", volume.Name, claimToClaimKey(claim))
 
+	volumeClone, dirty, err := ctrl.getBindVolumeToClaim(volume, claim)
+	if err != nil {
+		return nil, err
+	}
+
+	// Save the volume only if something was changed
+	if dirty {
+		return ctrl.updateBindVolumeToClaim(volumeClone, claim, true)
+	}
+
+	glog.V(4).Infof("updating PersistentVolume[%s]: already bound to %q", volume.Name, claimToClaimKey(claim))
+	return volume, nil
+}
+
+// bindVolumeToClaim modifies given volume to be bound to a claim and saves it to
+// API server. The claim is not modified in this method!
+func (ctrl *PersistentVolumeController) updateBindVolumeToClaim(volumeClone *v1.PersistentVolume, claim *v1.PersistentVolumeClaim, updateCache bool) (*v1.PersistentVolume, error) {
+	glog.V(2).Infof("claim %q bound to volume %q", claimToClaimKey(claim), volumeClone.Name)
+	newVol, err := ctrl.kubeClient.Core().PersistentVolumes().Update(volumeClone)
+	if err != nil {
+		glog.V(4).Infof("updating PersistentVolume[%s]: binding to %q failed: %v", volumeClone.Name, claimToClaimKey(claim), err)
+		return newVol, err
+	}
+	if updateCache {
+		_, err = ctrl.storeVolumeUpdate(newVol)
+		if err != nil {
+			glog.V(4).Infof("updating PersistentVolume[%s]: cannot update internal cache: %v", volumeClone.Name, err)
+			return newVol, err
+		}
+	}
+	glog.V(4).Infof("updating PersistentVolume[%s]: bound to %q", newVol.Name, claimToClaimKey(claim))
+	return newVol, nil
+}
+
+// Get new PV object only, no API or cache update
+func (ctrl *PersistentVolumeController) getBindVolumeToClaim(volume *v1.PersistentVolume, claim *v1.PersistentVolumeClaim) (*v1.PersistentVolume, bool, error) {
 	dirty := false
 
 	// Check if the volume was already bound (either by user or by controller)
@@ -768,7 +840,7 @@ func (ctrl *PersistentVolumeController) bindVolumeToClaim(volume *v1.PersistentV
 
 		claimRef, err := ref.GetReference(scheme.Scheme, claim)
 		if err != nil {
-			return nil, fmt.Errorf("Unexpected error getting claim reference: %v", err)
+			return nil, false, fmt.Errorf("Unexpected error getting claim reference: %v", err)
 		}
 		volumeClone.Spec.ClaimRef = claimRef
 		dirty = true
@@ -780,25 +852,7 @@ func (ctrl *PersistentVolumeController) bindVolumeToClaim(volume *v1.PersistentV
 		dirty = true
 	}
 
-	// Save the volume only if something was changed
-	if dirty {
-		glog.V(2).Infof("claim %q bound to volume %q", claimToClaimKey(claim), volume.Name)
-		newVol, err := ctrl.kubeClient.CoreV1().PersistentVolumes().Update(volumeClone)
-		if err != nil {
-			glog.V(4).Infof("updating PersistentVolume[%s]: binding to %q failed: %v", volume.Name, claimToClaimKey(claim), err)
-			return newVol, err
-		}
-		_, err = ctrl.storeVolumeUpdate(newVol)
-		if err != nil {
-			glog.V(4).Infof("updating PersistentVolume[%s]: cannot update internal cache: %v", volume.Name, err)
-			return newVol, err
-		}
-		glog.V(4).Infof("updating PersistentVolume[%s]: bound to %q", newVol.Name, claimToClaimKey(claim))
-		return newVol, nil
-	}
-
-	glog.V(4).Infof("updating PersistentVolume[%s]: already bound to %q", volume.Name, claimToClaimKey(claim))
-	return volume, nil
+	return volumeClone, dirty, nil
 }
 
 // bindClaimToVolume modifies the given claim to be bound to a volume and

pkg/controller/volume/persistentvolume/pv_controller_test.go

@@ -21,8 +21,12 @@ import (
 	"time"
 
 	"github.com/golang/glog"
+
 	"k8s.io/api/core/v1"
+	storagev1 "k8s.io/api/storage/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/watch"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/client-go/informers"
 	"k8s.io/client-go/kubernetes/fake"
 	core "k8s.io/client-go/testing"
@@ -232,3 +236,106 @@ func addVolumeAnnotation(volume *v1.PersistentVolume, annName, annValue string)
 	volume.Annotations[annName] = annValue
 	return volume
 }
+
+func makePVCClass(scName *string) *v1.PersistentVolumeClaim {
+	return &v1.PersistentVolumeClaim{
+		Spec: v1.PersistentVolumeClaimSpec{
+			StorageClassName: scName,
+		},
+	}
+}
+
+func makeStorageClass(scName string, mode *storagev1.VolumeBindingMode) *storagev1.StorageClass {
+	return &storagev1.StorageClass{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: scName,
+		},
+		VolumeBindingMode: mode,
+	}
+}
+
+func TestDelayBinding(t *testing.T) {
+	var (
+		classNotHere       = "not-here"
+		classNoMode        = "no-mode"
+		classImmediateMode = "immediate-mode"
+		classWaitMode      = "wait-mode"
+
+		modeImmediate = storagev1.VolumeBindingImmediate
+		modeWait      = storagev1.VolumeBindingWaitForFirstConsumer
+	)
+
+	tests := map[string]struct {
+		pvc         *v1.PersistentVolumeClaim
+		shouldDelay bool
+		shouldFail  bool
+	}{
+		"nil-class": {
+			pvc:         makePVCClass(nil),
+			shouldDelay: false,
+		},
+		"class-not-found": {
+			pvc:         makePVCClass(&classNotHere),
+			shouldDelay: false,
+		},
+		"no-mode-class": {
+			pvc:         makePVCClass(&classNoMode),
+			shouldDelay: false,
+			shouldFail:  true,
+		},
+		"immediate-mode-class": {
+			pvc:         makePVCClass(&classImmediateMode),
+			shouldDelay: false,
+		},
+		"wait-mode-class": {
+			pvc:         makePVCClass(&classWaitMode),
+			shouldDelay: true,
+		},
+	}
+
+	classes := []*storagev1.StorageClass{
+		makeStorageClass(classNoMode, nil),
+		makeStorageClass(classImmediateMode, &modeImmediate),
+		makeStorageClass(classWaitMode, &modeWait),
+	}
+
+	client := &fake.Clientset{}
+	informerFactory := informers.NewSharedInformerFactory(client, controller.NoResyncPeriodFunc())
+	classInformer := informerFactory.Storage().V1().StorageClasses()
+	ctrl := &PersistentVolumeController{
+		classLister: classInformer.Lister(),
+	}
+
+	for _, class := range classes {
+		if err := classInformer.Informer().GetIndexer().Add(class); err != nil {
+			t.Fatalf("Failed to add storage class %q: %v", class.Name, err)
+		}
+	}
+
+	// When feature gate is disabled, should always be delayed
+	name := "feature-disabled"
+	shouldDelay, err := ctrl.shouldDelayBinding(makePVCClass(&classWaitMode))
+	if err != nil {
+		t.Errorf("Test %q returned error: %v", name, err)
+	}
+	if shouldDelay {
+		t.Errorf("Test %q returned true, expected false", name)
+	}
+
+	// Enable feature gate
+	utilfeature.DefaultFeatureGate.Set("VolumeScheduling=true")
+	defer utilfeature.DefaultFeatureGate.Set("VolumeScheduling=false")
+
+	for name, test := range tests {
+		shouldDelay, err = ctrl.shouldDelayBinding(test.pvc)
+		if err != nil && !test.shouldFail {
+			t.Errorf("Test %q returned error: %v", name, err)
+		}
+		if err == nil && test.shouldFail {
+			t.Errorf("Test %q returned success, expected error", name)
+		}
+		if shouldDelay != test.shouldDelay {
+			t.Errorf("Test %q returned unexpected %v", name, test.shouldDelay)
+		}
+	}
+}

pkg/controller/volume/persistentvolume/scheduler_assume_cache.go

+/*
+Copyright 2017 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 persistentvolume
+
+import (
+	"fmt"
+	"strconv"
+	"sync"
+
+	"github.com/golang/glog"
+
+	"k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/meta"
+	"k8s.io/client-go/tools/cache"
+)
+
+// AssumeCache is a cache on top of the informer that allows for updating
+// objects outside of informer events and also restoring the informer
+// cache's version of the object.  Objects are assumed to be
+// Kubernetes API objects that implement meta.Interface
+type AssumeCache interface {
+	// Assume updates the object in-memory only
+	Assume(obj interface{}) error
+
+	// Restore the informer cache's version of the object
+	Restore(objName string)
+
+	// Get the object by name
+	Get(objName string) (interface{}, error)
+
+	// List all the objects in the cache
+	List() []interface{}
+}
+
+type errWrongType struct {
+	typeName string
+	object   interface{}
+}
+
+func (e *errWrongType) Error() string {
+	return fmt.Sprintf("could not convert object to type %v: %+v", e.typeName, e.object)
+}
+
+type errNotFound struct {
+	typeName   string
+	objectName string
+}
+
+func (e *errNotFound) Error() string {
+	return fmt.Sprintf("could not find %v %q", e.typeName, e.objectName)
+}
+
+type errObjectName struct {
+	detailedErr error
+}
+
+func (e *errObjectName) Error() string {
+	return fmt.Sprintf("failed to get object name: %v", e.detailedErr)
+}
+
+// assumeCache stores two pointers to represent a single object:
+// * The pointer to the informer object.
+// * The pointer to the latest object, which could be the same as
+//   the informer object, or an in-memory object.
+//
+// An informer update always overrides the latest object pointer.
+//
+// Assume() only updates the latest object pointer.
+// Restore() sets the latest object pointer back to the informer object.
+// Get/List() always returns the latest object pointer.
+type assumeCache struct {
+	mutex sync.Mutex
+
+	// describes the object stored
+	description string
+
+	// Stores objInfo pointers
+	store cache.Store
+}
+
+type objInfo struct {
+	// name of the object
+	name string
+
+	// Latest version of object could be cached-only or from informer
+	latestObj interface{}
+
+	// Latest object from informer
+	apiObj interface{}
+}
+
+func objInfoKeyFunc(obj interface{}) (string, error) {
+	objInfo, ok := obj.(*objInfo)
+	if !ok {
+		return "", &errWrongType{"objInfo", obj}
+	}
+	return objInfo.name, nil
+}
+
+func NewAssumeCache(informer cache.SharedIndexInformer, description string) *assumeCache {
+	// TODO: index by storageclass
+	c := &assumeCache{store: cache.NewStore(objInfoKeyFunc), description: description}
+
+	// Unit tests don't use informers
+	if informer != nil {
+		informer.AddEventHandler(
+			cache.ResourceEventHandlerFuncs{
+				AddFunc:    c.add,
+				UpdateFunc: c.update,
+				DeleteFunc: c.delete,
+			},
+		)
+	}
+	return c
+}
+
+func (c *assumeCache) add(obj interface{}) {
+	if obj == nil {
+		return
+	}
+
+	name, err := cache.MetaNamespaceKeyFunc(obj)
+	if err != nil {
+		glog.Errorf("add failed: %v", &errObjectName{err})
+		return
+	}
+
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+
+	objInfo := &objInfo{name: name, latestObj: obj, apiObj: obj}
+	c.store.Update(objInfo)
+}
+
+func (c *assumeCache) update(oldObj interface{}, newObj interface{}) {
+	c.add(newObj)
+}
+
+func (c *assumeCache) delete(obj interface{}) {
+	if obj == nil {
+		return
+	}
+
+	name, err := cache.MetaNamespaceKeyFunc(obj)
+	if err != nil {
+		glog.Errorf("delete failed: %v", &errObjectName{err})
+		return
+	}
+
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+
+	objInfo := &objInfo{name: name}
+	err = c.store.Delete(objInfo)
+	if err != nil {
+		glog.Errorf("delete: failed to delete %v %v: %v", c.description, name, err)
+	}
+}
+
+func (c *assumeCache) getObjVersion(name string, obj interface{}) (int64, error) {
+	objAccessor, err := meta.Accessor(obj)
+	if err != nil {
+		return -1, err
+	}
+
+	objResourceVersion, err := strconv.ParseInt(objAccessor.GetResourceVersion(), 10, 64)
+	if err != nil {
+		return -1, fmt.Errorf("error parsing ResourceVersion %q for %v %q: %s", objAccessor.GetResourceVersion(), c.description, name, err)
+	}
+	return objResourceVersion, nil
+}
+
+func (c *assumeCache) getObjInfo(name string) (*objInfo, error) {
+	obj, ok, err := c.store.GetByKey(name)
+	if err != nil {
+		return nil, err
+	}
+	if !ok {
+		return nil, &errNotFound{c.description, name}
+	}
+
+	objInfo, ok := obj.(*objInfo)
+	if !ok {
+		return nil, &errWrongType{"objInfo", obj}
+	}
+	return objInfo, nil
+}
+
+func (c *assumeCache) Get(objName string) (interface{}, error) {
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+
+	objInfo, err := c.getObjInfo(objName)
+	if err != nil {
+		return nil, err
+	}
+	return objInfo.latestObj, nil
+}
+
+func (c *assumeCache) List() []interface{} {
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+
+	allObjs := []interface{}{}
+	for _, obj := range c.store.List() {
+		objInfo, ok := obj.(*objInfo)
+		if !ok {
+			glog.Errorf("list error: %v", &errWrongType{"objInfo", obj})
+			continue
+		}
+		allObjs = append(allObjs, objInfo.latestObj)
+	}
+	return allObjs
+}
+
+func (c *assumeCache) Assume(obj interface{}) error {
+	name, err := cache.MetaNamespaceKeyFunc(obj)
+	if err != nil {
+		return &errObjectName{err}
+	}
+
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+
+	objInfo, err := c.getObjInfo(name)
+	if err != nil {
+		return err
+	}
+
+	newVersion, err := c.getObjVersion(name, obj)
+	if err != nil {
+		return err
+	}
+
+	storedVersion, err := c.getObjVersion(name, objInfo.latestObj)
+	if err != nil {
+		return err
+	}
+
+	if newVersion < storedVersion {
+		return fmt.Errorf("%v %q is out of sync", c.description, name)
+	}
+
+	// Only update the cached object
+	objInfo.latestObj = obj
+	glog.V(4).Infof("Assumed %v %q, version %v", c.description, name, newVersion)
+	return nil
+}
+
+func (c *assumeCache) Restore(objName string) {
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+
+	objInfo, err := c.getObjInfo(objName)
+	if err != nil {
+		// This could be expected if object got deleted
+		glog.V(5).Infof("Restore %v %q warning: %v", c.description, objName, err)
+	} else {
+		objInfo.latestObj = objInfo.apiObj
+		glog.V(4).Infof("Restored %v %q", c.description, objName)
+	}
+}
+
+// PVAssumeCache is a AssumeCache for PersistentVolume objects
+type PVAssumeCache interface {
+	AssumeCache
+
+	GetPV(pvName string) (*v1.PersistentVolume, error)
+	ListPVs() []*v1.PersistentVolume
+}
+
+type pvAssumeCache struct {
+	*assumeCache
+}
+
+func NewPVAssumeCache(informer cache.SharedIndexInformer) PVAssumeCache {
+	return &pvAssumeCache{assumeCache: NewAssumeCache(informer, "v1.PersistentVolume")}
+}
+
+func (c *pvAssumeCache) GetPV(pvName string) (*v1.PersistentVolume, error) {
+	obj, err := c.Get(pvName)
+	if err != nil {
+		return nil, err
+	}
+
+	pv, ok := obj.(*v1.PersistentVolume)
+	if !ok {
+		return nil, &errWrongType{"v1.PersistentVolume", obj}
+	}
+	return pv, nil
+}
+
+func (c *pvAssumeCache) ListPVs() []*v1.PersistentVolume {
+	objs := c.List()
+	pvs := []*v1.PersistentVolume{}
+	for _, obj := range objs {
+		pv, ok := obj.(*v1.PersistentVolume)
+		if !ok {
+			glog.Errorf("ListPVs: %v", &errWrongType{"v1.PersistentVolume", obj})
+		}
+		pvs = append(pvs, pv)
+	}
+	return pvs
+}

pkg/controller/volume/persistentvolume/scheduler_assume_cache_test.go

+/*
+Copyright 2017 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 persistentvolume
+
+import (
+	"fmt"
+	"testing"
+
+	"k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func makePV(name, version string) *v1.PersistentVolume {
+	return &v1.PersistentVolume{ObjectMeta: metav1.ObjectMeta{Name: name, ResourceVersion: version}}
+}
+
+func TestAssumePV(t *testing.T) {
+	scenarios := map[string]struct {
+		oldPV         *v1.PersistentVolume
+		newPV         *v1.PersistentVolume
+		shouldSucceed bool
+	}{
+		"success-same-version": {
+			oldPV:         makePV("pv1", "5"),
+			newPV:         makePV("pv1", "5"),
+			shouldSucceed: true,
+		},
+		"success-new-higher-version": {
+			oldPV:         makePV("pv1", "5"),
+			newPV:         makePV("pv1", "6"),
+			shouldSucceed: true,
+		},
+		"fail-old-not-found": {
+			oldPV:         makePV("pv2", "5"),
+			newPV:         makePV("pv1", "5"),
+			shouldSucceed: false,
+		},
+		"fail-new-lower-version": {
+			oldPV:         makePV("pv1", "5"),
+			newPV:         makePV("pv1", "4"),
+			shouldSucceed: false,
+		},
+		"fail-new-bad-version": {
+			oldPV:         makePV("pv1", "5"),
+			newPV:         makePV("pv1", "a"),
+			shouldSucceed: false,
+		},
+		"fail-old-bad-version": {
+			oldPV:         makePV("pv1", "a"),
+			newPV:         makePV("pv1", "5"),
+			shouldSucceed: false,
+		},
+	}
+
+	for name, scenario := range scenarios {
+		cache := NewPVAssumeCache(nil)
+		internal_cache, ok := cache.(*pvAssumeCache)
+		if !ok {
+			t.Fatalf("Failed to get internal cache")
+		}
+
+		// Add oldPV to cache
+		internal_cache.add(scenario.oldPV)
+		if err := getPV(cache, scenario.oldPV.Name, scenario.oldPV); err != nil {
+			t.Errorf("Failed to GetPV() after initial update: %v", err)
+			continue
+		}
+
+		// Assume newPV
+		err := cache.Assume(scenario.newPV)
+		if scenario.shouldSucceed && err != nil {
+			t.Errorf("Test %q failed: Assume() returned error %v", name, err)
+		}
+		if !scenario.shouldSucceed && err == nil {
+			t.Errorf("Test %q failed: Assume() returned success but expected error", name)
+		}
+
+		// Check that GetPV returns correct PV
+		expectedPV := scenario.newPV
+		if !scenario.shouldSucceed {
+			expectedPV = scenario.oldPV
+		}
+		if err := getPV(cache, scenario.oldPV.Name, expectedPV); err != nil {
+			t.Errorf("Failed to GetPV() after initial update: %v", err)
+		}
+	}
+}
+
+func TestRestorePV(t *testing.T) {
+	cache := NewPVAssumeCache(nil)
+	internal_cache, ok := cache.(*pvAssumeCache)
+	if !ok {
+		t.Fatalf("Failed to get internal cache")
+	}
+
+	oldPV := makePV("pv1", "5")
+	newPV := makePV("pv1", "5")
+
+	// Restore PV that doesn't exist
+	cache.Restore("nothing")
+
+	// Add oldPV to cache
+	internal_cache.add(oldPV)
+	if err := getPV(cache, oldPV.Name, oldPV); err != nil {
+		t.Fatalf("Failed to GetPV() after initial update: %v", err)
+	}
+
+	// Restore PV
+	cache.Restore(oldPV.Name)
+	if err := getPV(cache, oldPV.Name, oldPV); err != nil {
+		t.Fatalf("Failed to GetPV() after iniital restore: %v", err)
+	}
+
+	// Assume newPV
+	if err := cache.Assume(newPV); err != nil {
+		t.Fatalf("Assume() returned error %v", err)
+	}
+	if err := getPV(cache, oldPV.Name, newPV); err != nil {
+		t.Fatalf("Failed to GetPV() after Assume: %v", err)
+	}
+
+	// Restore PV
+	cache.Restore(oldPV.Name)
+	if err := getPV(cache, oldPV.Name, oldPV); err != nil {
+		t.Fatalf("Failed to GetPV() after restore: %v", err)
+	}
+}
+
+func TestBasicPVCache(t *testing.T) {
+	cache := NewPVAssumeCache(nil)
+	internal_cache, ok := cache.(*pvAssumeCache)
+	if !ok {
+		t.Fatalf("Failed to get internal cache")
+	}
+
+	// Get object that doesn't exist
+	pv, err := cache.GetPV("nothere")
+	if err == nil {
+		t.Errorf("GetPV() returned unexpected success")
+	}
+	if pv != nil {
+		t.Errorf("GetPV() returned unexpected PV %q", pv.Name)
+	}
+
+	// Add a bunch of PVs
+	pvs := map[string]*v1.PersistentVolume{}
+	for i := 0; i < 10; i++ {
+		pv := makePV(fmt.Sprintf("test-pv%v", i), "1")
+		pvs[pv.Name] = pv
+		internal_cache.add(pv)
+	}
+
+	// List them
+	verifyListPVs(t, cache, pvs)
+
+	// Update a PV
+	updatedPV := makePV("test-pv3", "2")
+	pvs[updatedPV.Name] = updatedPV
+	internal_cache.update(nil, updatedPV)
+
+	// List them
+	verifyListPVs(t, cache, pvs)
+
+	// Delete a PV
+	deletedPV := pvs["test-pv7"]
+	delete(pvs, deletedPV.Name)
+	internal_cache.delete(deletedPV)
+
+	// List them
+	verifyListPVs(t, cache, pvs)
+}
+
+func verifyListPVs(t *testing.T, cache PVAssumeCache, expectedPVs map[string]*v1.PersistentVolume) {
+	pvList := cache.ListPVs()
+	if len(pvList) != len(expectedPVs) {
+		t.Errorf("ListPVs() returned %v PVs, expected %v", len(pvList), len(expectedPVs))
+	}
+	for _, pv := range pvList {
+		expectedPV, ok := expectedPVs[pv.Name]
+		if !ok {
+			t.Errorf("ListPVs() returned unexpected PV %q", pv.Name)
+		}
+		if expectedPV != pv {
+			t.Errorf("ListPVs() returned PV %p, expected %p", pv, expectedPV)
+		}
+	}
+}
+
+func getPV(cache PVAssumeCache, name string, expectedPV *v1.PersistentVolume) error {
+	pv, err := cache.GetPV(name)
+	if err != nil {
+		return err
+	}
+	if pv != expectedPV {
+		return fmt.Errorf("GetPV() returned %p, expected %p", pv, expectedPV)
+	}
+	return nil
+}

pkg/controller/volume/persistentvolume/scheduler_binder_cache.go

+/*
+Copyright 2017 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 persistentvolume
+
+import (
+	"sync"
+
+	"k8s.io/api/core/v1"
+)
+
+// podBindingCache stores PV binding decisions per pod per node.
+// Pod entries are removed when the Pod is deleted or updated to
+// no longer be schedulable.
+type PodBindingCache interface {
+	// UpdateBindings will update the cache with the given bindings for the
+	// pod and node.
+	UpdateBindings(pod *v1.Pod, node string, bindings []*bindingInfo)
+
+	// DeleteBindings will remove all cached bindings for the given pod.
+	DeleteBindings(pod *v1.Pod)
+
+	// GetBindings will return the cached bindings for the given pod and node.
+	GetBindings(pod *v1.Pod, node string) []*bindingInfo
+}
+
+type podBindingCache struct {
+	mutex sync.Mutex
+
+	// Key = pod name
+	// Value = nodeBindings
+	bindings map[string]nodeBindings
+}
+
+// Key = nodeName
+// Value = array of bindingInfo
+type nodeBindings map[string][]*bindingInfo
+
+func NewPodBindingCache() PodBindingCache {
+	return &podBindingCache{bindings: map[string]nodeBindings{}}
+}
+
+func (c *podBindingCache) DeleteBindings(pod *v1.Pod) {
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+
+	podName := getPodName(pod)
+	delete(c.bindings, podName)
+}
+
+func (c *podBindingCache) UpdateBindings(pod *v1.Pod, node string, bindings []*bindingInfo) {
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+
+	podName := getPodName(pod)
+	nodeBinding, ok := c.bindings[podName]
+	if !ok {
+		nodeBinding = nodeBindings{}
+		c.bindings[podName] = nodeBinding
+	}
+	nodeBinding[node] = bindings
+}
+
+func (c *podBindingCache) GetBindings(pod *v1.Pod, node string) []*bindingInfo {
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+
+	podName := getPodName(pod)
+	nodeBindings, ok := c.bindings[podName]
+	if !ok {
+		return nil
+	}
+	return nodeBindings[node]
+}

pkg/controller/volume/persistentvolume/scheduler_binder_cache_test.go

+/*
+Copyright 2017 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 persistentvolume
+
+import (
+	"reflect"
+	"testing"
+
+	"k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func TestUpdateGetBindings(t *testing.T) {
+	scenarios := map[string]struct {
+		updateBindings []*bindingInfo
+		updatePod      string
+		updateNode     string
+
+		getBindings []*bindingInfo
+		getPod      string
+		getNode     string
+	}{
+		"no-pod": {
+			getPod:  "pod1",
+			getNode: "node1",
+		},
+		"no-node": {
+			updatePod:      "pod1",
+			updateNode:     "node1",
+			updateBindings: []*bindingInfo{},
+			getPod:         "pod1",
+			getNode:        "node2",
+		},
+		"binding-exists": {
+			updatePod:      "pod1",
+			updateNode:     "node1",
+			updateBindings: []*bindingInfo{{pvc: &v1.PersistentVolumeClaim{ObjectMeta: metav1.ObjectMeta{Name: "pvc1"}}}},
+			getPod:         "pod1",
+			getNode:        "node1",
+			getBindings:    []*bindingInfo{{pvc: &v1.PersistentVolumeClaim{ObjectMeta: metav1.ObjectMeta{Name: "pvc1"}}}},
+		},
+	}
+
+	for name, scenario := range scenarios {
+		cache := NewPodBindingCache()
+
+		// Perform updates
+		updatePod := &v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: scenario.updatePod, Namespace: "ns"}}
+		cache.UpdateBindings(updatePod, scenario.updateNode, scenario.updateBindings)
+
+		// Verify updated bindings
+		bindings := cache.GetBindings(updatePod, scenario.updateNode)
+		if !reflect.DeepEqual(bindings, scenario.updateBindings) {
+			t.Errorf("Test %v failed: returned bindings after update different. Got %+v, expected %+v", name, bindings, scenario.updateBindings)
+		}
+
+		// Get bindings
+		getPod := &v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: scenario.getPod, Namespace: "ns"}}
+		bindings = cache.GetBindings(getPod, scenario.getNode)
+		if !reflect.DeepEqual(bindings, scenario.getBindings) {
+			t.Errorf("Test %v failed: unexpected bindings returned. Got %+v, expected %+v", name, bindings, scenario.updateBindings)
+		}
+	}
+}
+
+func TestDeleteBindings(t *testing.T) {
+	initialBindings := []*bindingInfo{{pvc: &v1.PersistentVolumeClaim{ObjectMeta: metav1.ObjectMeta{Name: "pvc1"}}}}
+	cache := NewPodBindingCache()
+
+	pod := &v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod1", Namespace: "ns"}}
+
+	// Get nil bindings
+	bindings := cache.GetBindings(pod, "node1")
+	if bindings != nil {
+		t.Errorf("Test failed: expected inital nil bindings, got %+v", bindings)
+	}
+
+	// Delete nothing
+	cache.DeleteBindings(pod)
+
+	// Perform updates
+	cache.UpdateBindings(pod, "node1", initialBindings)
+
+	// Get bindings
+	bindings = cache.GetBindings(pod, "node1")
+	if !reflect.DeepEqual(bindings, initialBindings) {
+		t.Errorf("Test failed: expected bindings %+v, got %+v", initialBindings, bindings)
+	}
+
+	// Delete
+	cache.DeleteBindings(pod)
+
+	// Get bindings
+	bindings = cache.GetBindings(pod, "node1")
+	if bindings != nil {
+		t.Errorf("Test failed: expected nil bindings, got %+v", bindings)
+	}
+}

pkg/controller/volume/persistentvolume/scheduler_binder_fake.go

+/*
+Copyright 2017 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 persistentvolume
+
+import (
+	"k8s.io/api/core/v1"
+)
+
+type FakeVolumeBinderConfig struct {
+	AllBound              bool
+	FindUnboundSatsified  bool
+	FindBoundSatsified    bool
+	FindErr               error
+	AssumeBindingRequired bool
+	AssumeErr             error
+	BindErr               error
+}
+
+// NewVolumeBinder sets up all the caches needed for the scheduler to make
+// topology-aware volume binding decisions.
+func NewFakeVolumeBinder(config *FakeVolumeBinderConfig) *FakeVolumeBinder {
+	return &FakeVolumeBinder{
+		config: config,
+	}
+}
+
+type FakeVolumeBinder struct {
+	config       *FakeVolumeBinderConfig
+	AssumeCalled bool
+	BindCalled   bool
+}
+
+func (b *FakeVolumeBinder) FindPodVolumes(pod *v1.Pod, nodeName string) (unboundVolumesSatisfied, boundVolumesSatsified bool, err error) {
+	return b.config.FindUnboundSatsified, b.config.FindBoundSatsified, b.config.FindErr
+}
+
+func (b *FakeVolumeBinder) AssumePodVolumes(assumedPod *v1.Pod, nodeName string) (bool, bool, error) {
+	b.AssumeCalled = true
+	return b.config.AllBound, b.config.AssumeBindingRequired, b.config.AssumeErr
+}
+
+func (b *FakeVolumeBinder) BindPodVolumes(assumedPod *v1.Pod) error {
+	b.BindCalled = true
+	return b.config.BindErr
+}
+
+func (b *FakeVolumeBinder) GetBindingsCache() PodBindingCache {
+	return nil
+}

pkg/features/kube_features.go

@@ -180,6 +180,7 @@ const (
 	// alpha: v1.9
 	//
 	// Extend the default scheduler to be aware of PV topology and handle PV binding
+	// Before moving to beta, resolve Kubernetes issue #56180
 	VolumeScheduling utilfeature.Feature = "VolumeScheduling"
 
 	// owner: @vladimirvivien

pkg/kubectl/.import-restrictions

@@ -120,6 +120,7 @@
                 "k8s.io/kubernetes/pkg/security/apparmor",
                 "k8s.io/kubernetes/pkg/serviceaccount",
                 "k8s.io/kubernetes/pkg/util/file",
+                "k8s.io/kubernetes/pkg/util/goroutinemap",
                 "k8s.io/kubernetes/pkg/util/hash",
                 "k8s.io/kubernetes/pkg/util/interrupt",
                 "k8s.io/kubernetes/pkg/util/io",
@@ -146,4 +147,4 @@
 		],
 		"ForbiddenPrefixes": []
 	}]
-}
\ No newline at end of file
+}

plugin/cmd/kube-scheduler/app/BUILD

@@ -40,6 +40,7 @@ go_library(
         "//vendor/k8s.io/apiserver/pkg/util/feature:go_default_library",
         "//vendor/k8s.io/client-go/informers:go_default_library",
         "//vendor/k8s.io/client-go/informers/core/v1:go_default_library",
+        "//vendor/k8s.io/client-go/informers/storage/v1:go_default_library",
         "//vendor/k8s.io/client-go/kubernetes:go_default_library",
         "//vendor/k8s.io/client-go/kubernetes/typed/core/v1:go_default_library",
         "//vendor/k8s.io/client-go/rest:go_default_library",

plugin/cmd/kube-scheduler/app/server.go

@@ -40,6 +40,7 @@ import (
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/client-go/informers"
 	coreinformers "k8s.io/client-go/informers/core/v1"
+	storageinformers "k8s.io/client-go/informers/storage/v1"
 	clientset "k8s.io/client-go/kubernetes"
 	v1core "k8s.io/client-go/kubernetes/typed/core/v1"
 	restclient "k8s.io/client-go/rest"
@@ -625,6 +626,11 @@ func (s *SchedulerServer) Run(stop chan struct{}) error {
 // SchedulerConfig creates the scheduler configuration. This is exposed for use
 // by tests.
 func (s *SchedulerServer) SchedulerConfig() (*scheduler.Config, error) {
+	var storageClassInformer storageinformers.StorageClassInformer
+	if utilfeature.DefaultFeatureGate.Enabled(features.VolumeScheduling) {
+		storageClassInformer = s.InformerFactory.Storage().V1().StorageClasses()
+	}
+
 	// Set up the configurator which can create schedulers from configs.
 	configurator := factory.NewConfigFactory(
 		s.SchedulerName,
@@ -638,6 +644,7 @@ func (s *SchedulerServer) SchedulerConfig() (*scheduler.Config, error) {
 		s.InformerFactory.Apps().V1beta1().StatefulSets(),
 		s.InformerFactory.Core().V1().Services(),
 		s.InformerFactory.Policy().V1beta1().PodDisruptionBudgets(),
+		storageClassInformer,
 		s.HardPodAffinitySymmetricWeight,
 		utilfeature.DefaultFeatureGate.Enabled(features.EnableEquivalenceClassCache),
 	)

plugin/pkg/auth/authorizer/rbac/bootstrappolicy/policy.go

@@ -439,6 +439,18 @@ func ClusterRoles() []rbac.ClusterRole {
 		})
 	}
 
+	if utilfeature.DefaultFeatureGate.Enabled(features.VolumeScheduling) {
+		// Find the scheduler role
+		for i, role := range roles {
+			if role.Name == "system:kube-scheduler" {
+				pvRule := rbac.NewRule("update").Groups(legacyGroup).Resources("persistentvolumes").RuleOrDie()
+				scRule := rbac.NewRule(Read...).Groups(storageGroup).Resources("storageclasses").RuleOrDie()
+				roles[i].Rules = append(role.Rules, pvRule, scRule)
+				break
+			}
+		}
+	}
+
 	addClusterRoleLabel(roles)
 	return roles
 }

plugin/pkg/scheduler/BUILD

@@ -13,18 +13,21 @@ go_test(
     library = ":go_default_library",
     deps = [
         "//pkg/api/legacyscheme:go_default_library",
+        "//pkg/controller/volume/persistentvolume:go_default_library",
         "//plugin/pkg/scheduler/algorithm:go_default_library",
         "//plugin/pkg/scheduler/algorithm/predicates:go_default_library",
         "//plugin/pkg/scheduler/core:go_default_library",
         "//plugin/pkg/scheduler/schedulercache:go_default_library",
         "//plugin/pkg/scheduler/testing:go_default_library",
         "//plugin/pkg/scheduler/util:go_default_library",
+        "//plugin/pkg/scheduler/volumebinder:go_default_library",
         "//vendor/k8s.io/api/core/v1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/api/resource:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/labels:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/diff:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/wait:go_default_library",
+        "//vendor/k8s.io/apiserver/pkg/util/feature:go_default_library",
         "//vendor/k8s.io/client-go/tools/cache:go_default_library",
         "//vendor/k8s.io/client-go/tools/record:go_default_library",
     ],
@@ -38,17 +41,21 @@ go_library(
     ],
     importpath = "k8s.io/kubernetes/plugin/pkg/scheduler",
     deps = [
+        "//pkg/features:go_default_library",
         "//plugin/pkg/scheduler/algorithm:go_default_library",
+        "//plugin/pkg/scheduler/algorithm/predicates:go_default_library",
         "//plugin/pkg/scheduler/api:go_default_library",
         "//plugin/pkg/scheduler/core:go_default_library",
         "//plugin/pkg/scheduler/metrics:go_default_library",
         "//plugin/pkg/scheduler/schedulercache:go_default_library",
         "//plugin/pkg/scheduler/util:go_default_library",
+        "//plugin/pkg/scheduler/volumebinder:go_default_library",
         "//vendor/github.com/golang/glog:go_default_library",
         "//vendor/k8s.io/api/core/v1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/sets:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/wait:go_default_library",
+        "//vendor/k8s.io/apiserver/pkg/util/feature:go_default_library",
         "//vendor/k8s.io/client-go/kubernetes:go_default_library",
         "//vendor/k8s.io/client-go/listers/core/v1:go_default_library",
         "//vendor/k8s.io/client-go/tools/record:go_default_library",
@@ -75,6 +82,7 @@ filegroup(
         "//plugin/pkg/scheduler/schedulercache:all-srcs",
         "//plugin/pkg/scheduler/testing:all-srcs",
         "//plugin/pkg/scheduler/util:all-srcs",
+        "//plugin/pkg/scheduler/volumebinder:all-srcs",
     ],
     tags = ["automanaged"],
 )

plugin/pkg/scheduler/algorithm/predicates/BUILD

@@ -26,16 +26,18 @@ go_library(
         "//plugin/pkg/scheduler/algorithm/priorities/util:go_default_library",
         "//plugin/pkg/scheduler/schedulercache:go_default_library",
         "//plugin/pkg/scheduler/util:go_default_library",
+        "//plugin/pkg/scheduler/volumebinder:go_default_library",
         "//vendor/github.com/golang/glog:go_default_library",
         "//vendor/k8s.io/api/core/v1:go_default_library",
+        "//vendor/k8s.io/api/storage/v1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/api/errors:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/labels:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/rand:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/util/feature:go_default_library",
         "//vendor/k8s.io/client-go/listers/core/v1:go_default_library",
+        "//vendor/k8s.io/client-go/listers/storage/v1:go_default_library",
         "//vendor/k8s.io/client-go/util/workqueue:go_default_library",
-        "//vendor/k8s.io/metrics/pkg/client/clientset_generated/clientset:go_default_library",
     ],
 )
 
@@ -56,9 +58,11 @@ go_test(
         "//plugin/pkg/scheduler/testing:go_default_library",
         "//plugin/pkg/scheduler/util:go_default_library",
         "//vendor/k8s.io/api/core/v1:go_default_library",
+        "//vendor/k8s.io/api/storage/v1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/api/resource:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/labels:go_default_library",
+        "//vendor/k8s.io/apiserver/pkg/util/feature:go_default_library",
     ],
 )
 

plugin/pkg/scheduler/algorithm/predicates/error.go

@@ -50,7 +50,8 @@ var (
 	ErrNodeNetworkUnavailable                = newPredicateFailureError("NodeNetworkUnavailable")
 	ErrNodeUnschedulable                     = newPredicateFailureError("NodeUnschedulable")
 	ErrNodeUnknownCondition                  = newPredicateFailureError("NodeUnknownCondition")
-	ErrVolumeNodeConflict                    = newPredicateFailureError("NoVolumeNodeConflict")
+	ErrVolumeNodeConflict                    = newPredicateFailureError("VolumeNodeAffinityConflict")
+	ErrVolumeBindConflict                    = newPredicateFailureError("VolumeBindingNoMatch")
 	// ErrFakePredicate is used for test only. The fake predicates returning false also returns error
 	// as ErrFakePredicate.
 	ErrFakePredicate = newPredicateFailureError("FakePredicateError")

plugin/pkg/scheduler/algorithm/predicates/predicates.go

@@ -24,12 +24,14 @@ import (
 	"sync"
 
 	"k8s.io/api/core/v1"
+	storagev1 "k8s.io/api/storage/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/util/rand"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	corelisters "k8s.io/client-go/listers/core/v1"
+	storagelisters "k8s.io/client-go/listers/storage/v1"
 	"k8s.io/client-go/util/workqueue"
 	v1helper "k8s.io/kubernetes/pkg/apis/core/v1/helper"
 	v1qos "k8s.io/kubernetes/pkg/apis/core/v1/helper/qos"
@@ -41,13 +43,14 @@ import (
 	priorityutil "k8s.io/kubernetes/plugin/pkg/scheduler/algorithm/priorities/util"
 	"k8s.io/kubernetes/plugin/pkg/scheduler/schedulercache"
 	schedutil "k8s.io/kubernetes/plugin/pkg/scheduler/util"
-	"k8s.io/metrics/pkg/client/clientset_generated/clientset"
 
 	"github.com/golang/glog"
+	"k8s.io/kubernetes/plugin/pkg/scheduler/volumebinder"
 )
 
 const (
 	MatchInterPodAffinity = "MatchInterPodAffinity"
+	CheckVolumeBinding    = "CheckVolumeBinding"
 
 	// DefaultMaxGCEPDVolumes defines the maximum number of PD Volumes for GCE
 	// GCE instances can have up to 16 PD volumes attached.
@@ -127,6 +130,19 @@ func (c *CachedNodeInfo) GetNodeInfo(id string) (*v1.Node, error) {
 	return node, nil
 }
 
+type StorageClassInfo interface {
+	GetStorageClassInfo(className string) (*storagev1.StorageClass, error)
+}
+
+// CachedStorageClassInfo implements StorageClassInfo
+type CachedStorageClassInfo struct {
+	storagelisters.StorageClassLister
+}
+
+func (c *CachedStorageClassInfo) GetStorageClassInfo(className string) (*storagev1.StorageClass, error) {
+	return c.Get(className)
+}
+
 func isVolumeConflict(volume v1.Volume, pod *v1.Pod) bool {
 	// fast path if there is no conflict checking targets.
 	if volume.GCEPersistentDisk == nil && volume.AWSElasticBlockStore == nil && volume.RBD == nil && volume.ISCSI == nil {
@@ -416,8 +432,9 @@ var AzureDiskVolumeFilter VolumeFilter = VolumeFilter{
 }
 
 type VolumeZoneChecker struct {
-	pvInfo  PersistentVolumeInfo
-	pvcInfo PersistentVolumeClaimInfo
+	pvInfo    PersistentVolumeInfo
+	pvcInfo   PersistentVolumeClaimInfo
+	classInfo StorageClassInfo
 }
 
 // NewVolumeZonePredicate evaluates if a pod can fit due to the volumes it requests, given
@@ -434,10 +451,11 @@ type VolumeZoneChecker struct {
 // determining the zone of a volume during scheduling, and that is likely to
 // require calling out to the cloud provider.  It seems that we are moving away
 // from inline volume declarations anyway.
-func NewVolumeZonePredicate(pvInfo PersistentVolumeInfo, pvcInfo PersistentVolumeClaimInfo) algorithm.FitPredicate {
+func NewVolumeZonePredicate(pvInfo PersistentVolumeInfo, pvcInfo PersistentVolumeClaimInfo, classInfo StorageClassInfo) algorithm.FitPredicate {
 	c := &VolumeZoneChecker{
-		pvInfo:  pvInfo,
-		pvcInfo: pvcInfo,
+		pvInfo:    pvInfo,
+		pvcInfo:   pvcInfo,
+		classInfo: classInfo,
 	}
 	return c.predicate
 }
@@ -489,6 +507,21 @@ func (c *VolumeZoneChecker) predicate(pod *v1.Pod, meta algorithm.PredicateMetad
 
 			pvName := pvc.Spec.VolumeName
 			if pvName == "" {
+				if utilfeature.DefaultFeatureGate.Enabled(features.VolumeScheduling) {
+					scName := pvc.Spec.StorageClassName
+					if scName != nil && len(*scName) > 0 {
+						class, _ := c.classInfo.GetStorageClassInfo(*scName)
+						if class != nil {
+							if class.VolumeBindingMode == nil {
+								return false, nil, fmt.Errorf("VolumeBindingMode not set for StorageClass %q", scName)
+							}
+							if *class.VolumeBindingMode == storagev1.VolumeBindingWaitForFirstConsumer {
+								// Skip unbound volumes
+								continue
+							}
+						}
+					}
+				}
 				return false, nil, fmt.Errorf("PersistentVolumeClaim is not bound: %q", pvcName)
 			}
 
@@ -1403,33 +1436,30 @@ func CheckNodeConditionPredicate(pod *v1.Pod, meta algorithm.PredicateMetadata,
 	return len(reasons) == 0, reasons, nil
 }
 
-type VolumeNodeChecker struct {
-	pvInfo  PersistentVolumeInfo
-	pvcInfo PersistentVolumeClaimInfo
-	client  clientset.Interface
+type VolumeBindingChecker struct {
+	binder *volumebinder.VolumeBinder
 }
 
-// NewVolumeNodePredicate evaluates if a pod can fit due to the volumes it requests, given
-// that some volumes have node topology constraints, particularly when using Local PVs.
-// The requirement is that any pod that uses a PVC that is bound to a PV with topology constraints
-// must be scheduled to a node that satisfies the PV's topology labels.
-func NewVolumeNodePredicate(pvInfo PersistentVolumeInfo, pvcInfo PersistentVolumeClaimInfo, client clientset.Interface) algorithm.FitPredicate {
-	c := &VolumeNodeChecker{
-		pvInfo:  pvInfo,
-		pvcInfo: pvcInfo,
-		client:  client,
+// NewVolumeBindingPredicate evaluates if a pod can fit due to the volumes it requests,
+// for both bound and unbound PVCs.
+//
+// For PVCs that are bound, then it checks that the corresponding PV's node affinity is
+// satisfied by the given node.
+//
+// For PVCs that are unbound, it tries to find available PVs that can satisfy the PVC requirements
+// and that the PV node affinity is satisfied by the given node.
+//
+// The predicate returns true if all bound PVCs have compatible PVs with the node, and if all unbound
+// PVCs can be matched with an available and node-compatible PV.
+func NewVolumeBindingPredicate(binder *volumebinder.VolumeBinder) algorithm.FitPredicate {
+	c := &VolumeBindingChecker{
+		binder: binder,
 	}
 	return c.predicate
 }
 
-func (c *VolumeNodeChecker) predicate(pod *v1.Pod, meta algorithm.PredicateMetadata, nodeInfo *schedulercache.NodeInfo) (bool, []algorithm.PredicateFailureReason, error) {
-	if !utilfeature.DefaultFeatureGate.Enabled(features.PersistentLocalVolumes) {
-		return true, nil, nil
-	}
-
-	// If a pod doesn't have any volume attached to it, the predicate will always be true.
-	// Thus we make a fast path for it, to avoid unnecessary computations in this case.
-	if len(pod.Spec.Volumes) == 0 {
+func (c *VolumeBindingChecker) predicate(pod *v1.Pod, meta algorithm.PredicateMetadata, nodeInfo *schedulercache.NodeInfo) (bool, []algorithm.PredicateFailureReason, error) {
+	if !utilfeature.DefaultFeatureGate.Enabled(features.VolumeScheduling) {
 		return true, nil, nil
 	}
 
@@ -1438,45 +1468,27 @@ func (c *VolumeNodeChecker) predicate(pod *v1.Pod, meta algorithm.PredicateMetad
 		return false, nil, fmt.Errorf("node not found")
 	}
 
-	glog.V(2).Infof("Checking for prebound volumes with node affinity")
-	namespace := pod.Namespace
-	manifest := &(pod.Spec)
-	for i := range manifest.Volumes {
-		volume := &manifest.Volumes[i]
-		if volume.PersistentVolumeClaim == nil {
-			continue
-		}
-		pvcName := volume.PersistentVolumeClaim.ClaimName
-		if pvcName == "" {
-			return false, nil, fmt.Errorf("PersistentVolumeClaim had no name")
-		}
-		pvc, err := c.pvcInfo.GetPersistentVolumeClaimInfo(namespace, pvcName)
-		if err != nil {
-			return false, nil, err
-		}
+	unboundSatisfied, boundSatisfied, err := c.binder.Binder.FindPodVolumes(pod, node.Name)
+	if err != nil {
+		return false, nil, err
+	}
 
-		if pvc == nil {
-			return false, nil, fmt.Errorf("PersistentVolumeClaim was not found: %q", pvcName)
-		}
-		pvName := pvc.Spec.VolumeName
-		if pvName == "" {
-			return false, nil, fmt.Errorf("PersistentVolumeClaim is not bound: %q", pvcName)
-		}
+	failReasons := []algorithm.PredicateFailureReason{}
+	if !boundSatisfied {
+		glog.V(5).Info("Bound PVs not satisfied for pod %v/%v, node %q", pod.Namespace, pod.Name, node.Name)
+		failReasons = append(failReasons, ErrVolumeNodeConflict)
+	}
 
-		pv, err := c.pvInfo.GetPersistentVolumeInfo(pvName)
-		if err != nil {
-			return false, nil, err
-		}
-		if pv == nil {
-			return false, nil, fmt.Errorf("PersistentVolume not found: %q", pvName)
-		}
+	if !unboundSatisfied {
+		glog.V(5).Info("Couldn't find matching PVs for pod %v/%v, node %q", pod.Namespace, pod.Name, node.Name)
+		failReasons = append(failReasons, ErrVolumeBindConflict)
+	}
 
-		err = volumeutil.CheckNodeAffinity(pv, node.Labels)
-		if err != nil {
-			glog.V(2).Infof("Won't schedule pod %q onto node %q due to volume %q node mismatch: %v", pod.Name, node.Name, pvName, err.Error())
-			return false, []algorithm.PredicateFailureReason{ErrVolumeNodeConflict}, nil
-		}
-		glog.V(4).Infof("VolumeNode predicate allows node %q for pod %q due to volume %q", node.Name, pod.Name, pvName)
+	if len(failReasons) > 0 {
+		return false, failReasons, nil
 	}
+
+	// All volumes bound or matching PVs found for all unbound PVCs
+	glog.V(5).Info("All PVCs found matches for pod %v/%v, node %q", pod.Namespace, pod.Name, node.Name)
 	return true, nil, nil
 }

plugin/pkg/scheduler/algorithm/predicates/predicates_test.go

@@ -24,8 +24,10 @@ import (
 	"testing"
 
 	"k8s.io/api/core/v1"
+	storagev1 "k8s.io/api/storage/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	v1helper "k8s.io/kubernetes/pkg/apis/core/v1/helper"
 	kubeletapis "k8s.io/kubernetes/pkg/kubelet/apis"
 	"k8s.io/kubernetes/plugin/pkg/scheduler/algorithm"
@@ -74,6 +76,17 @@ func (pvs FakePersistentVolumeInfo) GetPersistentVolumeInfo(pvID string) (*v1.Pe
 	return nil, fmt.Errorf("Unable to find persistent volume: %s", pvID)
 }
 
+type FakeStorageClassInfo []storagev1.StorageClass
+
+func (classes FakeStorageClassInfo) GetStorageClassInfo(name string) (*storagev1.StorageClass, error) {
+	for _, sc := range classes {
+		if sc.Name == name {
+			return &sc, nil
+		}
+	}
+	return nil, fmt.Errorf("Unable to find storage class: %s", name)
+}
+
 var (
 	extendedResourceA = v1.ResourceName("example.com/aaa")
 	extendedResourceB = v1.ResourceName("example.com/bbb")
@@ -3834,7 +3847,7 @@ func TestVolumeZonePredicate(t *testing.T) {
 	expectedFailureReasons := []algorithm.PredicateFailureReason{ErrVolumeZoneConflict}
 
 	for _, test := range tests {
-		fit := NewVolumeZonePredicate(pvInfo, pvcInfo)
+		fit := NewVolumeZonePredicate(pvInfo, pvcInfo, nil)
 		node := &schedulercache.NodeInfo{}
 		node.SetNode(test.Node)
 
@@ -3927,7 +3940,7 @@ func TestVolumeZonePredicateMultiZone(t *testing.T) {
 	expectedFailureReasons := []algorithm.PredicateFailureReason{ErrVolumeZoneConflict}
 
 	for _, test := range tests {
-		fit := NewVolumeZonePredicate(pvInfo, pvcInfo)
+		fit := NewVolumeZonePredicate(pvInfo, pvcInfo, nil)
 		node := &schedulercache.NodeInfo{}
 		node.SetNode(test.Node)
 
@@ -3945,6 +3958,130 @@ func TestVolumeZonePredicateMultiZone(t *testing.T) {
 	}
 }
 
+func TestVolumeZonePredicateWithVolumeBinding(t *testing.T) {
+	var (
+		modeWait = storagev1.VolumeBindingWaitForFirstConsumer
+
+		class0         = "Class_0"
+		classWait      = "Class_Wait"
+		classImmediate = "Class_Immediate"
+	)
+
+	classInfo := FakeStorageClassInfo{
+		{
+			ObjectMeta: metav1.ObjectMeta{Name: classImmediate},
+		},
+		{
+			ObjectMeta:        metav1.ObjectMeta{Name: classWait},
+			VolumeBindingMode: &modeWait,
+		},
+	}
+
+	pvInfo := FakePersistentVolumeInfo{
+		{
+			ObjectMeta: metav1.ObjectMeta{Name: "Vol_1", Labels: map[string]string{kubeletapis.LabelZoneFailureDomain: "us-west1-a"}},
+		},
+	}
+
+	pvcInfo := FakePersistentVolumeClaimInfo{
+		{
+			ObjectMeta: metav1.ObjectMeta{Name: "PVC_1", Namespace: "default"},
+			Spec:       v1.PersistentVolumeClaimSpec{VolumeName: "Vol_1"},
+		},
+		{
+			ObjectMeta: metav1.ObjectMeta{Name: "PVC_NoSC", Namespace: "default"},
+			Spec:       v1.PersistentVolumeClaimSpec{StorageClassName: &class0},
+		},
+		{
+			ObjectMeta: metav1.ObjectMeta{Name: "PVC_EmptySC", Namespace: "default"},
+		},
+		{
+			ObjectMeta: metav1.ObjectMeta{Name: "PVC_WaitSC", Namespace: "default"},
+			Spec:       v1.PersistentVolumeClaimSpec{StorageClassName: &classWait},
+		},
+		{
+			ObjectMeta: metav1.ObjectMeta{Name: "PVC_ImmediateSC", Namespace: "default"},
+			Spec:       v1.PersistentVolumeClaimSpec{StorageClassName: &classImmediate},
+		},
+	}
+
+	testNode := &v1.Node{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:   "host1",
+			Labels: map[string]string{kubeletapis.LabelZoneFailureDomain: "us-west1-a", "uselessLabel": "none"},
+		},
+	}
+
+	tests := []struct {
+		Name          string
+		Pod           *v1.Pod
+		Fits          bool
+		Node          *v1.Node
+		ExpectFailure bool
+	}{
+		{
+			Name: "label zone failure domain matched",
+			Pod:  createPodWithVolume("pod_1", "vol_1", "PVC_1"),
+			Node: testNode,
+			Fits: true,
+		},
+		{
+			Name:          "unbound volume empty storage class",
+			Pod:           createPodWithVolume("pod_1", "vol_1", "PVC_EmptySC"),
+			Node:          testNode,
+			Fits:          false,
+			ExpectFailure: true,
+		},
+		{
+			Name:          "unbound volume no storage class",
+			Pod:           createPodWithVolume("pod_1", "vol_1", "PVC_NoSC"),
+			Node:          testNode,
+			Fits:          false,
+			ExpectFailure: true,
+		},
+		{
+			Name:          "unbound volume immediate binding mode",
+			Pod:           createPodWithVolume("pod_1", "vol_1", "PVC_ImmediateSC"),
+			Node:          testNode,
+			Fits:          false,
+			ExpectFailure: true,
+		},
+		{
+			Name: "unbound volume wait binding mode",
+			Pod:  createPodWithVolume("pod_1", "vol_1", "PVC_WaitSC"),
+			Node: testNode,
+			Fits: true,
+		},
+	}
+
+	err := utilfeature.DefaultFeatureGate.Set("VolumeScheduling=true")
+	if err != nil {
+		t.Fatalf("Failed to enable feature gate for VolumeScheduling: %v", err)
+	}
+
+	for _, test := range tests {
+		fit := NewVolumeZonePredicate(pvInfo, pvcInfo, classInfo)
+		node := &schedulercache.NodeInfo{}
+		node.SetNode(test.Node)
+
+		fits, _, err := fit(test.Pod, nil, node)
+		if !test.ExpectFailure && err != nil {
+			t.Errorf("%s: unexpected error: %v", test.Name, err)
+		}
+		if test.ExpectFailure && err == nil {
+			t.Errorf("%s: expected error, got success", test.Name)
+		}
+		if fits != test.Fits {
+			t.Errorf("%s: expected %v got %v", test.Name, test.Fits, fits)
+		}
+	}
+
+	err = utilfeature.DefaultFeatureGate.Set("VolumeScheduling=false")
+	if err != nil {
+		t.Fatalf("Failed to disable feature gate for VolumeScheduling: %v", err)
+	}
+}
+
 func TestGetMaxVols(t *testing.T) {
 	previousValue := os.Getenv(KubeMaxPDVols)
 	defaultValue := 39

plugin/pkg/scheduler/algorithmprovider/defaults/BUILD

@@ -34,6 +34,7 @@ go_test(
     deps = [
         "//pkg/api/legacyscheme:go_default_library",
         "//pkg/apis/core/install:go_default_library",
+        "//plugin/pkg/scheduler/algorithm/predicates:go_default_library",
         "//plugin/pkg/scheduler/api:go_default_library",
         "//plugin/pkg/scheduler/api/latest:go_default_library",
         "//plugin/pkg/scheduler/factory:go_default_library",

plugin/pkg/scheduler/algorithmprovider/defaults/compatibility_test.go

@@ -337,8 +337,7 @@ func TestCompatibility_v1_Scheduler(t *testing.T) {
 			{"name": "MatchInterPodAffinity"},
 			{"name": "GeneralPredicates"},
 			{"name": "TestServiceAffinity", "argument": {"serviceAffinity" : {"labels" : ["region"]}}},
-			{"name": "TestLabelsPresence",  "argument": {"labelsPresence"  : {"labels" : ["foo"], "presence":true}}},
-			{"name": "NoVolumeNodeConflict"}
+			{"name": "TestLabelsPresence",  "argument": {"labelsPresence"  : {"labels" : ["foo"], "presence":true}}}
 		  ],"priorities": [
 			{"name": "EqualPriority",   "weight": 2},
 			{"name": "ImageLocalityPriority",   "weight": 2},
@@ -370,7 +369,6 @@ func TestCompatibility_v1_Scheduler(t *testing.T) {
 					{Name: "GeneralPredicates"},
 					{Name: "TestServiceAffinity", Argument: &schedulerapi.PredicateArgument{ServiceAffinity: &schedulerapi.ServiceAffinity{Labels: []string{"region"}}}},
 					{Name: "TestLabelsPresence", Argument: &schedulerapi.PredicateArgument{LabelsPresence: &schedulerapi.LabelsPresence{Labels: []string{"foo"}, Presence: true}}},
-					{Name: "NoVolumeNodeConflict"},
 				},
 				Priorities: []schedulerapi.PriorityPolicy{
 					{Name: "EqualPriority", Weight: 2},
@@ -409,8 +407,80 @@ func TestCompatibility_v1_Scheduler(t *testing.T) {
 			{"name": "MatchInterPodAffinity"},
 			{"name": "GeneralPredicates"},
 			{"name": "TestServiceAffinity", "argument": {"serviceAffinity" : {"labels" : ["region"]}}},
-			{"name": "TestLabelsPresence",  "argument": {"labelsPresence"  : {"labels" : ["foo"], "presence":true}}},
-			{"name": "NoVolumeNodeConflict"}
+			{"name": "TestLabelsPresence",  "argument": {"labelsPresence"  : {"labels" : ["foo"], "presence":true}}}
+		  ],"priorities": [
+			{"name": "EqualPriority",   "weight": 2},
+			{"name": "ImageLocalityPriority",   "weight": 2},
+			{"name": "LeastRequestedPriority",   "weight": 2},
+			{"name": "BalancedResourceAllocation",   "weight": 2},
+			{"name": "SelectorSpreadPriority",   "weight": 2},
+			{"name": "NodePreferAvoidPodsPriority",   "weight": 2},
+			{"name": "NodeAffinityPriority",   "weight": 2},
+			{"name": "TaintTolerationPriority",   "weight": 2},
+			{"name": "InterPodAffinityPriority",   "weight": 2},
+			{"name": "MostRequestedPriority",   "weight": 2}
+		  ]
+		}`,
+			ExpectedPolicy: schedulerapi.Policy{
+				Predicates: []schedulerapi.PredicatePolicy{
+					{Name: "MatchNodeSelector"},
+					{Name: "PodFitsResources"},
+					{Name: "PodFitsHostPorts"},
+					{Name: "HostName"},
+					{Name: "NoDiskConflict"},
+					{Name: "NoVolumeZoneConflict"},
+					{Name: "PodToleratesNodeTaints"},
+					{Name: "CheckNodeMemoryPressure"},
+					{Name: "CheckNodeDiskPressure"},
+					{Name: "CheckNodeCondition"},
+					{Name: "MaxEBSVolumeCount"},
+					{Name: "MaxGCEPDVolumeCount"},
+					{Name: "MaxAzureDiskVolumeCount"},
+					{Name: "MatchInterPodAffinity"},
+					{Name: "GeneralPredicates"},
+					{Name: "TestServiceAffinity", Argument: &schedulerapi.PredicateArgument{ServiceAffinity: &schedulerapi.ServiceAffinity{Labels: []string{"region"}}}},
+					{Name: "TestLabelsPresence", Argument: &schedulerapi.PredicateArgument{LabelsPresence: &schedulerapi.LabelsPresence{Labels: []string{"foo"}, Presence: true}}},
+				},
+				Priorities: []schedulerapi.PriorityPolicy{
+					{Name: "EqualPriority", Weight: 2},
+					{Name: "ImageLocalityPriority", Weight: 2},
+					{Name: "LeastRequestedPriority", Weight: 2},
+					{Name: "BalancedResourceAllocation", Weight: 2},
+					{Name: "SelectorSpreadPriority", Weight: 2},
+					{Name: "NodePreferAvoidPodsPriority", Weight: 2},
+					{Name: "NodeAffinityPriority", Weight: 2},
+					{Name: "TaintTolerationPriority", Weight: 2},
+					{Name: "InterPodAffinityPriority", Weight: 2},
+					{Name: "MostRequestedPriority", Weight: 2},
+				},
+			},
+		},
+		// Do not change this JSON after the corresponding release has been tagged.
+		// A failure indicates backwards compatibility with the specified release was broken.
+		"1.9": {
+			JSON: `{
+		  "kind": "Policy",
+		  "apiVersion": "v1",
+		  "predicates": [
+			{"name": "MatchNodeSelector"},
+			{"name": "PodFitsResources"},
+			{"name": "PodFitsHostPorts"},
+			{"name": "HostName"},
+			{"name": "NoDiskConflict"},
+			{"name": "NoVolumeZoneConflict"},
+			{"name": "PodToleratesNodeTaints"},
+			{"name": "CheckNodeMemoryPressure"},
+			{"name": "CheckNodeDiskPressure"},
+			{"name": "CheckNodeCondition"},
+			{"name": "MaxEBSVolumeCount"},
+			{"name": "MaxGCEPDVolumeCount"},
+			{"name": "MaxAzureDiskVolumeCount"},
+			{"name": "MatchInterPodAffinity"},
+			{"name": "GeneralPredicates"},
+			{"name": "CheckVolumeBinding"},
+			{"name": "TestServiceAffinity", "argument": {"serviceAffinity" : {"labels" : ["region"]}}},
+			{"name": "TestLabelsPresence",  "argument": {"labelsPresence"  : {"labels" : ["foo"], "presence":true}}}
+
 		  ],"priorities": [
 			{"name": "EqualPriority",   "weight": 2},
 			{"name": "ImageLocalityPriority",   "weight": 2},
@@ -441,9 +511,9 @@ func TestCompatibility_v1_Scheduler(t *testing.T) {
 					{Name: "MaxAzureDiskVolumeCount"},
 					{Name: "MatchInterPodAffinity"},
 					{Name: "GeneralPredicates"},
+					{Name: "CheckVolumeBinding"},
 					{Name: "TestServiceAffinity", Argument: &schedulerapi.PredicateArgument{ServiceAffinity: &schedulerapi.ServiceAffinity{Labels: []string{"region"}}}},
 					{Name: "TestLabelsPresence", Argument: &schedulerapi.PredicateArgument{LabelsPresence: &schedulerapi.LabelsPresence{Labels: []string{"foo"}, Presence: true}}},
-					{Name: "NoVolumeNodeConflict"},
 				},
 				Priorities: []schedulerapi.PriorityPolicy{
 					{Name: "EqualPriority", Weight: 2},
@@ -506,6 +576,7 @@ func TestCompatibility_v1_Scheduler(t *testing.T) {
 			informerFactory.Apps().V1beta1().StatefulSets(),
 			informerFactory.Core().V1().Services(),
 			informerFactory.Policy().V1beta1().PodDisruptionBudgets(),
+			informerFactory.Storage().V1().StorageClasses(),
 			v1.DefaultHardPodAffinitySymmetricWeight,
 			enableEquivalenceCache,
 		).CreateFromConfig(policy); err != nil {

plugin/pkg/scheduler/algorithmprovider/defaults/defaults.go

@@ -114,7 +114,7 @@ func defaultPredicates() sets.String {
 		factory.RegisterFitPredicateFactory(
 			"NoVolumeZoneConflict",
 			func(args factory.PluginFactoryArgs) algorithm.FitPredicate {
-				return predicates.NewVolumeZonePredicate(args.PVInfo, args.PVCInfo)
+				return predicates.NewVolumeZonePredicate(args.PVInfo, args.PVCInfo, args.StorageClassInfo)
 			},
 		),
 		// Fit is determined by whether or not there would be too many AWS EBS volumes attached to the node
@@ -165,11 +165,11 @@ func defaultPredicates() sets.String {
 		// Fit is determined based on whether a pod can tolerate all of the node's taints
 		factory.RegisterFitPredicate("PodToleratesNodeTaints", predicates.PodToleratesNodeTaints),
 
-		// Fit is determined by volume zone requirements.
+		// Fit is determined by volume topology requirements.
 		factory.RegisterFitPredicateFactory(
-			"NoVolumeNodeConflict",
+			predicates.CheckVolumeBinding,
 			func(args factory.PluginFactoryArgs) algorithm.FitPredicate {
-				return predicates.NewVolumeNodePredicate(args.PVInfo, args.PVCInfo, nil)
+				return predicates.NewVolumeBindingPredicate(args.VolumeBinder)
 			},
 		),
 	)

plugin/pkg/scheduler/algorithmprovider/defaults/defaults_test.go

@@ -20,6 +20,7 @@ import (
 	"testing"
 
 	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/kubernetes/plugin/pkg/scheduler/algorithm/predicates"
 )
 
 func TestCopyAndReplace(t *testing.T) {
@@ -75,9 +76,9 @@ func TestDefaultPredicates(t *testing.T) {
 		"GeneralPredicates",
 		"CheckNodeMemoryPressure",
 		"CheckNodeDiskPressure",
-		"NoVolumeNodeConflict",
 		"CheckNodeCondition",
 		"PodToleratesNodeTaints",
+		predicates.CheckVolumeBinding,
 	)
 
 	if expected := defaultPredicates(); !result.Equal(expected) {

plugin/pkg/scheduler/core/BUILD

@@ -53,6 +53,7 @@ go_library(
         "//plugin/pkg/scheduler/api:go_default_library",
         "//plugin/pkg/scheduler/schedulercache:go_default_library",
         "//plugin/pkg/scheduler/util:go_default_library",
+        "//plugin/pkg/scheduler/volumebinder:go_default_library",
         "//vendor/github.com/golang/glog:go_default_library",
         "//vendor/github.com/golang/groupcache/lru:go_default_library",
         "//vendor/k8s.io/api/core/v1:go_default_library",

plugin/pkg/scheduler/core/extender_test.go

@@ -317,7 +317,7 @@ func TestGenericSchedulerWithExtenders(t *testing.T) {
 		}
 		queue := NewSchedulingQueue()
 		scheduler := NewGenericScheduler(
-			cache, nil, queue, test.predicates, algorithm.EmptyPredicateMetadataProducer, test.prioritizers, algorithm.EmptyMetadataProducer, extenders)
+			cache, nil, queue, test.predicates, algorithm.EmptyPredicateMetadataProducer, test.prioritizers, algorithm.EmptyMetadataProducer, extenders, nil)
 		podIgnored := &v1.Pod{}
 		machine, err := scheduler.Schedule(podIgnored, schedulertesting.FakeNodeLister(makeNodeList(test.nodes)))
 		if test.expectsErr {

plugin/pkg/scheduler/core/generic_scheduler.go

@@ -36,6 +36,7 @@ import (
 	"k8s.io/kubernetes/plugin/pkg/scheduler/util"
 
 	"github.com/golang/glog"
+	"k8s.io/kubernetes/plugin/pkg/scheduler/volumebinder"
 )
 
 type FailedPredicateMap map[string][]algorithm.PredicateFailureReason
@@ -91,6 +92,7 @@ type genericScheduler struct {
 	lastNodeIndex         uint64
 
 	cachedNodeInfoMap map[string]*schedulercache.NodeInfo
+	volumeBinder      *volumebinder.VolumeBinder
 }
 
 // Schedule tries to schedule the given pod to one of node in the node list.
@@ -867,7 +869,10 @@ func nodesWherePreemptionMightHelp(pod *v1.Pod, nodes []*v1.Node, failedPredicat
 				predicates.ErrNodeNotReady,
 				predicates.ErrNodeNetworkUnavailable,
 				predicates.ErrNodeUnschedulable,
-				predicates.ErrNodeUnknownCondition:
+				predicates.ErrNodeUnknownCondition,
+				predicates.ErrVolumeZoneConflict,
+				predicates.ErrVolumeNodeConflict,
+				predicates.ErrVolumeBindConflict:
 				unresolvableReasonExist = true
 				break
 				// TODO(bsalamat): Please add affinity failure cases once we have specific affinity failure errors.
@@ -909,7 +914,8 @@ func NewGenericScheduler(
 	predicateMetaProducer algorithm.PredicateMetadataProducer,
 	prioritizers []algorithm.PriorityConfig,
 	priorityMetaProducer algorithm.MetadataProducer,
-	extenders []algorithm.SchedulerExtender) algorithm.ScheduleAlgorithm {
+	extenders []algorithm.SchedulerExtender,
+	volumeBinder *volumebinder.VolumeBinder) algorithm.ScheduleAlgorithm {
 	return &genericScheduler{
 		cache:                 cache,
 		equivalenceCache:      eCache,
@@ -920,5 +926,6 @@ func NewGenericScheduler(
 		priorityMetaProducer:  priorityMetaProducer,
 		extenders:             extenders,
 		cachedNodeInfoMap:     make(map[string]*schedulercache.NodeInfo),
+		volumeBinder:          volumeBinder,
 	}
 }

plugin/pkg/scheduler/core/generic_scheduler_test.go

@@ -311,7 +311,7 @@ func TestGenericScheduler(t *testing.T) {
 		}
 
 		scheduler := NewGenericScheduler(
-			cache, nil, NewSchedulingQueue(), test.predicates, algorithm.EmptyPredicateMetadataProducer, test.prioritizers, algorithm.EmptyMetadataProducer, []algorithm.SchedulerExtender{})
+			cache, nil, NewSchedulingQueue(), test.predicates, algorithm.EmptyPredicateMetadataProducer, test.prioritizers, algorithm.EmptyMetadataProducer, []algorithm.SchedulerExtender{}, nil)
 		machine, err := scheduler.Schedule(test.pod, schedulertesting.FakeNodeLister(makeNodeList(test.nodes)))
 
 		if !reflect.DeepEqual(err, test.wErr) {
@@ -1190,7 +1190,7 @@ func TestPreempt(t *testing.T) {
 			extenders = append(extenders, extender)
 		}
 		scheduler := NewGenericScheduler(
-			cache, nil, NewSchedulingQueue(), map[string]algorithm.FitPredicate{"matches": algorithmpredicates.PodFitsResources}, algorithm.EmptyPredicateMetadataProducer, []algorithm.PriorityConfig{{Function: numericPriority, Weight: 1}}, algorithm.EmptyMetadataProducer, extenders)
+			cache, nil, NewSchedulingQueue(), map[string]algorithm.FitPredicate{"matches": algorithmpredicates.PodFitsResources}, algorithm.EmptyPredicateMetadataProducer, []algorithm.PriorityConfig{{Function: numericPriority, Weight: 1}}, algorithm.EmptyMetadataProducer, extenders, nil)
 		// Call Preempt and check the expected results.
 		node, victims, _, err := scheduler.Preempt(test.pod, schedulertesting.FakeNodeLister(makeNodeList(nodeNames)), error(&FitError{Pod: test.pod, FailedPredicates: failedPredMap}))
 		if err != nil {

plugin/pkg/scheduler/factory/BUILD

@@ -16,6 +16,7 @@ go_library(
     deps = [
         "//pkg/api/v1/pod:go_default_library",
         "//pkg/apis/core/helper:go_default_library",
+        "//pkg/features:go_default_library",
         "//pkg/kubelet/apis:go_default_library",
         "//plugin/pkg/scheduler:go_default_library",
         "//plugin/pkg/scheduler/algorithm:go_default_library",
@@ -26,6 +27,7 @@ go_library(
         "//plugin/pkg/scheduler/core:go_default_library",
         "//plugin/pkg/scheduler/schedulercache:go_default_library",
         "//plugin/pkg/scheduler/util:go_default_library",
+        "//plugin/pkg/scheduler/volumebinder:go_default_library",
         "//vendor/github.com/golang/glog:go_default_library",
         "//vendor/k8s.io/api/core/v1:go_default_library",
         "//vendor/k8s.io/api/policy/v1beta1:go_default_library",
@@ -38,15 +40,18 @@ go_library(
         "//vendor/k8s.io/apimachinery/pkg/types:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/runtime:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/sets:go_default_library",
+        "//vendor/k8s.io/apiserver/pkg/util/feature:go_default_library",
         "//vendor/k8s.io/client-go/informers/apps/v1beta1:go_default_library",
         "//vendor/k8s.io/client-go/informers/core/v1:go_default_library",
         "//vendor/k8s.io/client-go/informers/extensions/v1beta1:go_default_library",
         "//vendor/k8s.io/client-go/informers/policy/v1beta1:go_default_library",
+        "//vendor/k8s.io/client-go/informers/storage/v1:go_default_library",
         "//vendor/k8s.io/client-go/kubernetes:go_default_library",
         "//vendor/k8s.io/client-go/listers/apps/v1beta1:go_default_library",
         "//vendor/k8s.io/client-go/listers/core/v1:go_default_library",
         "//vendor/k8s.io/client-go/listers/extensions/v1beta1:go_default_library",
         "//vendor/k8s.io/client-go/listers/policy/v1beta1:go_default_library",
+        "//vendor/k8s.io/client-go/listers/storage/v1:go_default_library",
         "//vendor/k8s.io/client-go/tools/cache:go_default_library",
     ],
 )

plugin/pkg/scheduler/factory/factory_test.go

@@ -66,6 +66,7 @@ func TestCreate(t *testing.T) {
 		informerFactory.Apps().V1beta1().StatefulSets(),
 		informerFactory.Core().V1().Services(),
 		informerFactory.Policy().V1beta1().PodDisruptionBudgets(),
+		informerFactory.Storage().V1().StorageClasses(),
 		v1.DefaultHardPodAffinitySymmetricWeight,
 		enableEquivalenceCache,
 	)
@@ -99,6 +100,7 @@ func TestCreateFromConfig(t *testing.T) {
 		informerFactory.Apps().V1beta1().StatefulSets(),
 		informerFactory.Core().V1().Services(),
 		informerFactory.Policy().V1beta1().PodDisruptionBudgets(),
+		informerFactory.Storage().V1().StorageClasses(),
 		v1.DefaultHardPodAffinitySymmetricWeight,
 		enableEquivalenceCache,
 	)
@@ -159,6 +161,7 @@ func TestCreateFromConfigWithHardPodAffinitySymmetricWeight(t *testing.T) {
 		informerFactory.Apps().V1beta1().StatefulSets(),
 		informerFactory.Core().V1().Services(),
 		informerFactory.Policy().V1beta1().PodDisruptionBudgets(),
+		informerFactory.Storage().V1().StorageClasses(),
 		v1.DefaultHardPodAffinitySymmetricWeight,
 		enableEquivalenceCache,
 	)
@@ -220,6 +223,7 @@ func TestCreateFromEmptyConfig(t *testing.T) {
 		informerFactory.Apps().V1beta1().StatefulSets(),
 		informerFactory.Core().V1().Services(),
 		informerFactory.Policy().V1beta1().PodDisruptionBudgets(),
+		informerFactory.Storage().V1().StorageClasses(),
 		v1.DefaultHardPodAffinitySymmetricWeight,
 		enableEquivalenceCache,
 	)
@@ -278,6 +282,7 @@ func TestDefaultErrorFunc(t *testing.T) {
 		informerFactory.Apps().V1beta1().StatefulSets(),
 		informerFactory.Core().V1().Services(),
 		informerFactory.Policy().V1beta1().PodDisruptionBudgets(),
+		informerFactory.Storage().V1().StorageClasses(),
 		v1.DefaultHardPodAffinitySymmetricWeight,
 		enableEquivalenceCache,
 	)
@@ -388,6 +393,7 @@ func TestInvalidHardPodAffinitySymmetricWeight(t *testing.T) {
 		informerFactory.Apps().V1beta1().StatefulSets(),
 		informerFactory.Core().V1().Services(),
 		informerFactory.Policy().V1beta1().PodDisruptionBudgets(),
+		informerFactory.Storage().V1().StorageClasses(),
 		-1,
 		enableEquivalenceCache,
 	)
@@ -435,6 +441,7 @@ func TestInvalidFactoryArgs(t *testing.T) {
 			informerFactory.Apps().V1beta1().StatefulSets(),
 			informerFactory.Core().V1().Services(),
 			informerFactory.Policy().V1beta1().PodDisruptionBudgets(),
+			informerFactory.Storage().V1().StorageClasses(),
 			test.hardPodAffinitySymmetricWeight,
 			enableEquivalenceCache,
 		)

plugin/pkg/scheduler/factory/plugins.go

@@ -30,6 +30,7 @@ import (
 	schedulerapi "k8s.io/kubernetes/plugin/pkg/scheduler/api"
 
 	"github.com/golang/glog"
+	"k8s.io/kubernetes/plugin/pkg/scheduler/volumebinder"
 )
 
 // PluginFactoryArgs are passed to all plugin factory functions.
@@ -43,6 +44,8 @@ type PluginFactoryArgs struct {
 	NodeInfo                       predicates.NodeInfo
 	PVInfo                         predicates.PersistentVolumeInfo
 	PVCInfo                        predicates.PersistentVolumeClaimInfo
+	StorageClassInfo               predicates.StorageClassInfo
+	VolumeBinder                   *volumebinder.VolumeBinder
 	HardPodAffinitySymmetricWeight int32
 }
 

plugin/pkg/scheduler/scheduler.go

@@ -17,16 +17,20 @@ limitations under the License.
 package scheduler
 
 import (
+	"fmt"
 	"time"
 
 	"k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/wait"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	clientset "k8s.io/client-go/kubernetes"
 	corelisters "k8s.io/client-go/listers/core/v1"
 	"k8s.io/client-go/tools/record"
+	"k8s.io/kubernetes/pkg/features"
 	"k8s.io/kubernetes/plugin/pkg/scheduler/algorithm"
+	"k8s.io/kubernetes/plugin/pkg/scheduler/algorithm/predicates"
 	schedulerapi "k8s.io/kubernetes/plugin/pkg/scheduler/api"
 	"k8s.io/kubernetes/plugin/pkg/scheduler/core"
 	"k8s.io/kubernetes/plugin/pkg/scheduler/metrics"
@@ -34,6 +38,7 @@ import (
 	"k8s.io/kubernetes/plugin/pkg/scheduler/util"
 
 	"github.com/golang/glog"
+	"k8s.io/kubernetes/plugin/pkg/scheduler/volumebinder"
 )
 
 // Binder knows how to write a binding.
@@ -129,6 +134,9 @@ type Config struct {
 
 	// Close this to shut down the scheduler.
 	StopEverything chan struct{}
+
+	// VolumeBinder handles PVC/PV binding for the pod.
+	VolumeBinder *volumebinder.VolumeBinder
 }
 
 // NewFromConfigurator returns a new scheduler that is created entirely by the Configurator.  Assumes Create() is implemented.
@@ -164,6 +172,10 @@ func (sched *Scheduler) Run() {
 		return
 	}
 
+	if utilfeature.DefaultFeatureGate.Enabled(features.VolumeScheduling) {
+		go sched.config.VolumeBinder.Run(sched.bindVolumesWorker, sched.config.StopEverything)
+	}
+
 	go wait.Until(sched.scheduleOne, 0, sched.config.StopEverything)
 }
 
@@ -240,6 +252,114 @@ func (sched *Scheduler) preempt(preemptor *v1.Pod, scheduleErr error) (string, e
 	return nodeName, err
 }
 
+// assumeAndBindVolumes will update the volume cache and then asynchronously bind volumes if required.
+//
+// If volume binding is required, then the bind volumes routine will update the pod to send it back through
+// the scheduler.
+//
+// Otherwise, return nil error and continue to assume the pod.
+//
+// This function modifies assumed if volume binding is required.
+func (sched *Scheduler) assumeAndBindVolumes(assumed *v1.Pod, host string) error {
+	if utilfeature.DefaultFeatureGate.Enabled(features.VolumeScheduling) {
+		allBound, bindingRequired, err := sched.config.VolumeBinder.Binder.AssumePodVolumes(assumed, host)
+		if err != nil {
+			sched.config.Error(assumed, err)
+			sched.config.Recorder.Eventf(assumed, v1.EventTypeWarning, "FailedScheduling", "AssumePodVolumes failed: %v", err)
+			sched.config.PodConditionUpdater.Update(assumed, &v1.PodCondition{
+				Type:    v1.PodScheduled,
+				Status:  v1.ConditionFalse,
+				Reason:  "SchedulerError",
+				Message: err.Error(),
+			})
+			return err
+		}
+		if !allBound {
+			err = fmt.Errorf("Volume binding started, waiting for completion")
+			if bindingRequired {
+				if sched.config.Ecache != nil {
+					invalidPredicates := sets.NewString(predicates.CheckVolumeBinding)
+					sched.config.Ecache.InvalidateCachedPredicateItemOfAllNodes(invalidPredicates)
+				}
+
+				// bindVolumesWorker() will update the Pod object to put it back in the scheduler queue
+				sched.config.VolumeBinder.BindQueue.Add(assumed)
+			} else {
+				// We are just waiting for PV controller to finish binding, put it back in the
+				// scheduler queue
+				sched.config.Error(assumed, err)
+				sched.config.Recorder.Eventf(assumed, v1.EventTypeNormal, "FailedScheduling", "%v", err)
+				sched.config.PodConditionUpdater.Update(assumed, &v1.PodCondition{
+					Type:   v1.PodScheduled,
+					Status: v1.ConditionFalse,
+					Reason: "VolumeBindingWaiting",
+				})
+			}
+			return err
+		}
+	}
+	return nil
+}
+
+// bindVolumesWorker() processes pods queued in assumeAndBindVolumes() and tries to
+// make the API update for volume binding.
+// This function runs forever until the volume BindQueue is closed.
+func (sched *Scheduler) bindVolumesWorker() {
+	workFunc := func() bool {
+		keyObj, quit := sched.config.VolumeBinder.BindQueue.Get()
+		if quit {
+			return true
+		}
+		defer sched.config.VolumeBinder.BindQueue.Done(keyObj)
+
+		assumed, ok := keyObj.(*v1.Pod)
+		if !ok {
+			glog.V(4).Infof("Object is not a *v1.Pod")
+			return false
+		}
+
+		// TODO: add metrics
+		var reason string
+		var eventType string
+
+		glog.V(5).Infof("Trying to bind volumes for pod \"%v/%v\"", assumed.Namespace, assumed.Name)
+
+		// The Pod is always sent back to the scheduler afterwards.
+		err := sched.config.VolumeBinder.Binder.BindPodVolumes(assumed)
+		if err != nil {
+			glog.V(1).Infof("Failed to bind volumes for pod \"%v/%v\"", assumed.Namespace, assumed.Name, err)
+			reason = "VolumeBindingFailed"
+			eventType = v1.EventTypeWarning
+		} else {
+			glog.V(4).Infof("Successfully bound volumes for pod \"%v/%v\"", assumed.Namespace, assumed.Name)
+			reason = "VolumeBindingWaiting"
+			eventType = v1.EventTypeNormal
+			err = fmt.Errorf("Volume binding started, waiting for completion")
+		}
+
+		// Always fail scheduling regardless of binding success.
+		// The Pod needs to be sent back through the scheduler to:
+		// * Retry volume binding if it fails.
+		// * Retry volume binding if dynamic provisioning fails.
+		// * Bind the Pod to the Node once all volumes are bound.
+		sched.config.Error(assumed, err)
+		sched.config.Recorder.Eventf(assumed, eventType, "FailedScheduling", "%v", err)
+		sched.config.PodConditionUpdater.Update(assumed, &v1.PodCondition{
+			Type:   v1.PodScheduled,
+			Status: v1.ConditionFalse,
+			Reason: reason,
+		})
+		return false
+	}
+
+	for {
+		if quit := workFunc(); quit {
+			glog.V(4).Infof("bindVolumesWorker shutting down")
+			break
+		}
+	}
+}
+
 // assume signals to the cache that a pod is already in the cache, so that binding can be asynchronous.
 // assume modifies `assumed`.
 func (sched *Scheduler) assume(assumed *v1.Pod, host string) error {
@@ -334,15 +454,32 @@ func (sched *Scheduler) scheduleOne() {
 
 	// Tell the cache to assume that a pod now is running on a given node, even though it hasn't been bound yet.
 	// This allows us to keep scheduling without waiting on binding to occur.
-	assumedPod := *pod
+	assumedPod := pod.DeepCopy()
+
+	// Assume volumes first before assuming the pod.
+	//
+	// If no volumes need binding, then nil is returned, and continue to assume the pod.
+	//
+	// Otherwise, error is returned and volume binding is started asynchronously for all of the pod's volumes.
+	// scheduleOne() returns immediately on error, so that it doesn't continue to assume the pod.
+	//
+	// After the asynchronous volume binding updates are made, it will send the pod back through the scheduler for
+	// subsequent passes until all volumes are fully bound.
+	//
+	// This function modifies 'assumedPod' if volume binding is required.
+	err = sched.assumeAndBindVolumes(assumedPod, suggestedHost)
+	if err != nil {
+		return
+	}
+
 	// assume modifies `assumedPod` by setting NodeName=suggestedHost
-	err = sched.assume(&assumedPod, suggestedHost)
+	err = sched.assume(assumedPod, suggestedHost)
 	if err != nil {
 		return
 	}
 	// bind the pod to its host asynchronously (we can do this b/c of the assumption step above).
 	go func() {
-		err := sched.bind(&assumedPod, &v1.Binding{
+		err := sched.bind(assumedPod, &v1.Binding{
 			ObjectMeta: metav1.ObjectMeta{Namespace: assumedPod.Namespace, Name: assumedPod.Name, UID: assumedPod.UID},
 			Target: v1.ObjectReference{
 				Kind: "Node",

plugin/pkg/scheduler/volumebinder/BUILD

+load("@io_bazel_rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["volume_binder.go"],
+    importpath = "k8s.io/kubernetes/plugin/pkg/scheduler/volumebinder",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//pkg/controller/volume/persistentvolume:go_default_library",
+        "//vendor/k8s.io/api/core/v1:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/util/wait:go_default_library",
+        "//vendor/k8s.io/client-go/informers/core/v1:go_default_library",
+        "//vendor/k8s.io/client-go/informers/storage/v1:go_default_library",
+        "//vendor/k8s.io/client-go/kubernetes:go_default_library",
+        "//vendor/k8s.io/client-go/util/workqueue:go_default_library",
+    ],
+)
+
+filegroup(
+    name = "package-srcs",
+    srcs = glob(["**"]),
+    tags = ["automanaged"],
+    visibility = ["//visibility:private"],
+)
+
+filegroup(
+    name = "all-srcs",
+    srcs = [":package-srcs"],
+    tags = ["automanaged"],
+    visibility = ["//visibility:public"],
+)

plugin/pkg/scheduler/volumebinder/volume_binder.go

+/*
+Copyright 2017 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 volumebinder
+
+import (
+	"time"
+
+	"k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/util/wait"
+	coreinformers "k8s.io/client-go/informers/core/v1"
+	storageinformers "k8s.io/client-go/informers/storage/v1"
+	clientset "k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/util/workqueue"
+	"k8s.io/kubernetes/pkg/controller/volume/persistentvolume"
+)
+
+// VolumeBinder sets up the volume binding library and manages
+// the volume binding operations with a queue.
+type VolumeBinder struct {
+	Binder    persistentvolume.SchedulerVolumeBinder
+	BindQueue *workqueue.Type
+}
+
+// NewVolumeBinder sets up the volume binding library and binding queue
+func NewVolumeBinder(
+	client clientset.Interface,
+	pvcInformer coreinformers.PersistentVolumeClaimInformer,
+	pvInformer coreinformers.PersistentVolumeInformer,
+	nodeInformer coreinformers.NodeInformer,
+	storageClassInformer storageinformers.StorageClassInformer) *VolumeBinder {
+
+	return &VolumeBinder{
+		Binder:    persistentvolume.NewVolumeBinder(client, pvcInformer, pvInformer, nodeInformer, storageClassInformer),
+		BindQueue: workqueue.NewNamed("podsToBind"),
+	}
+}
+
+// NewFakeVolumeBinder sets up a fake volume binder and binding queue
+func NewFakeVolumeBinder(config *persistentvolume.FakeVolumeBinderConfig) *VolumeBinder {
+	return &VolumeBinder{
+		Binder:    persistentvolume.NewFakeVolumeBinder(config),
+		BindQueue: workqueue.NewNamed("podsToBind"),
+	}
+}
+
+// Run starts a goroutine to handle the binding queue with the given function.
+func (b *VolumeBinder) Run(bindWorkFunc func(), stopCh <-chan struct{}) {
+	go wait.Until(bindWorkFunc, time.Second, stopCh)
+
+	<-stopCh
+	b.BindQueue.ShutDown()
+}
+
+// DeletePodBindings will delete the cached volume bindings for the given pod.
+func (b *VolumeBinder) DeletePodBindings(pod *v1.Pod) {
+	cache := b.Binder.GetBindingsCache()
+	if cache != nil && pod != nil {
+		cache.DeleteBindings(pod)
+	}
+}

test/e2e/storage/BUILD

@@ -62,6 +62,7 @@ go_library(
         "//vendor/github.com/vmware/govmomi/vim25/types:go_default_library",
         "//vendor/golang.org/x/net/context:go_default_library",
         "//vendor/google.golang.org/api/googleapi:go_default_library",
+        "//vendor/k8s.io/api/apps/v1beta1:go_default_library",
         "//vendor/k8s.io/api/batch/v1:go_default_library",
         "//vendor/k8s.io/api/core/v1:go_default_library",
         "//vendor/k8s.io/api/extensions/v1beta1:go_default_library",

test/integration/scheduler/BUILD

@@ -17,6 +17,7 @@ go_test(
         "priorities_test.go",
         "scheduler_test.go",
         "taint_test.go",
+        "volume_binding_test.go",
     ],
     importpath = "k8s.io/kubernetes/test/integration/scheduler",
     library = ":go_default_library",
@@ -25,10 +26,12 @@ go_test(
         "//pkg/api/legacyscheme:go_default_library",
         "//pkg/api/testapi:go_default_library",
         "//pkg/apis/componentconfig:go_default_library",
+        "//pkg/apis/core/v1/helper:go_default_library",
         "//pkg/client/clientset_generated/internalclientset:go_default_library",
         "//pkg/client/informers/informers_generated/internalversion:go_default_library",
         "//pkg/controller/node:go_default_library",
         "//pkg/controller/node/ipam:go_default_library",
+        "//pkg/controller/volume/persistentvolume:go_default_library",
         "//pkg/features:go_default_library",
         "//pkg/kubeapiserver/admission:go_default_library",
         "//plugin/cmd/kube-scheduler/app:go_default_library",
@@ -47,6 +50,7 @@ go_test(
         "//vendor/github.com/golang/glog:go_default_library",
         "//vendor/k8s.io/api/core/v1:go_default_library",
         "//vendor/k8s.io/api/policy/v1beta1:go_default_library",
+        "//vendor/k8s.io/api/storage/v1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/api/errors:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/api/resource:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",

test/integration/scheduler/extender_test.go

@@ -369,6 +369,7 @@ func TestSchedulerExtender(t *testing.T) {
 		informerFactory.Apps().V1beta1().StatefulSets(),
 		informerFactory.Core().V1().Services(),
 		informerFactory.Policy().V1beta1().PodDisruptionBudgets(),
+		informerFactory.Storage().V1().StorageClasses(),
 		v1.DefaultHardPodAffinitySymmetricWeight,
 		enableEquivalenceCache,
 	)

test/integration/scheduler/scheduler_test.go

@@ -469,6 +469,7 @@ func TestMultiScheduler(t *testing.T) {
 		informerFactory2.Apps().V1beta1().StatefulSets(),
 		informerFactory2.Core().V1().Services(),
 		informerFactory2.Policy().V1beta1().PodDisruptionBudgets(),
+		informerFactory2.Storage().V1().StorageClasses(),
 		v1.DefaultHardPodAffinitySymmetricWeight,
 		enableEquivalenceCache,
 	)

test/integration/scheduler/taint_test.go

@@ -131,6 +131,7 @@ func TestTaintNodeByCondition(t *testing.T) {
 		informers.Apps().V1beta1().StatefulSets(),
 		informers.Core().V1().Services(),
 		informers.Policy().V1beta1().PodDisruptionBudgets(),
+		informers.Storage().V1().StorageClasses(),
 		v1.DefaultHardPodAffinitySymmetricWeight,
 		true, // Enable EqualCache by default.
 	)

test/integration/scheduler/util.go

@@ -78,6 +78,7 @@ func initTest(t *testing.T, nsPrefix string) *TestContext {
 		context.informerFactory.Apps().V1beta1().StatefulSets(),
 		context.informerFactory.Core().V1().Services(),
 		context.informerFactory.Policy().V1beta1().PodDisruptionBudgets(),
+		context.informerFactory.Storage().V1().StorageClasses(),
 		v1.DefaultHardPodAffinitySymmetricWeight,
 		true,
 	)

test/integration/scheduler_perf/util.go

@@ -75,6 +75,7 @@ func mustSetupScheduler() (schedulerConfigurator scheduler.Configurator, destroy
 		informerFactory.Apps().V1beta1().StatefulSets(),
 		informerFactory.Core().V1().Services(),
 		informerFactory.Policy().V1beta1().PodDisruptionBudgets(),
+		informerFactory.Storage().V1().StorageClasses(),
 		v1.DefaultHardPodAffinitySymmetricWeight,
 		enableEquivalenceCache,
 	)