add iscsi volume plugin

Signed-off-by: Huamin Chen <hchen@redhat.com>

pkg/volume/iscsi/disk_manager.go

@@ -0,0 +1,112 @@
+/*
+Copyright 2015 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 iscsi
+
+import (
+	"os"
+
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/util/mount"
+	"github.com/golang/glog"
+)
+
+// Abstract interface to disk operations.
+type diskManager interface {
+	MakeGlobalPDName(disk iscsiDisk) string
+	// Attaches the disk to the kubelet's host machine.
+	AttachDisk(disk iscsiDisk) error
+	// Detaches the disk from the kubelet's host machine.
+	DetachDisk(disk iscsiDisk, mntPath string) error
+}
+
+// utility to mount a disk based filesystem
+func diskSetUp(manager diskManager, disk iscsiDisk, volPath string, mounter mount.Interface) error {
+	globalPDPath := manager.MakeGlobalPDName(disk)
+	// TODO: handle failed mounts here.
+	mountpoint, err := mounter.IsMountPoint(volPath)
+
+	if err != nil && !os.IsNotExist(err) {
+		glog.Errorf("cannot validate mountpoint: %s", volPath)
+		return err
+	}
+	if mountpoint {
+		return nil
+	}
+	if err := manager.AttachDisk(disk); err != nil {
+		glog.Errorf("failed to attach disk")
+		return err
+	}
+
+	if err := os.MkdirAll(volPath, 0750); err != nil {
+		glog.Errorf("failed to mkdir:%s", volPath)
+		return err
+	}
+	// Perform a bind mount to the full path to allow duplicate mounts of the same disk.
+	flags := uintptr(0)
+	if disk.readOnly {
+		flags = mount.FlagReadOnly
+	}
+	err = mounter.Mount(globalPDPath, volPath, "", mount.FlagBind|flags, "")
+	if err != nil {
+		glog.Errorf("failed to bind mount:%s", globalPDPath)
+		return err
+	}
+	return nil
+}
+
+// utility to tear down a disk based filesystem
+func diskTearDown(manager diskManager, disk iscsiDisk, volPath string, mounter mount.Interface) error {
+	mountpoint, err := mounter.IsMountPoint(volPath)
+	if err != nil {
+		glog.Errorf("cannot validate mountpoint %s", volPath)
+		return err
+	}
+	if !mountpoint {
+		return os.Remove(volPath)
+	}
+
+	refs, err := mount.GetMountRefs(mounter, volPath)
+	if err != nil {
+		glog.Errorf("failed to get reference count %s", volPath)
+		return err
+	}
+	if err := mounter.Unmount(volPath, 0); err != nil {
+		glog.Errorf("failed to umount %s", volPath)
+		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 {
+		mntPath := refs[0]
+		if err := manager.DetachDisk(disk, mntPath); err != nil {
+			glog.Errorf("failed to detach disk from %s", mntPath)
+			return err
+		}
+	}
+
+	mountpoint, mntErr := mounter.IsMountPoint(volPath)
+	if mntErr != nil {
+		glog.Errorf("isMountpoint check failed: %v", mntErr)
+		return err
+	}
+	if !mountpoint {
+		if err := os.Remove(volPath); err != nil {
+			return err
+		}
+	}
+	return nil
+
+}

pkg/volume/iscsi/iscsi.go

@@ -0,0 +1,168 @@
+/*
+Copyright 2015 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 iscsi
+
+import (
+	"strconv"
+
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
+	"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{&ISCSIPlugin{nil, exec.New()}}
+}
+
+type ISCSIPlugin struct {
+	host volume.VolumeHost
+	exe  exec.Interface
+}
+
+var _ volume.VolumePlugin = &ISCSIPlugin{}
+
+const (
+	ISCSIPluginName = "kubernetes.io/iscsi"
+)
+
+func (plugin *ISCSIPlugin) Init(host volume.VolumeHost) {
+	plugin.host = host
+}
+
+func (plugin *ISCSIPlugin) Name() string {
+	return ISCSIPluginName
+}
+
+func (plugin *ISCSIPlugin) CanSupport(spec *api.Volume) bool {
+	if spec.ISCSI == nil {
+		return false
+	}
+	// see if iscsiadm is there
+	_, err := plugin.execCommand("iscsiadm", []string{"-h"})
+	if err == nil {
+		return true
+	}
+
+	return false
+}
+
+func (plugin *ISCSIPlugin) GetAccessModes() []api.AccessModeType {
+	return []api.AccessModeType{
+		api.ReadWriteOnce,
+		api.ReadOnlyMany,
+	}
+}
+
+func (plugin *ISCSIPlugin) 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, &ISCSIUtil{}, mount.New())
+}
+
+func (plugin *ISCSIPlugin) newBuilderInternal(spec *api.Volume, podUID types.UID, manager diskManager, mounter mount.Interface) (volume.Builder, error) {
+	lun := strconv.Itoa(spec.ISCSI.Lun)
+
+	return &iscsiDisk{
+		podUID:   podUID,
+		volName:  spec.Name,
+		portal:   spec.ISCSI.TargetPortal,
+		iqn:      spec.ISCSI.IQN,
+		lun:      lun,
+		fsType:   spec.ISCSI.FSType,
+		readOnly: spec.ISCSI.ReadOnly,
+		manager:  manager,
+		mounter:  mounter,
+		plugin:   plugin,
+	}, nil
+}
+
+func (plugin *ISCSIPlugin) NewCleaner(volName string, podUID types.UID) (volume.Cleaner, error) {
+	// Inject real implementations here, test through the internal function.
+	return plugin.newCleanerInternal(volName, podUID, &ISCSIUtil{}, mount.New())
+}
+
+func (plugin *ISCSIPlugin) newCleanerInternal(volName string, podUID types.UID, manager diskManager, mounter mount.Interface) (volume.Cleaner, error) {
+	return &iscsiDisk{
+		podUID:  podUID,
+		volName: volName,
+		manager: manager,
+		mounter: mounter,
+		plugin:  plugin,
+	}, nil
+}
+
+type iscsiDisk struct {
+	volName  string
+	podUID   types.UID
+	portal   string
+	iqn      string
+	readOnly bool
+	lun      string
+	fsType   string
+	plugin   *ISCSIPlugin
+	mounter  mount.Interface
+	// Utility interface that provides API calls to the provider to attach/detach disks.
+	manager diskManager
+}
+
+func (iscsi *iscsiDisk) GetPath() string {
+	name := ISCSIPluginName
+	// safe to use PodVolumeDir now: volume teardown occurs before pod is cleaned up
+	return iscsi.plugin.host.GetPodVolumeDir(iscsi.podUID, util.EscapeQualifiedNameForDisk(name), iscsi.volName)
+}
+
+func (iscsi *iscsiDisk) SetUp() error {
+	return iscsi.SetUpAt(iscsi.GetPath())
+}
+
+func (iscsi *iscsiDisk) SetUpAt(dir string) error {
+	// diskSetUp checks mountpoints and prevent repeated calls
+	err := diskSetUp(iscsi.manager, *iscsi, dir, iscsi.mounter)
+	if err != nil {
+		glog.Errorf("iscsi: failed to setup")
+		return err
+	}
+	globalPDPath := iscsi.manager.MakeGlobalPDName(*iscsi)
+	// make mountpoint rw/ro work as expected
+	//FIXME revisit pkg/util/mount and ensure rw/ro is implemented as expected
+	mode := "rw"
+	if iscsi.readOnly {
+		mode = "ro"
+	}
+	iscsi.plugin.execCommand("mount", []string{"-o", "remount," + mode, globalPDPath, dir})
+
+	return nil
+}
+
+// Unmounts the bind mount, and detaches the disk only if the disk
+// resource was the last reference to that disk on the kubelet.
+func (iscsi *iscsiDisk) TearDown() error {
+	return iscsi.TearDownAt(iscsi.GetPath())
+}
+
+func (iscsi *iscsiDisk) TearDownAt(dir string) error {
+	return diskTearDown(iscsi.manager, *iscsi, dir, iscsi.mounter)
+}
+
+func (plugin *ISCSIPlugin) execCommand(command string, args []string) ([]byte, error) {
+	cmd := plugin.exe.Command(command, args...)
+	return cmd.CombinedOutput()
+}

pkg/volume/iscsi/iscsi_test.go

@@ -0,0 +1,131 @@
+/*
+Copyright 2015 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 iscsi
+
+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/iscsi")
+	if err != nil {
+		t.Errorf("Can't find the plugin by name")
+	}
+	if plug.Name() != "kubernetes.io/iscsi" {
+		t.Errorf("Wrong name: %s", plug.Name())
+	}
+}
+
+type fakeDiskManager struct{}
+
+func (fake *fakeDiskManager) MakeGlobalPDName(disk iscsiDisk) string {
+	return "/tmp/fake_iscsi_path"
+}
+func (fake *fakeDiskManager) AttachDisk(disk iscsiDisk) error {
+	globalPath := disk.manager.MakeGlobalPDName(disk)
+	err := os.MkdirAll(globalPath, 0750)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (fake *fakeDiskManager) DetachDisk(disk iscsiDisk, mntPath string) error {
+	globalPath := disk.manager.MakeGlobalPDName(disk)
+	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/iscsi")
+	if err != nil {
+		t.Errorf("Can't find the plugin by name")
+	}
+	spec := &api.Volume{
+		Name: "vol1",
+		VolumeSource: api.VolumeSource{
+			ISCSI: &api.ISCSIVolumeSource{
+				TargetPortal: "127.0.0.1:3260",
+				IQN:          "iqn.2014-12.server:storage.target01",
+				FSType:       "ext4",
+				Lun:          0,
+			},
+		},
+	}
+	builder, err := plug.(*ISCSIPlugin).newBuilderInternal(spec, types.UID("poduid"), &fakeDiskManager{}, &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~iscsi/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.(*ISCSIPlugin).newCleanerInternal("vol1", types.UID("poduid"), &fakeDiskManager{}, &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/iscsi/iscsi_util.go

@@ -0,0 +1,156 @@
+/*
+Copyright 2015 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 iscsi
+
+import (
+	"errors"
+	"os"
+	"path"
+	"strings"
+	"time"
+
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/util/mount"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/volume"
+	"github.com/golang/glog"
+)
+
+// stat a path, if not exists, retry maxRetries times
+func waitForPathToExist(devicePath string, maxRetries int) bool {
+	for i := 0; i < maxRetries; i++ {
+		_, err := os.Stat(devicePath)
+		if err == nil {
+			return true
+		}
+		if err != nil && !os.IsNotExist(err) {
+			return false
+		}
+		time.Sleep(time.Second)
+	}
+	return false
+}
+
+// getDevicePrefixRefCount: given a prefix of device path, find its reference count from /proc/mounts
+// returns the reference count to the device and error code
+// for services like iscsi construct multiple device paths with the same prefix pattern.
+// this function aggregates all references to a service based on the prefix pattern
+// More specifically, this prefix semantics is to aggregate disk paths that belong to the same iSCSI target/iqn pair.
+// an iSCSI target could expose multiple LUNs through the same IQN, and Linux iSCSI initiator creates disk paths that start the same prefix but end with different LUN number
+// When we decide whether it is time to logout a target, we have to see if none of the LUNs are used any more.
+// That's where the prefix based ref count kicks in. If we only count the disks using exact match, we could log other disks out.
+func getDevicePrefixRefCount(mounter mount.Interface, deviceNamePrefix string) (int, error) {
+	mps, err := mounter.List()
+	if err != nil {
+		return -1, err
+	}
+
+	// Find the number of references to the device.
+	refCount := 0
+	for i := range mps {
+		if strings.HasPrefix(mps[i].Device, deviceNamePrefix) {
+			refCount++
+		}
+	}
+	return refCount, nil
+}
+
+// make a directory like /var/lib/kubelet/plugins/kubernetes.io/pod/iscsi/portal-iqn-some_iqn-lun-0
+func makePDNameInternal(host volume.VolumeHost, portal string, iqn string, lun string) string {
+	return path.Join(host.GetPluginDir(ISCSIPluginName), "iscsi", portal+"-iqn-"+iqn+"-lun-"+lun)
+}
+
+type ISCSIUtil struct{}
+
+func (util *ISCSIUtil) MakeGlobalPDName(iscsi iscsiDisk) string {
+	return makePDNameInternal(iscsi.plugin.host, iscsi.portal, iscsi.iqn, iscsi.lun)
+}
+
+func (util *ISCSIUtil) AttachDisk(iscsi iscsiDisk) error {
+	devicePath := strings.Join([]string{"/dev/disk/by-path/ip", iscsi.portal, "iscsi", iscsi.iqn, "lun", iscsi.lun}, "-")
+	exist := waitForPathToExist(devicePath, 1)
+	if exist == false {
+		// discover iscsi target
+		_, err := iscsi.plugin.execCommand("iscsiadm", []string{"-m", "discovery", "-t", "sendtargets", "-p", iscsi.portal})
+		if err != nil {
+			glog.Errorf("iscsi: failed to sendtargets to portal %s error:%v", iscsi.portal, err)
+			return err
+		}
+		// login to iscsi target
+		_, err = iscsi.plugin.execCommand("iscsiadm", []string{"-m", "node", "-p", iscsi.portal, "-T", iscsi.iqn, "--login"})
+		if err != nil {
+			glog.Errorf("iscsi: failed to attach disk:Error: %v", err)
+			return err
+		}
+		exist = waitForPathToExist(devicePath, 10)
+		if !exist {
+			return errors.New("Could not attach disk: Timeout after 10s")
+		}
+	}
+	// mount it
+	globalPDPath := iscsi.manager.MakeGlobalPDName(iscsi)
+	mountpoint, err := iscsi.mounter.IsMountPoint(globalPDPath)
+	if mountpoint {
+		glog.Infof("iscsi: %s already mounted", globalPDPath)
+		return nil
+	}
+
+	if err := os.MkdirAll(globalPDPath, 0750); err != nil {
+		glog.Errorf("iscsi: failed to mkdir %s, error", globalPDPath)
+		return err
+	}
+
+	err = iscsi.mounter.Mount(devicePath, globalPDPath, iscsi.fsType, uintptr(0), "")
+	if err != nil {
+		glog.Errorf("iscsi: failed to mount iscsi volume %s [%s] to %s, error %v", devicePath, iscsi.fsType, globalPDPath, err)
+	}
+
+	return err
+}
+
+func (util *ISCSIUtil) DetachDisk(iscsi iscsiDisk, mntPath string) error {
+	device, cnt, err := mount.GetDeviceNameFromMount(iscsi.mounter, mntPath)
+	if err != nil {
+		glog.Errorf("iscsi detach disk: failed to get device from mnt: %s\nError: %v", mntPath, err)
+		return err
+	}
+	if err = iscsi.mounter.Unmount(mntPath, 0); err != nil {
+		glog.Errorf("iscsi detach disk: failed to umount: %s\nError: %v", mntPath, err)
+		return err
+	}
+	cnt--
+	// if device is no longer used, see if need to logout the target
+	if cnt == 0 {
+		// strip -lun- from device path
+		ind := strings.LastIndex(device, "-lun-")
+		prefix := device[:(ind - 1)]
+		refCount, err := getDevicePrefixRefCount(iscsi.mounter, prefix)
+
+		if err == nil && refCount == 0 {
+			// this portal/iqn are no longer referenced, log out
+			// extract portal and iqn from device path
+			ind1 := strings.LastIndex(device, "-iscsi-")
+			portal := device[(len("/dev/disk/by-path/ip-")):ind1]
+			iqn := device[ind1+len("-iscsi-") : ind]
+
+			glog.Infof("iscsi: log out target %s iqn %s", portal, iqn)
+			_, err = iscsi.plugin.execCommand("iscsiadm", []string{"-m", "node", "-p", portal, "-T", iqn, "--logout"})
+			if err != nil {
+				glog.Errorf("iscsi: failed to detach disk Error: %v", err)
+			}
+		}
+	}
+	return nil
+}

pkg/volume/iscsi/iscsi_util_test.go

@@ -0,0 +1,54 @@
+/*
+Copyright 2015 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 iscsi
+
+import (
+	"testing"
+
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/util/mount"
+)
+
+func TestGetDevicePrefixRefCount(t *testing.T) {
+	fm := &mount.FakeMounter{
+		MountPoints: []mount.MountPoint{
+			{Device: "/dev/disk/by-path/prefix-lun-1",
+				Path: "/mnt/111"},
+			{Device: "/dev/disk/by-path/prefix-lun-1",
+				Path: "/mnt/222"},
+			{Device: "/dev/disk/by-path/prefix-lun-0",
+				Path: "/mnt/333"},
+			{Device: "/dev/disk/by-path/prefix-lun-0",
+				Path: "/mnt/444"},
+		},
+	}
+
+	tests := []struct {
+		devicePrefix string
+		expectedRefs int
+	}{
+		{
+			"/dev/disk/by-path/prefix",
+			4,
+		},
+	}
+
+	for i, test := range tests {
+		if refs, err := getDevicePrefixRefCount(fm, test.devicePrefix); err != nil || test.expectedRefs != refs {
+			t.Errorf("%d. GetDevicePrefixRefCount(%s) = %d, %v; expected %d, nil", i, test.devicePrefix, refs, err, test.expectedRefs)
+		}
+	}
+}