Merge pull request #67556 from msau42/fix-assume

cmd/kube-scheduler/app/options/options_test.go

@@ -144,6 +144,7 @@ users:
 	}
 
 	defaultSource := "DefaultProvider"
+	defaultBindTimeoutSeconds := int64(600)
 
 	testcases := []struct {
 		name             string
@@ -157,7 +158,10 @@ users:
 			options: &Options{
 				ConfigFile: configFile,
 				ComponentConfig: func() kubeschedulerconfig.KubeSchedulerConfiguration {
-					cfg, _ := newDefaultComponentConfig()
+					cfg, err := newDefaultComponentConfig()
+					if err != nil {
+						t.Fatal(err)
+					}
 					return *cfg
 				}(),
 			},
@@ -187,6 +191,7 @@ users:
 					ContentType: "application/vnd.kubernetes.protobuf",
 				},
 				PercentageOfNodesToScore: 50,
+				BindTimeoutSeconds:       &defaultBindTimeoutSeconds,
 			},
 		},
 		{
@@ -229,6 +234,7 @@ users:
 					ContentType: "application/vnd.kubernetes.protobuf",
 				},
 				PercentageOfNodesToScore: 50,
+				BindTimeoutSeconds:       &defaultBindTimeoutSeconds,
 			},
 		},
 		{

cmd/kube-scheduler/app/server.go

@@ -305,6 +305,7 @@ func NewSchedulerConfig(s schedulerserverconfig.CompletedConfig) (*scheduler.Con
 		EnableEquivalenceClassCache:    utilfeature.DefaultFeatureGate.Enabled(features.EnableEquivalenceClassCache),
 		DisablePreemption:              s.ComponentConfig.DisablePreemption,
 		PercentageOfNodesToScore:       s.ComponentConfig.PercentageOfNodesToScore,
+		BindTimeoutSeconds:             *s.ComponentConfig.BindTimeoutSeconds,
 	})
 
 	source := s.ComponentConfig.AlgorithmSource

pkg/controller/volume/persistentvolume/scheduler_assume_cache.go

@@ -83,7 +83,8 @@ func (e *errObjectName) Error() string {
 // 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
+	// Synchronizes updates to store
+	rwMutex sync.RWMutex
 
 	// describes the object stored
 	description string
@@ -155,8 +156,8 @@ func (c *assumeCache) add(obj interface{}) {
 		return
 	}
 
-	c.mutex.Lock()
-	defer c.mutex.Unlock()
+	c.rwMutex.Lock()
+	defer c.rwMutex.Unlock()
 
 	if objInfo, _ := c.getObjInfo(name); objInfo != nil {
 		newVersion, err := c.getObjVersion(name, obj)
@@ -199,8 +200,8 @@ func (c *assumeCache) delete(obj interface{}) {
 		return
 	}
 
-	c.mutex.Lock()
-	defer c.mutex.Unlock()
+	c.rwMutex.Lock()
+	defer c.rwMutex.Unlock()
 
 	objInfo := &objInfo{name: name}
 	err = c.store.Delete(objInfo)
@@ -239,8 +240,8 @@ func (c *assumeCache) getObjInfo(name string) (*objInfo, error) {
 }
 
 func (c *assumeCache) Get(objName string) (interface{}, error) {
-	c.mutex.Lock()
-	defer c.mutex.Unlock()
+	c.rwMutex.RLock()
+	defer c.rwMutex.RUnlock()
 
 	objInfo, err := c.getObjInfo(objName)
 	if err != nil {
@@ -250,8 +251,8 @@ func (c *assumeCache) Get(objName string) (interface{}, error) {
 }
 
 func (c *assumeCache) List(indexObj interface{}) []interface{} {
-	c.mutex.Lock()
-	defer c.mutex.Unlock()
+	c.rwMutex.RLock()
+	defer c.rwMutex.RUnlock()
 
 	allObjs := []interface{}{}
 	objs, err := c.store.Index(c.indexName, &objInfo{latestObj: indexObj})
@@ -277,8 +278,8 @@ func (c *assumeCache) Assume(obj interface{}) error {
 		return &errObjectName{err}
 	}
 
-	c.mutex.Lock()
-	defer c.mutex.Unlock()
+	c.rwMutex.Lock()
+	defer c.rwMutex.Unlock()
 
 	objInfo, err := c.getObjInfo(name)
 	if err != nil {
@@ -306,8 +307,8 @@ func (c *assumeCache) Assume(obj interface{}) error {
 }
 
 func (c *assumeCache) Restore(objName string) {
-	c.mutex.Lock()
-	defer c.mutex.Unlock()
+	c.rwMutex.Lock()
+	defer c.rwMutex.Unlock()
 
 	objInfo, err := c.getObjInfo(objName)
 	if err != nil {

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.
@@ -46,7 +50,8 @@ type PodBindingCache interface {
 }
 
 type podBindingCache struct {
-	mutex sync.Mutex
+	// synchronizes bindingDecisions
+	rwMutex sync.RWMutex
 
 	// Key = pod name
 	// Value = nodeDecisions
@@ -68,16 +73,16 @@ func NewPodBindingCache() PodBindingCache {
 }
 
 func (c *podBindingCache) DeleteBindings(pod *v1.Pod) {
-	c.mutex.Lock()
-	defer c.mutex.Unlock()
+	c.rwMutex.Lock()
+	defer c.rwMutex.Unlock()
 
 	podName := getPodName(pod)
 	delete(c.bindingDecisions, podName)
 }
 
 func (c *podBindingCache) UpdateBindings(pod *v1.Pod, node string, bindings []*bindingInfo) {
-	c.mutex.Lock()
-	defer c.mutex.Unlock()
+	c.rwMutex.Lock()
+	defer c.rwMutex.Unlock()
 
 	podName := getPodName(pod)
 	decisions, ok := c.bindingDecisions[podName]
@@ -97,8 +102,8 @@ func (c *podBindingCache) UpdateBindings(pod *v1.Pod, node string, bindings []*b
 }
 
 func (c *podBindingCache) GetBindings(pod *v1.Pod, node string) []*bindingInfo {
-	c.mutex.Lock()
-	defer c.mutex.Unlock()
+	c.rwMutex.RLock()
+	defer c.rwMutex.RUnlock()
 
 	podName := getPodName(pod)
 	decisions, ok := c.bindingDecisions[podName]
@@ -113,8 +118,8 @@ func (c *podBindingCache) GetBindings(pod *v1.Pod, node string) []*bindingInfo {
 }
 
 func (c *podBindingCache) UpdateProvisionedPVCs(pod *v1.Pod, node string, pvcs []*v1.PersistentVolumeClaim) {
-	c.mutex.Lock()
-	defer c.mutex.Unlock()
+	c.rwMutex.Lock()
+	defer c.rwMutex.Unlock()
 
 	podName := getPodName(pod)
 	decisions, ok := c.bindingDecisions[podName]
@@ -134,8 +139,8 @@ func (c *podBindingCache) UpdateProvisionedPVCs(pod *v1.Pod, node string, pvcs [
 }
 
 func (c *podBindingCache) GetProvisionedPVCs(pod *v1.Pod, node string) []*v1.PersistentVolumeClaim {
-	c.mutex.Lock()
-	defer c.mutex.Unlock()
+	c.rwMutex.RLock()
+	defer c.rwMutex.RUnlock()
 
 	podName := getPodName(pod)
 	decisions, ok := c.bindingDecisions[podName]

pkg/controller/volume/persistentvolume/scheduler_binder_fake.go

@@ -16,18 +16,15 @@ limitations under the License.
 
 package persistentvolume
 
-import (
-	"k8s.io/api/core/v1"
-)
+import "k8s.io/api/core/v1"
 
 type FakeVolumeBinderConfig struct {
-	AllBound              bool
-	FindUnboundSatsified  bool
-	FindBoundSatsified    bool
-	FindErr               error
-	AssumeBindingRequired bool
-	AssumeErr             error
-	BindErr               error
+	AllBound             bool
+	FindUnboundSatsified bool
+	FindBoundSatsified   bool
+	FindErr              error
+	AssumeErr            error
+	BindErr              error
 }
 
 // NewVolumeBinder sets up all the caches needed for the scheduler to make
@@ -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/scheduler/apis/config/types.go

@@ -85,6 +85,11 @@ type KubeSchedulerConfiguration struct {
 	// DEPRECATED.
 	// Indicate the "all topologies" set for empty topologyKey when it's used for PreferredDuringScheduling pod anti-affinity.
 	FailureDomains string
+
+	// Duration to wait for a binding operation to complete before timing out
+	// Value must be non-negative integer. The value zero indicates no waiting.
+	// If this value is nil, the default value will be used.
+	BindTimeoutSeconds *int64
 }
 
 // SchedulerAlgorithmSource is the source of a scheduler algorithm. One source

pkg/scheduler/apis/config/v1alpha1/defaults.go

@@ -23,6 +23,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	apiserverconfigv1alpha1 "k8s.io/apiserver/pkg/apis/config/v1alpha1"
 	kubescedulerconfigv1alpha1 "k8s.io/kube-scheduler/config/v1alpha1"
+
 	// this package shouldn't really depend on other k8s.io/kubernetes code
 	api "k8s.io/kubernetes/pkg/apis/core"
 	kubeletapis "k8s.io/kubernetes/pkg/kubelet/apis"
@@ -102,4 +103,9 @@ func SetDefaults_KubeSchedulerConfiguration(obj *kubescedulerconfigv1alpha1.Kube
 
 	// Use the default LeaderElectionConfiguration options
 	apiserverconfigv1alpha1.RecommendedDefaultLeaderElectionConfiguration(&obj.LeaderElection.LeaderElectionConfiguration)
+
+	if obj.BindTimeoutSeconds == nil {
+		defaultBindTimeoutSeconds := int64(600)
+		obj.BindTimeoutSeconds = &defaultBindTimeoutSeconds
+	}
 }

pkg/scheduler/apis/config/v1alpha1/zz_generated.conversion.go

@@ -121,6 +121,7 @@ func autoConvert_v1alpha1_KubeSchedulerConfiguration_To_config_KubeSchedulerConf
 	out.DisablePreemption = in.DisablePreemption
 	out.PercentageOfNodesToScore = in.PercentageOfNodesToScore
 	out.FailureDomains = in.FailureDomains
+	out.BindTimeoutSeconds = (*int64)(unsafe.Pointer(in.BindTimeoutSeconds))
 	return nil
 }
 
@@ -149,6 +150,7 @@ func autoConvert_config_KubeSchedulerConfiguration_To_v1alpha1_KubeSchedulerConf
 	out.DisablePreemption = in.DisablePreemption
 	out.PercentageOfNodesToScore = in.PercentageOfNodesToScore
 	out.FailureDomains = in.FailureDomains
+	out.BindTimeoutSeconds = (*int64)(unsafe.Pointer(in.BindTimeoutSeconds))
 	return nil
 }
 

pkg/scheduler/apis/config/validation/validation.go

@@ -41,6 +41,9 @@ func ValidateKubeSchedulerConfiguration(cc *config.KubeSchedulerConfiguration) f
 	if cc.HardPodAffinitySymmetricWeight < 0 || cc.HardPodAffinitySymmetricWeight > 100 {
 		allErrs = append(allErrs, field.Invalid(field.NewPath("hardPodAffinitySymmetricWeight"), cc.HardPodAffinitySymmetricWeight, "not in valid range 0-100"))
 	}
+	if cc.BindTimeoutSeconds == nil {
+		allErrs = append(allErrs, field.Required(field.NewPath("bindTimeoutSeconds"), ""))
+	}
 	return allErrs
 }
 

pkg/scheduler/apis/config/validation/validation_test.go

@@ -17,15 +17,17 @@ limitations under the License.
 package validation
 
 import (
+	"testing"
+	"time"
+
 	apimachinery "k8s.io/apimachinery/pkg/apis/config"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	apiserver "k8s.io/apiserver/pkg/apis/config"
 	"k8s.io/kubernetes/pkg/scheduler/apis/config"
-	"testing"
-	"time"
 )
 
 func TestValidateKubeSchedulerConfiguration(t *testing.T) {
+	testTimeout := int64(0)
 	validConfig := &config.KubeSchedulerConfiguration{
 		SchedulerName:                  "me",
 		HealthzBindAddress:             "0.0.0.0:10254",
@@ -56,6 +58,7 @@ func TestValidateKubeSchedulerConfiguration(t *testing.T) {
 				RetryPeriod:   metav1.Duration{Duration: 5 * time.Second},
 			},
 		},
+		BindTimeoutSeconds: &testTimeout,
 	}
 
 	HardPodAffinitySymmetricWeightGt100 := validConfig.DeepCopy()
@@ -86,6 +89,9 @@ func TestValidateKubeSchedulerConfiguration(t *testing.T) {
 	enableContentProfilingSetWithoutEnableProfiling.EnableProfiling = false
 	enableContentProfilingSetWithoutEnableProfiling.EnableContentionProfiling = true
 
+	bindTimeoutUnset := validConfig.DeepCopy()
+	bindTimeoutUnset.BindTimeoutSeconds = nil
+
 	scenarios := map[string]struct {
 		expectedToFail bool
 		config         *config.KubeSchedulerConfiguration
@@ -126,6 +132,10 @@ func TestValidateKubeSchedulerConfiguration(t *testing.T) {
 			expectedToFail: true,
 			config:         HardPodAffinitySymmetricWeightLt0,
 		},
+		"bind-timeout-unset": {
+			expectedToFail: true,
+			config:         bindTimeoutUnset,
+		},
 	}
 
 	for name, scenario := range scenarios {

pkg/scheduler/apis/config/zz_generated.deepcopy.go

@@ -32,6 +32,11 @@ func (in *KubeSchedulerConfiguration) DeepCopyInto(out *KubeSchedulerConfigurati
 	out.LeaderElection = in.LeaderElection
 	out.ClientConnection = in.ClientConnection
 	out.DebuggingConfiguration = in.DebuggingConfiguration
+	if in.BindTimeoutSeconds != nil {
+		in, out := &in.BindTimeoutSeconds, &out.BindTimeoutSeconds
+		*out = new(int64)
+		**out = **in
+	}
 	return
 }
 

pkg/scheduler/factory/factory.go

@@ -159,6 +159,7 @@ type ConfigFactoryArgs struct {
 	EnableEquivalenceClassCache    bool
 	DisablePreemption              bool
 	PercentageOfNodesToScore       int32
+	BindTimeoutSeconds             int64
 }
 
 // NewConfigFactory initializes the default implementation of a Configurator To encourage eventual privatization of the struct type, we only
@@ -305,7 +306,7 @@ func NewConfigFactory(args *ConfigFactoryArgs) scheduler.Configurator {
 
 	if utilfeature.DefaultFeatureGate.Enabled(features.VolumeScheduling) {
 		// Setup volume binder
-		c.volumeBinder = volumebinder.NewVolumeBinder(args.Client, args.PvcInformer, args.PvInformer, args.StorageClassInformer)
+		c.volumeBinder = volumebinder.NewVolumeBinder(args.Client, args.PvcInformer, args.PvInformer, args.StorageClassInformer, time.Duration(args.BindTimeoutSeconds)*time.Second)
 
 		args.StorageClassInformer.Informer().AddEventHandler(
 			cache.ResourceEventHandlerFuncs{

pkg/scheduler/factory/factory_test.go

@@ -49,6 +49,7 @@ import (
 const (
 	enableEquivalenceCache = true
 	disablePodPreemption   = false
+	bindTimeoutSeconds     = 600
 )
 
 func TestCreate(t *testing.T) {
@@ -557,6 +558,7 @@ func newConfigFactory(client *clientset.Clientset, hardPodAffinitySymmetricWeigh
 		enableEquivalenceCache,
 		disablePodPreemption,
 		schedulerapi.DefaultPercentageOfNodesToScore,
+		bindTimeoutSeconds,
 	})
 }
 

pkg/scheduler/scheduler.go

@@ -17,7 +17,6 @@ limitations under the License.
 package scheduler
 
 import (
-	"fmt"
 	"time"
 
 	"k8s.io/api/core/v1"
@@ -184,10 +183,6 @@ 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)
 }
 
@@ -265,17 +260,12 @@ 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.
+// assumeVolumes will update the volume cache with the chosen bindings
 //
 // This function modifies assumed if volume binding is required.
-func (sched *Scheduler) assumeAndBindVolumes(assumed *v1.Pod, host string) error {
+func (sched *Scheduler) assumeVolumes(assumed *v1.Pod, host string) (allBound bool, err error) {
 	if utilfeature.DefaultFeatureGate.Enabled(features.VolumeScheduling) {
-		allBound, bindingRequired, err := sched.config.VolumeBinder.Binder.AssumePodVolumes(assumed, host)
+		allBound, 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)
@@ -285,76 +275,38 @@ func (sched *Scheduler) assumeAndBindVolumes(assumed *v1.Pod, host string) error
 				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.CheckVolumeBindingPred)
-					sched.config.Ecache.InvalidatePredicates(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
+		// Invalidate ecache because assumed volumes could have affected the cached
+		// pvs for other pods
+		if sched.config.Ecache != nil {
+			invalidPredicates := sets.NewString(predicates.CheckVolumeBindingPred)
+			sched.config.Ecache.InvalidatePredicates(invalidPredicates)
 		}
 	}
-	return nil
+	return
 }
 
-// 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)
+// bindVolumes will make the API update with the assumed bindings and wait until
+// the PV controller has completely finished the binding operation.
+//
+// If binding errors, times out or gets undone, then an error will be returned to
+// retry scheduling.
+func (sched *Scheduler) bindVolumes(assumed *v1.Pod) error {
+	var reason string
+	var eventType string
+
+	glog.V(5).Infof("Trying to bind volumes for pod \"%v/%v\"", assumed.Namespace, assumed.Name)
+	err := sched.config.VolumeBinder.Binder.BindPodVolumes(assumed)
+	if err != nil {
+		glog.V(1).Infof("Failed to bind volumes for pod \"%v/%v\": %v", assumed.Namespace, assumed.Name, err)
 
-		// 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\": %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")
+		// Unassume the Pod and retry scheduling
+		if forgetErr := sched.config.SchedulerCache.ForgetPod(assumed); forgetErr != nil {
+			glog.Errorf("scheduler cache ForgetPod failed: %v", forgetErr)
 		}
 
-		// 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.
+		reason = "VolumeBindingFailed"
+		eventType = v1.EventTypeWarning
 		sched.config.Error(assumed, err)
 		sched.config.Recorder.Eventf(assumed, eventType, "FailedScheduling", "%v", err)
 		sched.config.PodConditionUpdater.Update(assumed, &v1.PodCondition{
@@ -362,15 +314,11 @@ func (sched *Scheduler) bindVolumesWorker() {
 			Status: v1.ConditionFalse,
 			Reason: reason,
 		})
-		return false
+		return err
 	}
 
-	for {
-		if quit := workFunc(); quit {
-			glog.V(4).Infof("bindVolumesWorker shutting down")
-			break
-		}
-	}
+	glog.V(5).Infof("Success binding volumes for pod \"%v/%v\"", assumed.Namespace, assumed.Name)
+	return nil
 }
 
 // assume signals to the cache that a pod is already in the cache, so that binding can be asynchronous.
@@ -478,16 +426,12 @@ func (sched *Scheduler) scheduleOne() {
 
 	// 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.
+	// If all volumes are completely bound, then allBound is true and binding will be skipped.
 	//
-	// 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.
+	// Otherwise, binding of volumes is started after the pod is assumed, but before pod binding.
 	//
 	// This function modifies 'assumedPod' if volume binding is required.
-	err = sched.assumeAndBindVolumes(assumedPod, suggestedHost)
+	allBound, err := sched.assumeVolumes(assumedPod, suggestedHost)
 	if err != nil {
 		return
 	}
@@ -499,6 +443,14 @@ func (sched *Scheduler) scheduleOne() {
 	}
 	// bind the pod to its host asynchronously (we can do this b/c of the assumption step above).
 	go func() {
+		// Bind volumes first before Pod
+		if !allBound {
+			err = sched.bindVolumes(assumedPod)
+			if err != nil {
+				return
+			}
+		}
+
 		err := sched.bind(assumedPod, &v1.Binding{
 			ObjectMeta: metav1.ObjectMeta{Namespace: assumedPod.Namespace, Name: assumedPod.Name, UID: assumedPod.UID},
 			Target: v1.ObjectReference{

pkg/scheduler/scheduler_test.go

@@ -707,8 +707,7 @@ func TestSchedulerWithVolumeBinding(t *testing.T) {
 			},
 			expectAssumeCalled: true,
 			expectPodBind:      &v1.Binding{ObjectMeta: metav1.ObjectMeta{Name: "foo", UID: types.UID("foo")}, Target: v1.ObjectReference{Kind: "Node", Name: "machine1"}},
-
-			eventReason: "Scheduled",
+			eventReason:        "Scheduled",
 		},
 		{
 			name: "bound/invalid pv affinity",
@@ -739,28 +738,15 @@ func TestSchedulerWithVolumeBinding(t *testing.T) {
 			expectError: makePredicateError("1 node(s) didn't find available persistent volumes to bind, 1 node(s) had volume node affinity conflict"),
 		},
 		{
-			name: "unbound/found matches",
+			name: "unbound/found matches/bind succeeds",
 			volumeBinderConfig: &persistentvolume.FakeVolumeBinderConfig{
-				FindUnboundSatsified:  true,
-				FindBoundSatsified:    true,
-				AssumeBindingRequired: true,
+				FindUnboundSatsified: true,
+				FindBoundSatsified:   true,
 			},
 			expectAssumeCalled: true,
 			expectBindCalled:   true,
-			eventReason:        "FailedScheduling",
-			expectError:        fmt.Errorf("Volume binding started, waiting for completion"),
-		},
-		{
-			name: "unbound/found matches/already-bound",
-			volumeBinderConfig: &persistentvolume.FakeVolumeBinderConfig{
-				FindUnboundSatsified:  true,
-				FindBoundSatsified:    true,
-				AssumeBindingRequired: false,
-			},
-			expectAssumeCalled: true,
-			expectBindCalled:   false,
-			eventReason:        "FailedScheduling",
-			expectError:        fmt.Errorf("Volume binding started, waiting for completion"),
+			expectPodBind:      &v1.Binding{ObjectMeta: metav1.ObjectMeta{Name: "foo", UID: types.UID("foo")}, Target: v1.ObjectReference{Kind: "Node", Name: "machine1"}},
+			eventReason:        "Scheduled",
 		},
 		{
 			name: "predicate error",
@@ -784,10 +770,9 @@ func TestSchedulerWithVolumeBinding(t *testing.T) {
 		{
 			name: "bind error",
 			volumeBinderConfig: &persistentvolume.FakeVolumeBinderConfig{
-				FindUnboundSatsified:  true,
-				FindBoundSatsified:    true,
-				AssumeBindingRequired: true,
-				BindErr:               bindErr,
+				FindUnboundSatsified: true,
+				FindBoundSatsified:   true,
+				BindErr:              bindErr,
 			},
 			expectAssumeCalled: true,
 			expectBindCalled:   true,
@@ -814,8 +799,6 @@ func TestSchedulerWithVolumeBinding(t *testing.T) {
 				close(eventChan)
 			})
 
-			go fakeVolumeBinder.Run(s.bindVolumesWorker, stop)
-
 			s.scheduleOne()
 
 			// Wait for pod to succeed or fail scheduling

pkg/scheduler/volumebinder/BUILD

@@ -8,11 +8,9 @@ go_library(
     deps = [
         "//pkg/controller/volume/persistentvolume:go_default_library",
         "//staging/src/k8s.io/api/core/v1:go_default_library",
-        "//staging/src/k8s.io/apimachinery/pkg/util/wait: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",
         "//staging/src/k8s.io/client-go/kubernetes:go_default_library",
-        "//staging/src/k8s.io/client-go/util/workqueue:go_default_library",
     ],
 )
 

pkg/scheduler/volumebinder/volume_binder.go

@@ -20,19 +20,15 @@ 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.
+// VolumeBinder sets up the volume binding library
 type VolumeBinder struct {
-	Binder    persistentvolume.SchedulerVolumeBinder
-	BindQueue *workqueue.Type
+	Binder persistentvolume.SchedulerVolumeBinder
 }
 
 // NewVolumeBinder sets up the volume binding library and binding queue
@@ -40,30 +36,21 @@ func NewVolumeBinder(
 	client clientset.Interface,
 	pvcInformer coreinformers.PersistentVolumeClaimInformer,
 	pvInformer coreinformers.PersistentVolumeInformer,
-	storageClassInformer storageinformers.StorageClassInformer) *VolumeBinder {
+	storageClassInformer storageinformers.StorageClassInformer,
+	bindTimeout time.Duration) *VolumeBinder {
 
 	return &VolumeBinder{
-		Binder:    persistentvolume.NewVolumeBinder(client, pvcInformer, pvInformer, storageClassInformer),
-		BindQueue: workqueue.NewNamed("podsToBind"),
+		Binder: persistentvolume.NewVolumeBinder(client, pvcInformer, pvInformer, storageClassInformer, bindTimeout),
 	}
 }
 
 // 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"),
+		Binder: persistentvolume.NewFakeVolumeBinder(config),
 	}
 }
 
-// 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()

pkg/volume/testing/testing.go

@@ -239,6 +239,7 @@ type FakeVolumePlugin struct {
 	VolumeLimits           map[string]int64
 	VolumeLimitsError      error
 	LimitKey               string
+	ProvisionDelaySeconds  int
 
 	Mounters             []*FakeVolume
 	Unmounters           []*FakeVolume
@@ -437,7 +438,7 @@ func (plugin *FakeVolumePlugin) NewProvisioner(options VolumeOptions) (Provision
 	plugin.Lock()
 	defer plugin.Unlock()
 	plugin.LastProvisionerOptions = options
-	return &FakeProvisioner{options, plugin.Host}, nil
+	return &FakeProvisioner{options, plugin.Host, plugin.ProvisionDelaySeconds}, nil
 }
 
 func (plugin *FakeVolumePlugin) GetAccessModes() []v1.PersistentVolumeAccessMode {
@@ -779,8 +780,9 @@ func (fd *FakeDeleter) GetPath() string {
 }
 
 type FakeProvisioner struct {
-	Options VolumeOptions
-	Host    VolumeHost
+	Options               VolumeOptions
+	Host                  VolumeHost
+	ProvisionDelaySeconds int
 }
 
 func (fc *FakeProvisioner) Provision(selectedNode *v1.Node, allowedTopologies []v1.TopologySelectorTerm) (*v1.PersistentVolume, error) {
@@ -807,6 +809,10 @@ func (fc *FakeProvisioner) Provision(selectedNode *v1.Node, allowedTopologies []
 		},
 	}
 
+	if fc.ProvisionDelaySeconds > 0 {
+		time.Sleep(time.Duration(fc.ProvisionDelaySeconds) * time.Second)
+	}
+
 	return pv, nil
 }
 

staging/src/k8s.io/kube-scheduler/config/v1alpha1/types.go

@@ -81,6 +81,11 @@ type KubeSchedulerConfiguration struct {
 	// DEPRECATED.
 	// Indicate the "all topologies" set for empty topologyKey when it's used for PreferredDuringScheduling pod anti-affinity.
 	FailureDomains string `json:"failureDomains"`
+
+	// Duration to wait for a binding operation to complete before timing out
+	// Value must be non-negative integer. The value zero indicates no waiting.
+	// If this value is nil, the default value will be used.
+	BindTimeoutSeconds *int64 `json:"bindTimeoutSeconds"`
 }
 
 // SchedulerAlgorithmSource is the source of a scheduler algorithm. One source

staging/src/k8s.io/kube-scheduler/config/v1alpha1/zz_generated.deepcopy.go

@@ -32,6 +32,11 @@ func (in *KubeSchedulerConfiguration) DeepCopyInto(out *KubeSchedulerConfigurati
 	in.LeaderElection.DeepCopyInto(&out.LeaderElection)
 	out.ClientConnection = in.ClientConnection
 	out.DebuggingConfiguration = in.DebuggingConfiguration
+	if in.BindTimeoutSeconds != nil {
+		in, out := &in.BindTimeoutSeconds, &out.BindTimeoutSeconds
+		*out = new(int64)
+		**out = **in
+	}
 	return
 }
 

test/e2e/storage/persistent_volumes-local.go

@@ -566,13 +566,28 @@ var _ = utils.SIGDescribe("PersistentVolumes-local ", func() {
 				framework.Skipf("Runs only when number of nodes >= %v", ssReplicas)
 			}
 			By("Creating a StatefulSet with pod anti-affinity on nodes")
-			ss := createStatefulSet(config, ssReplicas, volsPerNode, true)
+			ss := createStatefulSet(config, ssReplicas, volsPerNode, true, false)
 			validateStatefulSet(config, ss, true)
 		})
 
 		It("should use volumes on one node when pod has affinity", func() {
 			By("Creating a StatefulSet with pod affinity on nodes")
-			ss := createStatefulSet(config, ssReplicas, volsPerNode/ssReplicas, false)
+			ss := createStatefulSet(config, ssReplicas, volsPerNode/ssReplicas, false, false)
+			validateStatefulSet(config, ss, false)
+		})
+
+		It("should use volumes spread across nodes when pod management is parallel and pod has anti-affinity", func() {
+			if len(config.nodes) < ssReplicas {
+				framework.Skipf("Runs only when number of nodes >= %v", ssReplicas)
+			}
+			By("Creating a StatefulSet with pod anti-affinity on nodes")
+			ss := createStatefulSet(config, ssReplicas, 1, true, true)
+			validateStatefulSet(config, ss, true)
+		})
+
+		It("should use volumes on one node when pod management is parallel and pod has affinity", func() {
+			By("Creating a StatefulSet with pod affinity on nodes")
+			ss := createStatefulSet(config, ssReplicas, 1, false, true)
 			validateStatefulSet(config, ss, false)
 		})
 	})
@@ -1830,7 +1845,7 @@ func findLocalPersistentVolume(c clientset.Interface, volumePath string) (*v1.Pe
 	return nil, nil
 }
 
-func createStatefulSet(config *localTestConfig, ssReplicas int32, volumeCount int, anti bool) *appsv1.StatefulSet {
+func createStatefulSet(config *localTestConfig, ssReplicas int32, volumeCount int, anti, parallel bool) *appsv1.StatefulSet {
 	mounts := []v1.VolumeMount{}
 	claims := []v1.PersistentVolumeClaim{}
 	for i := 0; i < volumeCount; i++ {
@@ -1897,6 +1912,10 @@ func createStatefulSet(config *localTestConfig, ssReplicas int32, volumeCount in
 		},
 	}
 
+	if parallel {
+		spec.Spec.PodManagementPolicy = appsv1.ParallelPodManagement
+	}
+
 	ss, err := config.client.AppsV1().StatefulSets(config.ns).Create(spec)
 	Expect(err).NotTo(HaveOccurred())
 

test/integration/scheduler/BUILD

@@ -34,6 +34,7 @@ go_test(
         "//pkg/kubeapiserver/admission:go_default_library",
         "//pkg/scheduler:go_default_library",
         "//pkg/scheduler/algorithm:go_default_library",
+        "//pkg/scheduler/algorithm/predicates:go_default_library",
         "//pkg/scheduler/algorithmprovider:go_default_library",
         "//pkg/scheduler/api:go_default_library",
         "//pkg/scheduler/apis/config:go_default_library",

test/integration/scheduler/scheduler_test.go

@@ -183,6 +183,7 @@ func TestSchedulerCreationFromConfigMap(t *testing.T) {
 		eventBroadcaster := record.NewBroadcaster()
 		eventBroadcaster.StartRecordingToSink(&clientv1core.EventSinkImpl{Interface: clientSet.CoreV1().Events("")})
 
+		defaultBindTimeout := int64(30)
 		ss := &schedulerappconfig.Config{
 			ComponentConfig: kubeschedulerconfig.KubeSchedulerConfiguration{
 				HardPodAffinitySymmetricWeight: v1.DefaultHardPodAffinitySymmetricWeight,
@@ -195,6 +196,7 @@ func TestSchedulerCreationFromConfigMap(t *testing.T) {
 						},
 					},
 				},
+				BindTimeoutSeconds: &defaultBindTimeout,
 			},
 			Client:          clientSet,
 			InformerFactory: informerFactory,
@@ -244,6 +246,7 @@ func TestSchedulerCreationFromNonExistentConfigMap(t *testing.T) {
 	eventBroadcaster := record.NewBroadcaster()
 	eventBroadcaster.StartRecordingToSink(&clientv1core.EventSinkImpl{Interface: clientSet.CoreV1().Events("")})
 
+	defaultBindTimeout := int64(30)
 	ss := &schedulerappconfig.Config{
 		ComponentConfig: kubeschedulerconfig.KubeSchedulerConfiguration{
 			SchedulerName: v1.DefaultSchedulerName,
@@ -256,6 +259,7 @@ func TestSchedulerCreationFromNonExistentConfigMap(t *testing.T) {
 				},
 			},
 			HardPodAffinitySymmetricWeight: v1.DefaultHardPodAffinitySymmetricWeight,
+			BindTimeoutSeconds:             &defaultBindTimeout,
 		},
 		Client:          clientSet,
 		InformerFactory: informerFactory,

test/integration/scheduler/util.go

@@ -91,6 +91,7 @@ func createConfiguratorWithPodInformer(
 		EnableEquivalenceClassCache:    utilfeature.DefaultFeatureGate.Enabled(features.EnableEquivalenceClassCache),
 		DisablePreemption:              false,
 		PercentageOfNodesToScore:       schedulerapi.DefaultPercentageOfNodesToScore,
+		BindTimeoutSeconds:             600,
 	})
 }
 
@@ -143,7 +144,7 @@ func initTestScheduler(
 ) *TestContext {
 	// Pod preemption is enabled by default scheduler configuration, but preemption only happens when PodPriority
 	// feature gate is enabled at the same time.
-	return initTestSchedulerWithOptions(t, context, controllerCh, setPodInformer, policy, false, time.Second)
+	return initTestSchedulerWithOptions(t, context, controllerCh, setPodInformer, policy, false, false, time.Second)
 }
 
 // initTestSchedulerWithOptions initializes a test environment and creates a scheduler with default
@@ -155,13 +156,15 @@ func initTestSchedulerWithOptions(
 	setPodInformer bool,
 	policy *schedulerapi.Policy,
 	disablePreemption bool,
+	disableEquivalenceCache bool,
 	resyncPeriod time.Duration,
 ) *TestContext {
-	// Enable EnableEquivalenceClassCache for all integration tests.
-	defer utilfeaturetesting.SetFeatureGateDuringTest(
-		t,
-		utilfeature.DefaultFeatureGate,
-		features.EnableEquivalenceClassCache, true)()
+	if !disableEquivalenceCache {
+		defer utilfeaturetesting.SetFeatureGateDuringTest(
+			t,
+			utilfeature.DefaultFeatureGate,
+			features.EnableEquivalenceClassCache, true)()
+	}
 
 	// 1. Create scheduler
 	context.informerFactory = informers.NewSharedInformerFactory(context.clientSet, resyncPeriod)
@@ -256,7 +259,7 @@ func initTest(t *testing.T, nsPrefix string) *TestContext {
 // configuration but with pod preemption disabled.
 func initTestDisablePreemption(t *testing.T, nsPrefix string) *TestContext {
 	return initTestSchedulerWithOptions(
-		t, initTestMaster(t, nsPrefix, nil), nil, true, nil, true, time.Second)
+		t, initTestMaster(t, nsPrefix, nil), nil, true, nil, true, false, time.Second)
 }
 
 // cleanupTest deletes the scheduler and the test namespace. It should be called