Implement Creater and Deleter interfaces for AWS EBS.

cmd/kube-controller-manager/app/plugins.go

@@ -28,6 +28,7 @@ import (
 
 	// Volume plugins
 	"k8s.io/kubernetes/pkg/cloudprovider"
+	"k8s.io/kubernetes/pkg/cloudprovider/providers/aws"
 	"k8s.io/kubernetes/pkg/util/io"
 	"k8s.io/kubernetes/pkg/volume"
 	"k8s.io/kubernetes/pkg/volume/aws_ebs"
@@ -87,8 +88,8 @@ func NewVolumeProvisioner(cloud cloudprovider.Interface, flags VolumeConfigFlags
 	switch {
 	case cloud == nil && flags.EnableHostPathProvisioning:
 		return getProvisionablePluginFromVolumePlugins(host_path.ProbeVolumePlugins(volume.VolumeConfig{}))
-		//	case cloud != nil && aws.ProviderName == cloud.ProviderName():
-		//		return getProvisionablePluginFromVolumePlugins(aws_ebs.ProbeVolumePlugins())
+	case cloud != nil && aws.ProviderName == cloud.ProviderName():
+		return getProvisionablePluginFromVolumePlugins(aws_ebs.ProbeVolumePlugins())
 		//	case cloud != nil && gce.ProviderName == cloud.ProviderName():
 		//		return getProvisionablePluginFromVolumePlugins(gce_pd.ProbeVolumePlugins())
 		//	case cloud != nil && openstack.ProviderName == cloud.ProviderName():

pkg/cloudprovider/providers/aws/aws.go

@@ -136,6 +136,7 @@ type EC2Metadata interface {
 
 type VolumeOptions struct {
 	CapacityGB int
+	Tags       *map[string]string
 }
 
 // Volumes is an interface for managing cloud-provisioned volumes
@@ -1216,15 +1217,15 @@ func (aws *AWSCloud) DetachDisk(instanceName string, diskName string) error {
 }
 
 // Implements Volumes.CreateVolume
-func (aws *AWSCloud) CreateVolume(volumeOptions *VolumeOptions) (string, error) {
+func (s *AWSCloud) CreateVolume(volumeOptions *VolumeOptions) (string, error) {
 	// TODO: Should we tag this with the cluster id (so it gets deleted when the cluster does?)
 	// This is only used for testing right now
 
 	request := &ec2.CreateVolumeInput{}
-	request.AvailabilityZone = &aws.availabilityZone
+	request.AvailabilityZone = &s.availabilityZone
 	volSize := int64(volumeOptions.CapacityGB)
 	request.Size = &volSize
-	response, err := aws.ec2.CreateVolume(request)
+	response, err := s.ec2.CreateVolume(request)
 	if err != nil {
 		return "", err
 	}
@@ -1234,6 +1235,28 @@ func (aws *AWSCloud) CreateVolume(volumeOptions *VolumeOptions) (string, error)
 
 	volumeName := "aws://" + az + "/" + awsID
 
+	// apply tags
+	if volumeOptions.Tags != nil {
+		tags := []*ec2.Tag{}
+		for k, v := range *volumeOptions.Tags {
+			tag := &ec2.Tag{}
+			tag.Key = aws.String(k)
+			tag.Value = aws.String(v)
+			tags = append(tags, tag)
+		}
+		tagRequest := &ec2.CreateTagsInput{}
+		tagRequest.Resources = []*string{&awsID}
+		tagRequest.Tags = tags
+		if _, err := s.createTags(tagRequest); err != nil {
+			// delete the volume and hope it succeeds
+			delerr := s.DeleteVolume(volumeName)
+			if delerr != nil {
+				// delete did not succeed, we have a stray volume!
+				return "", fmt.Errorf("error tagging volume %s, could not delete the volume: %v", volumeName, delerr)
+			}
+			return "", fmt.Errorf("error tagging volume %s: %v", volumeName, err)
+		}
+	}
 	return volumeName, nil
 }
 

pkg/controller/persistentvolume/persistentvolume_provisioner_controller.go

@@ -335,6 +335,10 @@ func newProvisioner(plugin volume.ProvisionableVolumePlugin, claim *api.Persiste
 		Capacity:                      claim.Spec.Resources.Requests[api.ResourceName(api.ResourceStorage)],
 		AccessModes:                   claim.Spec.AccessModes,
 		PersistentVolumeReclaimPolicy: api.PersistentVolumeReclaimDelete,
+		CloudTags: &map[string]string{
+			cloudVolumeCreatedForNamespaceTag: claim.Namespace,
+			cloudVolumeCreatedForNameTag:      claim.Name,
+		},
 	}
 
 	provisioner, err := plugin.NewProvisioner(volumeOptions)

pkg/controller/persistentvolume/types.go

@@ -30,6 +30,12 @@ const (
 	// For example tiers might be gold, silver, and tin and the admin configures what that means for each volume plugin that can provision a volume.
 	// Values in the alpha version of this feature are not meaningful, but will be in the full version of this feature.
 	qosProvisioningKey = "volume.alpha.kubernetes.io/storage-class"
+	// Name of a tag attached to a real volume in cloud (e.g. AWS EBS or GCE PD)
+	// with namespace of a persistent volume claim used to create this volume.
+	cloudVolumeCreatedForNamespaceTag = "kubernetes.io/created-for/pvc/namespace"
+	// Name of a tag attached to a real volume in cloud (e.g. AWS EBS or GCE PD)
+	// with name of a persistent volume claim used to create this volume.
+	cloudVolumeCreatedForNameTag = "kubernetes.io/created-for/pvc/name"
 )
 
 // persistentVolumeOrderedIndex is a cache.Store that keeps persistent volumes indexed by AccessModes and ordered by storage capacity.

pkg/volume/aws_ebs/aws_ebs.go

@@ -26,6 +26,7 @@ import (
 
 	"github.com/golang/glog"
 	"k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/api/resource"
 	awscloud "k8s.io/kubernetes/pkg/cloudprovider/providers/aws"
 	"k8s.io/kubernetes/pkg/types"
 	"k8s.io/kubernetes/pkg/util"
@@ -45,6 +46,8 @@ type awsElasticBlockStorePlugin struct {
 
 var _ volume.VolumePlugin = &awsElasticBlockStorePlugin{}
 var _ volume.PersistentVolumePlugin = &awsElasticBlockStorePlugin{}
+var _ volume.DeletableVolumePlugin = &awsElasticBlockStorePlugin{}
+var _ volume.ProvisionableVolumePlugin = &awsElasticBlockStorePlugin{}
 
 const (
 	awsElasticBlockStorePluginName = "kubernetes.io/aws-ebs"
@@ -124,12 +127,50 @@ func (plugin *awsElasticBlockStorePlugin) newCleanerInternal(volName string, pod
 	}}, nil
 }
 
+func (plugin *awsElasticBlockStorePlugin) NewDeleter(spec *volume.Spec) (volume.Deleter, error) {
+	return plugin.newDeleterInternal(spec, &AWSDiskUtil{})
+}
+
+func (plugin *awsElasticBlockStorePlugin) newDeleterInternal(spec *volume.Spec, manager ebsManager) (volume.Deleter, error) {
+	if spec.PersistentVolume != nil && spec.PersistentVolume.Spec.AWSElasticBlockStore == nil {
+		return nil, fmt.Errorf("spec.PersistentVolumeSource.AWSElasticBlockStore is nil")
+	}
+	return &awsElasticBlockStoreDeleter{
+		awsElasticBlockStore: &awsElasticBlockStore{
+			volName:  spec.Name(),
+			volumeID: spec.PersistentVolume.Spec.AWSElasticBlockStore.VolumeID,
+			manager:  manager,
+			plugin:   plugin,
+		}}, nil
+}
+
+func (plugin *awsElasticBlockStorePlugin) NewProvisioner(options volume.VolumeOptions) (volume.Provisioner, error) {
+	if len(options.AccessModes) == 0 {
+		options.AccessModes = plugin.GetAccessModes()
+	}
+	return plugin.newProvisionerInternal(options, &AWSDiskUtil{})
+}
+
+func (plugin *awsElasticBlockStorePlugin) newProvisionerInternal(options volume.VolumeOptions, manager ebsManager) (volume.Provisioner, error) {
+	return &awsElasticBlockStoreProvisioner{
+		awsElasticBlockStore: &awsElasticBlockStore{
+			manager: manager,
+			plugin:  plugin,
+		},
+		options: options,
+	}, nil
+}
+
 // Abstract interface to PD operations.
 type ebsManager interface {
 	// Attaches the disk to the kubelet's host machine.
 	AttachAndMountDisk(b *awsElasticBlockStoreBuilder, globalPDPath string) error
 	// Detaches the disk from the kubelet's host machine.
 	DetachDisk(c *awsElasticBlockStoreCleaner) error
+	// Creates a volume
+	CreateVolume(provisioner *awsElasticBlockStoreProvisioner) (volumeID string, volumeSizeGB int, err error)
+	// Deletes a volume
+	DeleteVolume(deleter *awsElasticBlockStoreDeleter) error
 }
 
 // awsElasticBlockStore volumes are disk resources provided by Amazon Web Services
@@ -349,3 +390,67 @@ func (c *awsElasticBlockStoreCleaner) TearDownAt(dir string) error {
 	}
 	return nil
 }
+
+type awsElasticBlockStoreDeleter struct {
+	*awsElasticBlockStore
+}
+
+var _ volume.Deleter = &awsElasticBlockStoreDeleter{}
+
+func (d *awsElasticBlockStoreDeleter) GetPath() string {
+	name := awsElasticBlockStorePluginName
+	return d.plugin.host.GetPodVolumeDir(d.podUID, util.EscapeQualifiedNameForDisk(name), d.volName)
+}
+
+func (d *awsElasticBlockStoreDeleter) Delete() error {
+	return d.manager.DeleteVolume(d)
+}
+
+type awsElasticBlockStoreProvisioner struct {
+	*awsElasticBlockStore
+	options   volume.VolumeOptions
+	namespace string
+}
+
+var _ volume.Provisioner = &awsElasticBlockStoreProvisioner{}
+
+func (c *awsElasticBlockStoreProvisioner) Provision(pv *api.PersistentVolume) error {
+	volumeID, sizeGB, err := c.manager.CreateVolume(c)
+	if err != nil {
+		return err
+	}
+	pv.Spec.PersistentVolumeSource.AWSElasticBlockStore.VolumeID = volumeID
+	pv.Spec.Capacity = api.ResourceList{
+		api.ResourceName(api.ResourceStorage): resource.MustParse(fmt.Sprintf("%dGi", sizeGB)),
+	}
+	return nil
+}
+
+func (c *awsElasticBlockStoreProvisioner) NewPersistentVolumeTemplate() (*api.PersistentVolume, error) {
+	// Provide dummy api.PersistentVolume.Spec, it will be filled in
+	// awsElasticBlockStoreProvisioner.Provision()
+	return &api.PersistentVolume{
+		ObjectMeta: api.ObjectMeta{
+			GenerateName: "pv-aws-",
+			Labels:       map[string]string{},
+			Annotations: map[string]string{
+				"kubernetes.io/createdby": "aws-ebs-dynamic-provisioner",
+			},
+		},
+		Spec: api.PersistentVolumeSpec{
+			PersistentVolumeReclaimPolicy: c.options.PersistentVolumeReclaimPolicy,
+			AccessModes:                   c.options.AccessModes,
+			Capacity: api.ResourceList{
+				api.ResourceName(api.ResourceStorage): c.options.Capacity,
+			},
+			PersistentVolumeSource: api.PersistentVolumeSource{
+				AWSElasticBlockStore: &api.AWSElasticBlockStoreVolumeSource{
+					VolumeID:  "dummy",
+					FSType:    "ext4",
+					Partition: 0,
+					ReadOnly:  false,
+				},
+			},
+		},
+	}, nil
+}

pkg/volume/aws_ebs/aws_ebs_test.go

@@ -17,12 +17,14 @@ limitations under the License.
 package aws_ebs
 
 import (
+	"fmt"
 	"io/ioutil"
 	"os"
 	"path"
 	"testing"
 
 	"k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/api/resource"
 	"k8s.io/kubernetes/pkg/api/testapi"
 	"k8s.io/kubernetes/pkg/client/unversioned/testclient"
 	"k8s.io/kubernetes/pkg/types"
@@ -106,6 +108,17 @@ func (fake *fakePDManager) DetachDisk(c *awsElasticBlockStoreCleaner) error {
 	return nil
 }
 
+func (fake *fakePDManager) CreateVolume(c *awsElasticBlockStoreProvisioner) (volumeID string, volumeSizeGB int, err error) {
+	return "test-aws-volume-name", 100, nil
+}
+
+func (fake *fakePDManager) DeleteVolume(cd *awsElasticBlockStoreDeleter) error {
+	if cd.volumeID != "test-aws-volume-name" {
+		return fmt.Errorf("Deleter got unexpected volume name: %s", cd.volumeID)
+	}
+	return nil
+}
+
 func TestPlugin(t *testing.T) {
 	tmpDir, err := ioutil.TempDir(os.TempDir(), "awsebsTest")
 	if err != nil {
@@ -175,6 +188,47 @@ func TestPlugin(t *testing.T) {
 	} else if !os.IsNotExist(err) {
 		t.Errorf("SetUp() failed: %v", err)
 	}
+
+	// Test Provisioner
+	cap := resource.MustParse("100Gi")
+	options := volume.VolumeOptions{
+		Capacity: cap,
+		AccessModes: []api.PersistentVolumeAccessMode{
+			api.ReadWriteOnce,
+		},
+		PersistentVolumeReclaimPolicy: api.PersistentVolumeReclaimDelete,
+	}
+	provisioner, err := plug.(*awsElasticBlockStorePlugin).newProvisionerInternal(options, &fakePDManager{})
+	persistentSpec, err := provisioner.NewPersistentVolumeTemplate()
+	if err != nil {
+		t.Errorf("NewPersistentVolumeTemplate() failed: %v", err)
+	}
+
+	// get 2nd Provisioner - persistent volume controller will do the same
+	provisioner, err = plug.(*awsElasticBlockStorePlugin).newProvisionerInternal(options, &fakePDManager{})
+	err = provisioner.Provision(persistentSpec)
+	if err != nil {
+		t.Errorf("Provision() failed: %v", err)
+	}
+
+	if persistentSpec.Spec.PersistentVolumeSource.AWSElasticBlockStore.VolumeID != "test-aws-volume-name" {
+		t.Errorf("Provision() returned unexpected volume ID: %s", persistentSpec.Spec.PersistentVolumeSource.AWSElasticBlockStore.VolumeID)
+	}
+	cap = persistentSpec.Spec.Capacity[api.ResourceStorage]
+	size := cap.Value()
+	if size != 100*1024*1024*1024 {
+		t.Errorf("Provision() returned unexpected volume size: %v", size)
+	}
+
+	// Test Deleter
+	volSpec := &volume.Spec{
+		PersistentVolume: persistentSpec,
+	}
+	deleter, err := plug.(*awsElasticBlockStorePlugin).newDeleterInternal(volSpec, &fakePDManager{})
+	err = deleter.Delete()
+	if err != nil {
+		t.Errorf("Deleter() failed: %v", err)
+	}
 }
 
 func TestPersistentClaimReadOnlyFlag(t *testing.T) {

pkg/volume/aws_ebs/aws_util.go

@@ -22,6 +22,8 @@ import (
 	"time"
 
 	"github.com/golang/glog"
+	aws_cloud "k8s.io/kubernetes/pkg/cloudprovider/providers/aws"
+	"k8s.io/kubernetes/pkg/volume"
 )
 
 type AWSDiskUtil struct{}
@@ -107,3 +109,42 @@ func (util *AWSDiskUtil) DetachDisk(c *awsElasticBlockStoreCleaner) error {
 	}
 	return nil
 }
+
+func (util *AWSDiskUtil) DeleteVolume(d *awsElasticBlockStoreDeleter) error {
+	volumes, err := d.getVolumeProvider()
+	if err != nil {
+		glog.V(2).Info("Error getting volume provider: ", err)
+		return err
+	}
+
+	if err := volumes.DeleteVolume(d.volumeID); err != nil {
+		glog.V(2).Infof("Error deleting AWS EBS volume %s: %v", d.volumeID, err)
+		return err
+	}
+	glog.V(2).Infof("Successfully deleted AWS EBS volume %s", d.volumeID)
+	return nil
+}
+
+func (util *AWSDiskUtil) CreateVolume(c *awsElasticBlockStoreProvisioner) (volumeID string, volumeSizeGB int, err error) {
+	volumes, err := c.getVolumeProvider()
+	if err != nil {
+		glog.V(2).Info("Error getting volume provider: ", err)
+		return "", 0, err
+	}
+
+	requestBytes := c.options.Capacity.Value()
+	// AWS works with gigabytes, convert to GiB with rounding up
+	requestGB := int(volume.RoundUpSize(requestBytes, 1024*1024*1024))
+	volSpec := &aws_cloud.VolumeOptions{
+		CapacityGB: requestGB,
+		Tags:       c.options.CloudTags,
+	}
+
+	name, err := volumes.CreateVolume(volSpec)
+	if err != nil {
+		glog.V(2).Infof("Error creating AWS EBS volume: %v", err)
+		return "", 0, err
+	}
+	glog.V(2).Infof("Successfully created AWS EBS volume %s", name)
+	return name, requestGB, nil
+}

pkg/volume/plugins.go

@@ -49,6 +49,8 @@ type VolumeOptions struct {
 	AccessModes []api.PersistentVolumeAccessMode
 	// Reclamation policy for a persistent volume
 	PersistentVolumeReclaimPolicy api.PersistentVolumeReclaimPolicy
+	// Tags to attach to the real volume in the cloud provider - e.g. AWS EBS
+	CloudTags *map[string]string
 }
 
 // VolumePlugin is an interface to volume plugins that can be used on a

pkg/volume/util.go

@@ -139,3 +139,12 @@ func CalculateTimeoutForVolume(minimumTimeout, timeoutIncrement int, pv *api.Per
 		return timeout
 	}
 }
+
+// RoundUpSize calculates how many allocation units are needed to accomodate
+// a volume of given size. E.g. when user wants 1500MiB volume, while AWS EBS
+// allocates volumes in gibibyte-sized chunks,
+// RoundUpSize(1500 * 1024*1024, 1024*1024*1024) returns '2'
+// (2 GiB is the smallest allocatable volume that can hold 1500MiB)
+func RoundUpSize(volumeSizeBytes int64, allocationUnitBytes int64) int64 {
+	return (volumeSizeBytes + allocationUnitBytes - 1) / allocationUnitBytes
+}