In findMatchingVolumes:

pkg/controller/volume/events/event.go

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

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

@@ -822,6 +822,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
@@ -96,7 +97,7 @@ func (pvIndex *persistentVolumeOrderedIndex) findByClaim(claim *v1.PersistentVol
 			return nil, err
 		}
 
-		bestVol, err := findMatchingVolume(claim, volumes)
+		bestVol, err := findMatchingVolume(claim, volumes, nil /* node for topology binding*/, nil /* exclusion map */, delayBinding)
 		if err != nil {
 			return nil, err
 		}
@@ -108,7 +109,27 @@ func (pvIndex *persistentVolumeOrderedIndex) findByClaim(claim *v1.PersistentVol
 	return nil, nil
 }
 
-func findMatchingVolume(claim *v1.PersistentVolumeClaim, volumes []*v1.PersistentVolume) (*v1.PersistentVolume, error) {
+// 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)]
@@ -131,6 +152,11 @@ func findMatchingVolume(claim *v1.PersistentVolumeClaim, volumes []*v1.Persisten
 	// - 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)
@@ -143,6 +169,15 @@ func findMatchingVolume(claim *v1.PersistentVolumeClaim, volumes []*v1.Persisten
 			continue
 		}
 
+		if node != nil {
+			// Scheduler path, check that the PV NodeAffinity
+			// is satisfied by the node
+			err := volumeutil.CheckNodeAffinity(volume, node.Labels)
+			if err != nil {
+				continue
+			}
+		}
+
 		if isVolumeBoundToClaim(volume, claim) {
 			// this claim and volume are pre-bound; return
 			// the volume if the size request is satisfied,
@@ -153,6 +188,13 @@ func findMatchingVolume(claim *v1.PersistentVolumeClaim, volumes []*v1.Persisten
 			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
@@ -166,6 +208,14 @@ func findMatchingVolume(claim *v1.PersistentVolumeClaim, volumes []*v1.Persisten
 			continue
 		}
 
+		if node != nil {
+			// Scheduler path
+			// Check that the access modes match
+			if !checkAccessModes(claim, volume) {
+				continue
+			}
+		}
+
 		if volumeQty.Cmp(requestedQty) >= 0 {
 			if smallestVolume == nil || smallestVolumeQty.Cmp(volumeQty) > 0 {
 				smallestVolume = volume
@@ -204,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
@@ -287,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
 			}

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