Volume topology aware dynamic provisioning: work based on new API

cmd/kube-controller-manager/app/core.go

@@ -175,6 +175,7 @@ func startPersistentVolumeBinderController(ctx ControllerContext) (bool, error)
 		ClaimInformer:             ctx.InformerFactory.Core().V1().PersistentVolumeClaims(),
 		ClassInformer:             ctx.InformerFactory.Storage().V1().StorageClasses(),
 		PodInformer:               ctx.InformerFactory.Core().V1().Pods(),
+		NodeInformer:              ctx.InformerFactory.Core().V1().Nodes(),
 		EnableDynamicProvisioning: ctx.ComponentConfig.PersistentVolumeBinderController.VolumeConfiguration.EnableDynamicProvisioning,
 	}
 	volumeController, volumeControllerErr := persistentvolumecontroller.NewController(params)

pkg/controller/volume/persistentvolume/framework_test.go

@@ -612,6 +612,7 @@ func newTestController(kubeClient clientset.Interface, informerFactory informers
 		ClaimInformer:             informerFactory.Core().V1().PersistentVolumeClaims(),
 		ClassInformer:             informerFactory.Storage().V1().StorageClasses(),
 		PodInformer:               informerFactory.Core().V1().Pods(),
+		NodeInformer:              informerFactory.Core().V1().Nodes(),
 		EventRecorder:             record.NewFakeRecorder(1000),
 		EnableDynamicProvisioning: enableDynamicProvisioning,
 	}
@@ -1192,7 +1193,7 @@ func (plugin *mockVolumePlugin) NewProvisioner(options vol.VolumeOptions) (vol.P
 	}
 }
 
-func (plugin *mockVolumePlugin) Provision() (*v1.PersistentVolume, error) {
+func (plugin *mockVolumePlugin) Provision(selectedNode *v1.Node, allowedTopologies []v1.TopologySelectorTerm) (*v1.PersistentVolume, error) {
 	if len(plugin.provisionCalls) <= plugin.provisionCallCounter {
 		return nil, fmt.Errorf("Mock plugin error: unexpected provisioner call %d", plugin.provisionCallCounter)
 	}

pkg/controller/volume/persistentvolume/pv_controller.go

@@ -174,6 +174,8 @@ type PersistentVolumeController struct {
 	classListerSynced  cache.InformerSynced
 	podLister          corelisters.PodLister
 	podListerSynced    cache.InformerSynced
+	NodeLister         corelisters.NodeLister
+	NodeListerSynced   cache.InformerSynced
 
 	kubeClient                clientset.Interface
 	eventRecorder             record.EventRecorder
@@ -1434,13 +1436,26 @@ func (ctrl *PersistentVolumeController) provisionClaimOperation(claim *v1.Persis
 		return
 	}
 
+	var selectedNode *v1.Node = nil
+	var allowedTopologies []v1.TopologySelectorTerm = nil
+	if utilfeature.DefaultFeatureGate.Enabled(features.DynamicProvisioningScheduling) {
+		if nodeName, ok := claim.Annotations[annSelectedNode]; ok {
+			selectedNode, err = ctrl.NodeLister.Get(nodeName)
+			if err != nil {
+				strerr := fmt.Sprintf("Failed to get target node: %v", err)
+				glog.V(3).Infof("unexpected error getting target node %q for claim %q: %v", nodeName, claimToClaimKey(claim), err)
+				ctrl.eventRecorder.Event(claim, v1.EventTypeWarning, events.ProvisioningFailed, strerr)
+				return
+			}
+		}
+		allowedTopologies = storageClass.AllowedTopologies
+	}
+
 	opComplete := util.OperationCompleteHook(plugin.GetPluginName(), "volume_provision")
-	// TODO: modify the Provision() interface to pass in the allowed topology information
-	// of the provisioned volume.
-	volume, err = provisioner.Provision()
+	volume, err = provisioner.Provision(selectedNode, allowedTopologies)
 	opComplete(&err)
 	if err != nil {
-		// Other places of failure has nothing to do with DynamicProvisioningScheduling,
+		// Other places of failure have nothing to do with DynamicProvisioningScheduling,
 		// so just let controller retry in the next sync. We'll only call func
 		// rescheduleProvisioning here when the underlying provisioning actually failed.
 		ctrl.rescheduleProvisioning(claim)

pkg/controller/volume/persistentvolume/pv_controller_base.go

@@ -63,6 +63,7 @@ type ControllerParameters struct {
 	ClaimInformer             coreinformers.PersistentVolumeClaimInformer
 	ClassInformer             storageinformers.StorageClassInformer
 	PodInformer               coreinformers.PodInformer
+	NodeInformer              coreinformers.NodeInformer
 	EventRecorder             record.EventRecorder
 	EnableDynamicProvisioning bool
 }
@@ -122,6 +123,8 @@ func NewController(p ControllerParameters) (*PersistentVolumeController, error)
 	controller.classListerSynced = p.ClassInformer.Informer().HasSynced
 	controller.podLister = p.PodInformer.Lister()
 	controller.podListerSynced = p.PodInformer.Informer().HasSynced
+	controller.NodeLister = p.NodeInformer.Lister()
+	controller.NodeListerSynced = p.NodeInformer.Informer().HasSynced
 	return controller, nil
 }
 
@@ -268,7 +271,7 @@ func (ctrl *PersistentVolumeController) Run(stopCh <-chan struct{}) {
 	glog.Infof("Starting persistent volume controller")
 	defer glog.Infof("Shutting down persistent volume controller")
 
-	if !controller.WaitForCacheSync("persistent volume", stopCh, ctrl.volumeListerSynced, ctrl.claimListerSynced, ctrl.classListerSynced, ctrl.podListerSynced) {
+	if !controller.WaitForCacheSync("persistent volume", stopCh, ctrl.volumeListerSynced, ctrl.claimListerSynced, ctrl.classListerSynced, ctrl.podListerSynced, ctrl.NodeListerSynced) {
 		return
 	}
 

pkg/controller/volume/persistentvolume/scheduler_binder.go

@@ -24,6 +24,7 @@ import (
 
 	"k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	coreinformers "k8s.io/client-go/informers/core/v1"
 	storageinformers "k8s.io/client-go/informers/storage/v1"
@@ -481,7 +482,11 @@ func (b *volumeBinder) checkVolumeProvisions(pod *v1.Pod, claimsToProvision []*v
 			return false, nil
 		}
 
-		// TODO: Check if the node can satisfy the topology requirement in the class
+		// Check if the node can satisfy the topology requirement in the class
+		if !v1helper.MatchTopologySelectorTerms(class.AllowedTopologies, labels.Set(node.Labels)) {
+			glog.V(4).Infof("Node %q cannot satisfy provisioning topology requirements of claim %q", node.Name, getPVCName(claim))
+			return false, nil
+		}
 
 		// TODO: Check if capacity of the node domain in the storage class
 		// can satisfy resource requirement of given claim

pkg/controller/volume/persistentvolume/scheduler_binder_test.go

@@ -50,6 +50,7 @@ var (
 	provisionedPVC2             = makeTestPVC("provisioned-pvc2", "1Gi", pvcUnbound, "", "1", &waitClass)
 	provisionedPVCHigherVersion = makeTestPVC("provisioned-pvc2", "1Gi", pvcUnbound, "", "2", &waitClass)
 	noProvisionerPVC            = makeTestPVC("no-provisioner-pvc", "1Gi", pvcUnbound, "", "1", &provisionNotSupportClass)
+	topoMismatchPVC             = makeTestPVC("topo-mismatch-pvc", "1Gi", pvcUnbound, "", "1", &topoMismatchClass)
 
 	pvNoNode                   = makeTestPV("pv-no-node", "", "1G", "1", nil, waitClass)
 	pvNode1a                   = makeTestPV("pv-node1a", "node1", "5G", "1", nil, waitClass)
@@ -74,6 +75,7 @@ var (
 	waitClass                = "waitClass"
 	immediateClass           = "immediateClass"
 	provisionNotSupportClass = "provisionNotSupportedClass"
+	topoMismatchClass        = "topoMismatchClass"
 
 	nodeLabelKey   = "nodeKey"
 	nodeLabelValue = "node1"
@@ -112,6 +114,16 @@ func newTestBinder(t *testing.T) *testEnv {
 			},
 			VolumeBindingMode: &waitMode,
 			Provisioner:       "test-provisioner",
+			AllowedTopologies: []v1.TopologySelectorTerm{
+				{
+					MatchLabelExpressions: []v1.TopologySelectorLabelRequirement{
+						{
+							Key:    nodeLabelKey,
+							Values: []string{nodeLabelValue, "reference-value"},
+						},
+					},
+				},
+			},
 		},
 		{
 			ObjectMeta: metav1.ObjectMeta{
@@ -126,6 +138,23 @@ func newTestBinder(t *testing.T) *testEnv {
 			VolumeBindingMode: &waitMode,
 			Provisioner:       "kubernetes.io/no-provisioner",
 		},
+		{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: topoMismatchClass,
+			},
+			VolumeBindingMode: &waitMode,
+			Provisioner:       "test-provisioner",
+			AllowedTopologies: []v1.TopologySelectorTerm{
+				{
+					MatchLabelExpressions: []v1.TopologySelectorLabelRequirement{
+						{
+							Key:    nodeLabelKey,
+							Values: []string{"reference-value"},
+						},
+					},
+				},
+			},
+		},
 	}
 	for _, class := range classes {
 		if err := classInformer.Informer().GetIndexer().Add(class); err != nil {
@@ -740,6 +769,11 @@ func TestFindPodVolumesWithProvisioning(t *testing.T) {
 			expectedUnbound: false,
 			expectedBound:   true,
 		},
+		"volume-topology-unsatisfied": {
+			podPVCs:         []*v1.PersistentVolumeClaim{topoMismatchPVC},
+			expectedUnbound: false,
+			expectedBound:   true,
+		},
 	}
 
 	// Set VolumeScheduling and DynamicProvisioningScheduling feature gate

pkg/volume/aws_ebs/aws_ebs.go

@@ -491,7 +491,7 @@ type awsElasticBlockStoreProvisioner struct {
 
 var _ volume.Provisioner = &awsElasticBlockStoreProvisioner{}
 
-func (c *awsElasticBlockStoreProvisioner) Provision() (*v1.PersistentVolume, error) {
+func (c *awsElasticBlockStoreProvisioner) Provision(selectedNode *v1.Node, allowedTopologies []v1.TopologySelectorTerm) (*v1.PersistentVolume, error) {
 	if !util.AccessModesContainedInAll(c.plugin.GetAccessModes(), c.options.PVC.Spec.AccessModes) {
 		return nil, fmt.Errorf("invalid AccessModes %v: only AccessModes %v are supported", c.options.PVC.Spec.AccessModes, c.plugin.GetAccessModes())
 	}

pkg/volume/aws_ebs/aws_ebs_test.go

@@ -173,7 +173,7 @@ func TestPlugin(t *testing.T) {
 	if err != nil {
 		t.Errorf("Error creating new provisioner:%v", err)
 	}
-	persistentSpec, err := provisioner.Provision()
+	persistentSpec, err := provisioner.Provision(nil, nil)
 	if err != nil {
 		t.Errorf("Provision() failed: %v", err)
 	}

pkg/volume/azure_dd/azure_provision.go

@@ -67,7 +67,7 @@ func (d *azureDiskDeleter) Delete() error {
 	return diskController.DeleteBlobDisk(volumeSource.DataDiskURI)
 }
 
-func (p *azureDiskProvisioner) Provision() (*v1.PersistentVolume, error) {
+func (p *azureDiskProvisioner) Provision(selectedNode *v1.Node, allowedTopologies []v1.TopologySelectorTerm) (*v1.PersistentVolume, error) {
 	if !util.AccessModesContainedInAll(p.plugin.GetAccessModes(), p.options.PVC.Spec.AccessModes) {
 		return nil, fmt.Errorf("invalid AccessModes %v: only AccessModes %v are supported", p.options.PVC.Spec.AccessModes, p.plugin.GetAccessModes())
 	}

pkg/volume/azure_file/azure_provision.go

@@ -131,7 +131,7 @@ type azureFileProvisioner struct {
 
 var _ volume.Provisioner = &azureFileProvisioner{}
 
-func (a *azureFileProvisioner) Provision() (*v1.PersistentVolume, error) {
+func (a *azureFileProvisioner) Provision(selectedNode *v1.Node, allowedTopologies []v1.TopologySelectorTerm) (*v1.PersistentVolume, error) {
 	if !util.AccessModesContainedInAll(a.plugin.GetAccessModes(), a.options.PVC.Spec.AccessModes) {
 		return nil, fmt.Errorf("invalid AccessModes %v: only AccessModes %v are supported", a.options.PVC.Spec.AccessModes, a.plugin.GetAccessModes())
 	}

pkg/volume/cinder/cinder.go

@@ -500,7 +500,7 @@ type cinderVolumeProvisioner struct {
 
 var _ volume.Provisioner = &cinderVolumeProvisioner{}
 
-func (c *cinderVolumeProvisioner) Provision() (*v1.PersistentVolume, error) {
+func (c *cinderVolumeProvisioner) Provision(selectedNode *v1.Node, allowedTopologies []v1.TopologySelectorTerm) (*v1.PersistentVolume, error) {
 	if !util.AccessModesContainedInAll(c.plugin.GetAccessModes(), c.options.PVC.Spec.AccessModes) {
 		return nil, fmt.Errorf("invalid AccessModes %v: only AccessModes %v are supported", c.options.PVC.Spec.AccessModes, c.plugin.GetAccessModes())
 	}

pkg/volume/cinder/cinder_test.go

@@ -196,7 +196,7 @@ func TestPlugin(t *testing.T) {
 		PersistentVolumeReclaimPolicy: v1.PersistentVolumeReclaimDelete,
 	}
 	provisioner, err := plug.(*cinderPlugin).newProvisionerInternal(options, &fakePDManager{0})
-	persistentSpec, err := provisioner.Provision()
+	persistentSpec, err := provisioner.Provision(nil, nil)
 	if err != nil {
 		t.Errorf("Provision() failed: %v", err)
 	}

pkg/volume/flocker/flocker_volume.go

@@ -54,7 +54,7 @@ type flockerVolumeProvisioner struct {
 
 var _ volume.Provisioner = &flockerVolumeProvisioner{}
 
-func (c *flockerVolumeProvisioner) Provision() (*v1.PersistentVolume, error) {
+func (c *flockerVolumeProvisioner) Provision(selectedNode *v1.Node, allowedTopologies []v1.TopologySelectorTerm) (*v1.PersistentVolume, error) {
 	if !util.AccessModesContainedInAll(c.plugin.GetAccessModes(), c.options.PVC.Spec.AccessModes) {
 		return nil, fmt.Errorf("invalid AccessModes %v: only AccessModes %v are supported", c.options.PVC.Spec.AccessModes, c.plugin.GetAccessModes())
 	}

pkg/volume/flocker/flocker_volume_test.go

@@ -57,7 +57,7 @@ func TestProvision(t *testing.T) {
 	dir, provisioner := newTestableProvisioner(assert, options)
 	defer os.RemoveAll(dir)
 
-	persistentSpec, err := provisioner.Provision()
+	persistentSpec, err := provisioner.Provision(nil, nil)
 	assert.NoError(err, "Provision() failed: ", err)
 
 	cap := persistentSpec.Spec.Capacity[v1.ResourceStorage]
@@ -85,7 +85,7 @@ func TestProvision(t *testing.T) {
 
 	dir, provisioner = newTestableProvisioner(assert, options)
 	defer os.RemoveAll(dir)
-	persistentSpec, err = provisioner.Provision()
+	persistentSpec, err = provisioner.Provision(nil, nil)
 	assert.Error(err, "Provision() did not fail with Parameters specified")
 
 	// selectors are not supported
@@ -97,6 +97,6 @@ func TestProvision(t *testing.T) {
 
 	dir, provisioner = newTestableProvisioner(assert, options)
 	defer os.RemoveAll(dir)
-	persistentSpec, err = provisioner.Provision()
+	persistentSpec, err = provisioner.Provision(nil, nil)
 	assert.Error(err, "Provision() did not fail with Selector specified")
 }

pkg/volume/gce_pd/gce_pd.go

@@ -417,7 +417,7 @@ type gcePersistentDiskProvisioner struct {
 
 var _ volume.Provisioner = &gcePersistentDiskProvisioner{}
 
-func (c *gcePersistentDiskProvisioner) Provision() (*v1.PersistentVolume, error) {
+func (c *gcePersistentDiskProvisioner) Provision(selectedNode *v1.Node, allowedTopologies []v1.TopologySelectorTerm) (*v1.PersistentVolume, error) {
 	if !util.AccessModesContainedInAll(c.plugin.GetAccessModes(), c.options.PVC.Spec.AccessModes) {
 		return nil, fmt.Errorf("invalid AccessModes %v: only AccessModes %v are supported", c.options.PVC.Spec.AccessModes, c.plugin.GetAccessModes())
 	}

pkg/volume/gce_pd/gce_pd_test.go

@@ -167,7 +167,7 @@ func TestPlugin(t *testing.T) {
 	if err != nil {
 		t.Errorf("Error creating new provisioner:%v", err)
 	}
-	persistentSpec, err := provisioner.Provision()
+	persistentSpec, err := provisioner.Provision(nil, nil)
 	if err != nil {
 		t.Errorf("Provision() failed: %v", err)
 	}

pkg/volume/glusterfs/glusterfs.go

@@ -664,7 +664,7 @@ func (d *glusterfsVolumeDeleter) Delete() error {
 	return nil
 }
 
-func (p *glusterfsVolumeProvisioner) Provision() (*v1.PersistentVolume, error) {
+func (p *glusterfsVolumeProvisioner) Provision(selectedNode *v1.Node, allowedTopologies []v1.TopologySelectorTerm) (*v1.PersistentVolume, error) {
 	if !volutil.AccessModesContainedInAll(p.plugin.GetAccessModes(), p.options.PVC.Spec.AccessModes) {
 		return nil, fmt.Errorf("invalid AccessModes %v: only AccessModes %v are supported", p.options.PVC.Spec.AccessModes, p.plugin.GetAccessModes())
 	}

pkg/volume/host_path/host_path.go

@@ -265,7 +265,7 @@ type hostPathProvisioner struct {
 
 // Create for hostPath simply creates a local /tmp/hostpath_pv/%s directory as a new PersistentVolume.
 // This Provisioner is meant for development and testing only and WILL NOT WORK in a multi-node cluster.
-func (r *hostPathProvisioner) Provision() (*v1.PersistentVolume, error) {
+func (r *hostPathProvisioner) Provision(selectedNode *v1.Node, allowedTopologies []v1.TopologySelectorTerm) (*v1.PersistentVolume, error) {
 	if util.CheckPersistentVolumeClaimModeBlock(r.options.PVC) {
 		return nil, fmt.Errorf("%s does not support block volume provisioning", r.plugin.GetPluginName())
 	}

pkg/volume/host_path/host_path_test.go

@@ -178,7 +178,7 @@ func TestProvisioner(t *testing.T) {
 	if err != nil {
 		t.Errorf("Failed to make a new Provisioner: %v", err)
 	}
-	pv, err := creater.Provision()
+	pv, err := creater.Provision(nil, nil)
 	if err != nil {
 		t.Errorf("Unexpected error creating volume: %v", err)
 	}

pkg/volume/photon_pd/photon_pd.go

@@ -340,7 +340,7 @@ func (plugin *photonPersistentDiskPlugin) newProvisionerInternal(options volume.
 	}, nil
 }
 
-func (p *photonPersistentDiskProvisioner) Provision() (*v1.PersistentVolume, error) {
+func (p *photonPersistentDiskProvisioner) Provision(selectedNode *v1.Node, allowedTopologies []v1.TopologySelectorTerm) (*v1.PersistentVolume, error) {
 	if !util.AccessModesContainedInAll(p.plugin.GetAccessModes(), p.options.PVC.Spec.AccessModes) {
 		return nil, fmt.Errorf("invalid AccessModes %v: only AccessModes %v are supported", p.options.PVC.Spec.AccessModes, p.plugin.GetAccessModes())
 	}

pkg/volume/photon_pd/photon_pd_test.go

@@ -166,7 +166,7 @@ func TestPlugin(t *testing.T) {
 	if err != nil {
 		t.Fatalf("Error creating new provisioner:%v", err)
 	}
-	persistentSpec, err := provisioner.Provision()
+	persistentSpec, err := provisioner.Provision(nil, nil)
 	if err != nil {
 		t.Errorf("Provision() failed: %v", err)
 	}

pkg/volume/portworx/portworx.go

@@ -378,7 +378,7 @@ type portworxVolumeProvisioner struct {
 
 var _ volume.Provisioner = &portworxVolumeProvisioner{}
 
-func (c *portworxVolumeProvisioner) Provision() (*v1.PersistentVolume, error) {
+func (c *portworxVolumeProvisioner) Provision(selectedNode *v1.Node, allowedTopologies []v1.TopologySelectorTerm) (*v1.PersistentVolume, error) {
 	if !util.AccessModesContainedInAll(c.plugin.GetAccessModes(), c.options.PVC.Spec.AccessModes) {
 		return nil, fmt.Errorf("invalid AccessModes %v: only AccessModes %v are supported", c.options.PVC.Spec.AccessModes, c.plugin.GetAccessModes())
 	}

pkg/volume/portworx/portworx_test.go

@@ -204,7 +204,7 @@ func TestPlugin(t *testing.T) {
 	if err != nil {
 		t.Errorf("Error creating a new provisioner:%v", err)
 	}
-	persistentSpec, err := provisioner.Provision()
+	persistentSpec, err := provisioner.Provision(nil, nil)
 	if err != nil {
 		t.Errorf("Provision() failed: %v", err)
 	}

pkg/volume/quobyte/quobyte.go

@@ -354,7 +354,7 @@ type quobyteVolumeProvisioner struct {
 	options volume.VolumeOptions
 }
 
-func (provisioner *quobyteVolumeProvisioner) Provision() (*v1.PersistentVolume, error) {
+func (provisioner *quobyteVolumeProvisioner) Provision(selectedNode *v1.Node, allowedTopologies []v1.TopologySelectorTerm) (*v1.PersistentVolume, error) {
 	if !util.AccessModesContainedInAll(provisioner.plugin.GetAccessModes(), provisioner.options.PVC.Spec.AccessModes) {
 		return nil, fmt.Errorf("invalid AccessModes %v: only AccessModes %v are supported", provisioner.options.PVC.Spec.AccessModes, provisioner.plugin.GetAccessModes())
 	}

pkg/volume/rbd/rbd.go

@@ -579,7 +579,7 @@ type rbdVolumeProvisioner struct {
 
 var _ volume.Provisioner = &rbdVolumeProvisioner{}
 
-func (r *rbdVolumeProvisioner) Provision() (*v1.PersistentVolume, error) {
+func (r *rbdVolumeProvisioner) Provision(selectedNode *v1.Node, allowedTopologies []v1.TopologySelectorTerm) (*v1.PersistentVolume, error) {
 	if !volutil.AccessModesContainedInAll(r.plugin.GetAccessModes(), r.options.PVC.Spec.AccessModes) {
 		return nil, fmt.Errorf("invalid AccessModes %v: only AccessModes %v are supported", r.options.PVC.Spec.AccessModes, r.plugin.GetAccessModes())
 	}

pkg/volume/scaleio/sio_volume.go

@@ -252,7 +252,7 @@ func (v *sioVolume) Delete() error {
 // ************************
 var _ volume.Provisioner = &sioVolume{}
 
-func (v *sioVolume) Provision() (*api.PersistentVolume, error) {
+func (v *sioVolume) Provision(selectedNode *api.Node, allowedTopologies []api.TopologySelectorTerm) (*api.PersistentVolume, error) {
 	glog.V(4).Info(log("attempting to dynamically provision pvc %v", v.options.PVC.Name))
 
 	if !util.AccessModesContainedInAll(v.plugin.GetAccessModes(), v.options.PVC.Spec.AccessModes) {