Statefulsets for cinder: allow multi-AZ deployments, spread pods across zones

pkg/cloudprovider/providers/openstack/openstack_test.go

@@ -354,7 +354,7 @@ func TestVolumes(t *testing.T) {
 	tags := map[string]string{
 		"test": "value",
 	}
-	vol, err := os.CreateVolume("kubernetes-test-volume-"+rand.String(10), 1, "", "", &tags)
+	vol, _, err := os.CreateVolume("kubernetes-test-volume-"+rand.String(10), 1, "", "", &tags)
 	if err != nil {
 		t.Fatalf("Cannot create a new Cinder volume: %v", err)
 	}

pkg/cloudprovider/providers/openstack/openstack_volumes.go

@@ -36,7 +36,7 @@ import (
 )
 
 type volumeService interface {
-	createVolume(opts VolumeCreateOpts) (string, error)
+	createVolume(opts VolumeCreateOpts) (string, string, error)
 	getVolume(diskName string) (Volume, error)
 	deleteVolume(volumeName string) error
 }
@@ -74,7 +74,7 @@ type VolumeCreateOpts struct {
 	Metadata     map[string]string
 }
 
-func (volumes *VolumesV1) createVolume(opts VolumeCreateOpts) (string, error) {
+func (volumes *VolumesV1) createVolume(opts VolumeCreateOpts) (string, string, error) {
 
 	create_opts := volumes_v1.CreateOpts{
 		Name:             opts.Name,
@@ -86,12 +86,12 @@ func (volumes *VolumesV1) createVolume(opts VolumeCreateOpts) (string, error) {
 
 	vol, err := volumes_v1.Create(volumes.blockstorage, create_opts).Extract()
 	if err != nil {
-		return "", err
+		return "", "", err
 	}
-	return vol.ID, nil
+	return vol.ID, vol.AvailabilityZone, nil
 }
 
-func (volumes *VolumesV2) createVolume(opts VolumeCreateOpts) (string, error) {
+func (volumes *VolumesV2) createVolume(opts VolumeCreateOpts) (string, string, error) {
 
 	create_opts := volumes_v2.CreateOpts{
 		Name:             opts.Name,
@@ -103,9 +103,9 @@ func (volumes *VolumesV2) createVolume(opts VolumeCreateOpts) (string, error) {
 
 	vol, err := volumes_v2.Create(volumes.blockstorage, create_opts).Extract()
 	if err != nil {
-		return "", err
+		return "", "", err
 	}
-	return vol.ID, nil
+	return vol.ID, vol.AvailabilityZone, nil
 }
 
 func (volumes *VolumesV1) getVolume(diskName string) (Volume, error) {
@@ -283,12 +283,12 @@ func (os *OpenStack) getVolume(diskName string) (Volume, error) {
 }
 
 // Create a volume of given size (in GiB)
-func (os *OpenStack) CreateVolume(name string, size int, vtype, availability string, tags *map[string]string) (volumeName string, err error) {
+func (os *OpenStack) CreateVolume(name string, size int, vtype, availability string, tags *map[string]string) (string, string, error) {
 
 	volumes, err := os.volumeService("")
 	if err != nil || volumes == nil {
 		glog.Errorf("Unable to initialize cinder client for region: %s", os.region)
-		return "", err
+		return "", "", err
 	}
 	opts := VolumeCreateOpts{
 		Name:         name,
@@ -299,15 +299,15 @@ func (os *OpenStack) CreateVolume(name string, size int, vtype, availability str
 	if tags != nil {
 		opts.Metadata = *tags
 	}
-	volume_id, err := volumes.createVolume(opts)
+	volumeId, volumeAZ, err := volumes.createVolume(opts)
 
 	if err != nil {
 		glog.Errorf("Failed to create a %d GB volume: %v", size, err)
-		return "", err
+		return "", "", err
 	}
 
-	glog.Infof("Created volume %v", volume_id)
-	return volume_id, nil
+	glog.Infof("Created volume %v in Availability Zone: %v", volumeId, volumeAZ)
+	return volumeId, volumeAZ, nil
 }
 
 // GetDevicePath returns the path of an attached block storage volume, specified by its id.

pkg/cloudprovider/providers/rackspace/rackspace.go

@@ -480,8 +480,8 @@ func (os *Rackspace) GetZone() (cloudprovider.Zone, error) {
 }
 
 // Create a volume of given size (in GiB)
-func (rs *Rackspace) CreateVolume(name string, size int, vtype, availability string, tags *map[string]string) (volumeName string, err error) {
-	return "", errors.New("unimplemented")
+func (rs *Rackspace) CreateVolume(name string, size int, vtype, availability string, tags *map[string]string) (volumeName string, volumeAZ string, err error) {
+	return "", "", errors.New("unimplemented")
 }
 
 func (rs *Rackspace) DeleteVolume(volumeName string) error {

pkg/volume/cinder/BUILD

@@ -19,6 +19,7 @@ go_library(
     tags = ["automanaged"],
     deps = [
         "//pkg/api/v1:go_default_library",
+        "//pkg/client/clientset_generated/clientset:go_default_library",
         "//pkg/cloudprovider:go_default_library",
         "//pkg/cloudprovider/providers/openstack:go_default_library",
         "//pkg/cloudprovider/providers/rackspace:go_default_library",
@@ -32,6 +33,7 @@ go_library(
         "//vendor/k8s.io/apimachinery/pkg/api/resource:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/types:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/util/sets:go_default_library",
     ],
 )
 

pkg/volume/cinder/attacher_test.go

@@ -506,8 +506,8 @@ func (testcase *testcase) ShouldTrustDevicePath() bool {
 	return true
 }
 
-func (testcase *testcase) CreateVolume(name string, size int, vtype, availability string, tags *map[string]string) (volumeName string, err error) {
-	return "", errors.New("Not implemented")
+func (testcase *testcase) CreateVolume(name string, size int, vtype, availability string, tags *map[string]string) (volumeId string, volumeAZ string, err error) {
+	return "", "", errors.New("Not implemented")
 }
 
 func (testcase *testcase) GetDevicePath(diskId string) string {

pkg/volume/cinder/cinder.go

@@ -47,7 +47,7 @@ type CinderProvider interface {
 	AttachDisk(instanceID string, diskName string) (string, error)
 	DetachDisk(instanceID string, partialDiskId string) error
 	DeleteVolume(volumeName string) error
-	CreateVolume(name string, size int, vtype, availability string, tags *map[string]string) (volumeName string, err error)
+	CreateVolume(name string, size int, vtype, availability string, tags *map[string]string) (string, string, error)
 	GetDevicePath(diskId string) string
 	InstanceID() (string, error)
 	GetAttachmentDiskPath(instanceID string, diskName string) (string, error)
@@ -239,7 +239,7 @@ type cdManager interface {
 	// Detaches the disk from the kubelet's host machine.
 	DetachDisk(unmounter *cinderVolumeUnmounter) error
 	// Creates a volume
-	CreateVolume(provisioner *cinderVolumeProvisioner) (volumeID string, volumeSizeGB int, err error)
+	CreateVolume(provisioner *cinderVolumeProvisioner) (volumeID string, volumeSizeGB int, labels map[string]string, err error)
 	// Deletes a volume
 	DeleteVolume(deleter *cinderVolumeDeleter) error
 }
@@ -482,15 +482,15 @@ type cinderVolumeProvisioner struct {
 var _ volume.Provisioner = &cinderVolumeProvisioner{}
 
 func (c *cinderVolumeProvisioner) Provision() (*v1.PersistentVolume, error) {
-	volumeID, sizeGB, err := c.manager.CreateVolume(c)
+	volumeID, sizeGB, labels, err := c.manager.CreateVolume(c)
 	if err != nil {
 		return nil, err
 	}
 
 	pv := &v1.PersistentVolume{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:   c.options.PVName,
-			Labels: map[string]string{},
+			Labels: labels,
 			Annotations: map[string]string{
 				"kubernetes.io/createdby": "cinder-dynamic-provisioner",
 			},

pkg/volume/cinder/cinder_test.go

@@ -116,8 +116,8 @@ func (fake *fakePDManager) DetachDisk(c *cinderVolumeUnmounter) error {
 	return nil
 }
 
-func (fake *fakePDManager) CreateVolume(c *cinderVolumeProvisioner) (volumeID string, volumeSizeGB int, err error) {
-	return "test-volume-name", 1, nil
+func (fake *fakePDManager) CreateVolume(c *cinderVolumeProvisioner) (volumeID string, volumeSizeGB int, labels map[string]string, err error) {
+	return "test-volume-name", 1, nil, nil
 }
 
 func (fake *fakePDManager) DeleteVolume(cd *cinderVolumeDeleter) error {

pkg/volume/cinder/cinder_util.go

@@ -25,7 +25,11 @@ import (
 	"time"
 
 	"github.com/golang/glog"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/kubernetes/pkg/api/v1"
+	"k8s.io/kubernetes/pkg/client/clientset_generated/clientset"
 	"k8s.io/kubernetes/pkg/util/exec"
 	"k8s.io/kubernetes/pkg/volume"
 )
@@ -135,10 +139,28 @@ func (util *CinderDiskUtil) DeleteVolume(cd *cinderVolumeDeleter) error {
 	return nil
 }
 
-func (util *CinderDiskUtil) CreateVolume(c *cinderVolumeProvisioner) (volumeID string, volumeSizeGB int, err error) {
+func getZonesFromNodes(kubeClient clientset.Interface) (sets.String, error) {
+	// TODO: caching, currently it is overkill because it calls this function
+	// only when it creates dynamic PV
+	zones := make(sets.String)
+	nodes, err := kubeClient.Core().Nodes().List(metav1.ListOptions{})
+	if err != nil {
+		glog.V(2).Infof("Error listing nodes")
+		return zones, err
+	}
+	for _, node := range nodes.Items {
+		if zone, ok := node.Labels[metav1.LabelZoneFailureDomain]; ok {
+			zones.Insert(zone)
+		}
+	}
+	glog.V(4).Infof("zones found: %v", zones)
+	return zones, nil
+}
+
+func (util *CinderDiskUtil) CreateVolume(c *cinderVolumeProvisioner) (volumeID string, volumeSizeGB int, volumeLabels map[string]string, err error) {
 	cloud, err := c.plugin.getCloudProvider()
 	if err != nil {
-		return "", 0, err
+		return "", 0, nil, err
 	}
 
 	capacity := c.options.PVC.Spec.Resources.Requests[v1.ResourceName(v1.ResourceStorage)]
@@ -157,21 +179,40 @@ func (util *CinderDiskUtil) CreateVolume(c *cinderVolumeProvisioner) (volumeID s
 		case "availability":
 			availability = v
 		default:
-			return "", 0, fmt.Errorf("invalid option %q for volume plugin %s", k, c.plugin.GetPluginName())
+			return "", 0, nil, fmt.Errorf("invalid option %q for volume plugin %s", k, c.plugin.GetPluginName())
 		}
 	}
 	// TODO: implement PVC.Selector parsing
 	if c.options.PVC.Spec.Selector != nil {
-		return "", 0, fmt.Errorf("claim.Spec.Selector is not supported for dynamic provisioning on Cinder")
+		return "", 0, nil, fmt.Errorf("claim.Spec.Selector is not supported for dynamic provisioning on Cinder")
 	}
 
-	name, err = cloud.CreateVolume(name, volSizeGB, vtype, availability, c.options.CloudTags)
-	if err != nil {
-		glog.V(2).Infof("Error creating cinder volume: %v", err)
-		return "", 0, err
+	if availability == "" {
+		// No zone specified, choose one randomly in the same region
+		zones, err := getZonesFromNodes(c.plugin.host.GetKubeClient())
+		if err != nil {
+			glog.V(2).Infof("error getting zone information: %v", err)
+			return "", 0, nil, err
+		}
+		// if we did not get any zones, lets leave it blank and gophercloud will
+		// use zone "nova" as default
+		if len(zones) > 0 {
+			availability = volume.ChooseZoneForVolume(zones, c.options.PVC.Name)
+		}
 	}
-	glog.V(2).Infof("Successfully created cinder volume %s", name)
-	return name, volSizeGB, nil
+
+	volumeId, volumeAZ, errr := cloud.CreateVolume(name, volSizeGB, vtype, availability, c.options.CloudTags)
+	if errr != nil {
+		glog.V(2).Infof("Error creating cinder volume: %v", errr)
+		return "", 0, nil, errr
+	}
+	glog.V(2).Infof("Successfully created cinder volume %s", volumeId)
+
+	// these are needed that pod is spawning to same AZ
+	volumeLabels = make(map[string]string)
+	volumeLabels[metav1.LabelZoneFailureDomain] = volumeAZ
+
+	return volumeId, volSizeGB, volumeLabels, nil
 }
 
 func probeAttachedVolume() error {