Merge pull request #67121 from feiskyer/azdisk-affinity

pkg/cloudprovider/providers/azure/azure_managedDiskController.go

@@ -29,7 +29,6 @@ import (
 
 	"k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
-	"k8s.io/apimachinery/pkg/util/sets"
 	kwait "k8s.io/apimachinery/pkg/util/wait"
 	kubeletapis "k8s.io/kubernetes/pkg/kubelet/apis"
 	"k8s.io/kubernetes/pkg/volume"
@@ -51,16 +50,8 @@ type ManagedDiskOptions struct {
 	PVCName string
 	// The name of resource group.
 	ResourceGroup string
-	// Wether the disk is zoned.
-	Zoned bool
-	// Wether AvailabilityZone is set.
-	ZonePresent bool
-	// Wether AvailabilityZones is set.
-	ZonesPresent bool
 	// The AvailabilityZone to create the disk.
 	AvailabilityZone string
-	// List of AvailabilityZone to create the disk.
-	AvailabilityZones string
 	// The tags of the disk.
 	Tags map[string]string
 	// The SKU of storage account.
@@ -73,44 +64,12 @@ func newManagedDiskController(common *controllerCommon) (*ManagedDiskController,
 
 //CreateManagedDisk : create managed disk
 func (c *ManagedDiskController) CreateManagedDisk(options *ManagedDiskOptions) (string, error) {
-	var zones sets.String
-	var activeZones sets.String
 	var err error
 	glog.V(4).Infof("azureDisk - creating new managed Name:%s StorageAccountType:%s Size:%v", options.DiskName, options.StorageAccountType, options.SizeGB)
 
-	// Get active zones which have nodes running on.
-	activeZones, err = c.common.cloud.GetActiveZones()
-	if err != nil {
-		return "", fmt.Errorf("error querying active zones: %v", err)
-	}
-
-	// Validate and choose availability zone for creating disk.
-	if options.Zoned && !options.ZonePresent && !options.ZonesPresent {
-		// Neither "zone" or "zones" specified. Pick a zone randomly selected
-		// from all active zones where Kubernetes cluster has a node.
-		zones = activeZones
-	} else if !options.ZonePresent && options.ZonesPresent {
-		// Choose zone from specified zones.
-		if zones, err = util.ZonesToSet(options.AvailabilityZones); err != nil {
-			return "", err
-		}
-	} else if options.ZonePresent && !options.ZonesPresent {
-		if err := util.ValidateZone(options.AvailabilityZone); err != nil {
-			return "", err
-		}
-		zones = make(sets.String)
-		zones.Insert(options.AvailabilityZone)
-	}
 	var createZones *[]string
-	if len(zones.List()) > 0 {
-		createAZ := util.ChooseZoneForVolume(zones, options.PVCName)
-		// Do not allow creation of disks in zones that are do not have nodes. Such disks
-		// are not currently usable.
-		if !activeZones.Has(createAZ) {
-			return "", fmt.Errorf("kubernetes does not have a node in zone %q", createAZ)
-		}
-
-		zoneList := []string{c.common.cloud.GetZoneID(createAZ)}
+	if len(options.AvailabilityZone) > 0 {
+		zoneList := []string{c.common.cloud.GetZoneID(options.AvailabilityZone)}
 		createZones = &zoneList
 	}
 

pkg/volume/azure_dd/azure_dd.go

@@ -22,9 +22,11 @@ import (
 	"github.com/Azure/azure-sdk-for-go/services/compute/mgmt/2018-04-01/compute"
 	"github.com/Azure/azure-sdk-for-go/services/storage/mgmt/2017-10-01/storage"
 	"github.com/golang/glog"
+
 	"k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/kubernetes/pkg/cloudprovider/providers/azure"
 	"k8s.io/kubernetes/pkg/volume"
 	"k8s.io/kubernetes/pkg/volume/util"
@@ -61,6 +63,9 @@ type DiskController interface {
 
 	// GetAzureDiskLabels gets availability zone labels for Azuredisk.
 	GetAzureDiskLabels(diskURI string) (map[string]string, error)
+
+	// GetActiveZones returns all the zones in which k8s nodes are currently running.
+	GetActiveZones() (sets.String, error)
 }
 
 type azureDataDiskPlugin struct {

pkg/volume/azure_dd/azure_provision.go

@@ -25,6 +25,7 @@ import (
 	"k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/sets"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/kubernetes/pkg/cloudprovider/providers/azure"
 	"k8s.io/kubernetes/pkg/features"
@@ -118,12 +119,13 @@ func (p *azureDiskProvisioner) Provision(selectedNode *v1.Node, allowedTopologie
 		err                        error
 		resourceGroup              string
 
-		zoned             bool
-		zonePresent       bool
-		zonesPresent      bool
-		strZoned          string
-		availabilityZone  string
-		availabilityZones string
+		zoned                    bool
+		zonePresent              bool
+		zonesPresent             bool
+		strZoned                 string
+		availabilityZone         string
+		availabilityZones        sets.String
+		selectedAvailabilityZone string
 	)
 	// maxLength = 79 - (4 for ".vhd") = 75
 	name := util.GenerateVolumeName(p.options.ClusterName, p.options.PVName, 75)
@@ -156,7 +158,10 @@ func (p *azureDiskProvisioner) Provision(selectedNode *v1.Node, allowedTopologie
 			availabilityZone = v
 		case "zones":
 			zonesPresent = true
-			availabilityZones = v
+			availabilityZones, err = util.ZonesToSet(v)
+			if err != nil {
+				return nil, fmt.Errorf("error parsing zones %s, must be strings separated by commas: %v", v, err)
+			}
 		case "zoned":
 			strZoned = v
 		default:
@@ -175,6 +180,16 @@ func (p *azureDiskProvisioner) Provision(selectedNode *v1.Node, allowedTopologie
 		return nil, err
 	}
 
+	if kind != v1.AzureManagedDisk {
+		if resourceGroup != "" {
+			return nil, errors.New("StorageClass option 'resourceGroup' can be used only for managed disks")
+		}
+
+		if zoned {
+			return nil, errors.New("StorageClass option 'zoned' parameter is only supported for managed disks")
+		}
+	}
+
 	zoned, err = parseZoned(strZoned, kind)
 	if err != nil {
 		return nil, err
@@ -184,10 +199,6 @@ func (p *azureDiskProvisioner) Provision(selectedNode *v1.Node, allowedTopologie
 		return nil, fmt.Errorf("zone or zones StorageClass parameters must be used together with zoned parameter")
 	}
 
-	if zonePresent && zonesPresent {
-		return nil, fmt.Errorf("both zone and zones StorageClass parameters must not be used at the same time")
-	}
-
 	if cachingMode, err = normalizeCachingMode(cachingMode); err != nil {
 		return nil, err
 	}
@@ -197,8 +208,17 @@ func (p *azureDiskProvisioner) Provision(selectedNode *v1.Node, allowedTopologie
 		return nil, err
 	}
 
-	if resourceGroup != "" && kind != v1.AzureManagedDisk {
-		return nil, errors.New("StorageClass option 'resourceGroup' can be used only for managed disks")
+	// Select zone for managed disks based on zone, zones and allowedTopologies.
+	if zoned {
+		activeZones, err := diskController.GetActiveZones()
+		if err != nil {
+			return nil, fmt.Errorf("error querying active zones: %v", err)
+		}
+
+		selectedAvailabilityZone, err = util.SelectZoneForVolume(zonePresent, zonesPresent, availabilityZone, availabilityZones, activeZones, selectedNode, allowedTopologies, p.options.PVC.Name)
+		if err != nil {
+			return nil, err
+		}
 	}
 
 	// create disk
@@ -217,11 +237,7 @@ func (p *azureDiskProvisioner) Provision(selectedNode *v1.Node, allowedTopologie
 			PVCName:            p.options.PVC.Name,
 			SizeGB:             requestGiB,
 			Tags:               tags,
-			Zoned:              zoned,
-			ZonePresent:        zonePresent,
-			ZonesPresent:       zonesPresent,
-			AvailabilityZone:   availabilityZone,
-			AvailabilityZones:  availabilityZones,
+			AvailabilityZone:   selectedAvailabilityZone,
 		}
 		diskURI, err = diskController.CreateManagedDisk(volumeOptions)
 		if err != nil {
@@ -232,10 +248,6 @@ func (p *azureDiskProvisioner) Provision(selectedNode *v1.Node, allowedTopologie
 			return nil, err
 		}
 	} else {
-		if zoned {
-			return nil, errors.New("zoned parameter is only supported for managed disks")
-		}
-
 		if kind == v1.AzureDedicatedBlobDisk {
 			_, diskURI, _, err = diskController.CreateVolume(name, account, storageAccountType, location, requestGiB)
 			if err != nil {
@@ -286,5 +298,21 @@ func (p *azureDiskProvisioner) Provision(selectedNode *v1.Node, allowedTopologie
 		},
 	}
 
+	if zoned && utilfeature.DefaultFeatureGate.Enabled(features.VolumeScheduling) {
+		requirements := make([]v1.NodeSelectorRequirement, 0)
+		for k, v := range labels {
+			requirements = append(requirements, v1.NodeSelectorRequirement{Key: k, Operator: v1.NodeSelectorOpIn, Values: []string{v}})
+		}
+
+		nodeSelectorTerm := v1.NodeSelectorTerm{
+			MatchExpressions: requirements,
+		}
+		pv.Spec.NodeAffinity = &v1.VolumeNodeAffinity{
+			Required: &v1.NodeSelector{
+				NodeSelectorTerms: []v1.NodeSelectorTerm{nodeSelectorTerm},
+			},
+		}
+	}
+
 	return pv, nil
 }