Volume scheduling library changes:

* FindPodVolumes
    * Prebound PVCs are treated like unbound immediate PVCs and will error
    * Always check for fully bound PVCs and cache bindings for not fully
      bound PVCs
* BindPodVolumes
    * Retry API updates for not fully bound PVCs even if the assume cache
      already marked it
    * Wait for PVCs to be fully bound after making the API updates
    * Error when detecting binding/provisioning failure conditions

pkg/controller/volume/persistentvolume/scheduler_binder.go

@@ -19,12 +19,14 @@ package persistentvolume
 import (
 	"fmt"
 	"sort"
+	"time"
 
 	"github.com/golang/glog"
 
 	"k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
+	"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"
@@ -42,16 +44,19 @@ import (
 //    a. Invokes all predicate functions, parallelized across nodes.  FindPodVolumes() is invoked here.
 //    b. Invokes all priority functions.  Future/TBD
 //    c. Selects the best node for the Pod.
-//    d. Cache the node selection for the Pod. (Assume phase)
+//    d. Cache the node selection for the Pod. AssumePodVolumes() is invoked here.
 //       i.  If PVC binding is required, cache in-memory only:
-//           * Updated PV objects for prebinding to the corresponding PVCs.
+//           * For manual binding: update PV objects for prebinding to the corresponding PVCs.
-//           * For the pod, which PVs need API updates.
+//           * For dynamic provisioning: update PVC object with a selected node from c)
-//           AssumePodVolumes() is invoked here.  Then BindPodVolumes() is called asynchronously by the
+//           * For the pod, which PVCs and PVs need API updates.
-//           scheduler.  After BindPodVolumes() is complete, the Pod is added back to the scheduler queue
+//       ii. Afterwards, the main scheduler caches the Pod->Node binding in the scheduler's pod cache,
-//           to be processed again until all PVCs are bound.
+//           This is handled in the scheduler and not here.
-//       ii. If PVC binding is not required, cache the Pod->Node binding in the scheduler's pod cache,
+//    e. Asynchronously bind volumes and pod in a separate goroutine
-//           and asynchronously bind the Pod to the Node.  This is handled in the scheduler and not here.
+//        i.  BindPodVolumes() is called first. It makes all the necessary API updates and waits for
-// 2. Once the assume operation is done, the scheduler processes the next Pod in the scheduler queue
+//            PV controller to fully bind and provision the PVCs. If binding fails, the Pod is sent
+//            back through the scheduler.
+//        ii. After BindPodVolumes() is complete, then the scheduler does the final Pod->Node binding.
+// 2. Once all the assume operations are done in d), the scheduler processes the next Pod in the scheduler queue
 //    while the actual binding operation occurs in the background.
 type SchedulerVolumeBinder interface {
 	// FindPodVolumes checks if all of a Pod's PVCs can be satisfied by the node.
@@ -71,18 +76,18 @@ type SchedulerVolumeBinder interface {
 	// 2. Take the PVCs that need provisioning and update the PVC cache with related
 	// annotations set.
 	//
-	// It returns true if all volumes are fully bound, and returns true if any volume binding/provisioning
+	// It returns true if all volumes are fully bound
-	// API operation needs to be done afterwards.
 	//
 	// This function will modify assumedPod with the node name.
 	// This function is called serially.
-	AssumePodVolumes(assumedPod *v1.Pod, nodeName string) (allFullyBound bool, bindingRequired bool, err error)
+	AssumePodVolumes(assumedPod *v1.Pod, nodeName string) (allFullyBound bool, err error)
 
 	// BindPodVolumes will:
 	// 1. Initiate the volume binding by making the API call to prebind the PV
 	// to its matching PVC.
 	// 2. Trigger the volume provisioning by making the API call to set related
 	// annotations on the PVC
+	// 3. Wait for PVCs to be completely bound by the PV controller
 	//
 	// This function can be called in parallel.
 	BindPodVolumes(assumedPod *v1.Pod) error
@@ -100,6 +105,9 @@ type volumeBinder struct {
 	// Stores binding decisions that were made in FindPodVolumes for use in AssumePodVolumes.
 	// AssumePodVolumes modifies the bindings again for use in BindPodVolumes.
 	podBindingCache PodBindingCache
+
+	// Amount of time to wait for the bind operation to succeed
+	bindTimeout time.Duration
 }
 
 // NewVolumeBinder sets up all the caches needed for the scheduler to make volume binding decisions.
@@ -107,7 +115,8 @@ func NewVolumeBinder(
 	kubeClient clientset.Interface,
 	pvcInformer coreinformers.PersistentVolumeClaimInformer,
 	pvInformer coreinformers.PersistentVolumeInformer,
-	storageClassInformer storageinformers.StorageClassInformer) SchedulerVolumeBinder {
+	storageClassInformer storageinformers.StorageClassInformer,
+	bindTimeout time.Duration) SchedulerVolumeBinder {
 
 	// TODO: find better way...
 	ctrl := &PersistentVolumeController{
@@ -120,6 +129,7 @@ func NewVolumeBinder(
 		pvcCache:        NewPVCAssumeCache(pvcInformer.Informer()),
 		pvCache:         NewPVAssumeCache(pvInformer.Informer()),
 		podBindingCache: NewPodBindingCache(),
+		bindTimeout:     bindTimeout,
 	}
 
 	return b
@@ -183,22 +193,24 @@ func (b *volumeBinder) FindPodVolumes(pod *v1.Pod, node *v1.Node) (unboundVolume
 // in podBindingCache for the chosen node, and:
 // 1. Update the pvCache with the new prebound PV.
 // 2. Update the pvcCache with the new PVCs with annotations set
-// It will update podBindingCache again with the PVs and PVCs that need an API update.
+// 3. Update podBindingCache again with cached API updates for PVs and PVCs.
-func (b *volumeBinder) AssumePodVolumes(assumedPod *v1.Pod, nodeName string) (allFullyBound, bindingRequired bool, err error) {
+func (b *volumeBinder) AssumePodVolumes(assumedPod *v1.Pod, nodeName string) (allFullyBound bool, err error) {
 	podName := getPodName(assumedPod)
 
 	glog.V(4).Infof("AssumePodVolumes for pod %q, node %q", podName, nodeName)
 
 	if allBound := b.arePodVolumesBound(assumedPod); allBound {
 		glog.V(4).Infof("AssumePodVolumes for pod %q, node %q: all PVCs bound and nothing to do", podName, nodeName)
-		return true, false, nil
+		return true, nil
 	}
 
 	assumedPod.Spec.NodeName = nodeName
-	// Assume PV
-	claimsToBind := b.podBindingCache.GetBindings(assumedPod, nodeName)
-	newBindings := []*bindingInfo{}
 
+	claimsToBind := b.podBindingCache.GetBindings(assumedPod, nodeName)
+	claimsToProvision := b.podBindingCache.GetProvisionedPVCs(assumedPod, nodeName)
+
+	// Assume PV
+	newBindings := []*bindingInfo{}
 	for _, binding := range claimsToBind {
 		newPV, dirty, err := b.ctrl.getBindVolumeToClaim(binding.pv, binding.pvc)
 		glog.V(5).Infof("AssumePodVolumes: getBindVolumeToClaim for pod %q, PV %q, PVC %q.  newPV %p, dirty %v, err: %v",
@@ -210,29 +222,20 @@ func (b *volumeBinder) AssumePodVolumes(assumedPod *v1.Pod, nodeName string) (al
 			err)
 		if err != nil {
 			b.revertAssumedPVs(newBindings)
-			return false, true, err
+			return false, err
 		}
+		// TODO: can we assume everytime?
 		if dirty {
 			err = b.pvCache.Assume(newPV)
 			if err != nil {
 				b.revertAssumedPVs(newBindings)
-				return false, true, err
+				return false, err
+			}
 		}
-
 		newBindings = append(newBindings, &bindingInfo{pv: newPV, pvc: binding.pvc})
 	}
-	}
-
-	// Don't update cached bindings if no API updates are needed.  This can happen if we
-	// previously updated the PV object and are waiting for the PV controller to finish binding.
-	if len(newBindings) != 0 {
-		bindingRequired = true
-		b.podBindingCache.UpdateBindings(assumedPod, nodeName, newBindings)
-	}
 
 	// Assume PVCs
-	claimsToProvision := b.podBindingCache.GetProvisionedPVCs(assumedPod, nodeName)
-
 	newProvisionedPVCs := []*v1.PersistentVolumeClaim{}
 	for _, claim := range claimsToProvision {
 		// The claims from method args can be pointing to watcher cache. We must not
@@ -249,50 +252,37 @@ func (b *volumeBinder) AssumePodVolumes(assumedPod *v1.Pod, nodeName string) (al
 		newProvisionedPVCs = append(newProvisionedPVCs, claimClone)
 	}
 
-	if len(newProvisionedPVCs) != 0 {
+	// Update cache with the assumed pvcs and pvs
-		bindingRequired = true
+	// Even if length is zero, update the cache with an empty slice to indicate that no
+	// operations are needed
+	b.podBindingCache.UpdateBindings(assumedPod, nodeName, newBindings)
 	b.podBindingCache.UpdateProvisionedPVCs(assumedPod, nodeName, newProvisionedPVCs)
-	}
 
 	return
 }
 
-// BindPodVolumes gets the cached bindings and PVCs to provision in podBindingCache
+// BindPodVolumes gets the cached bindings and PVCs to provision in podBindingCache,
-// and makes the API update for those PVs/PVCs.
+// makes the API update for those PVs/PVCs, and waits for the PVCs to be completely bound
+// by the PV controller.
 func (b *volumeBinder) BindPodVolumes(assumedPod *v1.Pod) error {
 	podName := getPodName(assumedPod)
-	glog.V(4).Infof("BindPodVolumes for pod %q", podName)
+	glog.V(4).Infof("BindPodVolumes for pod %q, node %q", podName, assumedPod.Spec.NodeName)
 
 	bindings := b.podBindingCache.GetBindings(assumedPod, assumedPod.Spec.NodeName)
 	claimsToProvision := b.podBindingCache.GetProvisionedPVCs(assumedPod, assumedPod.Spec.NodeName)
 
-	// Do the actual prebinding. Let the PV controller take care of the rest
+	// Start API operations
-	// There is no API rollback if the actual binding fails
+	err := b.bindAPIUpdate(podName, bindings, claimsToProvision)
-	for i, bindingInfo := range bindings {
-		glog.V(5).Infof("BindPodVolumes: Pod %q, binding PV %q to PVC %q", podName, bindingInfo.pv.Name, bindingInfo.pvc.Name)
-		_, err := b.ctrl.updateBindVolumeToClaim(bindingInfo.pv, bindingInfo.pvc, false)
 	if err != nil {
-			// only revert assumed cached updates for volumes we haven't successfully bound
-			b.revertAssumedPVs(bindings[i:])
-			// Revert all of the assumed cached updates for claims,
-			// since no actual API update will be done
-			b.revertAssumedPVCs(claimsToProvision)
 		return err
 	}
-	}
 
-	// Update claims objects to trigger volume provisioning. Let the PV controller take care of the rest
+	return wait.Poll(time.Second, b.bindTimeout, func() (bool, error) {
-	// PV controller is expect to signal back by removing related annotations if actual provisioning fails
+		// Get cached values every time in case the pod gets deleted
-	for i, claim := range claimsToProvision {
+		bindings = b.podBindingCache.GetBindings(assumedPod, assumedPod.Spec.NodeName)
-		if _, err := b.ctrl.kubeClient.CoreV1().PersistentVolumeClaims(claim.Namespace).Update(claim); err != nil {
+		claimsToProvision = b.podBindingCache.GetProvisionedPVCs(assumedPod, assumedPod.Spec.NodeName)
-			glog.V(4).Infof("updating PersistentVolumeClaim[%s] failed: %v", getPVCName(claim), err)
+		return b.checkBindings(assumedPod, bindings, claimsToProvision)
-			// only revert assumed cached updates for claims we haven't successfully updated
+	})
-			b.revertAssumedPVCs(claimsToProvision[i:])
-			return err
-		}
-	}
-
-	return nil
 }
 
 func getPodName(pod *v1.Pod) string {
@@ -303,12 +293,131 @@ func getPVCName(pvc *v1.PersistentVolumeClaim) string {
 	return pvc.Namespace + "/" + pvc.Name
 }
 
-func (b *volumeBinder) isVolumeBound(namespace string, vol *v1.Volume, checkFullyBound bool) (bool, *v1.PersistentVolumeClaim, error) {
+// bindAPIUpdate gets the cached bindings and PVCs to provision in podBindingCache
+// and makes the API update for those PVs/PVCs.
+func (b *volumeBinder) bindAPIUpdate(podName string, bindings []*bindingInfo, claimsToProvision []*v1.PersistentVolumeClaim) error {
+	if bindings == nil {
+		return fmt.Errorf("failed to get cached bindings for pod %q", podName)
+	}
+	if claimsToProvision == nil {
+		return fmt.Errorf("failed to get cached claims to provision for pod %q", podName)
+	}
+
+	lastProcessedBinding := 0
+	lastProcessedProvisioning := 0
+	defer func() {
+		// only revert assumed cached updates for volumes we haven't successfully bound
+		if lastProcessedBinding < len(bindings) {
+			b.revertAssumedPVs(bindings[lastProcessedBinding:])
+		}
+		// only revert assumed cached updates for claims we haven't updated,
+		if lastProcessedProvisioning < len(claimsToProvision) {
+			b.revertAssumedPVCs(claimsToProvision[lastProcessedProvisioning:])
+		}
+	}()
+
+	var (
+		binding *bindingInfo
+		claim   *v1.PersistentVolumeClaim
+	)
+
+	// Do the actual prebinding. Let the PV controller take care of the rest
+	// There is no API rollback if the actual binding fails
+	for _, binding = range bindings {
+		glog.V(5).Infof("bindAPIUpdate: Pod %q, binding PV %q to PVC %q", podName, binding.pv.Name, binding.pvc.Name)
+		// TODO: does it hurt if we make an api call and nothing needs to be updated?
+		if _, err := b.ctrl.updateBindVolumeToClaim(binding.pv, binding.pvc, false); err != nil {
+			return err
+		}
+		lastProcessedBinding++
+	}
+
+	// Update claims objects to trigger volume provisioning. Let the PV controller take care of the rest
+	// PV controller is expect to signal back by removing related annotations if actual provisioning fails
+	for _, claim = range claimsToProvision {
+		glog.V(5).Infof("bindAPIUpdate: Pod %q, PVC %q", podName, getPVCName(claim))
+		if _, err := b.ctrl.kubeClient.CoreV1().PersistentVolumeClaims(claim.Namespace).Update(claim); err != nil {
+			return err
+		}
+		lastProcessedProvisioning++
+	}
+
+	return nil
+}
+
+// checkBindings runs through all the PVCs in the Pod and checks:
+// * if the PVC is fully bound
+// * if there are any conditions that require binding to fail and be retried
+//
+// It returns true when all of the Pod's PVCs are fully bound, and error if
+// binding (and scheduling) needs to be retried
+func (b *volumeBinder) checkBindings(pod *v1.Pod, bindings []*bindingInfo, claimsToProvision []*v1.PersistentVolumeClaim) (bool, error) {
+	podName := getPodName(pod)
+	if bindings == nil {
+		return false, fmt.Errorf("failed to get cached bindings for pod %q", podName)
+	}
+	if claimsToProvision == nil {
+		return false, fmt.Errorf("failed to get cached claims to provision for pod %q", podName)
+	}
+
+	for _, binding := range bindings {
+		// Check for any conditions that might require scheduling retry
+
+		// Check if pv still exists
+		pv, err := b.pvCache.GetPV(binding.pv.Name)
+		if err != nil || pv == nil {
+			return false, fmt.Errorf("failed to check pv binding: %v", err)
+		}
+
+		// Check if pv.ClaimRef got dropped by unbindVolume()
+		if pv.Spec.ClaimRef == nil || pv.Spec.ClaimRef.UID == "" {
+			return false, fmt.Errorf("ClaimRef got reset for pv %q", pv.Name)
+		}
+
+		// Check if pvc is fully bound
+		if isBound, _, err := b.isPVCBound(binding.pvc.Namespace, binding.pvc.Name); !isBound || err != nil {
+			return false, err
+		}
+
+		// TODO; what if pvc is bound to the wrong pv? It means our assume cache should be reverted.
+		// Or will pv controller cleanup the pv.ClaimRef?
+	}
+
+	for _, claim := range claimsToProvision {
+		bound, pvc, err := b.isPVCBound(claim.Namespace, claim.Name)
+		if err != nil || pvc == nil {
+			return false, fmt.Errorf("failed to check pvc binding: %v", err)
+		}
+
+		// Check if selectedNode annotation is still set
+		if pvc.Annotations == nil {
+			return false, fmt.Errorf("selectedNode annotation reset for PVC %q", pvc.Name)
+		}
+		selectedNode := pvc.Annotations[annSelectedNode]
+		if selectedNode != pod.Spec.NodeName {
+			return false, fmt.Errorf("selectedNode annotation value %q not set to scheduled node %q", selectedNode, pod.Spec.NodeName)
+		}
+
+		if !bound {
+			return false, nil
+		}
+	}
+
+	// All pvs and pvcs that we operated on are bound
+	glog.V(4).Infof("All PVCs for pod %q are bound", podName)
+	return true, nil
+}
+
+func (b *volumeBinder) isVolumeBound(namespace string, vol *v1.Volume) (bool, *v1.PersistentVolumeClaim, error) {
 	if vol.PersistentVolumeClaim == nil {
 		return true, nil, nil
 	}
 
 	pvcName := vol.PersistentVolumeClaim.ClaimName
+	return b.isPVCBound(namespace, pvcName)
+}
+
+func (b *volumeBinder) isPVCBound(namespace, pvcName string) (bool, *v1.PersistentVolumeClaim, error) {
 	claim := &v1.PersistentVolumeClaim{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      pvcName,
@@ -322,7 +431,6 @@ func (b *volumeBinder) isVolumeBound(namespace string, vol *v1.Volume, checkFull
 
 	pvName := pvc.Spec.VolumeName
 	if pvName != "" {
-		if checkFullyBound {
 		if metav1.HasAnnotation(pvc.ObjectMeta, annBindCompleted) {
 			glog.V(5).Infof("PVC %q is fully bound to PV %q", getPVCName(pvc), pvName)
 			return true, pvc, nil
@@ -331,9 +439,6 @@ func (b *volumeBinder) isVolumeBound(namespace string, vol *v1.Volume, checkFull
 			return false, pvc, nil
 		}
 	}
-		glog.V(5).Infof("PVC %q is bound or prebound to PV %q", getPVCName(pvc), pvName)
-		return true, pvc, nil
-	}
 
 	glog.V(5).Infof("PVC %q is not bound", getPVCName(pvc))
 	return false, pvc, nil
@@ -342,7 +447,7 @@ func (b *volumeBinder) isVolumeBound(namespace string, vol *v1.Volume, checkFull
 // arePodVolumesBound returns true if all volumes are fully bound
 func (b *volumeBinder) arePodVolumesBound(pod *v1.Pod) bool {
 	for _, vol := range pod.Spec.Volumes {
-		if isBound, _, _ := b.isVolumeBound(pod.Namespace, &vol, true); !isBound {
+		if isBound, _, _ := b.isVolumeBound(pod.Namespace, &vol); !isBound {
 			// Pod has at least one PVC that needs binding
 			return false
 		}
@@ -358,7 +463,7 @@ func (b *volumeBinder) getPodVolumes(pod *v1.Pod) (boundClaims []*v1.PersistentV
 	unboundClaims = []*bindingInfo{}
 
 	for _, vol := range pod.Spec.Volumes {
-		volumeBound, pvc, err := b.isVolumeBound(pod.Namespace, &vol, false)
+		volumeBound, pvc, err := b.isVolumeBound(pod.Namespace, &vol)
 		if err != nil {
 			return nil, nil, nil, err
 		}
@@ -372,7 +477,8 @@ func (b *volumeBinder) getPodVolumes(pod *v1.Pod) (boundClaims []*v1.PersistentV
 			if err != nil {
 				return nil, nil, nil, err
 			}
-			if delayBinding {
+			// Prebound PVCs are treated as unbound immediate binding
+			if delayBinding && pvc.Spec.VolumeName == "" {
 				// Scheduler path
 				unboundClaims = append(unboundClaims, &bindingInfo{pvc: pvc})
 			} else {
@@ -518,19 +624,6 @@ type bindingInfo struct {
 	pv *v1.PersistentVolume
 }
 
-// Used in unit test errors
-func (b bindingInfo) String() string {
-	pvcName := ""
-	pvName := ""
-	if b.pvc != nil {
-		pvcName = getPVCName(b.pvc)
-	}
-	if b.pv != nil {
-		pvName = b.pv.Name
-	}
-	return fmt.Sprintf("[PVC %q, PV %q]", pvcName, pvName)
-}
-
 type byPVCSize []*bindingInfo
 
 func (a byPVCSize) Len() int {

pkg/controller/volume/persistentvolume/scheduler_binder_cache.go

@@ -31,6 +31,8 @@ type PodBindingCache interface {
 	UpdateBindings(pod *v1.Pod, node string, bindings []*bindingInfo)
 
 	// GetBindings will return the cached bindings for the given pod and node.
+	// A nil return value means that the entry was not found. An empty slice
+	// means that no binding operations are needed.
 	GetBindings(pod *v1.Pod, node string) []*bindingInfo
 
 	// UpdateProvisionedPVCs will update the cache with the given provisioning decisions
@@ -38,6 +40,8 @@ type PodBindingCache interface {
 	UpdateProvisionedPVCs(pod *v1.Pod, node string, provisionings []*v1.PersistentVolumeClaim)
 
 	// GetProvisionedPVCs will return the cached provisioning decisions for the given pod and node.
+	// A nil return value means that the entry was not found. An empty slice
+	// means that no provisioning operations are needed.
 	GetProvisionedPVCs(pod *v1.Pod, node string) []*v1.PersistentVolumeClaim
 
 	// DeleteBindings will remove all cached bindings and provisionings for the given pod.

pkg/controller/volume/persistentvolume/scheduler_binder_fake.go

@@ -16,16 +16,13 @@ limitations under the License.
 
 package persistentvolume
 
-import (
+import "k8s.io/api/core/v1"
-	"k8s.io/api/core/v1"
-)
 
 type FakeVolumeBinderConfig struct {
 	AllBound             bool
 	FindUnboundSatsified bool
 	FindBoundSatsified   bool
 	FindErr              error
-	AssumeBindingRequired bool
 	AssumeErr            error
 	BindErr              error
 }
@@ -48,9 +45,9 @@ func (b *FakeVolumeBinder) FindPodVolumes(pod *v1.Pod, node *v1.Node) (unboundVo
 	return b.config.FindUnboundSatsified, b.config.FindBoundSatsified, b.config.FindErr
 }
 
-func (b *FakeVolumeBinder) AssumePodVolumes(assumedPod *v1.Pod, nodeName string) (bool, bool, error) {
+func (b *FakeVolumeBinder) AssumePodVolumes(assumedPod *v1.Pod, nodeName string) (bool, error) {
 	b.AssumeCalled = true
-	return b.config.AllBound, b.config.AssumeBindingRequired, b.config.AssumeErr
+	return b.config.AllBound, b.config.AssumeErr
 }
 
 func (b *FakeVolumeBinder) BindPodVolumes(assumedPod *v1.Pod) error {

pkg/controller/volume/persistentvolume/scheduler_binder_test.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"reflect"
 	"testing"
+	"time"
 
 	"github.com/golang/glog"
 
@@ -38,20 +39,30 @@ import (
 )
 
 var (
-	unboundPVC                  = makeTestPVC("unbound-pvc", "1G", pvcUnbound, "", "1", &waitClass)
+	// PVCs for manual binding
-	unboundPVC2                 = makeTestPVC("unbound-pvc2", "5G", pvcUnbound, "", "1", &waitClass)
+	// TODO: clean up all of these
-	preboundPVC                 = makeTestPVC("prebound-pvc", "1G", pvcPrebound, "pv-node1a", "1", &waitClass)
+	unboundPVC          = makeTestPVC("unbound-pvc", "1G", "", pvcUnbound, "", "1", &waitClass)
-	boundPVC                    = makeTestPVC("bound-pvc", "1G", pvcBound, "pv-bound", "1", &waitClass)
+	unboundPVC2         = makeTestPVC("unbound-pvc2", "5G", "", pvcUnbound, "", "1", &waitClass)
-	boundPVC2                   = makeTestPVC("bound-pvc2", "1G", pvcBound, "pv-bound2", "1", &waitClass)
+	preboundPVC         = makeTestPVC("prebound-pvc", "1G", "", pvcPrebound, "pv-node1a", "1", &waitClass)
+	preboundPVCNode1a   = makeTestPVC("unbound-pvc", "1G", "", pvcPrebound, "pv-node1a", "1", &waitClass)
+	boundPVC            = makeTestPVC("bound-pvc", "1G", "", pvcBound, "pv-bound", "1", &waitClass)
+	boundPVC2           = makeTestPVC("bound-pvc2", "1G", "", pvcBound, "pv-bound2", "1", &waitClass)
+	boundPVCNode1a      = makeTestPVC("unbound-pvc", "1G", "", pvcBound, "pv-node1a", "1", &waitClass)
 	badPVC              = makeBadPVC()
-	immediateUnboundPVC         = makeTestPVC("immediate-unbound-pvc", "1G", pvcUnbound, "", "1", &immediateClass)
+	immediateUnboundPVC = makeTestPVC("immediate-unbound-pvc", "1G", "", pvcUnbound, "", "1", &immediateClass)
-	immediateBoundPVC           = makeTestPVC("immediate-bound-pvc", "1G", pvcBound, "pv-bound-immediate", "1", &immediateClass)
+	immediateBoundPVC   = makeTestPVC("immediate-bound-pvc", "1G", "", pvcBound, "pv-bound-immediate", "1", &immediateClass)
-	provisionedPVC              = makeTestPVC("provisioned-pvc", "1Gi", pvcUnbound, "", "1", &waitClassWithProvisioner)
-	provisionedPVC2             = makeTestPVC("provisioned-pvc2", "1Gi", pvcUnbound, "", "1", &waitClassWithProvisioner)
-	provisionedPVCHigherVersion = makeTestPVC("provisioned-pvc2", "1Gi", pvcUnbound, "", "2", &waitClassWithProvisioner)
-	noProvisionerPVC            = makeTestPVC("no-provisioner-pvc", "1Gi", pvcUnbound, "", "1", &waitClass)
-	topoMismatchPVC             = makeTestPVC("topo-mismatch-pvc", "1Gi", pvcUnbound, "", "1", &topoMismatchClass)
 
+	// PVCs for dynamic provisioning
+	provisionedPVC              = makeTestPVC("provisioned-pvc", "1Gi", "", pvcUnbound, "", "1", &waitClassWithProvisioner)
+	provisionedPVC2             = makeTestPVC("provisioned-pvc2", "1Gi", "", pvcUnbound, "", "1", &waitClassWithProvisioner)
+	provisionedPVCHigherVersion = makeTestPVC("provisioned-pvc2", "1Gi", "", pvcUnbound, "", "2", &waitClassWithProvisioner)
+	provisionedPVCBound         = makeTestPVC("provisioned-pvc", "1Gi", "", pvcBound, "some-pv", "1", &waitClassWithProvisioner)
+	noProvisionerPVC            = makeTestPVC("no-provisioner-pvc", "1Gi", "", pvcUnbound, "", "1", &waitClass)
+	topoMismatchPVC             = makeTestPVC("topo-mismatch-pvc", "1Gi", "", pvcUnbound, "", "1", &topoMismatchClass)
+
+	selectedNodePVC = makeTestPVC("provisioned-pvc", "1Gi", nodeLabelValue, pvcSelectedNode, "", "1", &waitClassWithProvisioner)
+
+	// PVs for manual binding
 	pvNoNode                   = makeTestPV("pv-no-node", "", "1G", "1", nil, waitClass)
 	pvNode1a                   = makeTestPV("pv-node1a", "node1", "5G", "1", nil, waitClass)
 	pvNode1b                   = makeTestPV("pv-node1b", "node1", "10G", "1", nil, waitClass)
@@ -59,12 +70,13 @@ var (
 	pvNode2                    = makeTestPV("pv-node2", "node2", "1G", "1", nil, waitClass)
 	pvPrebound                 = makeTestPV("pv-prebound", "node1", "1G", "1", unboundPVC, waitClass)
 	pvBound                    = makeTestPV("pv-bound", "node1", "1G", "1", boundPVC, waitClass)
-	pvNode1aBound              = makeTestPV("pv-node1a", "node1", "1G", "1", unboundPVC, waitClass)
+	pvNode1aBound              = makeTestPV("pv-node1a", "node1", "5G", "1", unboundPVC, waitClass)
-	pvNode1bBound              = makeTestPV("pv-node1b", "node1", "5G", "1", unboundPVC2, waitClass)
+	pvNode1bBound              = makeTestPV("pv-node1b", "node1", "10G", "1", unboundPVC2, waitClass)
-	pvNode1bBoundHigherVersion = makeTestPV("pv-node1b", "node1", "5G", "2", unboundPVC2, waitClass)
+	pvNode1bBoundHigherVersion = makeTestPV("pv-node1b", "node1", "10G", "2", unboundPVC2, waitClass)
 	pvBoundImmediate           = makeTestPV("pv-bound-immediate", "node1", "1G", "1", immediateBoundPVC, immediateClass)
 	pvBoundImmediateNode2      = makeTestPV("pv-bound-immediate", "node2", "1G", "1", immediateBoundPVC, immediateClass)
 
+	// PVC/PV bindings for manual binding
 	binding1a      = makeBinding(unboundPVC, pvNode1a)
 	binding1b      = makeBinding(unboundPVC2, pvNode1b)
 	bindingNoNode  = makeBinding(unboundPVC, pvNoNode)
@@ -72,11 +84,13 @@ var (
 	binding1aBound = makeBinding(unboundPVC, pvNode1aBound)
 	binding1bBound = makeBinding(unboundPVC2, pvNode1bBound)
 
+	// storage class names
 	waitClass                = "waitClass"
 	immediateClass           = "immediateClass"
 	waitClassWithProvisioner = "waitClassWithProvisioner"
 	topoMismatchClass        = "topoMismatchClass"
 
+	// node topology
 	nodeLabelKey   = "nodeKey"
 	nodeLabelValue = "node1"
 )
@@ -102,7 +116,8 @@ func newTestBinder(t *testing.T) *testEnv {
 		client,
 		pvcInformer,
 		informerFactory.Core().V1().PersistentVolumes(),
-		classInformer)
+		classInformer,
+		10*time.Second)
 
 	// Add storageclasses
 	waitMode := storagev1.VolumeBindingWaitForFirstConsumer
@@ -247,17 +262,44 @@ func (env *testEnv) initPodCache(pod *v1.Pod, node string, bindings []*bindingIn
 func (env *testEnv) validatePodCache(t *testing.T, name, node string, pod *v1.Pod, expectedBindings []*bindingInfo, expectedProvisionings []*v1.PersistentVolumeClaim) {
 	cache := env.internalBinder.podBindingCache
 	bindings := cache.GetBindings(pod, node)
+	if aLen, eLen := len(bindings), len(expectedBindings); aLen != eLen {
+		t.Errorf("Test %q failed. expected %v bindings, got %v", name, eLen, aLen)
+	} else if expectedBindings == nil && bindings != nil {
+		// nil and empty are different
+		t.Errorf("Test %q failed. expected nil bindings, got empty", name)
+	} else if expectedBindings != nil && bindings == nil {
+		// nil and empty are different
+		t.Errorf("Test %q failed. expected empty bindings, got nil", name)
+	} else {
+		for i := 0; i < aLen; i++ {
+			// Validate PV
+			if !reflect.DeepEqual(expectedBindings[i].pv, bindings[i].pv) {
+				t.Errorf("Test %q failed. binding.pv doesn't match [A-expected, B-got]: %s", name, diff.ObjectDiff(expectedBindings[i].pv, bindings[i].pv))
+			}
 
-	if !reflect.DeepEqual(expectedBindings, bindings) {
+			// Validate PVC
-		t.Errorf("Test %q failed: Expected bindings %+v, got %+v", name, expectedBindings, bindings)
+			if !reflect.DeepEqual(expectedBindings[i].pvc, bindings[i].pvc) {
+				t.Errorf("Test %q failed. binding.pvc doesn't match [A-expected, B-got]: %s", name, diff.ObjectDiff(expectedBindings[i].pvc, bindings[i].pvc))
+			}
+		}
 	}
 
 	provisionedClaims := cache.GetProvisionedPVCs(pod, node)
-
+	if aLen, eLen := len(provisionedClaims), len(expectedProvisionings); aLen != eLen {
-	if !reflect.DeepEqual(expectedProvisionings, provisionedClaims) {
+		t.Errorf("Test %q failed. expected %v provisioned claims, got %v", name, eLen, aLen)
-		t.Errorf("Test %q failed: Expected provisionings %+v, got %+v", name, expectedProvisionings, provisionedClaims)
+	} else if expectedProvisionings == nil && provisionedClaims != nil {
+		// nil and empty are different
+		t.Errorf("Test %q failed. expected nil provisionings, got empty", name)
+	} else if expectedProvisionings != nil && provisionedClaims == nil {
+		// nil and empty are different
+		t.Errorf("Test %q failed. expected empty provisionings, got nil", name)
+	} else {
+		for i := 0; i < aLen; i++ {
+			if !reflect.DeepEqual(expectedProvisionings[i], provisionedClaims[i]) {
+				t.Errorf("Test %q failed. provisioned claims doesn't match [A-expected, B-got]: %s", name, diff.ObjectDiff(expectedProvisionings[i], provisionedClaims[i]))
+			}
+		}
 	}
-
 }
 
 func (env *testEnv) getPodBindings(t *testing.T, name, node string, pod *v1.Pod) []*bindingInfo {
@@ -266,8 +308,6 @@ func (env *testEnv) getPodBindings(t *testing.T, name, node string, pod *v1.Pod)
 }
 
 func (env *testEnv) validateAssume(t *testing.T, name string, pod *v1.Pod, bindings []*bindingInfo, provisionings []*v1.PersistentVolumeClaim) {
-	// TODO: Check binding cache
-
 	// Check pv cache
 	pvCache := env.internalBinder.pvCache
 	for _, b := range bindings {
@@ -383,17 +423,21 @@ const (
 	pvcUnbound = iota
 	pvcPrebound
 	pvcBound
+	pvcSelectedNode
 )
 
-func makeTestPVC(name, size string, pvcBoundState int, pvName, resourceVersion string, className *string) *v1.PersistentVolumeClaim {
+func makeTestPVC(name, size, node string, pvcBoundState int, pvName, resourceVersion string, className *string) *v1.PersistentVolumeClaim {
 	pvc := &v1.PersistentVolumeClaim{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "PersistentVolumeClaim",
+			APIVersion: "v1",
+		},
 		ObjectMeta: metav1.ObjectMeta{
 			Name:            name,
 			Namespace:       "testns",
 			UID:             types.UID("pvc-uid"),
 			ResourceVersion: resourceVersion,
 			SelfLink:        testapi.Default.SelfLink("pvc", name),
-			Annotations:     map[string]string{},
 		},
 		Spec: v1.PersistentVolumeClaimSpec{
 			Resources: v1.ResourceRequirements{
@@ -406,6 +450,9 @@ func makeTestPVC(name, size string, pvcBoundState int, pvName, resourceVersion s
 	}
 
 	switch pvcBoundState {
+	case pvcSelectedNode:
+		metav1.SetMetaDataAnnotation(&pvc.ObjectMeta, annSelectedNode, node)
+		// don't fallthrough
 	case pvcBound:
 		metav1.SetMetaDataAnnotation(&pvc.ObjectMeta, annBindCompleted, "yes")
 		fallthrough
@@ -454,6 +501,9 @@ func makeTestPV(name, node, capacity, version string, boundToPVC *v1.PersistentV
 
 	if boundToPVC != nil {
 		pv.Spec.ClaimRef = &v1.ObjectReference{
+			Kind:            boundToPVC.Kind,
+			APIVersion:      boundToPVC.APIVersion,
+			ResourceVersion: boundToPVC.ResourceVersion,
 			Name:            boundToPVC.Name,
 			Namespace:       boundToPVC.Namespace,
 			UID:             boundToPVC.UID,
@@ -464,6 +514,24 @@ func makeTestPV(name, node, capacity, version string, boundToPVC *v1.PersistentV
 	return pv
 }
 
+func pvcSetSelectedNode(pvc *v1.PersistentVolumeClaim, node string) *v1.PersistentVolumeClaim {
+	newPVC := pvc.DeepCopy()
+	metav1.SetMetaDataAnnotation(&pvc.ObjectMeta, annSelectedNode, node)
+	return newPVC
+}
+
+func pvcSetEmptyAnnotations(pvc *v1.PersistentVolumeClaim) *v1.PersistentVolumeClaim {
+	newPVC := pvc.DeepCopy()
+	newPVC.Annotations = map[string]string{}
+	return newPVC
+}
+
+func pvRemoveClaimUID(pv *v1.PersistentVolume) *v1.PersistentVolume {
+	newPV := pv.DeepCopy()
+	newPV.Spec.ClaimRef.UID = ""
+	return newPV
+}
+
 func makePod(pvcs []*v1.PersistentVolumeClaim) *v1.Pod {
 	pod := &v1.Pod{
 		ObjectMeta: metav1.ObjectMeta{
@@ -515,7 +583,7 @@ func makeBinding(pvc *v1.PersistentVolumeClaim, pv *v1.PersistentVolume) *bindin
 func addProvisionAnn(pvc *v1.PersistentVolumeClaim) *v1.PersistentVolumeClaim {
 	res := pvc.DeepCopy()
 	// Add provision related annotations
-	res.Annotations[annSelectedNode] = nodeLabelValue
+	metav1.SetMetaDataAnnotation(&res.ObjectMeta, annSelectedNode, nodeLabelValue)
 
 	return res
 }
@@ -570,8 +638,7 @@ func TestFindPodVolumesWithoutProvisioning(t *testing.T) {
 		"prebound-pvc": {
 			podPVCs:    []*v1.PersistentVolumeClaim{preboundPVC},
 			pvs:        []*v1.PersistentVolume{pvNode1aBound},
-			expectedUnbound: true,
+			shouldFail: true,
-			expectedBound:   true,
 		},
 		"unbound-pvc,pv-same-node": {
 			podPVCs:          []*v1.PersistentVolumeClaim{unboundPVC},
@@ -623,9 +690,7 @@ func TestFindPodVolumesWithoutProvisioning(t *testing.T) {
 		"one-prebound,one-unbound": {
 			podPVCs:    []*v1.PersistentVolumeClaim{unboundPVC, preboundPVC},
 			pvs:        []*v1.PersistentVolume{pvNode1a, pvNode1b},
-			expectedBindings: []*bindingInfo{binding1a},
+			shouldFail: true,
-			expectedUnbound:  true,
-			expectedBound:    true,
 		},
 		"immediate-bound-pvc": {
 			podPVCs:         []*v1.PersistentVolumeClaim{immediateBoundPVC},
@@ -835,60 +900,56 @@ func TestAssumePodVolumes(t *testing.T) {
 
 		// Expected return values
 		shouldFail       bool
-		expectedBindingRequired bool
 		expectedAllBound bool
 
-		// if nil, use bindings
 		expectedBindings      []*bindingInfo
+		expectedProvisionings []*v1.PersistentVolumeClaim
 	}{
 		"all-bound": {
 			podPVCs:          []*v1.PersistentVolumeClaim{boundPVC},
 			pvs:              []*v1.PersistentVolume{pvBound},
 			expectedAllBound: true,
 		},
-		"prebound-pvc": {
-			podPVCs: []*v1.PersistentVolumeClaim{preboundPVC},
-			pvs:     []*v1.PersistentVolume{pvNode1a},
-		},
 		"one-binding": {
 			podPVCs:               []*v1.PersistentVolumeClaim{unboundPVC},
 			bindings:              []*bindingInfo{binding1a},
 			pvs:                   []*v1.PersistentVolume{pvNode1a},
-			expectedBindingRequired: true,
+			expectedBindings:      []*bindingInfo{binding1aBound},
+			expectedProvisionings: []*v1.PersistentVolumeClaim{},
 		},
 		"two-bindings": {
 			podPVCs:               []*v1.PersistentVolumeClaim{unboundPVC, unboundPVC2},
 			bindings:              []*bindingInfo{binding1a, binding1b},
 			pvs:                   []*v1.PersistentVolume{pvNode1a, pvNode1b},
-			expectedBindingRequired: true,
+			expectedBindings:      []*bindingInfo{binding1aBound, binding1bBound},
+			expectedProvisionings: []*v1.PersistentVolumeClaim{},
 		},
 		"pv-already-bound": {
 			podPVCs:               []*v1.PersistentVolumeClaim{unboundPVC},
 			bindings:              []*bindingInfo{binding1aBound},
 			pvs:                   []*v1.PersistentVolume{pvNode1aBound},
-			expectedBindingRequired: false,
+			expectedBindings:      []*bindingInfo{binding1aBound},
-			expectedBindings:        []*bindingInfo{},
+			expectedProvisionings: []*v1.PersistentVolumeClaim{},
 		},
 		"claimref-failed": {
 			podPVCs:    []*v1.PersistentVolumeClaim{unboundPVC},
 			bindings:   []*bindingInfo{binding1a, bindingBad},
 			pvs:        []*v1.PersistentVolume{pvNode1a, pvNode1b},
 			shouldFail: true,
-			expectedBindingRequired: true,
 		},
 		"tmpupdate-failed": {
 			podPVCs:    []*v1.PersistentVolumeClaim{unboundPVC},
 			bindings:   []*bindingInfo{binding1a, binding1b},
 			pvs:        []*v1.PersistentVolume{pvNode1a},
 			shouldFail: true,
-			expectedBindingRequired: true,
 		},
 		"one-binding, one-pvc-provisioned": {
 			podPVCs:               []*v1.PersistentVolumeClaim{unboundPVC, provisionedPVC},
 			bindings:              []*bindingInfo{binding1a},
 			pvs:                   []*v1.PersistentVolume{pvNode1a},
 			provisionedPVCs:       []*v1.PersistentVolumeClaim{provisionedPVC},
-			expectedBindingRequired: true,
+			expectedBindings:      []*bindingInfo{binding1aBound},
+			expectedProvisionings: []*v1.PersistentVolumeClaim{selectedNodePVC},
 		},
 		"one-binding, one-provision-tmpupdate-failed": {
 			podPVCs:         []*v1.PersistentVolumeClaim{unboundPVC, provisionedPVCHigherVersion},
@@ -896,7 +957,6 @@ func TestAssumePodVolumes(t *testing.T) {
 			pvs:             []*v1.PersistentVolume{pvNode1a},
 			provisionedPVCs: []*v1.PersistentVolumeClaim{provisionedPVC2},
 			shouldFail:      true,
-			expectedBindingRequired: true,
 		},
 	}
 
@@ -911,7 +971,7 @@ func TestAssumePodVolumes(t *testing.T) {
 		testEnv.initVolumes(scenario.pvs, scenario.pvs)
 
 		// Execute
-		allBound, bindingRequired, err := testEnv.binder.AssumePodVolumes(pod, "node1")
+		allBound, err := testEnv.binder.AssumePodVolumes(pod, "node1")
 
 		// Validate
 		if !scenario.shouldFail && err != nil {
@@ -920,24 +980,25 @@ func TestAssumePodVolumes(t *testing.T) {
 		if scenario.shouldFail && err == nil {
 			t.Errorf("Test %q failed: returned success but expected error", name)
 		}
-		if scenario.expectedBindingRequired != bindingRequired {
-			t.Errorf("Test %q failed: returned unexpected bindingRequired: %v", name, bindingRequired)
-		}
 		if scenario.expectedAllBound != allBound {
 			t.Errorf("Test %q failed: returned unexpected allBound: %v", name, allBound)
 		}
 		if scenario.expectedBindings == nil {
 			scenario.expectedBindings = scenario.bindings
 		}
-		if scenario.shouldFail {
+		if scenario.expectedProvisionings == nil {
-			testEnv.validateFailedAssume(t, name, pod, scenario.expectedBindings, scenario.provisionedPVCs)
+			scenario.expectedProvisionings = scenario.provisionedPVCs
-		} else {
-			testEnv.validateAssume(t, name, pod, scenario.expectedBindings, scenario.provisionedPVCs)
 		}
+		if scenario.shouldFail {
+			testEnv.validateFailedAssume(t, name, pod, scenario.expectedBindings, scenario.expectedProvisionings)
+		} else {
+			testEnv.validateAssume(t, name, pod, scenario.expectedBindings, scenario.expectedProvisionings)
+		}
+		testEnv.validatePodCache(t, name, pod.Spec.NodeName, pod, scenario.expectedBindings, scenario.expectedProvisionings)
 	}
 }
 
-func TestBindPodVolumes(t *testing.T) {
+func TestBindAPIUpdate(t *testing.T) {
 	scenarios := map[string]struct {
 		// Inputs
 		bindings  []*bindingInfo
@@ -960,19 +1021,38 @@ func TestBindPodVolumes(t *testing.T) {
 		// if nil, use expectedPVCs
 		expectedAPIPVCs []*v1.PersistentVolumeClaim
 	}{
-		"all-bound": {},
+		"nothing-to-bind-nil": {
-		"not-fully-bound": {
+			shouldFail: true,
+		},
+		"nothing-to-bind-bindings-nil": {
+			provisionedPVCs: []*v1.PersistentVolumeClaim{},
+			shouldFail:      true,
+		},
+		"nothing-to-bind-provisionings-nil": {
 			bindings:   []*bindingInfo{},
+			shouldFail: true,
+		},
+		"nothing-to-bind-empty": {
+			bindings:        []*bindingInfo{},
+			provisionedPVCs: []*v1.PersistentVolumeClaim{},
 		},
 		"one-binding": {
 			bindings:        []*bindingInfo{binding1aBound},
 			cachedPVs:       []*v1.PersistentVolume{pvNode1a},
 			expectedPVs:     []*v1.PersistentVolume{pvNode1aBound},
+			provisionedPVCs: []*v1.PersistentVolumeClaim{},
 		},
 		"two-bindings": {
 			bindings:        []*bindingInfo{binding1aBound, binding1bBound},
 			cachedPVs:       []*v1.PersistentVolume{pvNode1a, pvNode1b},
 			expectedPVs:     []*v1.PersistentVolume{pvNode1aBound, pvNode1bBound},
+			provisionedPVCs: []*v1.PersistentVolumeClaim{},
+		},
+		"api-already-updated": {
+			bindings:        []*bindingInfo{binding1aBound},
+			cachedPVs:       []*v1.PersistentVolume{pvNode1aBound},
+			expectedPVs:     []*v1.PersistentVolume{pvNode1aBound},
+			provisionedPVCs: []*v1.PersistentVolumeClaim{},
 		},
 		"api-update-failed": {
 			bindings:        []*bindingInfo{binding1aBound, binding1bBound},
@@ -980,14 +1060,17 @@ func TestBindPodVolumes(t *testing.T) {
 			apiPVs:          []*v1.PersistentVolume{pvNode1a, pvNode1bBoundHigherVersion},
 			expectedPVs:     []*v1.PersistentVolume{pvNode1aBound, pvNode1b},
 			expectedAPIPVs:  []*v1.PersistentVolume{pvNode1aBound, pvNode1bBoundHigherVersion},
+			provisionedPVCs: []*v1.PersistentVolumeClaim{},
 			shouldFail:      true,
 		},
 		"one-provisioned-pvc": {
+			bindings:        []*bindingInfo{},
 			provisionedPVCs: []*v1.PersistentVolumeClaim{addProvisionAnn(provisionedPVC)},
 			cachedPVCs:      []*v1.PersistentVolumeClaim{provisionedPVC},
 			expectedPVCs:    []*v1.PersistentVolumeClaim{addProvisionAnn(provisionedPVC)},
 		},
 		"provision-api-update-failed": {
+			bindings:        []*bindingInfo{},
 			provisionedPVCs: []*v1.PersistentVolumeClaim{addProvisionAnn(provisionedPVC), addProvisionAnn(provisionedPVC2)},
 			cachedPVCs:      []*v1.PersistentVolumeClaim{provisionedPVC, provisionedPVC2},
 			apiPVCs:         []*v1.PersistentVolumeClaim{provisionedPVC, provisionedPVCHigherVersion},
@@ -995,7 +1078,7 @@ func TestBindPodVolumes(t *testing.T) {
 			expectedAPIPVCs: []*v1.PersistentVolumeClaim{addProvisionAnn(provisionedPVC), provisionedPVCHigherVersion},
 			shouldFail:      true,
 		},
-		"bingding-succeed, provision-api-update-failed": {
+		"binding-succeed, provision-api-update-failed": {
 			bindings:        []*bindingInfo{binding1aBound},
 			cachedPVs:       []*v1.PersistentVolume{pvNode1a},
 			expectedPVs:     []*v1.PersistentVolume{pvNode1aBound},
@@ -1008,7 +1091,7 @@ func TestBindPodVolumes(t *testing.T) {
 		},
 	}
 	for name, scenario := range scenarios {
-		glog.V(5).Infof("Running test case %q", name)
+		glog.V(4).Infof("Running test case %q", name)
 
 		// Setup
 		testEnv := newTestBinder(t)
@@ -1024,7 +1107,7 @@ func TestBindPodVolumes(t *testing.T) {
 		testEnv.assumeVolumes(t, name, "node1", pod, scenario.bindings, scenario.provisionedPVCs)
 
 		// Execute
-		err := testEnv.binder.BindPodVolumes(pod)
+		err := testEnv.internalBinder.bindAPIUpdate(pod.Name, scenario.bindings, scenario.provisionedPVCs)
 
 		// Validate
 		if !scenario.shouldFail && err != nil {
@@ -1044,6 +1127,301 @@ func TestBindPodVolumes(t *testing.T) {
 	}
 }
 
+func TestCheckBindings(t *testing.T) {
+	scenarios := map[string]struct {
+		// Inputs
+		bindings  []*bindingInfo
+		cachedPVs []*v1.PersistentVolume
+
+		provisionedPVCs []*v1.PersistentVolumeClaim
+		cachedPVCs      []*v1.PersistentVolumeClaim
+
+		// Expected return values
+		shouldFail    bool
+		expectedBound bool
+	}{
+		"nothing-to-bind-nil": {
+			shouldFail: true,
+		},
+		"nothing-to-bind-bindings-nil": {
+			provisionedPVCs: []*v1.PersistentVolumeClaim{},
+			shouldFail:      true,
+		},
+		"nothing-to-bind-provisionings-nil": {
+			bindings:   []*bindingInfo{},
+			shouldFail: true,
+		},
+		"nothing-to-bind": {
+			bindings:        []*bindingInfo{},
+			provisionedPVCs: []*v1.PersistentVolumeClaim{},
+			expectedBound:   true,
+		},
+		"binding-bound": {
+			bindings:        []*bindingInfo{binding1aBound},
+			provisionedPVCs: []*v1.PersistentVolumeClaim{},
+			cachedPVs:       []*v1.PersistentVolume{pvNode1aBound},
+			cachedPVCs:      []*v1.PersistentVolumeClaim{boundPVCNode1a},
+			expectedBound:   true,
+		},
+		"binding-prebound": {
+			bindings:        []*bindingInfo{binding1aBound},
+			provisionedPVCs: []*v1.PersistentVolumeClaim{},
+			cachedPVs:       []*v1.PersistentVolume{pvNode1aBound},
+			cachedPVCs:      []*v1.PersistentVolumeClaim{preboundPVCNode1a},
+		},
+		"binding-unbound": {
+			bindings:        []*bindingInfo{binding1aBound},
+			provisionedPVCs: []*v1.PersistentVolumeClaim{},
+			cachedPVs:       []*v1.PersistentVolume{pvNode1aBound},
+			cachedPVCs:      []*v1.PersistentVolumeClaim{unboundPVC},
+		},
+		"binding-pvc-not-exists": {
+			bindings:        []*bindingInfo{binding1aBound},
+			provisionedPVCs: []*v1.PersistentVolumeClaim{},
+			cachedPVs:       []*v1.PersistentVolume{pvNode1aBound},
+			shouldFail:      true,
+		},
+		"binding-pv-not-exists": {
+			bindings:        []*bindingInfo{binding1aBound},
+			provisionedPVCs: []*v1.PersistentVolumeClaim{},
+			cachedPVCs:      []*v1.PersistentVolumeClaim{boundPVCNode1a},
+			shouldFail:      true,
+		},
+		"binding-claimref-nil": {
+			bindings:        []*bindingInfo{binding1aBound},
+			provisionedPVCs: []*v1.PersistentVolumeClaim{},
+			cachedPVs:       []*v1.PersistentVolume{pvNode1a},
+			cachedPVCs:      []*v1.PersistentVolumeClaim{boundPVCNode1a},
+			shouldFail:      true,
+		},
+		"binding-claimref-uid-empty": {
+			bindings:        []*bindingInfo{binding1aBound},
+			provisionedPVCs: []*v1.PersistentVolumeClaim{},
+			cachedPVs:       []*v1.PersistentVolume{pvRemoveClaimUID(pvNode1aBound)},
+			cachedPVCs:      []*v1.PersistentVolumeClaim{boundPVCNode1a},
+			shouldFail:      true,
+		},
+		"binding-one-bound,one-unbound": {
+			bindings:        []*bindingInfo{binding1aBound, binding1bBound},
+			provisionedPVCs: []*v1.PersistentVolumeClaim{},
+			cachedPVs:       []*v1.PersistentVolume{pvNode1aBound, pvNode1bBound},
+			cachedPVCs:      []*v1.PersistentVolumeClaim{boundPVCNode1a, unboundPVC2},
+		},
+		"provisioning-pvc-bound": {
+			bindings:        []*bindingInfo{},
+			provisionedPVCs: []*v1.PersistentVolumeClaim{addProvisionAnn(provisionedPVC)},
+			cachedPVCs:      []*v1.PersistentVolumeClaim{addProvisionAnn(provisionedPVCBound)},
+			expectedBound:   true,
+		},
+		"provisioning-pvc-unbound": {
+			bindings:        []*bindingInfo{},
+			provisionedPVCs: []*v1.PersistentVolumeClaim{addProvisionAnn(provisionedPVC)},
+			cachedPVCs:      []*v1.PersistentVolumeClaim{addProvisionAnn(provisionedPVC)},
+		},
+		"provisioning-pvc-not-exists": {
+			bindings:        []*bindingInfo{},
+			provisionedPVCs: []*v1.PersistentVolumeClaim{addProvisionAnn(provisionedPVC)},
+			shouldFail:      true,
+		},
+		"provisioning-pvc-annotations-nil": {
+			bindings:        []*bindingInfo{},
+			provisionedPVCs: []*v1.PersistentVolumeClaim{addProvisionAnn(provisionedPVC)},
+			cachedPVCs:      []*v1.PersistentVolumeClaim{provisionedPVC},
+			shouldFail:      true,
+		},
+		"provisioning-pvc-selected-node-dropped": {
+			bindings:        []*bindingInfo{},
+			provisionedPVCs: []*v1.PersistentVolumeClaim{addProvisionAnn(provisionedPVC)},
+			cachedPVCs:      []*v1.PersistentVolumeClaim{pvcSetEmptyAnnotations(provisionedPVC)},
+			shouldFail:      true,
+		},
+		"provisioning-pvc-selected-node-wrong-node": {
+			bindings:        []*bindingInfo{},
+			provisionedPVCs: []*v1.PersistentVolumeClaim{addProvisionAnn(provisionedPVC)},
+			cachedPVCs:      []*v1.PersistentVolumeClaim{pvcSetSelectedNode(provisionedPVC, "wrong-node")},
+			shouldFail:      true,
+		},
+		"binding-bound-provisioning-unbound": {
+			bindings:        []*bindingInfo{binding1aBound},
+			provisionedPVCs: []*v1.PersistentVolumeClaim{addProvisionAnn(provisionedPVC)},
+			cachedPVs:       []*v1.PersistentVolume{pvNode1aBound},
+			cachedPVCs:      []*v1.PersistentVolumeClaim{boundPVCNode1a, addProvisionAnn(provisionedPVC)},
+		},
+	}
+
+	for name, scenario := range scenarios {
+		glog.V(4).Infof("Running test case %q", name)
+
+		// Setup
+		pod := makePod(nil)
+		testEnv := newTestBinder(t)
+		testEnv.initVolumes(scenario.cachedPVs, nil)
+		testEnv.initClaims(scenario.cachedPVCs, nil)
+
+		// Execute
+		allBound, err := testEnv.internalBinder.checkBindings(pod, scenario.bindings, scenario.provisionedPVCs)
+
+		// Validate
+		if !scenario.shouldFail && err != nil {
+			t.Errorf("Test %q failed: returned error: %v", name, err)
+		}
+		if scenario.shouldFail && err == nil {
+			t.Errorf("Test %q failed: returned success but expected error", name)
+		}
+		if scenario.expectedBound != allBound {
+			t.Errorf("Test %q failed: returned bound %v", name, allBound)
+		}
+	}
+}
+
+func TestBindPodVolumes(t *testing.T) {
+	scenarios := map[string]struct {
+		// Inputs
+		// These tests only support a single pv and pvc and static binding
+		bindingsNil bool // Pass in nil bindings slice
+		binding     *bindingInfo
+		cachedPV    *v1.PersistentVolume
+		cachedPVC   *v1.PersistentVolumeClaim
+		apiPV       *v1.PersistentVolume
+
+		// This function runs with a delay of 5 seconds
+		delayFunc func(*testing.T, *testEnv, *v1.Pod, *v1.PersistentVolume, *v1.PersistentVolumeClaim)
+
+		// Expected return values
+		shouldFail bool
+	}{
+		"nothing-to-bind-nil": {
+			bindingsNil: true,
+			shouldFail:  true,
+		},
+		"nothing-to-bind-empty": {},
+		"already-bound": {
+			binding:   binding1aBound,
+			cachedPV:  pvNode1aBound,
+			cachedPVC: boundPVCNode1a,
+		},
+		"binding-succeeds-after-time": {
+			binding:   binding1aBound,
+			cachedPV:  pvNode1a,
+			cachedPVC: unboundPVC,
+			delayFunc: func(t *testing.T, testEnv *testEnv, pod *v1.Pod, pv *v1.PersistentVolume, pvc *v1.PersistentVolumeClaim) {
+				// Update PVC to be fully bound to PV
+				newPVC := pvc.DeepCopy()
+				newPVC.ResourceVersion = "100"
+				newPVC.Spec.VolumeName = pv.Name
+				metav1.SetMetaDataAnnotation(&newPVC.ObjectMeta, annBindCompleted, "yes")
+
+				// Update pvc cache, fake client doesn't invoke informers
+				internalBinder, ok := testEnv.binder.(*volumeBinder)
+				if !ok {
+					t.Fatalf("Failed to convert to internal binder")
+				}
+
+				pvcCache := internalBinder.pvcCache
+				internalPVCCache, ok := pvcCache.(*pvcAssumeCache)
+				if !ok {
+					t.Fatalf("Failed to convert to internal PVC cache")
+				}
+				internalPVCCache.add(newPVC)
+			},
+		},
+		"pod-deleted-after-time": {
+			binding:   binding1aBound,
+			cachedPV:  pvNode1a,
+			cachedPVC: unboundPVC,
+			delayFunc: func(t *testing.T, testEnv *testEnv, pod *v1.Pod, pv *v1.PersistentVolume, pvc *v1.PersistentVolumeClaim) {
+				bindingsCache := testEnv.binder.GetBindingsCache()
+				if bindingsCache == nil {
+					t.Fatalf("Failed to get bindings cache")
+				}
+
+				// Delete the pod from the cache
+				bindingsCache.DeleteBindings(pod)
+
+				// Check that it's deleted
+				bindings := bindingsCache.GetBindings(pod, "node1")
+				if bindings != nil {
+					t.Fatalf("Failed to delete bindings")
+				}
+			},
+			shouldFail: true,
+		},
+		"binding-times-out": {
+			binding:    binding1aBound,
+			cachedPV:   pvNode1a,
+			cachedPVC:  unboundPVC,
+			shouldFail: true,
+		},
+		"binding-fails": {
+			binding:    binding1bBound,
+			cachedPV:   pvNode1b,
+			apiPV:      pvNode1bBoundHigherVersion,
+			cachedPVC:  unboundPVC2,
+			shouldFail: true,
+		},
+		"check-fails": {
+			binding:   binding1aBound,
+			cachedPV:  pvNode1a,
+			cachedPVC: unboundPVC,
+			delayFunc: func(t *testing.T, testEnv *testEnv, pod *v1.Pod, pv *v1.PersistentVolume, pvc *v1.PersistentVolumeClaim) {
+				// Delete PVC
+				// Update pvc cache, fake client doesn't invoke informers
+				internalBinder, ok := testEnv.binder.(*volumeBinder)
+				if !ok {
+					t.Fatalf("Failed to convert to internal binder")
+				}
+
+				pvcCache := internalBinder.pvcCache
+				internalPVCCache, ok := pvcCache.(*pvcAssumeCache)
+				if !ok {
+					t.Fatalf("Failed to convert to internal PVC cache")
+				}
+				internalPVCCache.delete(pvc)
+			},
+			shouldFail: true,
+		},
+	}
+
+	for name, scenario := range scenarios {
+		glog.V(4).Infof("Running test case %q", name)
+
+		// Setup
+		pod := makePod(nil)
+		if scenario.apiPV == nil {
+			scenario.apiPV = scenario.cachedPV
+		}
+		testEnv := newTestBinder(t)
+		if !scenario.bindingsNil {
+			if scenario.binding != nil {
+				testEnv.initVolumes([]*v1.PersistentVolume{scenario.cachedPV}, []*v1.PersistentVolume{scenario.apiPV})
+				testEnv.initClaims([]*v1.PersistentVolumeClaim{scenario.cachedPVC}, nil)
+				testEnv.assumeVolumes(t, name, "node1", pod, []*bindingInfo{scenario.binding}, []*v1.PersistentVolumeClaim{})
+			} else {
+				testEnv.assumeVolumes(t, name, "node1", pod, []*bindingInfo{}, []*v1.PersistentVolumeClaim{})
+			}
+		}
+
+		if scenario.delayFunc != nil {
+			go func() {
+				time.Sleep(5 * time.Second)
+				glog.V(5).Infof("Running delay function")
+				scenario.delayFunc(t, testEnv, pod, scenario.binding.pv, scenario.binding.pvc)
+			}()
+		}
+
+		// Execute
+		err := testEnv.binder.BindPodVolumes(pod)
+
+		// Validate
+		if !scenario.shouldFail && err != nil {
+			t.Errorf("Test %q failed: returned error: %v", name, err)
+		}
+		if scenario.shouldFail && err == nil {
+			t.Errorf("Test %q failed: returned success but expected error", name)
+		}
+	}
+}
+
 func TestFindAssumeVolumes(t *testing.T) {
 	// Set feature gate
 	utilfeature.DefaultFeatureGate.Set("VolumeScheduling=true")
@@ -1080,17 +1458,15 @@ func TestFindAssumeVolumes(t *testing.T) {
 	expectedBindings := testEnv.getPodBindings(t, "before-assume", testNode.Name, pod)
 
 	// 2. Assume matches
-	allBound, bindingRequired, err := testEnv.binder.AssumePodVolumes(pod, testNode.Name)
+	allBound, err := testEnv.binder.AssumePodVolumes(pod, testNode.Name)
 	if err != nil {
 		t.Errorf("Test failed: AssumePodVolumes returned error: %v", err)
 	}
 	if allBound {
 		t.Errorf("Test failed: detected unbound volumes as bound")
 	}
-	if !bindingRequired {
-		t.Errorf("Test failed: binding not required")
-	}
 	testEnv.validateAssume(t, "assume", pod, expectedBindings, nil)
+
 	// After assume, claimref should be set on pv
 	expectedBindings = testEnv.getPodBindings(t, "after-assume", testNode.Name, pod)
 
@@ -1106,6 +1482,6 @@ func TestFindAssumeVolumes(t *testing.T) {
 		if !unboundSatisfied {
 			t.Errorf("Test failed: couldn't find PVs for all PVCs")
 		}
-		testEnv.validatePodCache(t, "after-assume", testNode.Name, pod, expectedBindings, nil)
+		testEnv.validatePodCache(t, "after-assume", testNode.Name, pod, expectedBindings, []*v1.PersistentVolumeClaim{})
 	}
 }