Adding fsGroup check before mounting a volume

fsGroup check will be enforcing that if a volume has already been
mounted by one pod and another pod wants to mount it but has a different
fsGroup value, this mount operation will not be allowed.

pkg/kubelet/events/event.go

@@ -46,6 +46,7 @@ const (
 	FailedDetachVolume                   = "FailedDetachVolume"
 	FailedMountVolume                    = "FailedMount"
 	FailedUnMountVolume                  = "FailedUnMount"
+	WarnAlreadyMountedVolume             = "AlreadyMountedVolume"
 	SuccessfulDetachVolume               = "SuccessfulDetachVolume"
 	SuccessfulMountVolume                = "SuccessfulMountVolume"
 	SuccessfulUnMountVolume              = "SuccessfulUnMountVolume"

pkg/util/mount/mount.go

@@ -123,7 +123,6 @@ func GetMountRefs(mounter Interface, mountPath string) ([]string, error) {
 	if err != nil {
 		return nil, err
 	}
-
 	// Find the device name.
 	deviceName := ""
 	// If mountPath is symlink, need get its target path.
@@ -152,6 +151,39 @@ func GetMountRefs(mounter Interface, mountPath string) ([]string, error) {
 	return refs, nil
 }
 
+// GetMountRefsByDev finds all references to the device provided
+// by mountPath; returns a list of paths.
+func GetMountRefsByDev(mounter Interface, mountPath string) ([]string, error) {
+	mps, err := mounter.List()
+	if err != nil {
+		return nil, err
+	}
+	slTarget, err := filepath.EvalSymlinks(mountPath)
+	if err != nil {
+		slTarget = mountPath
+	}
+
+	// Finding the device mounted to mountPath
+	diskDev := ""
+	for i := range mps {
+		if slTarget == mps[i].Path {
+			diskDev = mps[i].Device
+			break
+		}
+	}
+
+	// Find all references to the device.
+	var refs []string
+	for i := range mps {
+		if mps[i].Device == diskDev || mps[i].Device == slTarget {
+			if mps[i].Path != slTarget {
+				refs = append(refs, mps[i].Path)
+			}
+		}
+	}
+	return refs, nil
+}
+
 // GetDeviceNameFromMount: given a mnt point, find the device from /proc/mounts
 // returns the device name, reference count, and error code
 func GetDeviceNameFromMount(mounter Interface, mountPath string) (string, int, error) {

pkg/util/mount/mount_linux_test.go

@@ -170,3 +170,41 @@ func TestGetDeviceNameFromMount(t *testing.T) {
 		}
 	}
 }
+
+func TestGetMountRefsByDev(t *testing.T) {
+	fm := &FakeMounter{
+		MountPoints: []MountPoint{
+			{Device: "/dev/sdb", Path: "/var/lib/kubelet/plugins/kubernetes.io/gce-pd/mounts/gce-pd"},
+			{Device: "/dev/sdb", Path: "/var/lib/kubelet/pods/some-pod/volumes/kubernetes.io~gce-pd/gce-pd-in-pod"},
+			{Device: "/dev/sdc", Path: "/var/lib/kubelet/plugins/kubernetes.io/gce-pd/mounts/gce-pd2"},
+			{Device: "/dev/sdc", Path: "/var/lib/kubelet/pods/some-pod/volumes/kubernetes.io~gce-pd/gce-pd2-in-pod1"},
+			{Device: "/dev/sdc", Path: "/var/lib/kubelet/pods/some-pod/volumes/kubernetes.io~gce-pd/gce-pd2-in-pod2"},
+		},
+	}
+
+	tests := []struct {
+		mountPath    string
+		expectedRefs []string
+	}{
+		{
+			"/var/lib/kubelet/plugins/kubernetes.io/gce-pd/mounts/gce-pd",
+			[]string{
+				"/var/lib/kubelet/pods/some-pod/volumes/kubernetes.io~gce-pd/gce-pd-in-pod",
+			},
+		},
+		{
+			"/var/lib/kubelet/plugins/kubernetes.io/gce-pd/mounts/gce-pd2",
+			[]string{
+				"/var/lib/kubelet/pods/some-pod/volumes/kubernetes.io~gce-pd/gce-pd2-in-pod1",
+				"/var/lib/kubelet/pods/some-pod/volumes/kubernetes.io~gce-pd/gce-pd2-in-pod2",
+			},
+		},
+	}
+
+	for i, test := range tests {
+
+		if refs, err := GetMountRefsByDev(fm, test.mountPath); err != nil || !setEquivalent(test.expectedRefs, refs) {
+			t.Errorf("%d. getMountRefsByDev(%q) = %v, %v; expected %v, nil", i, test.mountPath, refs, err, test.expectedRefs)
+		}
+	}
+}

pkg/volume/local/BUILD

@@ -13,6 +13,8 @@ go_library(
         "local.go",
     ],
     deps = [
+        "//pkg/kubelet/events:go_default_library",
+        "//pkg/util/keymutex:go_default_library",
         "//pkg/util/mount:go_default_library",
         "//pkg/util/strings:go_default_library",
         "//pkg/volume:go_default_library",
@@ -22,6 +24,8 @@ go_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/client-go/kubernetes/scheme:go_default_library",
+        "//vendor/k8s.io/client-go/tools/record:go_default_library",
     ],
 )
 

pkg/volume/local/local.go

@@ -18,13 +18,17 @@ package local
 
 import (
 	"fmt"
-	"os"
-
 	"github.com/golang/glog"
+	"os"
+	"syscall"
 
 	"k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/client-go/tools/record"
+	"k8s.io/kubernetes/pkg/kubelet/events"
+	"k8s.io/kubernetes/pkg/util/keymutex"
 	"k8s.io/kubernetes/pkg/util/mount"
 	"k8s.io/kubernetes/pkg/util/strings"
 	"k8s.io/kubernetes/pkg/volume"
@@ -38,7 +42,9 @@ func ProbeVolumePlugins() []volume.VolumePlugin {
 }
 
 type localVolumePlugin struct {
-	host volume.VolumeHost
+	host        volume.VolumeHost
+	volumeLocks keymutex.KeyMutex
+	recorder    record.EventRecorder
 }
 
 var _ volume.VolumePlugin = &localVolumePlugin{}
@@ -50,6 +56,11 @@ const (
 
 func (plugin *localVolumePlugin) Init(host volume.VolumeHost) error {
 	plugin.host = host
+	plugin.volumeLocks = keymutex.NewKeyMutex()
+	eventBroadcaster := record.NewBroadcaster()
+	eventBroadcaster.StartLogging(glog.Infof)
+	recorder := eventBroadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: "localvolume"})
+	plugin.recorder = recorder
 	return nil
 }
 
@@ -102,6 +113,7 @@ func (plugin *localVolumePlugin) NewMounter(spec *volume.Spec, pod *v1.Pod, _ vo
 
 	return &localVolumeMounter{
 		localVolume: &localVolume{
+			pod:             pod,
 			podUID:          pod.UID,
 			volName:         spec.Name(),
 			mounter:         plugin.host.GetMounter(plugin.GetPluginName()),
@@ -146,6 +158,7 @@ func (plugin *localVolumePlugin) ConstructVolumeSpec(volumeName, mountPath strin
 // The directory at the globalPath will be bind-mounted to the pod's directory
 type localVolume struct {
 	volName string
+	pod     *v1.Pod
 	podUID  types.UID
 	// Global path to the volume
 	globalPath string
@@ -188,6 +201,9 @@ func (m *localVolumeMounter) SetUp(fsGroup *int64) error {
 
 // SetUpAt bind mounts the directory to the volume path and sets up volume ownership
 func (m *localVolumeMounter) SetUpAt(dir string, fsGroup *int64) error {
+	m.plugin.volumeLocks.LockKey(m.globalPath)
+	defer m.plugin.volumeLocks.UnlockKey(m.globalPath)
+
 	if m.globalPath == "" {
 		err := fmt.Errorf("LocalVolume volume %q path is empty", m.volName)
 		return err
@@ -204,9 +220,30 @@ func (m *localVolumeMounter) SetUpAt(dir string, fsGroup *int64) error {
 		glog.Errorf("cannot validate mount point: %s %v", dir, err)
 		return err
 	}
+
 	if !notMnt {
 		return nil
 	}
+	refs, err := mount.GetMountRefsByDev(m.mounter, m.globalPath)
+	if fsGroup != nil {
+		if err != nil {
+			glog.Errorf("cannot collect mounting information: %s %v", m.globalPath, err)
+			return err
+		}
+
+		if len(refs) > 0 {
+			fsGroupNew := int64(*fsGroup)
+			fsGroupSame, fsGroupOld, err := isSameFSGroup(m.globalPath, fsGroupNew)
+			if err != nil {
+				err = fmt.Errorf("failed to check fsGroup for %s (%v)", m.globalPath, err)
+				return err
+			}
+			if !fsGroupSame {
+				m.plugin.recorder.Eventf(m.pod, v1.EventTypeWarning, events.WarnAlreadyMountedVolume, "The requested fsGroup is %d, but the volume %s has GID %d. The volume may not be shareable.", fsGroupNew, m.volName, fsGroupOld)
+			}
+		}
+
+	}
 
 	if err := os.MkdirAll(dir, 0750); err != nil {
 		glog.Errorf("mkdir failed on disk %s (%v)", dir, err)
@@ -247,14 +284,29 @@ func (m *localVolumeMounter) SetUpAt(dir string, fsGroup *int64) error {
 		os.Remove(dir)
 		return err
 	}
-
 	if !m.readOnly {
-		// TODO: how to prevent multiple mounts with conflicting fsGroup?
-		return volume.SetVolumeOwnership(m, fsGroup)
+		// Volume owner will be written only once on the first volume mount
+		if len(refs) == 0 {
+			return volume.SetVolumeOwnership(m, fsGroup)
+		}
 	}
 	return nil
 }
 
+// isSameFSGroup is called only for requests to mount an already mounted
+// volume. It checks if fsGroup of new mount request is the same or not.
+// It returns false if it not the same. It also returns current Gid of a path
+// provided for dir variable.
+func isSameFSGroup(dir string, fsGroup int64) (bool, int, error) {
+	info, err := os.Stat(dir)
+	if err != nil {
+		glog.Errorf("Error getting stats for %s (%v)", dir, err)
+		return false, 0, err
+	}
+	s := info.Sys().(*syscall.Stat_t)
+	return int(s.Gid) == int(fsGroup), int(s.Gid), nil
+}
+
 type localVolumeUnmounter struct {
 	*localVolume
 }

pkg/volume/local/local_test.go

@@ -17,8 +17,10 @@ limitations under the License.
 package local
 
 import (
+	"fmt"
 	"os"
 	"path"
+	"syscall"
 	"testing"
 
 	"k8s.io/api/core/v1"
@@ -104,7 +106,7 @@ func TestCanSupport(t *testing.T) {
 	tmpDir, plug := getPlugin(t)
 	defer os.RemoveAll(tmpDir)
 
-	if !plug.CanSupport(getTestVolume(false, "/test-vol")) {
+	if !plug.CanSupport(getTestVolume(false, tmpDir)) {
 		t.Errorf("Expected true")
 	}
 }
@@ -130,7 +132,7 @@ func TestGetVolumeName(t *testing.T) {
 	tmpDir, plug := getPersistentPlugin(t)
 	defer os.RemoveAll(tmpDir)
 
-	volName, err := plug.GetVolumeName(getTestVolume(false, "/test-vol"))
+	volName, err := plug.GetVolumeName(getTestVolume(false, tmpDir))
 	if err != nil {
 		t.Errorf("Failed to get volume name: %v", err)
 	}
@@ -161,7 +163,7 @@ func TestMountUnmount(t *testing.T) {
 	defer os.RemoveAll(tmpDir)
 
 	pod := &v1.Pod{ObjectMeta: metav1.ObjectMeta{UID: types.UID("poduid")}}
-	mounter, err := plug.NewMounter(getTestVolume(false, "/test-vol"), pod, volume.VolumeOptions{})
+	mounter, err := plug.NewMounter(getTestVolume(false, tmpDir), pod, volume.VolumeOptions{})
 	if err != nil {
 		t.Errorf("Failed to make a new Mounter: %v", err)
 	}
@@ -204,6 +206,66 @@ func TestMountUnmount(t *testing.T) {
 	}
 }
 
+func testFSGroupMount(plug volume.VolumePlugin, pod *v1.Pod, tmpDir string, fsGroup int64) error {
+	mounter, err := plug.NewMounter(getTestVolume(false, tmpDir), pod, volume.VolumeOptions{})
+	if err != nil {
+		return err
+	}
+	if mounter == nil {
+		return fmt.Errorf("Got a nil Mounter")
+	}
+
+	volPath := path.Join(tmpDir, testMountPath)
+	path := mounter.GetPath()
+	if path != volPath {
+		return fmt.Errorf("Got unexpected path: %s", path)
+	}
+
+	if err := mounter.SetUp(&fsGroup); err != nil {
+		return err
+	}
+	return nil
+}
+
+func TestFSGroupMount(t *testing.T) {
+	tmpDir, plug := getPlugin(t)
+	defer os.RemoveAll(tmpDir)
+	info, err := os.Stat(tmpDir)
+	if err != nil {
+		t.Errorf("Error getting stats for %s (%v)", tmpDir, err)
+	}
+	s := info.Sys().(*syscall.Stat_t)
+	if s == nil {
+		t.Errorf("Error getting stats for %s (%v)", tmpDir, err)
+	}
+	fsGroup1 := int64(s.Gid)
+	fsGroup2 := fsGroup1 + 1
+	pod1 := &v1.Pod{ObjectMeta: metav1.ObjectMeta{UID: types.UID("poduid")}}
+	pod1.Spec.SecurityContext = &v1.PodSecurityContext{
+		FSGroup: &fsGroup1,
+	}
+	pod2 := &v1.Pod{ObjectMeta: metav1.ObjectMeta{UID: types.UID("poduid")}}
+	pod2.Spec.SecurityContext = &v1.PodSecurityContext{
+		FSGroup: &fsGroup2,
+	}
+	err = testFSGroupMount(plug, pod1, tmpDir, fsGroup1)
+	if err != nil {
+		t.Errorf("Failed to make a new Mounter: %v", err)
+	}
+	err = testFSGroupMount(plug, pod2, tmpDir, fsGroup2)
+	if err != nil {
+		t.Errorf("Failed to make a new Mounter: %v", err)
+	}
+	//Checking if GID of tmpDir has not been changed by mounting it by second pod
+	s = info.Sys().(*syscall.Stat_t)
+	if s == nil {
+		t.Errorf("Error getting stats for %s (%v)", tmpDir, err)
+	}
+	if fsGroup1 != int64(s.Gid) {
+		t.Errorf("Old Gid %d for volume %s got overwritten by new Gid %d", fsGroup1, tmpDir, int64(s.Gid))
+	}
+}
+
 func TestConstructVolumeSpec(t *testing.T) {
 	tmpDir, plug := getPlugin(t)
 	defer os.RemoveAll(tmpDir)
@@ -243,7 +305,7 @@ func TestPersistentClaimReadOnlyFlag(t *testing.T) {
 
 	// Read only == true
 	pod := &v1.Pod{ObjectMeta: metav1.ObjectMeta{UID: types.UID("poduid")}}
-	mounter, err := plug.NewMounter(getTestVolume(true, "/test-vol"), pod, volume.VolumeOptions{})
+	mounter, err := plug.NewMounter(getTestVolume(true, tmpDir), pod, volume.VolumeOptions{})
 	if err != nil {
 		t.Errorf("Failed to make a new Mounter: %v", err)
 	}
@@ -255,7 +317,7 @@ func TestPersistentClaimReadOnlyFlag(t *testing.T) {
 	}
 
 	// Read only == false
-	mounter, err = plug.NewMounter(getTestVolume(false, "/test-vol"), pod, volume.VolumeOptions{})
+	mounter, err = plug.NewMounter(getTestVolume(false, tmpDir), pod, volume.VolumeOptions{})
 	if err != nil {
 		t.Errorf("Failed to make a new Mounter: %v", err)
 	}
@@ -276,7 +338,7 @@ func TestUnsupportedPlugins(t *testing.T) {
 
 	plugMgr := volume.VolumePluginMgr{}
 	plugMgr.InitPlugins(ProbeVolumePlugins(), volumetest.NewFakeVolumeHost(tmpDir, nil, nil))
-	spec := getTestVolume(false, "/test-vol")
+	spec := getTestVolume(false, tmpDir)
 
 	recyclePlug, err := plugMgr.FindRecyclablePluginBySpec(spec)
 	if err == nil && recyclePlug != nil {