Merge pull request #51608 from cofyc/rbd_attach_detach

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

@@ -78,6 +78,7 @@ func ProbeAttachableVolumePlugins() []volume.VolumePlugin {
 	allPlugins = append(allPlugins, storageos.ProbeVolumePlugins()...)
 	allPlugins = append(allPlugins, fc.ProbeVolumePlugins()...)
 	allPlugins = append(allPlugins, iscsi.ProbeVolumePlugins()...)
+	allPlugins = append(allPlugins, rbd.ProbeVolumePlugins()...)
 	return allPlugins
 }
 

pkg/util/mount/fake.go

@@ -17,6 +17,7 @@ limitations under the License.
 package mount
 
 import (
+	"os"
 	"path/filepath"
 	"sync"
 
@@ -136,6 +137,11 @@ func (f *FakeMounter) IsLikelyNotMountPoint(file string) (bool, error) {
 	f.mutex.Lock()
 	defer f.mutex.Unlock()
 
+	_, err := os.Stat(file)
+	if err != nil {
+		return true, err
+	}
+
 	// If file is a symlink, get its absolute path
 	absFile, err := filepath.EvalSymlinks(file)
 	if err != nil {

pkg/util/mount/safe_format_and_mount_test.go

@@ -18,6 +18,8 @@ package mount
 
 import (
 	"fmt"
+	"io/ioutil"
+	"os"
 	"runtime"
 	"testing"
 
@@ -50,6 +52,11 @@ func TestSafeFormatAndMount(t *testing.T) {
 	if runtime.GOOS == "darwin" || runtime.GOOS == "windows" {
 		t.Skipf("not supported on GOOS=%s", runtime.GOOS)
 	}
+	mntDir, err := ioutil.TempDir(os.TempDir(), "mount")
+	if err != nil {
+		t.Fatalf("failed to create tmp dir: %v", err)
+	}
+	defer os.RemoveAll(mntDir)
 	tests := []struct {
 		description   string
 		fstype        string
@@ -207,7 +214,7 @@ func TestSafeFormatAndMount(t *testing.T) {
 		}
 
 		device := "/dev/foo"
-		dest := "/mnt/bar"
+		dest := mntDir
 		err := mounter.FormatAndMount(device, dest, test.fstype, test.mountOptions)
 		if test.expectedError == nil {
 			if err != nil {

pkg/volume/rbd/BUILD

@@ -9,6 +9,7 @@ load(
 go_library(
     name = "go_default_library",
     srcs = [
+        "attacher.go",
         "disk_manager.go",
         "doc.go",
         "rbd.go",
@@ -16,6 +17,7 @@ go_library(
     ],
     importpath = "k8s.io/kubernetes/pkg/volume/rbd",
     deps = [
+        "//pkg/util/file:go_default_library",
         "//pkg/util/mount:go_default_library",
         "//pkg/util/node:go_default_library",
         "//pkg/util/strings:go_default_library",
@@ -42,10 +44,10 @@ go_test(
         "//pkg/util/mount:go_default_library",
         "//pkg/volume:go_default_library",
         "//pkg/volume/testing:go_default_library",
-        "//vendor/github.com/stretchr/testify/assert:go_default_library",
         "//vendor/k8s.io/api/core/v1: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/uuid:go_default_library",
         "//vendor/k8s.io/client-go/kubernetes/fake:go_default_library",
         "//vendor/k8s.io/client-go/util/testing:go_default_library",
     ],

pkg/volume/rbd/attacher.go

+/*
+Copyright 2017 The Kubernetes Authors.
+
+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 rbd
+
+import (
+	"fmt"
+	"os"
+	"time"
+
+	"github.com/golang/glog"
+	"k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/kubernetes/pkg/util/mount"
+	"k8s.io/kubernetes/pkg/volume"
+	volutil "k8s.io/kubernetes/pkg/volume/util"
+)
+
+// NewAttacher implements AttachableVolumePlugin.NewAttacher.
+func (plugin *rbdPlugin) NewAttacher() (volume.Attacher, error) {
+	return plugin.newAttacherInternal(&RBDUtil{})
+}
+
+func (plugin *rbdPlugin) newAttacherInternal(manager diskManager) (volume.Attacher, error) {
+	return &rbdAttacher{
+		plugin:  plugin,
+		manager: manager,
+	}, nil
+}
+
+// NewDetacher implements AttachableVolumePlugin.NewDetacher.
+func (plugin *rbdPlugin) NewDetacher() (volume.Detacher, error) {
+	return plugin.newDetacherInternal(&RBDUtil{})
+}
+
+func (plugin *rbdPlugin) newDetacherInternal(manager diskManager) (volume.Detacher, error) {
+	return &rbdDetacher{
+		plugin:  plugin,
+		manager: manager,
+	}, nil
+}
+
+// GetDeviceMountRefs implements AttachableVolumePlugin.GetDeviceMountRefs.
+func (plugin *rbdPlugin) GetDeviceMountRefs(deviceMountPath string) ([]string, error) {
+	mounter := plugin.host.GetMounter(plugin.GetPluginName())
+	return mount.GetMountRefs(mounter, deviceMountPath)
+}
+
+// rbdAttacher implements volume.Attacher interface.
+type rbdAttacher struct {
+	plugin  *rbdPlugin
+	manager diskManager
+}
+
+var _ volume.Attacher = &rbdAttacher{}
+
+// Attach implements Attacher.Attach.
+// We do not lock image here, because it requires kube-controller-manager to
+// access external `rbd` utility. And there is no need since AttachDetach
+// controller will not try to attach RWO volumes which are already attached to
+// other nodes.
+func (attacher *rbdAttacher) Attach(spec *volume.Spec, nodeName types.NodeName) (string, error) {
+	return "", nil
+}
+
+// VolumesAreAttached implements Attacher.VolumesAreAttached.
+// There is no way to confirm whether the volume is attached or not from
+// outside of the kubelet node. This method needs to return true always, like
+// iSCSI, FC plugin.
+func (attacher *rbdAttacher) VolumesAreAttached(specs []*volume.Spec, nodeName types.NodeName) (map[*volume.Spec]bool, error) {
+	volumesAttachedCheck := make(map[*volume.Spec]bool)
+	for _, spec := range specs {
+		volumesAttachedCheck[spec] = true
+	}
+	return volumesAttachedCheck, nil
+}
+
+// WaitForAttach implements Attacher.WaitForAttach. It's called by kublet to
+// attach volume onto the node.
+// This method is idempotent, callers are responsible for retrying on failure.
+func (attacher *rbdAttacher) WaitForAttach(spec *volume.Spec, devicePath string, pod *v1.Pod, timeout time.Duration) (string, error) {
+	glog.V(4).Infof("rbd: waiting for attach volume (name: %s) for pod (name: %s, uid: %s)", spec.Name(), pod.Name, pod.UID)
+	mounter, err := attacher.plugin.createMounterFromVolumeSpecAndPod(spec, pod)
+	if err != nil {
+		glog.Warningf("failed to create mounter: %v", spec)
+		return "", err
+	}
+	realDevicePath, err := attacher.manager.AttachDisk(*mounter)
+	if err != nil {
+		return "", err
+	}
+	glog.V(3).Infof("rbd: successfully wait for attach volume (spec: %s, pool: %s, image: %s) at %s", spec.Name(), mounter.Pool, mounter.Image, realDevicePath)
+	return realDevicePath, nil
+}
+
+// GetDeviceMountPath implements Attacher.GetDeviceMountPath.
+func (attacher *rbdAttacher) GetDeviceMountPath(spec *volume.Spec) (string, error) {
+	img, err := getVolumeSourceImage(spec)
+	if err != nil {
+		return "", err
+	}
+	pool, err := getVolumeSourcePool(spec)
+	if err != nil {
+		return "", err
+	}
+	return makePDNameInternal(attacher.plugin.host, pool, img), nil
+}
+
+// MountDevice implements Attacher.MountDevice. It is called by the kubelet to
+// mount device at the given mount path.
+// This method is idempotent, callers are responsible for retrying on failure.
+func (attacher *rbdAttacher) MountDevice(spec *volume.Spec, devicePath string, deviceMountPath string) error {
+	glog.V(4).Infof("rbd: mouting device %s to %s", devicePath, deviceMountPath)
+	notMnt, err := attacher.plugin.mounter.IsLikelyNotMountPoint(deviceMountPath)
+	if err != nil {
+		if os.IsNotExist(err) {
+			if err := os.MkdirAll(deviceMountPath, 0750); err != nil {
+				return err
+			}
+			notMnt = true
+		} else {
+			return err
+		}
+	}
+	if !notMnt {
+		return nil
+	}
+	fstype, err := getVolumeSourceFSType(spec)
+	if err != nil {
+		return err
+	}
+	ro, err := getVolumeSourceReadOnly(spec)
+	if err != nil {
+		return err
+	}
+	options := []string{}
+	if ro {
+		options = append(options, "ro")
+	}
+	mountOptions := volume.MountOptionFromSpec(spec, options...)
+	err = attacher.plugin.mounter.FormatAndMount(devicePath, deviceMountPath, fstype, mountOptions)
+	if err != nil {
+		os.Remove(deviceMountPath)
+		return fmt.Errorf("rbd: failed to mount device %s at %s (fstype: %s), error %v", devicePath, deviceMountPath, fstype, err)
+	}
+	glog.V(3).Infof("rbd: successfully mount device %s at %s (fstype: %s)", devicePath, deviceMountPath, fstype)
+	return nil
+}
+
+// rbdDetacher implements volume.Detacher interface.
+type rbdDetacher struct {
+	plugin  *rbdPlugin
+	manager diskManager
+}
+
+var _ volume.Detacher = &rbdDetacher{}
+
+// UnmountDevice implements Detacher.UnmountDevice. It unmounts the global
+// mount of the RBD image. This is called once all bind mounts have been
+// unmounted.
+// Internally, it does four things:
+//  - Unmount device from deviceMountPath.
+//  - Detach device from the node.
+//  - Remove lock if found. (No need to check volume readonly or not, because
+//  device is not on the node anymore, it's safe to remove lock.)
+//  - Remove the deviceMountPath at last.
+// This method is idempotent, callers are responsible for retrying on failure.
+func (detacher *rbdDetacher) UnmountDevice(deviceMountPath string) error {
+	if pathExists, pathErr := volutil.PathExists(deviceMountPath); pathErr != nil {
+		return fmt.Errorf("Error checking if path exists: %v", pathErr)
+	} else if !pathExists {
+		glog.Warningf("Warning: Unmount skipped because path does not exist: %v", deviceMountPath)
+		return nil
+	}
+	devicePath, cnt, err := mount.GetDeviceNameFromMount(detacher.plugin.mounter, deviceMountPath)
+	if err != nil {
+		return err
+	}
+	if cnt > 1 {
+		return fmt.Errorf("rbd: more than 1 reference counts at %s", deviceMountPath)
+	}
+	if cnt == 1 {
+		// Unmount the device from the device mount point.
+		glog.V(4).Infof("rbd: unmouting device mountpoint %s", deviceMountPath)
+		if err = detacher.plugin.mounter.Unmount(deviceMountPath); err != nil {
+			return err
+		}
+		glog.V(3).Infof("rbd: successfully umount device mountpath %s", deviceMountPath)
+	}
+	glog.V(4).Infof("rbd: detaching device %s", devicePath)
+	err = detacher.manager.DetachDisk(detacher.plugin, deviceMountPath, devicePath)
+	if err != nil {
+		return err
+	}
+	glog.V(3).Infof("rbd: successfully detach device %s", devicePath)
+	err = os.Remove(deviceMountPath)
+	if err != nil {
+		return err
+	}
+	glog.V(3).Infof("rbd: successfully remove device mount point %s", deviceMountPath)
+	return nil
+}
+
+// Detach implements Detacher.Detach.
+func (detacher *rbdDetacher) Detach(deviceName string, nodeName types.NodeName) error {
+	return nil
+}

pkg/volume/rbd/disk_manager.go

@@ -23,6 +23,7 @@ limitations under the License.
 package rbd
 
 import (
+	"fmt"
 	"os"
 
 	"github.com/golang/glog"
@@ -33,23 +34,33 @@ import (
 
 // Abstract interface to disk operations.
 type diskManager interface {
+	// MakeGlobalPDName creates global persistent disk path.
 	MakeGlobalPDName(disk rbd) string
 	// Attaches the disk to the kubelet's host machine.
-	AttachDisk(disk rbdMounter) error
+	// If it successfully attaches, the path to the device
+	// is returned. Otherwise, an error will be returned.
+	AttachDisk(disk rbdMounter) (string, error)
 	// Detaches the disk from the kubelet's host machine.
-	DetachDisk(disk rbdUnmounter, mntPath string) error
-	// Creates a rbd image
+	DetachDisk(plugin *rbdPlugin, deviceMountPath string, device string) error
+	// Creates a rbd image.
 	CreateImage(provisioner *rbdVolumeProvisioner) (r *v1.RBDPersistentVolumeSource, volumeSizeGB int, err error)
-	// Deletes a rbd image
+	// Deletes a rbd image.
 	DeleteImage(deleter *rbdVolumeDeleter) error
 }
 
 // utility to mount a disk based filesystem
 func diskSetUp(manager diskManager, b rbdMounter, volPath string, mounter mount.Interface, fsGroup *int64) error {
 	globalPDPath := manager.MakeGlobalPDName(*b.rbd)
-	// TODO: handle failed mounts here.
-	notMnt, err := mounter.IsLikelyNotMountPoint(volPath)
+	notMnt, err := mounter.IsLikelyNotMountPoint(globalPDPath)
+	if err != nil && !os.IsNotExist(err) {
+		glog.Errorf("cannot validate mountpoint: %s", globalPDPath)
+		return err
+	}
+	if notMnt {
+		return fmt.Errorf("no device is mounted at %s", globalPDPath)
+	}
 
+	notMnt, err = mounter.IsLikelyNotMountPoint(volPath)
 	if err != nil && !os.IsNotExist(err) {
 		glog.Errorf("cannot validate mountpoint: %s", volPath)
 		return err
@@ -57,10 +68,6 @@ func diskSetUp(manager diskManager, b rbdMounter, volPath string, mounter mount.
 	if !notMnt {
 		return nil
 	}
-	if err := manager.AttachDisk(b); 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)
@@ -89,43 +96,31 @@ func diskSetUp(manager diskManager, b rbdMounter, volPath string, mounter mount.
 // utility to tear down a disk based filesystem
 func diskTearDown(manager diskManager, c rbdUnmounter, volPath string, mounter mount.Interface) error {
 	notMnt, err := mounter.IsLikelyNotMountPoint(volPath)
-	if err != nil {
-		glog.Errorf("cannot validate mountpoint %s", volPath)
+	if err != nil && !os.IsNotExist(err) {
+		glog.Errorf("cannot validate mountpoint: %s", volPath)
 		return err
 	}
 	if notMnt {
+		glog.V(3).Infof("volume path %s is not a mountpoint, deleting", volPath)
 		return os.Remove(volPath)
 	}
 
-	refs, err := mount.GetMountRefs(mounter, volPath)
-	if err != nil {
-		glog.Errorf("failed to get reference count %s", volPath)
-		return err
-	}
+	// Unmount the bind-mount inside this pod.
 	if err := mounter.Unmount(volPath); 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(c, mntPath); err != nil {
-			glog.Errorf("failed to detach disk from %s", mntPath)
-			return err
-		}
-	}
 
 	notMnt, mntErr := mounter.IsLikelyNotMountPoint(volPath)
-	if mntErr != nil {
+	if err != nil && !os.IsNotExist(err) {
 		glog.Errorf("IsLikelyNotMountPoint check failed: %v", mntErr)
 		return err
 	}
 	if notMnt {
 		if err := os.Remove(volPath); err != nil {
+			glog.V(2).Info("Error removing mountpoint ", volPath, ": ", err)
 			return err
 		}
 	}
 	return nil
-
 }