Added Block Storage support to Rackspace provider, improved Node discovery.

pkg/cloudprovider/providers/openstack/openstack.go

@@ -23,6 +23,7 @@ import (
 	"io"
 	"io/ioutil"
 	"net/http"
+	"path"
 	"regexp"
 	"strings"
 	"time"
@@ -468,6 +469,11 @@ func (i *Instances) ExternalID(name string) (string, error) {
 	return srv.ID, nil
 }
 
+// InstanceID returns the kubelet's cloud provider ID.
+func (os *OpenStack) InstanceID() (string, error) {
+	return os.localInstanceID, nil
+}
+
 // InstanceID returns the cloud provider ID of the specified instance.
 func (i *Instances) InstanceID(name string) (string, error) {
 	srv, err := getServerByName(i.compute, name)
@@ -956,7 +962,7 @@ func (os *OpenStack) Routes() (cloudprovider.Routes, bool) {
 }
 
 // Attaches given cinder volume to the compute running kubelet
-func (os *OpenStack) AttachDisk(diskName string) (string, error) {
+func (os *OpenStack) AttachDisk(instanceID string, diskName string) (string, error) {
 	disk, err := os.getVolume(diskName)
 	if err != nil {
 		return "", err
@@ -970,8 +976,8 @@ func (os *OpenStack) AttachDisk(diskName string) (string, error) {
 	}
 
 	if len(disk.Attachments) > 0 && disk.Attachments[0]["server_id"] != nil {
-		if os.localInstanceID == disk.Attachments[0]["server_id"] {
-			glog.V(4).Infof("Disk: %q is already attached to compute: %q", diskName, os.localInstanceID)
+		if instanceID == disk.Attachments[0]["server_id"] {
+			glog.V(4).Infof("Disk: %q is already attached to compute: %q", diskName, instanceID)
 			return disk.ID, nil
 		} else {
 			errMsg := fmt.Sprintf("Disk %q is attached to a different compute: %q, should be detached before proceeding", diskName, disk.Attachments[0]["server_id"])
@@ -980,19 +986,19 @@ func (os *OpenStack) AttachDisk(diskName string) (string, error) {
 		}
 	}
 	// add read only flag here if possible spothanis
-	_, err = volumeattach.Create(cClient, os.localInstanceID, &volumeattach.CreateOpts{
+	_, err = volumeattach.Create(cClient, instanceID, &volumeattach.CreateOpts{
 		VolumeID: disk.ID,
 	}).Extract()
 	if err != nil {
-		glog.Errorf("Failed to attach %s volume to %s compute", diskName, os.localInstanceID)
+		glog.Errorf("Failed to attach %s volume to %s compute", diskName, instanceID)
 		return "", err
 	}
-	glog.V(2).Infof("Successfully attached %s volume to %s compute", diskName, os.localInstanceID)
+	glog.V(2).Infof("Successfully attached %s volume to %s compute", diskName, instanceID)
 	return disk.ID, nil
 }
 
 // Detaches given cinder volume from the compute running kubelet
-func (os *OpenStack) DetachDisk(partialDiskId string) error {
+func (os *OpenStack) DetachDisk(instanceID string, partialDiskId string) error {
 	disk, err := os.getVolume(partialDiskId)
 	if err != nil {
 		return err
@@ -1004,17 +1010,17 @@ func (os *OpenStack) DetachDisk(partialDiskId string) error {
 		glog.Errorf("Unable to initialize nova client for region: %s", os.region)
 		return err
 	}
-	if len(disk.Attachments) > 0 && disk.Attachments[0]["server_id"] != nil && os.localInstanceID == disk.Attachments[0]["server_id"] {
+	if len(disk.Attachments) > 0 && disk.Attachments[0]["server_id"] != nil && instanceID == disk.Attachments[0]["server_id"] {
 		// This is a blocking call and effects kubelet's performance directly.
 		// We should consider kicking it out into a separate routine, if it is bad.
-		err = volumeattach.Delete(cClient, os.localInstanceID, disk.ID).ExtractErr()
+		err = volumeattach.Delete(cClient, instanceID, disk.ID).ExtractErr()
 		if err != nil {
-			glog.Errorf("Failed to delete volume %s from compute %s attached %v", disk.ID, os.localInstanceID, err)
+			glog.Errorf("Failed to delete volume %s from compute %s attached %v", disk.ID, instanceID, err)
 			return err
 		}
-		glog.V(2).Infof("Successfully detached volume: %s from compute: %s", disk.ID, os.localInstanceID)
+		glog.V(2).Infof("Successfully detached volume: %s from compute: %s", disk.ID, instanceID)
 	} else {
-		errMsg := fmt.Sprintf("Disk: %s has no attachments or is not attached to compute: %s", disk.Name, os.localInstanceID)
+		errMsg := fmt.Sprintf("Disk: %s has no attachments or is not attached to compute: %s", disk.Name, instanceID)
 		glog.Errorf(errMsg)
 		return errors.New(errMsg)
 	}
@@ -1086,6 +1092,22 @@ func (os *OpenStack) CreateVolume(name string, size int, tags *map[string]string
 	return vol.ID, err
 }
 
+// GetDevicePath returns the path of an attached block storage volume, specified by its id.
+func (os *OpenStack) GetDevicePath(diskId string) string {
+	files, _ := ioutil.ReadDir("/dev/disk/by-id/")
+	for _, f := range files {
+		if strings.Contains(f.Name(), "virtio-") {
+			devid_prefix := f.Name()[len("virtio-"):len(f.Name())]
+			if strings.Contains(diskId, devid_prefix) {
+				glog.V(4).Infof("Found disk attached as %q; full devicepath: %s\n", f.Name(), path.Join("/dev/disk/by-id/", f.Name()))
+				return path.Join("/dev/disk/by-id/", f.Name())
+			}
+		}
+	}
+	glog.Warningf("Failed to find device for the diskid: %q\n", diskId)
+	return ""
+}
+
 func (os *OpenStack) DeleteVolume(volumeName string) error {
 	sClient, err := openstack.NewBlockStorageV1(os.provider, gophercloud.EndpointOpts{
 		Region: os.region,

pkg/cloudprovider/providers/openstack/openstack_test.go

@@ -268,7 +268,7 @@ func TestVolumes(t *testing.T) {
 
 	WaitForVolumeStatus(t, os, vol, volumeAvailableStatus, volumeCreateTimeoutSeconds)
 
-	diskId, err := os.AttachDisk(vol)
+	diskId, err := os.AttachDisk(os.localInstanceID, vol)
 	if err != nil {
 		t.Fatalf("Cannot AttachDisk Cinder volume %s: %v", vol, err)
 	}
@@ -276,7 +276,7 @@ func TestVolumes(t *testing.T) {
 
 	WaitForVolumeStatus(t, os, vol, volumeInUseStatus, volumeCreateTimeoutSeconds)
 
-	err = os.DetachDisk(vol)
+	err = os.DetachDisk(os.localInstanceID, vol)
 	if err != nil {
 		t.Fatalf("Cannot DetachDisk Cinder volume %s: %v", vol, err)
 	}

pkg/cloudprovider/providers/rackspace/rackspace_test.go

@@ -107,6 +107,27 @@ func configFromEnv() (cfg Config, ok bool) {
 	return
 }
 
+func TestParseMetaData(t *testing.T) {
+	_, err := parseMetaData(strings.NewReader(""))
+	if err == nil {
+		t.Errorf("Should fail when invalid meta data is provided: %s", err)
+	}
+
+	id, err := parseMetaData(strings.NewReader(`
+	{
+		"UUID":"someuuid",
+		"name":"somename",
+		"project_id":"someprojectid"
+	}
+	`))
+	if err != nil {
+		t.Fatalf("Should succeed when valid meta data is provided: %s", err)
+	}
+	if id != "someuuid" {
+		t.Errorf("incorrect uuid: %s", id)
+	}
+}
+
 func TestNewRackspace(t *testing.T) {
 	cfg, ok := configFromEnv()
 	if !ok {

pkg/volume/cinder/cinder.go

@@ -26,6 +26,7 @@ import (
 	"k8s.io/kubernetes/pkg/api"
 	"k8s.io/kubernetes/pkg/api/resource"
 	"k8s.io/kubernetes/pkg/cloudprovider/providers/openstack"
+	"k8s.io/kubernetes/pkg/cloudprovider/providers/rackspace"
 	"k8s.io/kubernetes/pkg/types"
 	"k8s.io/kubernetes/pkg/util/exec"
 	"k8s.io/kubernetes/pkg/util/keymutex"
@@ -39,6 +40,15 @@ func ProbeVolumePlugins() []volume.VolumePlugin {
 	return []volume.VolumePlugin{&cinderPlugin{}}
 }
 
+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, tags *map[string]string) (volumeName string, err error)
+	GetDevicePath(diskId string) string
+	InstanceID() (string, error)
+}
+
 type cinderPlugin struct {
 	host volume.VolumeHost
 	// Guarding SetUp and TearDown operations
@@ -153,18 +163,21 @@ func (plugin *cinderPlugin) newProvisionerInternal(options volume.VolumeOptions,
 	}, nil
 }
 
-func (plugin *cinderPlugin) getCloudProvider() (*openstack.OpenStack, error) {
+func (plugin *cinderPlugin) getCloudProvider() (CinderProvider, error) {
 	cloud := plugin.host.GetCloudProvider()
 	if cloud == nil {
 		glog.Errorf("Cloud provider not initialized properly")
 		return nil, errors.New("Cloud provider not initialized properly")
 	}
 
-	os := cloud.(*openstack.OpenStack)
-	if os == nil {
-		return nil, errors.New("Invalid cloud provider: expected OpenStack")
+	switch cloud := cloud.(type) {
+	case *rackspace.Rackspace:
+		return cloud, nil
+	case *openstack.OpenStack:
+		return cloud, nil
+	default:
+		return nil, errors.New("Invalid cloud provider: expected OpenStack or Rackspace.")
 	}
-	return os, nil
 }
 
 // Abstract interface to PD operations.

pkg/volume/cinder/cinder_util.go

@@ -18,10 +18,7 @@ package cinder
 
 import (
 	"errors"
-	"io/ioutil"
 	"os"
-	"path"
-	"strings"
 	"time"
 
 	"github.com/golang/glog"
@@ -42,7 +39,11 @@ func (util *CinderDiskUtil) AttachDisk(b *cinderVolumeMounter, globalPDPath stri
 	if err != nil {
 		return err
 	}
-	diskid, err := cloud.AttachDisk(b.pdName)
+	instanceid, err := cloud.InstanceID()
+	if err != nil {
+		return err
+	}
+	diskid, err := cloud.AttachDisk(instanceid, b.pdName)
 	if err != nil {
 		return err
 	}
@@ -50,8 +51,7 @@ func (util *CinderDiskUtil) AttachDisk(b *cinderVolumeMounter, globalPDPath stri
 	var devicePath string
 	numTries := 0
 	for {
-		devicePath = makeDevicePath(diskid)
-		// probe the attached vol so that symlink in /dev/disk/by-id is created
+		devicePath = cloud.GetDevicePath(diskid)
 		probeAttachedVolume()
 
 		_, err := os.Stat(devicePath)
@@ -89,21 +89,6 @@ func (util *CinderDiskUtil) AttachDisk(b *cinderVolumeMounter, globalPDPath stri
 	return nil
 }
 
-func makeDevicePath(diskid string) string {
-	files, _ := ioutil.ReadDir("/dev/disk/by-id/")
-	for _, f := range files {
-		if strings.Contains(f.Name(), "virtio-") {
-			devid_prefix := f.Name()[len("virtio-"):len(f.Name())]
-			if strings.Contains(diskid, devid_prefix) {
-				glog.V(4).Infof("Found disk attached as %q; full devicepath: %s\n", f.Name(), path.Join("/dev/disk/by-id/", f.Name()))
-				return path.Join("/dev/disk/by-id/", f.Name())
-			}
-		}
-	}
-	glog.Warningf("Failed to find device for the diskid: %q\n", diskid)
-	return ""
-}
-
 // Unmounts the device and detaches the disk from the kubelet's host machine.
 func (util *CinderDiskUtil) DetachDisk(cd *cinderVolumeUnmounter) error {
 	globalPDPath := makeGlobalPDName(cd.plugin.host, cd.pdName)
@@ -119,8 +104,11 @@ func (util *CinderDiskUtil) DetachDisk(cd *cinderVolumeUnmounter) error {
 	if err != nil {
 		return err
 	}
-
-	if err = cloud.DetachDisk(cd.pdName); err != nil {
+	instanceid, err := cloud.InstanceID()
+	if err != nil {
+		return err
+	}
+	if err = cloud.DetachDisk(instanceid, cd.pdName); err != nil {
 		return err
 	}
 	glog.V(2).Infof("Successfully detached cinder volume %s", cd.pdName)