Make volume binder resilient to races

pkg/controller/volume/persistentvolume/BUILD

@@ -44,6 +44,7 @@ go_library(
         "//staging/src/k8s.io/apimachinery/pkg/util/runtime:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/sets:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/wait:go_default_library",
+        "//staging/src/k8s.io/apiserver/pkg/storage/etcd:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/util/feature:go_default_library",
         "//staging/src/k8s.io/client-go/informers/core/v1:go_default_library",
         "//staging/src/k8s.io/client-go/informers/storage/v1:go_default_library",
@@ -92,6 +93,7 @@ go_test(
         "//staging/src/k8s.io/apimachinery/pkg/api/resource:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/runtime:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/diff:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/wait:go_default_library",

pkg/controller/volume/persistentvolume/pv_controller.go

@@ -285,15 +285,16 @@ func checkVolumeSatisfyClaim(volume *v1.PersistentVolume, claim *v1.PersistentVo
 	return nil
 }
 
-func (ctrl *PersistentVolumeController) shouldDelayBinding(claim *v1.PersistentVolumeClaim) (bool, error) {
+func (ctrl *PersistentVolumeController) isDelayBindingProvisioning(claim *v1.PersistentVolumeClaim) bool {
 	// When feature VolumeScheduling enabled,
 	// Scheduler signal to the PV controller to start dynamic
 	// provisioning by setting the "annSelectedNode" annotation
 	// in the PVC
-	if _, ok := claim.Annotations[annSelectedNode]; ok {
-		return false, nil
-	}
+	_, ok := claim.Annotations[annSelectedNode]
+	return ok
+}
 
+func (ctrl *PersistentVolumeController) isDelayBindingMode(claim *v1.PersistentVolumeClaim) (bool, error) {
 	className := v1helper.GetPersistentVolumeClaimClass(claim)
 	if className == "" {
 		return false, nil
@@ -311,6 +312,18 @@ func (ctrl *PersistentVolumeController) shouldDelayBinding(claim *v1.PersistentV
 	return *class.VolumeBindingMode == storage.VolumeBindingWaitForFirstConsumer, nil
 }
 
+// shouldDelayBinding returns true if binding of claim should be delayed, false otherwise.
+// If binding of claim should be delayed, only claims pbound by scheduler
+func (ctrl *PersistentVolumeController) shouldDelayBinding(claim *v1.PersistentVolumeClaim) (bool, error) {
+	// If claim has already been assigned a node by scheduler for dynamic provisioning.
+	if ctrl.isDelayBindingProvisioning(claim) {
+		return false, nil
+	}
+
+	// If claim is in delay binding mode.
+	return ctrl.isDelayBindingMode(claim)
+}
+
 // syncUnboundClaim is the main controller method to decide what to do with an
 // unbound claim.
 func (ctrl *PersistentVolumeController) syncUnboundClaim(claim *v1.PersistentVolumeClaim) error {

pkg/controller/volume/persistentvolume/scheduler_assume_cache.go

@@ -42,6 +42,9 @@ type AssumeCache interface {
 	// Get the object by name
 	Get(objName string) (interface{}, error)
 
+	// Get the API object by name
+	GetAPIObj(objName string) (interface{}, error)
+
 	// List all the objects in the cache
 	List(indexObj interface{}) []interface{}
 }
@@ -250,6 +253,17 @@ func (c *assumeCache) Get(objName string) (interface{}, error) {
 	return objInfo.latestObj, nil
 }
 
+func (c *assumeCache) GetAPIObj(objName string) (interface{}, error) {
+	c.rwMutex.RLock()
+	defer c.rwMutex.RUnlock()
+
+	objInfo, err := c.getObjInfo(objName)
+	if err != nil {
+		return nil, err
+	}
+	return objInfo.apiObj, nil
+}
+
 func (c *assumeCache) List(indexObj interface{}) []interface{} {
 	c.rwMutex.RLock()
 	defer c.rwMutex.RUnlock()
@@ -297,7 +311,7 @@ func (c *assumeCache) Assume(obj interface{}) error {
 	}
 
 	if newVersion < storedVersion {
-		return fmt.Errorf("%v %q is out of sync", c.description, name)
+		return fmt.Errorf("%v %q is out of sync (stored: %d, assume: %d)", c.description, name, storedVersion, newVersion)
 	}
 
 	// Only update the cached object
@@ -325,6 +339,7 @@ type PVAssumeCache interface {
 	AssumeCache
 
 	GetPV(pvName string) (*v1.PersistentVolume, error)
+	GetAPIPV(pvName string) (*v1.PersistentVolume, error)
 	ListPVs(storageClassName string) []*v1.PersistentVolume
 }
 
@@ -356,6 +371,18 @@ func (c *pvAssumeCache) GetPV(pvName string) (*v1.PersistentVolume, error) {
 	return pv, nil
 }
 
+func (c *pvAssumeCache) GetAPIPV(pvName string) (*v1.PersistentVolume, error) {
+	obj, err := c.GetAPIObj(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(storageClassName string) []*v1.PersistentVolume {
 	objs := c.List(&v1.PersistentVolume{
 		Spec: v1.PersistentVolumeSpec{
@@ -380,6 +407,7 @@ type PVCAssumeCache interface {
 	// GetPVC returns the PVC from the cache with given pvcKey.
 	// pvcKey is the result of MetaNamespaceKeyFunc on PVC obj
 	GetPVC(pvcKey string) (*v1.PersistentVolumeClaim, error)
+	GetAPIPVC(pvcKey string) (*v1.PersistentVolumeClaim, error)
 }
 
 type pvcAssumeCache struct {
@@ -402,3 +430,15 @@ func (c *pvcAssumeCache) GetPVC(pvcKey string) (*v1.PersistentVolumeClaim, error
 	}
 	return pvc, nil
 }
+
+func (c *pvcAssumeCache) GetAPIPVC(pvcKey string) (*v1.PersistentVolumeClaim, error) {
+	obj, err := c.GetAPIObj(pvcKey)
+	if err != nil {
+		return nil, err
+	}
+	pvc, ok := obj.(*v1.PersistentVolumeClaim)
+	if !ok {
+		return nil, &errWrongType{"v1.PersistentVolumeClaim", obj}
+	}
+	return pvc, nil
+}

pkg/controller/volume/persistentvolume/scheduler_binder_cache.go

@@ -44,6 +44,9 @@ type PodBindingCache interface {
 	// means that no provisioning operations are needed.
 	GetProvisionedPVCs(pod *v1.Pod, node string) []*v1.PersistentVolumeClaim
 
+	// GetDecisions will return all cached decisions for the given pod.
+	GetDecisions(pod *v1.Pod) nodeDecisions
+
 	// DeleteBindings will remove all cached bindings and provisionings for the given pod.
 	// TODO: separate the func if it is needed to delete bindings/provisionings individually
 	DeleteBindings(pod *v1.Pod)
@@ -72,6 +75,17 @@ func NewPodBindingCache() PodBindingCache {
 	return &podBindingCache{bindingDecisions: map[string]nodeDecisions{}}
 }
 
+func (c *podBindingCache) GetDecisions(pod *v1.Pod) nodeDecisions {
+	c.rwMutex.Lock()
+	defer c.rwMutex.Unlock()
+	podName := getPodName(pod)
+	decisions, ok := c.bindingDecisions[podName]
+	if !ok {
+		return nil
+	}
+	return decisions
+}
+
 func (c *podBindingCache) DeleteBindings(pod *v1.Pod) {
 	c.rwMutex.Lock()
 	defer c.rwMutex.Unlock()