Add DynamicProvisioningScheduling support for EBS

pkg/cloudprovider/providers/aws/aws.go

@@ -415,12 +415,8 @@ const (
 type VolumeOptions struct {
 	CapacityGB       int
 	Tags             map[string]string
-	PVCName           string
 	VolumeType       string
-	ZonePresent       bool
-	ZonesPresent      bool
 	AvailabilityZone string
-	AvailabilityZones string
 	// IOPSPerGB x CapacityGB will give total IOPS of the volume to create.
 	// Calculated total IOPS will be capped at MaxTotalIOPS.
 	IOPSPerGB int
@@ -1431,9 +1427,9 @@ func (c *Cloud) InstanceType(ctx context.Context, nodeName types.NodeName) (stri
 	return aws.StringValue(inst.InstanceType), nil
 }
 
-// getCandidateZonesForDynamicVolume retrieves  a list of all the zones in which nodes are running
+// GetCandidateZonesForDynamicVolume retrieves  a list of all the zones in which nodes are running
 // It currently involves querying all instances
-func (c *Cloud) getCandidateZonesForDynamicVolume() (sets.String, error) {
+func (c *Cloud) GetCandidateZonesForDynamicVolume() (sets.String, error) {
 	// We don't currently cache this; it is currently used only in volume
 	// creation which is expected to be a comparatively rare occurrence.
 
@@ -2157,29 +2153,6 @@ func (c *Cloud) DetachDisk(diskName KubernetesVolumeID, nodeName types.NodeName)
 
 // CreateDisk implements Volumes.CreateDisk
 func (c *Cloud) CreateDisk(volumeOptions *VolumeOptions) (KubernetesVolumeID, error) {
-	var createAZ string
-	if !volumeOptions.ZonePresent && !volumeOptions.ZonesPresent {
-		// querry for candidate zones only if zone parameters absent
-		allZones, err := c.getCandidateZonesForDynamicVolume()
-		if err != nil {
-			return "", fmt.Errorf("error querying for all zones: %v", err)
-		}
-		createAZ = volumeutil.ChooseZoneForVolume(allZones, volumeOptions.PVCName)
-	}
-	if !volumeOptions.ZonePresent && volumeOptions.ZonesPresent {
-		if adminSetOfZones, err := volumeutil.ZonesToSet(volumeOptions.AvailabilityZones); err != nil {
-			return "", err
-		} else {
-			createAZ = volumeutil.ChooseZoneForVolume(adminSetOfZones, volumeOptions.PVCName)
-		}
-	}
-	if volumeOptions.ZonePresent && !volumeOptions.ZonesPresent {
-		if err := volumeutil.ValidateZone(volumeOptions.AvailabilityZone); err != nil {
-			return "", err
-		}
-		createAZ = volumeOptions.AvailabilityZone
-	}
-
 	var createType string
 	var iops int64
 	switch volumeOptions.VolumeType {
@@ -2211,7 +2184,7 @@ func (c *Cloud) CreateDisk(volumeOptions *VolumeOptions) (KubernetesVolumeID, er
 
 	// TODO: Should we tag this with the cluster id (so it gets deleted when the cluster does?)
 	request := &ec2.CreateVolumeInput{}
-	request.AvailabilityZone = aws.String(createAZ)
+	request.AvailabilityZone = aws.String(volumeOptions.AvailabilityZone)
 	request.Size = aws.Int64(int64(volumeOptions.CapacityGB))
 	request.VolumeType = aws.String(createType)
 	request.Encrypted = aws.Bool(volumeOptions.Encrypted)

pkg/volume/aws_ebs/BUILD

@@ -29,6 +29,7 @@ go_library(
         "//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/types:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/util/sets:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/util/feature:go_default_library",
         "//vendor/github.com/golang/glog:go_default_library",
     ],

pkg/volume/aws_ebs/aws_ebs.go

@@ -314,7 +314,7 @@ var _ volume.VolumePluginWithAttachLimits = &awsElasticBlockStorePlugin{}
 
 // Abstract interface to PD operations.
 type ebsManager interface {
-	CreateVolume(provisioner *awsElasticBlockStoreProvisioner) (volumeID aws.KubernetesVolumeID, volumeSizeGB int, labels map[string]string, fstype string, err error)
+	CreateVolume(provisioner *awsElasticBlockStoreProvisioner, node *v1.Node, allowedTopologies []v1.TopologySelectorTerm) (volumeID aws.KubernetesVolumeID, volumeSizeGB int, labels map[string]string, fstype string, err error)
 	// Deletes a volume
 	DeleteVolume(deleter *awsElasticBlockStoreDeleter) error
 }
@@ -504,7 +504,7 @@ func (c *awsElasticBlockStoreProvisioner) Provision(selectedNode *v1.Node, allow
 		return nil, fmt.Errorf("invalid AccessModes %v: only AccessModes %v are supported", c.options.PVC.Spec.AccessModes, c.plugin.GetAccessModes())
 	}
 
-	volumeID, sizeGB, labels, fstype, err := c.manager.CreateVolume(c)
+	volumeID, sizeGB, labels, fstype, err := c.manager.CreateVolume(c, selectedNode, allowedTopologies)
 	if err != nil {
 		glog.Errorf("Provision failed: %v", err)
 		return nil, err
@@ -554,14 +554,23 @@ func (c *awsElasticBlockStoreProvisioner) Provision(selectedNode *v1.Node, allow
 		pv.Spec.AccessModes = c.plugin.GetAccessModes()
 	}
 
+	requirements := make([]v1.NodeSelectorRequirement, 0)
 	if len(labels) != 0 {
 		if pv.Labels == nil {
 			pv.Labels = make(map[string]string)
 		}
 		for k, v := range labels {
 			pv.Labels[k] = v
+			requirements = append(requirements, v1.NodeSelectorRequirement{Key: k, Operator: v1.NodeSelectorOpIn, Values: []string{v}})
 		}
 	}
 
+	if utilfeature.DefaultFeatureGate.Enabled(features.VolumeScheduling) {
+		pv.Spec.NodeAffinity = new(v1.VolumeNodeAffinity)
+		pv.Spec.NodeAffinity.Required = new(v1.NodeSelector)
+		pv.Spec.NodeAffinity.Required.NodeSelectorTerms = make([]v1.NodeSelectorTerm, 1)
+		pv.Spec.NodeAffinity.Required.NodeSelectorTerms[0].MatchExpressions = requirements
+	}
+
 	return pv, nil
 }

pkg/volume/aws_ebs/aws_ebs_test.go

@@ -84,7 +84,7 @@ type fakePDManager struct {
 
 // TODO(jonesdl) To fully test this, we could create a loopback device
 // and mount that instead.
-func (fake *fakePDManager) CreateVolume(c *awsElasticBlockStoreProvisioner) (volumeID aws.KubernetesVolumeID, volumeSizeGB int, labels map[string]string, fstype string, err error) {
+func (fake *fakePDManager) CreateVolume(c *awsElasticBlockStoreProvisioner, node *v1.Node, allowedTopologies []v1.TopologySelectorTerm) (volumeID aws.KubernetesVolumeID, volumeSizeGB int, labels map[string]string, fstype string, err error) {
 	labels = make(map[string]string)
 	labels["fakepdmanager"] = "yes"
 	return "test-aws-volume-name", 100, labels, "", nil
@@ -173,6 +173,7 @@ func TestPlugin(t *testing.T) {
 	if err != nil {
 		t.Errorf("Error creating new provisioner:%v", err)
 	}
+
 	persistentSpec, err := provisioner.Provision(nil, nil)
 	if err != nil {
 		t.Errorf("Provision() failed: %v", err)
@@ -191,6 +192,37 @@ func TestPlugin(t *testing.T) {
 		t.Errorf("Provision() returned unexpected labels: %v", persistentSpec.Labels)
 	}
 
+	// check nodeaffinity members
+	if persistentSpec.Spec.NodeAffinity == nil {
+		t.Errorf("Provision() returned unexpected nil NodeAffinity")
+	}
+
+	if persistentSpec.Spec.NodeAffinity.Required == nil {
+		t.Errorf("Provision() returned unexpected nil NodeAffinity.Required")
+	}
+
+	n := len(persistentSpec.Spec.NodeAffinity.Required.NodeSelectorTerms)
+	if n != 1 {
+		t.Errorf("Provision() returned unexpected number of NodeSelectorTerms %d. Expected %d", n, 1)
+	}
+
+	n = len(persistentSpec.Spec.NodeAffinity.Required.NodeSelectorTerms[0].MatchExpressions)
+	if n != 1 {
+		t.Errorf("Provision() returned unexpected number of MatchExpressions %d. Expected %d", n, 1)
+	}
+
+	req := persistentSpec.Spec.NodeAffinity.Required.NodeSelectorTerms[0].MatchExpressions[0]
+	if req.Key != "fakepdmanager" {
+		t.Errorf("Provision() returned unexpected requirement key in NodeAffinity %v", req.Key)
+	}
+
+	if req.Operator != v1.NodeSelectorOpIn {
+		t.Errorf("Provision() returned unexpected requirement operator in NodeAffinity %v", req.Operator)
+	}
+
+	if len(req.Values) != 1 || req.Values[0] != "yes" {
+		t.Errorf("Provision() returned unexpected requirement value in NodeAffinity %v", req.Values)
+	}
 	// Test Deleter
 	volSpec := &volume.Spec{
 		PersistentVolume: persistentSpec,

pkg/volume/aws_ebs/aws_util.go

@@ -26,6 +26,8 @@ import (
 
 	"github.com/golang/glog"
 	"k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
+	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/kubernetes/pkg/cloudprovider"
 	"k8s.io/kubernetes/pkg/cloudprovider/providers/aws"
 	"k8s.io/kubernetes/pkg/volume"
@@ -40,6 +42,7 @@ const (
 	maxRetries          = 10
 	checkSleepDuration  = time.Second
 	errorSleepDuration  = 5 * time.Second
+	ebsMaxReplicasInAZ  = 1
 )
 
 type AWSDiskUtil struct{}
@@ -67,7 +70,7 @@ func (util *AWSDiskUtil) DeleteVolume(d *awsElasticBlockStoreDeleter) error {
 
 // CreateVolume creates an AWS EBS volume.
 // Returns: volumeID, volumeSizeGB, labels, error
-func (util *AWSDiskUtil) CreateVolume(c *awsElasticBlockStoreProvisioner) (aws.KubernetesVolumeID, int, map[string]string, string, error) {
+func (util *AWSDiskUtil) CreateVolume(c *awsElasticBlockStoreProvisioner, node *v1.Node, allowedTopologies []v1.TopologySelectorTerm) (aws.KubernetesVolumeID, int, map[string]string, string, error) {
 	cloud, err := getCloudProvider(c.awsElasticBlockStore.plugin.host.GetCloudProvider())
 	if err != nil {
 		return "", 0, nil, "", err
@@ -83,73 +86,97 @@ func (util *AWSDiskUtil) CreateVolume(c *awsElasticBlockStoreProvisioner) (aws.K
 	tags["Name"] = volumeutil.GenerateVolumeName(c.options.ClusterName, c.options.PVName, 255) // AWS tags can have 255 characters
 
 	capacity := c.options.PVC.Spec.Resources.Requests[v1.ResourceName(v1.ResourceStorage)]
-	// AWS works with gigabytes, convert to GiB with rounding up
-	requestGiB, err := volumeutil.RoundUpToGiBInt(capacity)
+
+	zonesWithNodes, err := cloud.GetCandidateZonesForDynamicVolume()
+	if err != nil {
+		return "", 0, nil, "", fmt.Errorf("error querying for all zones: %v", err)
+	}
+
+	volumeOptions, err := populateVolumeOptions(c.plugin.GetPluginName(), c.options.PVC.Name, capacity, tags, c.options.Parameters, node, allowedTopologies, zonesWithNodes)
+	if err != nil {
+		glog.V(2).Infof("Error populating EBS options: %v", err)
+		return "", 0, nil, "", err
+	}
+
+	// TODO: implement PVC.Selector parsing
+	if c.options.PVC.Spec.Selector != nil {
+		return "", 0, nil, "", fmt.Errorf("claim.Spec.Selector is not supported for dynamic provisioning on AWS")
+	}
+
+	name, err := cloud.CreateDisk(volumeOptions)
 	if err != nil {
+		glog.V(2).Infof("Error creating EBS Disk volume: %v", err)
 		return "", 0, nil, "", err
 	}
+	glog.V(2).Infof("Successfully created EBS Disk volume %s", name)
+
+	labels, err := cloud.GetVolumeLabels(name)
+	if err != nil {
+		// We don't really want to leak the volume here...
+		glog.Errorf("error building labels for new EBS volume %q: %v", name, err)
+	}
+
+	fstype := ""
+	if v, ok := c.options.Parameters[volume.VolumeParameterFSType]; ok {
+		fstype = v
+	}
+
+	return name, volumeOptions.CapacityGB, labels, fstype, nil
+}
+
+// returns volumeOptions for EBS based on storageclass parameters and node configuration
+func populateVolumeOptions(pluginName, pvcName string, capacityGB resource.Quantity, tags map[string]string, storageParams map[string]string, node *v1.Node, allowedTopologies []v1.TopologySelectorTerm, zonesWithNodes sets.String) (*aws.VolumeOptions, error) {
+	requestGiB, err := volumeutil.RoundUpToGiBInt(capacityGB)
+	if err != nil {
+		return nil, err
+	}
+
 	volumeOptions := &aws.VolumeOptions{
 		CapacityGB: requestGiB,
 		Tags:       tags,
-		PVCName:    c.options.PVC.Name,
 	}
-	fstype := ""
+
 	// Apply Parameters (case-insensitive). We leave validation of
 	// the values to the cloud provider.
-	volumeOptions.ZonePresent = false
-	volumeOptions.ZonesPresent = false
-	for k, v := range c.options.Parameters {
+	zonePresent := false
+	zonesPresent := false
+	var zone string
+	var zones sets.String
+	for k, v := range storageParams {
 		switch strings.ToLower(k) {
 		case "type":
 			volumeOptions.VolumeType = v
 		case "zone":
-			volumeOptions.ZonePresent = true
-			volumeOptions.AvailabilityZone = v
+			zonePresent = true
+			zone = v
 		case "zones":
-			volumeOptions.ZonesPresent = true
-			volumeOptions.AvailabilityZones = v
+			zonesPresent = true
+			zones, err = volumeutil.ZonesToSet(v)
+			if err != nil {
+				return nil, fmt.Errorf("error parsing zones %s, must be strings separated by commas: %v", zones, err)
+			}
 		case "iopspergb":
 			volumeOptions.IOPSPerGB, err = strconv.Atoi(v)
 			if err != nil {
-				return "", 0, nil, "", fmt.Errorf("invalid iopsPerGB value %q, must be integer between 1 and 30: %v", v, err)
+				return nil, fmt.Errorf("invalid iopsPerGB value %q, must be integer between 1 and 30: %v", v, err)
 			}
 		case "encrypted":
 			volumeOptions.Encrypted, err = strconv.ParseBool(v)
 			if err != nil {
-				return "", 0, nil, "", fmt.Errorf("invalid encrypted boolean value %q, must be true or false: %v", v, err)
+				return nil, fmt.Errorf("invalid encrypted boolean value %q, must be true or false: %v", v, err)
 			}
 		case "kmskeyid":
 			volumeOptions.KmsKeyId = v
-		case volume.VolumeParameterFSType:
-			fstype = v
 		default:
-			return "", 0, nil, "", fmt.Errorf("invalid option %q for volume plugin %s", k, c.plugin.GetPluginName())
+			return nil, fmt.Errorf("invalid option %q for volume plugin %s", k, pluginName)
 		}
 	}
 
-	if volumeOptions.ZonePresent && volumeOptions.ZonesPresent {
-		return "", 0, nil, "", fmt.Errorf("both zone and zones StorageClass parameters must not be used at the same time")
-	}
-
-	// TODO: implement PVC.Selector parsing
-	if c.options.PVC.Spec.Selector != nil {
-		return "", 0, nil, "", fmt.Errorf("claim.Spec.Selector is not supported for dynamic provisioning on AWS")
-	}
-
-	name, err := cloud.CreateDisk(volumeOptions)
+	volumeOptions.AvailabilityZone, err = volumeutil.SelectZoneForVolume(zonePresent, zonesPresent, zone, zones, zonesWithNodes, node, allowedTopologies, pvcName)
 	if err != nil {
-		glog.V(2).Infof("Error creating EBS Disk volume: %v", err)
-		return "", 0, nil, "", err
+		return nil, err
 	}
-	glog.V(2).Infof("Successfully created EBS Disk volume %s", name)
-
-	labels, err := cloud.GetVolumeLabels(name)
-	if err != nil {
-		// We don't really want to leak the volume here...
-		glog.Errorf("error building labels for new EBS volume %q: %v", name, err)
-	}
-
-	return name, requestGiB, labels, fstype, nil
+	return volumeOptions, nil
 }
 
 // Returns the first path that exists, or empty string if none exist.

pkg/volume/util/BUILD

@@ -56,6 +56,7 @@ go_test(
     embed = [":go_default_library"],
     deps = [
         "//pkg/apis/core/install:go_default_library",
+        "//pkg/kubelet/apis:go_default_library",
         "//pkg/util/mount:go_default_library",
         "//pkg/util/slice:go_default_library",
         "//pkg/volume:go_default_library",
@@ -64,6 +65,7 @@ go_test(
         "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/sets:go_default_library",
+        "//staging/src/k8s.io/apiserver/pkg/util/feature:go_default_library",
         "//staging/src/k8s.io/client-go/util/testing:go_default_library",
     ],
 )

pkg/volume/util/util.go

@@ -79,6 +79,16 @@ const (
 	VolumeDynamicallyCreatedByKey = "kubernetes.io/createdby"
 )
 
+// VolumeZoneConfig contains config information about zonal volume.
+type VolumeZoneConfig struct {
+	ZonePresent                bool
+	ZonesPresent               bool
+	ReplicaZoneFromNodePresent bool
+	Zone                       string
+	Zones                      string
+	ReplicaZoneFromNode        string
+}
+
 // IsReady checks for the existence of a regular file
 // called 'ready' in the given directory and returns
 // true if that file exists.
@@ -308,6 +318,77 @@ func LoadPodFromFile(filePath string) (*v1.Pod, error) {
 	return pod, nil
 }
 
+// SelectZone selects a zone for a volume based on several factors:
+// node.zone, allowedTopologies, zone/zones parameters from storageclass
+// and zones with active nodes from the cluster
+func SelectZoneForVolume(zoneParameterPresent, zonesParameterPresent bool, zoneParameter string, zonesParameter, zonesWithNodes sets.String, node *v1.Node, allowedTopologies []v1.TopologySelectorTerm, pvcName string) (string, error) {
+	if zoneParameterPresent && zonesParameterPresent {
+		return "", fmt.Errorf("both zone and zones StorageClass parameters must not be used at the same time")
+	}
+
+	// pick zone from node if present
+	if node != nil {
+		// DynamicProvisioningScheduling implicit since node is not nil
+		if zoneParameterPresent || zonesParameterPresent {
+			return "", fmt.Errorf("zone[s] cannot be specified in StorageClass if VolumeBindingMode is set to WaitForFirstConsumer. Please specify allowedTopologies in StorageClass for constraining zones.")
+		}
+
+		// pick node's zone and ignore any allowedTopologies (since scheduler is assumed to have accounted for it already)
+		z, ok := node.ObjectMeta.Labels[kubeletapis.LabelZoneFailureDomain]
+		if !ok {
+			return "", fmt.Errorf("%s Label for node missing", kubeletapis.LabelZoneFailureDomain)
+		}
+		return z, nil
+	}
+
+	// pick zone from allowedZones if specified
+	allowedZones, err := ZonesFromAllowedTopologies(allowedTopologies)
+	if err != nil {
+		return "", err
+	}
+
+	if (len(allowedTopologies) > 0) && (allowedZones.Len() == 0) {
+		return "", fmt.Errorf("no matchLabelExpressions with %s key found in allowedTopologies. Please specify matchLabelExpressions with %s key", kubeletapis.LabelZoneFailureDomain, kubeletapis.LabelZoneFailureDomain)
+	}
+
+	if allowedZones.Len() > 0 {
+		// DynamicProvisioningScheduling implicit since allowedZones present
+		if zoneParameterPresent || zonesParameterPresent {
+			return "", fmt.Errorf("zone[s] cannot be specified in StorageClass if allowedTopologies specified")
+		}
+		return ChooseZoneForVolume(*allowedZones, pvcName), nil
+	}
+
+	// pick zone from parameters if present
+	if zoneParameterPresent {
+		return zoneParameter, nil
+	}
+
+	if zonesParameterPresent {
+		return ChooseZoneForVolume(zonesParameter, pvcName), nil
+	}
+
+	// pick zone from zones with nodes
+	return ChooseZoneForVolume(zonesWithNodes, pvcName), nil
+}
+
+// ZonesFromAllowedTopologies returns a list of zones specified in allowedTopologies
+func ZonesFromAllowedTopologies(allowedTopologies []v1.TopologySelectorTerm) (*sets.String, error) {
+	zones := make(sets.String)
+	for _, term := range allowedTopologies {
+		for _, exp := range term.MatchLabelExpressions {
+			if exp.Key == kubeletapis.LabelZoneFailureDomain {
+				for _, value := range exp.Values {
+					zones.Insert(value)
+				}
+			} else {
+				return nil, fmt.Errorf("unsupported key found in matchLabelExpressions: %s", exp.Key)
+			}
+		}
+	}
+	return &zones, nil
+}
+
 func ZonesSetToLabelValue(strSet sets.String) string {
 	return strings.Join(strSet.UnsortedList(), kubeletapis.LabelMultiZoneDelimiter)
 }