Skip to content

K
k3s
Commit a2de7d2d authored by Deep Debroy's avatar Deep Debroy
Browse files
Options

Add DynamicProvisioningScheduling support for GCE PD and RePD

Signed-off-by: 's avatarDeep Debroy <ddebroy@docker.com>
parent 99c17d89
Expand all Hide whitespace changes
Inline Side-by-side
Showing with 202 additions and 124 deletions
pkg/volume/gce_pd/gce_pd.go
View file @ a2de7d2d
......@@ -32,6 +32,7 @@ import (
utilfeature "k8s.io/apiserver/pkg/util/feature"
gcecloud "k8s.io/kubernetes/pkg/cloudprovider/providers/gce"
"k8s.io/kubernetes/pkg/features"
kubeletapis "k8s.io/kubernetes/pkg/kubelet/apis"
"k8s.io/kubernetes/pkg/util/mount"
kstrings "k8s.io/kubernetes/pkg/util/strings"
"k8s.io/kubernetes/pkg/volume"
......@@ -303,7 +304,7 @@ func (plugin *gcePersistentDiskPlugin) ConstructVolumeSpec(volumeName, mountPath
// Abstract interface to PD operations.
type pdManager interface {
// Creates a volume
CreateVolume(provisioner *gcePersistentDiskProvisioner) (volumeID string, volumeSizeGB int, labels map[string]string, fstype string, err error)
CreateVolume(provisioner *gcePersistentDiskProvisioner, node *v1.Node, allowedTopologies []v1.TopologySelectorTerm) (volumeID string, volumeSizeGB int, labels map[string]string, fstype string, err error)
// Deletes a volume
DeleteVolume(deleter *gcePersistentDiskDeleter) error
}
......@@ -471,7 +472,7 @@ func (c *gcePersistentDiskProvisioner) Provision(selectedNode *v1.Node, allowedT
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 {
return nil, err
}
......@@ -519,14 +520,32 @@ func (c *gcePersistentDiskProvisioner) Provision(selectedNode *v1.Node, allowedT
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
var values []string
if k == kubeletapis.LabelZoneFailureDomain {
values, err = util.LabelZonesToList(v)
if err != nil {
return nil, fmt.Errorf("failed to convert label string for Zone: %s to a List: %v", v, err)
}
} else {
values = []string{v}
}
requirements = append(requirements, v1.NodeSelectorRequirement{Key: k, Operator: v1.NodeSelectorOpIn, Values: values})
}
}
if utilfeature.DefaultFeatureGate.Enabled(features.VolumeScheduling) && len(requirements) > 0 {
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/gce_pd/gce_pd_test.go
View file @ a2de7d2d
......@@ -20,6 +20,8 @@ import (
"fmt"
"os"
"path"
"reflect"
"sort"
"testing"
"k8s.io/api/core/v1"
......@@ -27,10 +29,12 @@ import (
"k8s.io/apimachinery/pkg/types"
"k8s.io/client-go/kubernetes/fake"
utiltesting "k8s.io/client-go/util/testing"
kubeletapis "k8s.io/kubernetes/pkg/kubelet/apis"
"k8s.io/kubernetes/pkg/util/mount"
"k8s.io/kubernetes/pkg/volume"
volumetest "k8s.io/kubernetes/pkg/volume/testing"
"k8s.io/kubernetes/pkg/volume/util"
volumeutil "k8s.io/kubernetes/pkg/volume/util"
)
func TestCanSupport(t *testing.T) {
......@@ -78,9 +82,10 @@ func TestGetAccessModes(t *testing.T) {
type fakePDManager struct {
}
func (fake *fakePDManager) CreateVolume(c *gcePersistentDiskProvisioner) (volumeID string, volumeSizeGB int, labels map[string]string, fstype string, err error) {
func (fake *fakePDManager) CreateVolume(c *gcePersistentDiskProvisioner, node *v1.Node, allowedTopologies []v1.TopologySelectorTerm) (volumeID string, volumeSizeGB int, labels map[string]string, fstype string, err error) {
labels = make(map[string]string)
labels["fakepdmanager"] = "yes"
labels[kubeletapis.LabelZoneFailureDomain] = "zone1__zone2"
return "test-gce-volume-name", 100, labels, "", nil
}
......@@ -91,6 +96,15 @@ func (fake *fakePDManager) DeleteVolume(cd *gcePersistentDiskDeleter) error {
return nil
}
func getNodeSelectorRequirementWithKey(key string, term v1.NodeSelectorTerm) (*v1.NodeSelectorRequirement, error) {
for _, r := range term.MatchExpressions {
if r.Key == key {
return &r, nil
}
}
return nil, fmt.Errorf("key %s not found", key)
}
func TestPlugin(t *testing.T) {
tmpDir, err := utiltesting.MkTmpdir("gcepdTest")
if err != nil {
......@@ -182,7 +196,35 @@ func TestPlugin(t *testing.T) {
}
if persistentSpec.Labels["fakepdmanager"] != "yes" {
t.Errorf("Provision() returned unexpected labels: %v", persistentSpec.Labels)
t.Errorf("Provision() returned unexpected value for fakepdmanager: %v", persistentSpec.Labels["fakepdmanager"])
}
if persistentSpec.Labels[kubeletapis.LabelZoneFailureDomain] != "zone1__zone2" {
t.Errorf("Provision() returned unexpected value for %s: %v", kubeletapis.LabelZoneFailureDomain, persistentSpec.Labels[kubeletapis.LabelZoneFailureDomain])
}
if persistentSpec.Spec.NodeAffinity == nil {
t.Errorf("Unexpected nil NodeAffinity found")
}
if len(persistentSpec.Spec.NodeAffinity.Required.NodeSelectorTerms) != 1 {
t.Errorf("Unexpected number of NodeSelectorTerms")
}
term := persistentSpec.Spec.NodeAffinity.Required.NodeSelectorTerms[0]
if len(term.MatchExpressions) != 2 {
t.Errorf("Unexpected number of NodeSelectorRequirements in volume NodeAffinity: %d", len(term.MatchExpressions))
}
r, _ := getNodeSelectorRequirementWithKey("fakepdmanager", term)
if r == nil || r.Values[0] != "yes" || r.Operator != v1.NodeSelectorOpIn {
t.Errorf("NodeSelectorRequirement fakepdmanager-in-yes not found in volume NodeAffinity")
}
zones, _ := volumeutil.ZonesToSet("zone1,zone2")
r, _ = getNodeSelectorRequirementWithKey(kubeletapis.LabelZoneFailureDomain, term)
if r == nil {
t.Errorf("NodeSelectorRequirement %s-in-%v not found in volume NodeAffinity", kubeletapis.LabelZoneFailureDomain, zones)
}
sort.Strings(r.Values)
if !reflect.DeepEqual(r.Values, zones.List()) {
t.Errorf("ZoneFailureDomain elements %v does not match zone labels %v", r.Values, zones)
}
// Test Deleter
......
pkg/volume/gce_pd/gce_util.go
View file @ a2de7d2d
......@@ -87,7 +87,7 @@ func (util *GCEDiskUtil) DeleteVolume(d *gcePersistentDiskDeleter) error {
// CreateVolume creates a GCE PD.
// Returns: gcePDName, volumeSizeGB, labels, fsType, error
func (gceutil *GCEDiskUtil) CreateVolume(c *gcePersistentDiskProvisioner) (string, int, map[string]string, string, error) {
func (gceutil *GCEDiskUtil) CreateVolume(c *gcePersistentDiskProvisioner, node *v1.Node, allowedTopologies []v1.TopologySelectorTerm) (string, int, map[string]string, string, error) {
cloud, err := getCloudProvider(c.gcePersistentDisk.plugin.host.GetCloudProvider())
if err != nil {
return "", 0, nil, "", err
......@@ -104,7 +104,7 @@ func (gceutil *GCEDiskUtil) CreateVolume(c *gcePersistentDiskProvisioner) (strin
// to the cloud provider.
diskType := ""
configuredZone := ""
configuredZones := ""
var configuredZones sets.String
zonePresent := false
zonesPresent := false
replicationType := replicationTypeNone
......@@ -118,7 +118,10 @@ func (gceutil *GCEDiskUtil) CreateVolume(c *gcePersistentDiskProvisioner) (strin
configuredZone = v
case "zones":
zonesPresent = true
configuredZones = v
configuredZones, err = volumeutil.ZonesToSet(v)
if err != nil {
return "", 0, nil, "", err
}
case "replication-type":
if !utilfeature.DefaultFeatureGate.Enabled(features.GCERegionalPersistentDisk) {
return "", 0, nil, "",
......@@ -133,76 +136,49 @@ func (gceutil *GCEDiskUtil) CreateVolume(c *gcePersistentDiskProvisioner) (strin
}
}
if zonePresent && zonesPresent {
return "", 0, nil, "", fmt.Errorf("the 'zone' and 'zones' StorageClass parameters must not be used at the same time")
}
if replicationType == replicationTypeRegionalPD && zonePresent {
// If a user accidentally types 'zone' instead of 'zones', we want to throw an error
// instead of assuming that 'zones' is empty and proceed by randomly selecting zones.
return "", 0, nil, "", fmt.Errorf("the '%s' replication type does not support the 'zone' parameter; use 'zones' instead", replicationTypeRegionalPD)
}
// 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 GCE")
}
var activezones sets.String
activezones, err = cloud.GetAllCurrentZones()
if err != nil {
return "", 0, nil, "", err
}
switch replicationType {
case replicationTypeRegionalPD:
err = createRegionalPD(
selectedZones, err := volumeutil.SelectZonesForVolume(zonePresent, zonesPresent, configuredZone, configuredZones, activezones, node, allowedTopologies, c.options.PVC.Name, maxRegionalPDZones)
if err != nil {
glog.V(2).Infof("Error selecting zones for regional GCE PD volume: %v", err)
return "", 0, nil, "", err
}
if err = cloud.CreateRegionalDisk(
name,
c.options.PVC.Name,
diskType,
configuredZones,
requestGB,
c.options.CloudTags,
cloud)
if err != nil {
selectedZones,
int64(requestGB),
*c.options.CloudTags); err != nil {
glog.V(2).Infof("Error creating regional GCE PD volume: %v", err)
return "", 0, nil, "", err
}
glog.V(2).Infof("Successfully created Regional GCE PD volume %s", name)
case replicationTypeNone:
var zones sets.String
if !zonePresent && !zonesPresent {
// 00 - neither "zone" or "zones" specified
// Pick a zone randomly selected from all active zones where
// Kubernetes cluster has a node.
zones, err = cloud.GetAllCurrentZones()
if err != nil {
glog.V(2).Infof("error getting zone information from GCE: %v", err)
return "", 0, nil, "", err
}
} else if !zonePresent && zonesPresent {
// 01 - "zones" specified
// Pick a zone randomly selected from specified set.
if zones, err = volumeutil.ZonesToSet(configuredZones); err != nil {
return "", 0, nil, "", err
}
} else if zonePresent && !zonesPresent {
// 10 - "zone" specified
// Use specified zone
if err := volumeutil.ValidateZone(configuredZone); err != nil {
return "", 0, nil, "", err
}
zones = make(sets.String)
zones.Insert(configuredZone)
selectedZone, err := volumeutil.SelectZoneForVolume(zonePresent, zonesPresent, configuredZone, configuredZones, activezones, node, allowedTopologies, c.options.PVC.Name)
if err != nil {
return "", 0, nil, "", err
}
zone := volumeutil.ChooseZoneForVolume(zones, c.options.PVC.Name)
if err := cloud.CreateDisk(
name,
diskType,
zone,
selectedZone,
int64(requestGB),
*c.options.CloudTags); err != nil {
glog.V(2).Infof("Error creating single-zone GCE PD volume: %v", err)
return "", 0, nil, "", err
}
glog.V(2).Infof("Successfully created single-zone GCE PD volume %s", name)
default:
......@@ -218,57 +194,6 @@ func (gceutil *GCEDiskUtil) CreateVolume(c *gcePersistentDiskProvisioner) (strin
return name, int(requestGB), labels, fstype, nil
}
// Creates a Regional PD
func createRegionalPD(
diskName string,
pvcName string,
diskType string,
zonesString string,
requestGB int64,
cloudTags *map[string]string,
cloud *gcecloud.GCECloud) error {
var replicaZones sets.String
var err error
if zonesString == "" {
// Consider all zones
replicaZones, err = cloud.GetAllCurrentZones()
if err != nil {
glog.V(2).Infof("error getting zone information from GCE: %v", err)
return err
}
} else {
replicaZones, err = volumeutil.ZonesToSet(zonesString)
if err != nil {
return err
}
}
zoneCount := replicaZones.Len()
var selectedReplicaZones sets.String
if zoneCount < maxRegionalPDZones {
return fmt.Errorf("cannot specify only %d zone(s) for Regional PDs.", zoneCount)
} else if zoneCount == maxRegionalPDZones {
selectedReplicaZones = replicaZones
} else {
// Must randomly select zones
selectedReplicaZones = volumeutil.ChooseZonesForVolume(
replicaZones, pvcName, maxRegionalPDZones)
}
if err = cloud.CreateRegionalDisk(
diskName,
diskType,
selectedReplicaZones,
int64(requestGB),
*cloudTags); err != nil {
return err
}
return nil
}
// Returns the first path that exists, or empty string if none exist.
func verifyDevicePath(devicePaths []string, sdBeforeSet sets.String, diskName string) (string, error) {
if err := udevadmChangeToNewDrives(sdBeforeSet); err != nil {
......
pkg/volume/util/util.go
View file @ a2de7d2d
......@@ -320,62 +320,102 @@ 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
// SelectZoneForVolume is a wrapper around SelectZonesForVolume
// to select a single zone for a volume based on parameters
func SelectZoneForVolume(zoneParameterPresent, zonesParameterPresent bool, zoneParameter string, zonesParameter, zonesWithNodes sets.String, node *v1.Node, allowedTopologies []v1.TopologySelectorTerm, pvcName string) (string, error) {
zones, err := SelectZonesForVolume(zoneParameterPresent, zonesParameterPresent, zoneParameter, zonesParameter, zonesWithNodes, node, allowedTopologies, pvcName, 1)
if err != nil {
return "", err
}
zone, ok := zones.PopAny()
if !ok {
return "", fmt.Errorf("could not determine a zone to provision volume in")
}
return zone, nil
}
// SelectZonesForVolume selects zones for a volume based on several factors:
// node.zone, allowedTopologies, zone/zones parameters from storageclass,
// zones with active nodes from the cluster. The number of zones = replicas.
func SelectZonesForVolume(zoneParameterPresent, zonesParameterPresent bool, zoneParameter string, zonesParameter, zonesWithNodes sets.String, node *v1.Node, allowedTopologies []v1.TopologySelectorTerm, pvcName string, numReplicas uint32) (sets.String, error) {
if zoneParameterPresent && zonesParameterPresent {
return "", fmt.Errorf("both zone and zones StorageClass parameters must not be used at the same time")
return nil, fmt.Errorf("both zone and zones StorageClass parameters must not be used at the same time")
}
// pick zone from node if present
var zoneFromNode string
// pick one 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.")
return nil, 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]
// pick node's zone for one of the replicas
var ok bool
zoneFromNode, ok = node.ObjectMeta.Labels[kubeletapis.LabelZoneFailureDomain]
if !ok {
return "", fmt.Errorf("%s Label for node missing", kubeletapis.LabelZoneFailureDomain)
return nil, fmt.Errorf("%s Label for node missing", kubeletapis.LabelZoneFailureDomain)
}
// if single replica volume and node with zone found, return immediately
if numReplicas == 1 {
return sets.NewString(zoneFromNode), nil
}
return z, nil
}
// pick zone from allowedZones if specified
allowedZones, err := ZonesFromAllowedTopologies(allowedTopologies)
if err != nil {
return "", err
return nil, 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)
return nil, 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 nil, fmt.Errorf("zone[s] cannot be specified in StorageClass if allowedTopologies specified")
}
// scheduler will guarantee if node != null above, zoneFromNode is member of allowedZones.
// so if zoneFromNode != "", we can safely assume it is part of allowedZones.
if zones, err := chooseZonesForVolumeIncludingZone(allowedZones, pvcName, zoneFromNode, numReplicas); err != nil {
return nil, fmt.Errorf("cannot process zones in allowedTopologies: %v", err)
} else {
return zones, nil
}
return ChooseZoneForVolume(*allowedZones, pvcName), nil
}
// pick zone from parameters if present
if zoneParameterPresent {
return zoneParameter, nil
if numReplicas > 1 {
return nil, fmt.Errorf("zone cannot be specified if desired number of replicas for pv is greather than 1. Please specify zones or allowedTopologies to specify desired zones")
}
return sets.NewString(zoneParameter), nil
}
if zonesParameterPresent {
return ChooseZoneForVolume(zonesParameter, pvcName), nil
if uint32(zonesParameter.Len()) < numReplicas {
return nil, fmt.Errorf("not enough zones found in zones parameter to provision a volume with %d replicas. Found %d zones, need %d zones", numReplicas, zonesParameter.Len(), numReplicas)
}
// directly choose from zones parameter; no zone from node need to be considered
return ChooseZonesForVolume(zonesParameter, pvcName, numReplicas), nil
}
// pick zone from zones with nodes
return ChooseZoneForVolume(zonesWithNodes, pvcName), nil
if zonesWithNodes.Len() > 0 {
// If node != null (and thus zoneFromNode != ""), zoneFromNode will be member of zonesWithNodes
if zones, err := chooseZonesForVolumeIncludingZone(zonesWithNodes, pvcName, zoneFromNode, numReplicas); err != nil {
return nil, fmt.Errorf("cannot process zones where nodes exist in the cluster: %v", err)
} else {
return zones, nil
}
}
return nil, fmt.Errorf("cannot determine zones to provision volume in")
}
// ZonesFromAllowedTopologies returns a list of zones specified in allowedTopologies
func ZonesFromAllowedTopologies(allowedTopologies []v1.TopologySelectorTerm) (*sets.String, error) {
func ZonesFromAllowedTopologies(allowedTopologies []v1.TopologySelectorTerm) (sets.String, error) {
zones := make(sets.String)
for _, term := range allowedTopologies {
for _, exp := range term.MatchLabelExpressions {
......@@ -388,7 +428,7 @@ func ZonesFromAllowedTopologies(allowedTopologies []v1.TopologySelectorTerm) (*s
}
}
}
return &zones, nil
return zones, nil
}
func ZonesSetToLabelValue(strSet sets.String) string {
......@@ -397,7 +437,11 @@ func ZonesSetToLabelValue(strSet sets.String) string {
// ZonesToSet converts a string containing a comma separated list of zones to set
func ZonesToSet(zonesString string) (sets.String, error) {
return stringToSet(zonesString, ",")
zones, err := stringToSet(zonesString, ",")
if err != nil {
return nil, fmt.Errorf("error parsing zones %s, must be strings separated by commas: %v", zonesString, err)
}
return zones, nil
}
// LabelZonesToSet converts a PV label value from string containing a delimited list of zones to set
......@@ -422,6 +466,27 @@ func stringToSet(str, delimiter string) (sets.String, error) {
return zonesSet, nil
}
// LabelZonesToList converts a PV label value from string containing a delimited list of zones to list
func LabelZonesToList(labelZonesValue string) ([]string, error) {
return stringToList(labelZonesValue, kubeletapis.LabelMultiZoneDelimiter)
}
// StringToList converts a string containing list separated by specified delimiter to a list
func stringToList(str, delimiter string) ([]string, error) {
zonesSlice := make([]string, 0)
for _, zone := range strings.Split(str, delimiter) {
trimmedZone := strings.TrimSpace(zone)
if trimmedZone == "" {
return nil, fmt.Errorf(
"%q separated list (%q) must not contain an empty string",
delimiter,
str)
}
zonesSlice = append(zonesSlice, trimmedZone)
}
return zonesSlice, nil
}
// CalculateTimeoutForVolume calculates time for a Recycler pod to complete a
// recycle operation. The calculation and return value is either the
// minimumTimeout or the timeoutIncrement per Gi of storage size, whichever is
......@@ -539,6 +604,33 @@ func ChooseZoneForVolume(zones sets.String, pvcName string) string {
return zone
}
// chooseZonesForVolumeIncludingZone is a wrapper around ChooseZonesForVolume that ensures zoneToInclude is chosen
// zoneToInclude can either be empty in which case it is ignored. If non-empty, zoneToInclude is expected to be member of zones.
// numReplicas is expected to be > 0 and <= zones.Len()
func chooseZonesForVolumeIncludingZone(zones sets.String, pvcName, zoneToInclude string, numReplicas uint32) (sets.String, error) {
if numReplicas == 0 {
return nil, fmt.Errorf("invalid number of replicas passed")
}
if uint32(zones.Len()) < numReplicas {
return nil, fmt.Errorf("not enough zones found to provision a volume with %d replicas. Need at least %d distinct zones for a volume with %d replicas", numReplicas, numReplicas, numReplicas)
}
if zoneToInclude != "" && !zones.Has(zoneToInclude) {
return nil, fmt.Errorf("zone to be included: %s needs to be member of set: %v", zoneToInclude, zones)
}
if uint32(zones.Len()) == numReplicas {
return zones, nil
}
if zoneToInclude != "" {
zones.Delete(zoneToInclude)
numReplicas = numReplicas - 1
}
zonesChosen := ChooseZonesForVolume(zones, pvcName, numReplicas)
if zoneToInclude != "" {
zonesChosen.Insert(zoneToInclude)
}
return zonesChosen, nil
}
// ChooseZonesForVolume is identical to ChooseZoneForVolume, but selects a multiple zones, for multi-zone disks.
func ChooseZonesForVolume(zones sets.String, pvcName string, numZones uint32) sets.String {
// We create the volume in a zone determined by the name
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment