Add initial support for Volumes to AWS

pkg/volume/aws_pd/aws_pd.go

@@ -0,0 +1,287 @@
+/*
+Copyright 2014 Google Inc. All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package aws_pd
+
+import (
+	"fmt"
+	"os"
+	"path"
+	"strconv"
+
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/cloudprovider"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/cloudprovider/aws"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/types"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/util/exec"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/util/mount"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/volume"
+	"github.com/golang/glog"
+)
+
+// This is the primary entrypoint for volume plugins.
+func ProbeVolumePlugins() []volume.VolumePlugin {
+	return []volume.VolumePlugin{&awsPersistentDiskPlugin{nil}}
+}
+
+type awsPersistentDiskPlugin struct {
+	host volume.VolumeHost
+}
+
+var _ volume.VolumePlugin = &awsPersistentDiskPlugin{}
+
+const (
+	awsPersistentDiskPluginName = "kubernetes.io/aws-pd"
+)
+
+func (plugin *awsPersistentDiskPlugin) Init(host volume.VolumeHost) {
+	plugin.host = host
+}
+
+func (plugin *awsPersistentDiskPlugin) Name() string {
+	return awsPersistentDiskPluginName
+}
+
+func (plugin *awsPersistentDiskPlugin) CanSupport(spec *api.Volume) bool {
+	if spec.AWSPersistentDisk != nil {
+		return true
+	}
+	return false
+}
+
+func (plugin *awsPersistentDiskPlugin) GetAccessModes() []api.AccessModeType {
+	return []api.AccessModeType{
+		api.ReadWriteOnce,
+	}
+}
+
+func (plugin *awsPersistentDiskPlugin) NewBuilder(spec *api.Volume, podRef *api.ObjectReference) (volume.Builder, error) {
+	// Inject real implementations here, test through the internal function.
+	return plugin.newBuilderInternal(spec, podRef.UID, &AWSDiskUtil{}, mount.New())
+}
+
+func (plugin *awsPersistentDiskPlugin) newBuilderInternal(spec *api.Volume, podUID types.UID, manager pdManager, mounter mount.Interface) (volume.Builder, error) {
+	pdName := spec.AWSPersistentDisk.PDName
+	fsType := spec.AWSPersistentDisk.FSType
+	partition := ""
+	if spec.AWSPersistentDisk.Partition != 0 {
+		partition = strconv.Itoa(spec.AWSPersistentDisk.Partition)
+	}
+	readOnly := spec.AWSPersistentDisk.ReadOnly
+
+	return &awsPersistentDisk{
+		podUID:      podUID,
+		volName:     spec.Name,
+		pdName:      pdName,
+		fsType:      fsType,
+		partition:   partition,
+		readOnly:    readOnly,
+		manager:     manager,
+		mounter:     mounter,
+		diskMounter: &awsSafeFormatAndMount{mounter, exec.New()},
+		plugin:      plugin,
+	}, nil
+}
+
+func (plugin *awsPersistentDiskPlugin) NewCleaner(volName string, podUID types.UID) (volume.Cleaner, error) {
+	// Inject real implementations here, test through the internal function.
+	return plugin.newCleanerInternal(volName, podUID, &AWSDiskUtil{}, mount.New())
+}
+
+func (plugin *awsPersistentDiskPlugin) newCleanerInternal(volName string, podUID types.UID, manager pdManager, mounter mount.Interface) (volume.Cleaner, error) {
+	return &awsPersistentDisk{
+		podUID:      podUID,
+		volName:     volName,
+		manager:     manager,
+		mounter:     mounter,
+		diskMounter: &awsSafeFormatAndMount{mounter, exec.New()},
+		plugin:      plugin,
+	}, nil
+}
+
+// Abstract interface to PD operations.
+type pdManager interface {
+	// Attaches the disk to the kubelet's host machine.
+	AttachAndMountDisk(pd *awsPersistentDisk, globalPDPath string) error
+	// Detaches the disk from the kubelet's host machine.
+	DetachDisk(pd *awsPersistentDisk) error
+}
+
+// awsPersistentDisk volumes are disk resources provided by Google Compute Engine
+// that are attached to the kubelet's host machine and exposed to the pod.
+type awsPersistentDisk struct {
+	volName string
+	podUID  types.UID
+	// Unique name of the PD, used to find the disk resource in the provider.
+	pdName string
+	// Filesystem type, optional.
+	fsType string
+	// Specifies the partition to mount
+	partition string
+	// Specifies whether the disk will be attached as read-only.
+	readOnly bool
+	// Utility interface that provides API calls to the provider to attach/detach disks.
+	manager pdManager
+	// Mounter interface that provides system calls to mount the global path to the pod local path.
+	mounter mount.Interface
+	// diskMounter provides the interface that is used to mount the actual block device.
+	diskMounter mount.Interface
+	plugin      *awsPersistentDiskPlugin
+}
+
+func detachDiskLogError(pd *awsPersistentDisk) {
+	err := pd.manager.DetachDisk(pd)
+	if err != nil {
+		glog.Warningf("Failed to detach disk: %v (%v)", pd, err)
+	}
+}
+
+// getVolumeProvider returns the AWS Volumes interface
+func (pd *awsPersistentDisk) getVolumeProvider() (aws_cloud.Volumes, error) {
+	name := "aws"
+	cloud, err := cloudprovider.GetCloudProvider(name, nil)
+	if err != nil {
+		return nil, err
+	}
+	volumes, ok := cloud.(aws_cloud.Volumes)
+	if !ok {
+		return nil, fmt.Errorf("Cloud provider does not support volumes")
+	}
+	return volumes, nil
+}
+
+// SetUp attaches the disk and bind mounts to the volume path.
+func (pd *awsPersistentDisk) SetUp() error {
+	return pd.SetUpAt(pd.GetPath())
+}
+
+// SetUpAt attaches the disk and bind mounts to the volume path.
+func (pd *awsPersistentDisk) SetUpAt(dir string) error {
+	// TODO: handle failed mounts here.
+	mountpoint, err := mount.IsMountPoint(dir)
+	glog.V(4).Infof("PersistentDisk set up: %s %v %v", dir, mountpoint, err)
+	if err != nil && !os.IsNotExist(err) {
+		return err
+	}
+	if mountpoint {
+		return nil
+	}
+
+	globalPDPath := makeGlobalPDName(pd.plugin.host, pd.pdName)
+	if err := pd.manager.AttachAndMountDisk(pd, globalPDPath); err != nil {
+		return err
+	}
+
+	flags := uintptr(0)
+	if pd.readOnly {
+		flags = mount.FlagReadOnly
+	}
+
+	if err := os.MkdirAll(dir, 0750); err != nil {
+		// TODO: we should really eject the attach/detach out into its own control loop.
+		detachDiskLogError(pd)
+		return err
+	}
+
+	// Perform a bind mount to the full path to allow duplicate mounts of the same PD.
+	err = pd.mounter.Mount(globalPDPath, dir, "", mount.FlagBind|flags, "")
+	if err != nil {
+		mountpoint, mntErr := mount.IsMountPoint(dir)
+		if mntErr != nil {
+			glog.Errorf("isMountpoint check failed: %v", mntErr)
+			return err
+		}
+		if mountpoint {
+			if mntErr = pd.mounter.Unmount(dir, 0); mntErr != nil {
+				glog.Errorf("Failed to unmount: %v", mntErr)
+				return err
+			}
+			mountpoint, mntErr := mount.IsMountPoint(dir)
+			if mntErr != nil {
+				glog.Errorf("isMountpoint check failed: %v", mntErr)
+				return err
+			}
+			if mountpoint {
+				// This is very odd, we don't expect it.  We'll try again next sync loop.
+				glog.Errorf("%s is still mounted, despite call to unmount().  Will try again next sync loop.", dir)
+				return err
+			}
+		}
+		os.Remove(dir)
+		// TODO: we should really eject the attach/detach out into its own control loop.
+		detachDiskLogError(pd)
+		return err
+	}
+
+	return nil
+}
+
+func makeGlobalPDName(host volume.VolumeHost, devName string) string {
+	return path.Join(host.GetPluginDir(awsPersistentDiskPluginName), "mounts", devName)
+}
+
+func (pd *awsPersistentDisk) GetPath() string {
+	name := awsPersistentDiskPluginName
+	return pd.plugin.host.GetPodVolumeDir(pd.podUID, util.EscapeQualifiedNameForDisk(name), pd.volName)
+}
+
+// Unmounts the bind mount, and detaches the disk only if the PD
+// resource was the last reference to that disk on the kubelet.
+func (pd *awsPersistentDisk) TearDown() error {
+	return pd.TearDownAt(pd.GetPath())
+}
+
+// Unmounts the bind mount, and detaches the disk only if the PD
+// resource was the last reference to that disk on the kubelet.
+func (pd *awsPersistentDisk) TearDownAt(dir string) error {
+	mountpoint, err := mount.IsMountPoint(dir)
+	if err != nil {
+		return err
+	}
+	if !mountpoint {
+		return os.Remove(dir)
+	}
+
+	refs, err := mount.GetMountRefs(pd.mounter, dir)
+	if err != nil {
+		return err
+	}
+	// Unmount the bind-mount inside this pod
+	if err := pd.mounter.Unmount(dir, 0); err != nil {
+		return err
+	}
+	// If len(refs) is 1, then all bind mounts have been removed, and the
+	// remaining reference is the global mount. It is safe to detach.
+	if len(refs) == 1 {
+		// pd.pdName is not initially set for volume-cleaners, so set it here.
+		pd.pdName = path.Base(refs[0])
+		if err := pd.manager.DetachDisk(pd); err != nil {
+			return err
+		}
+	}
+	mountpoint, mntErr := mount.IsMountPoint(dir)
+	if mntErr != nil {
+		glog.Errorf("isMountpoint check failed: %v", mntErr)
+		return err
+	}
+	if !mountpoint {
+		if err := os.Remove(dir); err != nil {
+			return err
+		}
+	}
+	return nil
+}

pkg/volume/aws_pd/aws_pd_test.go

@@ -0,0 +1,153 @@
+/*
+Copyright 2014 Google Inc. All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package aws_pd
+
+import (
+	"os"
+	"testing"
+
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/types"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/util/mount"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/volume"
+)
+
+func TestCanSupport(t *testing.T) {
+	plugMgr := volume.VolumePluginMgr{}
+	plugMgr.InitPlugins(ProbeVolumePlugins(), volume.NewFakeVolumeHost("/tmp/fake", nil, nil))
+
+	plug, err := plugMgr.FindPluginByName("kubernetes.io/aws-pd")
+	if err != nil {
+		t.Errorf("Can't find the plugin by name")
+	}
+	if plug.Name() != "kubernetes.io/aws-pd" {
+		t.Errorf("Wrong name: %s", plug.Name())
+	}
+	if !plug.CanSupport(&api.Volume{VolumeSource: api.VolumeSource{AWSPersistentDisk: &api.AWSPersistentDiskVolumeSource{}}}) {
+		t.Errorf("Expected true")
+	}
+}
+
+func TestGetAccessModes(t *testing.T) {
+	plugMgr := volume.VolumePluginMgr{}
+	plugMgr.InitPlugins(ProbeVolumePlugins(), volume.NewFakeVolumeHost("/tmp/fake", nil, nil))
+
+	plug, err := plugMgr.FindPersistentPluginByName("kubernetes.io/aws-pd")
+	if err != nil {
+		t.Errorf("Can't find the plugin by name")
+	}
+	if !contains(plug.GetAccessModes(), api.ReadWriteOnce) || !contains(plug.GetAccessModes(), api.ReadOnlyMany) {
+		t.Errorf("Expected two AccessModeTypes:  %s and %s", api.ReadWriteOnce, api.ReadOnlyMany)
+	}
+}
+
+func contains(modes []api.AccessModeType, mode api.AccessModeType) bool {
+	for _, m := range modes {
+		if m == mode {
+			return true
+		}
+	}
+	return false
+}
+
+type fakePDManager struct{}
+
+// TODO(jonesdl) To fully test this, we could create a loopback device
+// and mount that instead.
+func (fake *fakePDManager) AttachAndMountDisk(pd *awsPersistentDisk, globalPDPath string) error {
+	globalPath := makeGlobalPDName(pd.plugin.host, pd.pdName)
+	err := os.MkdirAll(globalPath, 0750)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (fake *fakePDManager) DetachDisk(pd *awsPersistentDisk) error {
+	globalPath := makeGlobalPDName(pd.plugin.host, pd.pdName)
+	err := os.RemoveAll(globalPath)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func TestPlugin(t *testing.T) {
+	plugMgr := volume.VolumePluginMgr{}
+	plugMgr.InitPlugins(ProbeVolumePlugins(), volume.NewFakeVolumeHost("/tmp/fake", nil, nil))
+
+	plug, err := plugMgr.FindPluginByName("kubernetes.io/aws-pd")
+	if err != nil {
+		t.Errorf("Can't find the plugin by name")
+	}
+	spec := &api.Volume{
+		Name: "vol1",
+		VolumeSource: api.VolumeSource{
+			AWSPersistentDisk: &api.AWSPersistentDiskVolumeSource{
+				PDName: "pd",
+				FSType: "ext4",
+			},
+		},
+	}
+	builder, err := plug.(*awsPersistentDiskPlugin).newBuilderInternal(spec, types.UID("poduid"), &fakePDManager{}, &mount.FakeMounter{})
+	if err != nil {
+		t.Errorf("Failed to make a new Builder: %v", err)
+	}
+	if builder == nil {
+		t.Errorf("Got a nil Builder: %v")
+	}
+
+	path := builder.GetPath()
+	if path != "/tmp/fake/pods/poduid/volumes/kubernetes.io~aws-pd/vol1" {
+		t.Errorf("Got unexpected path: %s", path)
+	}
+
+	if err := builder.SetUp(); err != nil {
+		t.Errorf("Expected success, got: %v", err)
+	}
+	if _, err := os.Stat(path); err != nil {
+		if os.IsNotExist(err) {
+			t.Errorf("SetUp() failed, volume path not created: %s", path)
+		} else {
+			t.Errorf("SetUp() failed: %v", err)
+		}
+	}
+	if _, err := os.Stat(path); err != nil {
+		if os.IsNotExist(err) {
+			t.Errorf("SetUp() failed, volume path not created: %s", path)
+		} else {
+			t.Errorf("SetUp() failed: %v", err)
+		}
+	}
+
+	cleaner, err := plug.(*awsPersistentDiskPlugin).newCleanerInternal("vol1", types.UID("poduid"), &fakePDManager{}, &mount.FakeMounter{})
+	if err != nil {
+		t.Errorf("Failed to make a new Cleaner: %v", err)
+	}
+	if cleaner == nil {
+		t.Errorf("Got a nil Cleaner: %v")
+	}
+
+	if err := cleaner.TearDown(); err != nil {
+		t.Errorf("Expected success, got: %v", err)
+	}
+	if _, err := os.Stat(path); err == nil {
+		t.Errorf("TearDown() failed, volume path still exists: %s", path)
+	} else if !os.IsNotExist(err) {
+		t.Errorf("SetUp() failed: %v", err)
+	}
+}

pkg/volume/aws_pd/aws_util.go

@@ -0,0 +1,140 @@
+/*
+Copyright 2014 Google Inc. All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package aws_pd
+
+import (
+	"errors"
+	"fmt"
+	"os"
+	"path"
+	"time"
+
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/util/exec"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/util/mount"
+	"github.com/golang/glog"
+)
+
+type AWSDiskUtil struct{}
+
+// Attaches a disk specified by a volume.AWSPersistentDisk to the current kubelet.
+// Mounts the disk to it's global path.
+func (util *AWSDiskUtil) AttachAndMountDisk(pd *awsPersistentDisk, globalPDPath string) error {
+	volumes, err := pd.getVolumeProvider()
+	if err != nil {
+		return err
+	}
+	flags := uintptr(0)
+	if pd.readOnly {
+		flags = mount.FlagReadOnly
+	}
+	if err := volumes.AttachDisk("", pd.pdName, pd.readOnly); err != nil {
+		return err
+	}
+	devicePath := path.Join("/dev/disk/by-id/", "aws-"+pd.pdName)
+	if pd.partition != "" {
+		devicePath = devicePath + "-part" + pd.partition
+	}
+	//TODO(jonesdl) There should probably be better method than busy-waiting here.
+	numTries := 0
+	for {
+		_, err := os.Stat(devicePath)
+		if err == nil {
+			break
+		}
+		if err != nil && !os.IsNotExist(err) {
+			return err
+		}
+		numTries++
+		if numTries == 10 {
+			return errors.New("Could not attach disk: Timeout after 10s")
+		}
+		time.Sleep(time.Second)
+	}
+
+	// Only mount the PD globally once.
+	mountpoint, err := mount.IsMountPoint(globalPDPath)
+	if err != nil {
+		if os.IsNotExist(err) {
+			if err := os.MkdirAll(globalPDPath, 0750); err != nil {
+				return err
+			}
+			mountpoint = false
+		} else {
+			return err
+		}
+	}
+	if !mountpoint {
+		err = pd.diskMounter.Mount(devicePath, globalPDPath, pd.fsType, flags, "")
+		if err != nil {
+			os.Remove(globalPDPath)
+			return err
+		}
+	}
+	return nil
+}
+
+// Unmounts the device and detaches the disk from the kubelet's host machine.
+func (util *AWSDiskUtil) DetachDisk(pd *awsPersistentDisk) error {
+	// Unmount the global PD mount, which should be the only one.
+	globalPDPath := makeGlobalPDName(pd.plugin.host, pd.pdName)
+	if err := pd.mounter.Unmount(globalPDPath, 0); err != nil {
+		return err
+	}
+	if err := os.Remove(globalPDPath); err != nil {
+		return err
+	}
+	// Detach the disk
+	volumes, err := pd.getVolumeProvider()
+	if err != nil {
+		return err
+	}
+	if err := volumes.DetachDisk("", pd.pdName); err != nil {
+		return err
+	}
+	return nil
+}
+
+// safe_format_and_mount is a utility script on AWS VMs that probes a persistent disk, and if
+// necessary formats it before mounting it.
+// This eliminates the necesisty to format a PD before it is used with a Pod on AWS.
+// TODO: port this script into Go and use it for all Linux platforms
+type awsSafeFormatAndMount struct {
+	mount.Interface
+	runner exec.Interface
+}
+
+// uses /usr/share/google/safe_format_and_mount to optionally mount, and format a disk
+func (mounter *awsSafeFormatAndMount) Mount(source string, target string, fstype string, flags uintptr, data string) error {
+	// Don't attempt to format if mounting as readonly. Go straight to mounting.
+	if (flags & mount.FlagReadOnly) != 0 {
+		return mounter.Interface.Mount(source, target, fstype, flags, data)
+	}
+	args := []string{}
+	// ext4 is the default for safe_format_and_mount
+	if len(fstype) > 0 && fstype != "ext4" {
+		args = append(args, "-m", fmt.Sprintf("mkfs.%s", fstype))
+	}
+	args = append(args, source, target)
+	// TODO: Accept other options here?
+	glog.V(5).Infof("exec-ing: /usr/share/google/safe_format_and_mount %v", args)
+	cmd := mounter.runner.Command("/usr/share/google/safe_format_and_mount", args...)
+	dataOut, err := cmd.CombinedOutput()
+	if err != nil {
+		glog.V(5).Infof("error running /usr/share/google/safe_format_and_mount\n%s", string(dataOut))
+	}
+	return err
+}

pkg/volume/aws_pd/aws_util_test.go

@@ -0,0 +1,84 @@
+/*
+Copyright 2014 Google Inc. All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package aws_pd
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/util/exec"
+)
+
+func TestSafeFormatAndMount(t *testing.T) {
+	tests := []struct {
+		fstype       string
+		expectedArgs []string
+		err          error
+	}{
+		{
+			fstype:       "ext4",
+			expectedArgs: []string{"/dev/foo", "/mnt/bar"},
+		},
+		{
+			fstype:       "vfat",
+			expectedArgs: []string{"-m", "mkfs.vfat", "/dev/foo", "/mnt/bar"},
+		},
+		{
+			err: fmt.Errorf("test error"),
+		},
+	}
+	for _, test := range tests {
+
+		var cmdOut string
+		var argsOut []string
+		fake := exec.FakeExec{
+			CommandScript: []exec.FakeCommandAction{
+				func(cmd string, args ...string) exec.Cmd {
+					cmdOut = cmd
+					argsOut = args
+					fake := exec.FakeCmd{
+						CombinedOutputScript: []exec.FakeCombinedOutputAction{
+							func() ([]byte, error) { return []byte{}, test.err },
+						},
+					}
+					return exec.InitFakeCmd(&fake, cmd, args...)
+				},
+			},
+		}
+
+		mounter := awsSafeFormatAndMount{
+			runner: &fake,
+		}
+
+		err := mounter.Mount("/dev/foo", "/mnt/bar", test.fstype, 0, "")
+		if test.err == nil && err != nil {
+			t.Errorf("unexpected error: %v", err)
+		}
+		if test.err != nil {
+			if err == nil {
+				t.Errorf("unexpected non-error")
+			}
+			return
+		}
+		if cmdOut != "/usr/share/google/safe_format_and_mount" {
+			t.Errorf("unexpected command: %s", cmdOut)
+		}
+		if len(argsOut) != len(test.expectedArgs) {
+			t.Errorf("unexpected args: %v, expected: %v", argsOut, test.expectedArgs)
+		}
+	}
+}