Merge pull request #4718 from cjcullen/pdtest

pkg/kubelet/volume/gce_pd/gce_pd.go

@@ -128,9 +128,9 @@ func (plugin *gcePersistentDiskPlugin) newCleanerInternal(volName string, podUID
 // Abstract interface to PD operations.
 type pdManager interface {
 	// Attaches the disk to the kubelet's host machine.
-	AttachDisk(pd *gcePersistentDisk) error
+	AttachAndMountDisk(pd *gcePersistentDisk, globalPDPath string) error
 	// Detaches the disk from the kubelet's host machine.
-	DetachDisk(pd *gcePersistentDisk, devicePath string) error
+	DetachDisk(pd *gcePersistentDisk) error
 }
 
 // gcePersistentDisk volumes are disk resources provided by Google Compute Engine
@@ -157,7 +157,7 @@ type gcePersistentDisk struct {
 }
 
 func detachDiskLogError(pd *gcePersistentDisk) {
-	err := pd.manager.DetachDisk(pd, "/dev/disk/by-id/google-"+pd.pdName)
+	err := pd.manager.DetachDisk(pd)
 	if err != nil {
 		glog.Warningf("Failed to detach disk: %v (%v)", pd, err)
 	}
@@ -179,7 +179,8 @@ func (pd *gcePersistentDisk) SetUp() error {
 		return nil
 	}
 
-	if err := pd.manager.AttachDisk(pd); err != nil {
+	globalPDPath := makeGlobalPDName(pd.plugin.host, pd.pdName)
+	if err := pd.manager.AttachAndMountDisk(pd, globalPDPath); err != nil {
 		return err
 	}
 
@@ -196,7 +197,6 @@ func (pd *gcePersistentDisk) SetUp() error {
 	}
 
 	// Perform a bind mount to the full path to allow duplicate mounts of the same PD.
-	globalPDPath := makeGlobalPDName(pd.plugin.host, pd.pdName, pd.readOnly)
 	err = pd.mounter.Mount(globalPDPath, pd.GetPath(), "", mount.FlagBind|flags, "")
 	if err != nil {
 		mountpoint, mntErr := isMountPoint(pd.GetPath())
@@ -229,7 +229,7 @@ func (pd *gcePersistentDisk) SetUp() error {
 	return nil
 }
 
-func makeGlobalPDName(host volume.Host, devName string, readOnly bool) string {
+func makeGlobalPDName(host volume.Host, devName string) string {
 	return path.Join(host.GetPluginDir(gcePersistentDiskPluginName), "mounts", devName)
 }
 
@@ -252,18 +252,20 @@ func (pd *gcePersistentDisk) TearDown() error {
 		return os.Remove(pd.GetPath())
 	}
 
-	devicePath, refCount, err := getMountRefCount(pd.mounter, pd.GetPath())
+	refs, err := getMountRefs(pd.mounter, pd.GetPath())
 	if err != nil {
 		return err
 	}
+	// Unmount the bind-mount inside this pod
 	if err := pd.mounter.Unmount(pd.GetPath(), 0); err != nil {
 		return err
 	}
-	refCount--
-	// If refCount is 1, then all bind mounts have been removed, and the
+	// 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 refCount == 1 {
-		if err := pd.manager.DetachDisk(pd, devicePath); err != nil {
+	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
 		}
 	}

pkg/kubelet/volume/gce_pd/gce_pd_test.go

@@ -23,7 +23,6 @@ import (
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/volume"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/types"
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/util/mount"
 )
 
 func TestCanSupport(t *testing.T) {
@@ -46,8 +45,8 @@ type fakePDManager struct{}
 
 // TODO(jonesdl) To fully test this, we could create a loopback device
 // and mount that instead.
-func (fake *fakePDManager) AttachDisk(pd *gcePersistentDisk) error {
-	globalPath := makeGlobalPDName(pd.plugin.host, pd.pdName, pd.readOnly)
+func (fake *fakePDManager) AttachAndMountDisk(pd *gcePersistentDisk, globalPDPath string) error {
+	globalPath := makeGlobalPDName(pd.plugin.host, pd.pdName)
 	err := os.MkdirAll(globalPath, 0750)
 	if err != nil {
 		return err
@@ -55,8 +54,8 @@ func (fake *fakePDManager) AttachDisk(pd *gcePersistentDisk) error {
 	return nil
 }
 
-func (fake *fakePDManager) DetachDisk(pd *gcePersistentDisk, devicePath string) error {
-	globalPath := makeGlobalPDName(pd.plugin.host, pd.pdName, pd.readOnly)
+func (fake *fakePDManager) DetachDisk(pd *gcePersistentDisk) error {
+	globalPath := makeGlobalPDName(pd.plugin.host, pd.pdName)
 	err := os.RemoveAll(globalPath)
 	if err != nil {
 		return err
@@ -64,20 +63,6 @@ func (fake *fakePDManager) DetachDisk(pd *gcePersistentDisk, devicePath string)
 	return nil
 }
 
-type fakeMounter struct{}
-
-func (fake *fakeMounter) Mount(source string, target string, fstype string, flags uintptr, data string) error {
-	return nil
-}
-
-func (fake *fakeMounter) Unmount(target string, flags int) error {
-	return nil
-}
-
-func (fake *fakeMounter) List() ([]mount.MountPoint, error) {
-	return []mount.MountPoint{}, nil
-}
-
 func TestPlugin(t *testing.T) {
 	plugMgr := volume.PluginMgr{}
 	plugMgr.InitPlugins(ProbeVolumePlugins(), &volume.FakeHost{"/tmp/fake", nil})

pkg/kubelet/volume/gce_pd/gce_util.go

@@ -21,9 +21,6 @@ import (
 	"fmt"
 	"os"
 	"path"
-	"path/filepath"
-	"regexp"
-	"strings"
 	"time"
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/cloudprovider"
@@ -33,15 +30,11 @@ import (
 	"github.com/golang/glog"
 )
 
-const partitionRegex = "[a-z][a-z]*(?P<partition>[0-9][0-9]*)?"
-
-var regexMatcher = regexp.MustCompile(partitionRegex)
-
 type GCEDiskUtil struct{}
 
 // Attaches a disk specified by a volume.GCEPersistentDisk to the current kubelet.
 // Mounts the disk to it's global path.
-func (util *GCEDiskUtil) AttachDisk(pd *gcePersistentDisk) error {
+func (util *GCEDiskUtil) AttachAndMountDisk(pd *gcePersistentDisk, globalPDPath string) error {
 	gce, err := cloudprovider.GetCloudProvider("gce", nil)
 	if err != nil {
 		return err
@@ -73,7 +66,7 @@ func (util *GCEDiskUtil) AttachDisk(pd *gcePersistentDisk) error {
 		}
 		time.Sleep(time.Second)
 	}
-	globalPDPath := makeGlobalPDName(pd.plugin.host, pd.pdName, pd.readOnly)
+
 	// Only mount the PD globally once.
 	mountpoint, err := isMountPoint(globalPDPath)
 	if err != nil {
@@ -96,47 +89,22 @@ func (util *GCEDiskUtil) AttachDisk(pd *gcePersistentDisk) error {
 	return nil
 }
 
-func getDeviceName(devicePath, canonicalDevicePath string) (string, error) {
-	isMatch := regexMatcher.MatchString(path.Base(canonicalDevicePath))
-	if !isMatch {
-		return "", fmt.Errorf("unexpected device: %s", canonicalDevicePath)
-	}
-	if isMatch {
-		result := make(map[string]string)
-		substrings := regexMatcher.FindStringSubmatch(path.Base(canonicalDevicePath))
-		for i, label := range regexMatcher.SubexpNames() {
-			result[label] = substrings[i]
-		}
-		partition := result["partition"]
-		devicePath = strings.TrimSuffix(devicePath, "-part"+partition)
-	}
-	return strings.TrimPrefix(path.Base(devicePath), "google-"), nil
-}
-
 // Unmounts the device and detaches the disk from the kubelet's host machine.
-// Expects a GCE device path symlink. Ex: /dev/disk/by-id/google-mydisk-part1
-func (util *GCEDiskUtil) DetachDisk(pd *gcePersistentDisk, devicePath string) error {
-	// Follow the symlink to the actual device path.
-	canonicalDevicePath, err := filepath.EvalSymlinks(devicePath)
-	if err != nil {
-		return err
-	}
-	deviceName, err := getDeviceName(devicePath, canonicalDevicePath)
-	if err != nil {
-		return err
-	}
-	globalPDPath := makeGlobalPDName(pd.plugin.host, deviceName, pd.readOnly)
+func (util *GCEDiskUtil) DetachDisk(pd *gcePersistentDisk) 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
 	gce, err := cloudprovider.GetCloudProvider("gce", nil)
 	if err != nil {
 		return err
 	}
-	if err := gce.(*gce_cloud.GCECloud).DetachDisk(deviceName); err != nil {
+	if err := gce.(*gce_cloud.GCECloud).DetachDisk(pd.pdName); err != nil {
 		return err
 	}
 	return nil
@@ -153,6 +121,10 @@ type gceSafeFormatAndMount struct {
 
 // uses /usr/share/google/safe_format_and_mount to optionally mount, and format a disk
 func (mounter *gceSafeFormatAndMount) 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" {

pkg/kubelet/volume/gce_pd/gce_util_test.go

@@ -23,40 +23,6 @@ import (
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/util/exec"
 )
 
-func TestGetDeviceName(t *testing.T) {
-	tests := []struct {
-		deviceName    string
-		canonicalName string
-		expectedName  string
-		expectError   bool
-	}{
-		{
-			deviceName:    "/dev/google-sd0-part0",
-			canonicalName: "/dev/google/sd0P1",
-			expectedName:  "sd0",
-		},
-		{
-			canonicalName: "0123456",
-			expectError:   true,
-		},
-	}
-	for _, test := range tests {
-		name, err := getDeviceName(test.deviceName, test.canonicalName)
-		if test.expectError {
-			if err == nil {
-				t.Error("unexpected non-error")
-			}
-			continue
-		}
-		if err != nil {
-			t.Errorf("unexpected error: %v", err)
-		}
-		if name != test.expectedName {
-			t.Errorf("expected: %s, got %s", test.expectedName, name)
-		}
-	}
-}
-
 func TestSafeFormatAndMount(t *testing.T) {
 	tests := []struct {
 		fstype       string

pkg/kubelet/volume/gce_pd/mount_util.go

@@ -20,17 +20,12 @@ import (
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/util/mount"
 )
 
-// Examines /proc/mounts to find the source device of the PD resource and the
-// number of references to that device. Returns both the full device path under
-// the /dev tree and the number of references.
-func getMountRefCount(mounter mount.Interface, mountPath string) (string, int, error) {
-	// TODO(jonesdl) This can be split up into two procedures, finding the device path
-	// and finding the number of references. The parsing could also be separated and another
-	// utility could determine if a path is an active mount point.
-
+// Examines /proc/mounts to find all other references to the device referenced
+// by mountPath.
+func getMountRefs(mounter mount.Interface, mountPath string) ([]string, error) {
 	mps, err := mounter.List()
 	if err != nil {
-		return "", -1, err
+		return nil, err
 	}
 
 	// Find the device name.
@@ -42,12 +37,12 @@ func getMountRefCount(mounter mount.Interface, mountPath string) (string, int, e
 		}
 	}
 
-	// Find the number of references to the device.
-	refCount := 0
+	// Find all references to the device.
+	var refs []string
 	for i := range mps {
-		if mps[i].Device == deviceName {
-			refCount++
+		if mps[i].Device == deviceName && mps[i].Path != mountPath {
+			refs = append(refs, mps[i].Path)
 		}
 	}
-	return deviceName, refCount, nil
+	return refs, nil
 }

pkg/kubelet/volume/gce_pd/mount_util_test.go

+/*
+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 gce_pd
+
+import (
+	"testing"
+
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/util/mount"
+)
+
+type fakeMounter struct {
+	mountPoints []mount.MountPoint
+}
+
+func (f *fakeMounter) Mount(source string, target string, fstype string, flags uintptr, data string) error {
+	return nil
+}
+
+func (f *fakeMounter) Unmount(target string, flags int) error {
+	return nil
+}
+
+func (f *fakeMounter) List() ([]mount.MountPoint, error) {
+	return f.mountPoints, nil
+}
+
+func TestGetMountRefs(t *testing.T) {
+	fm := &fakeMounter{
+		[]mount.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-pod"},
+			{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/pods/some-pod/volumes/kubernetes.io~gce-pd/gce-pd-in-pod",
+			[]string{
+				"/var/lib/kubelet/plugins/kubernetes.io/gce-pd/mounts/gce-pd",
+			},
+		},
+		{
+			"/var/lib/kubelet/pods/some-pod/volumes/kubernetes.io~gce-pd/gce-pd2-in-pod",
+			[]string{
+				"/var/lib/kubelet/pods/some-pod/volumes/kubernetes.io~gce-pd/gce-pd2-in-pod2",
+				"/var/lib/kubelet/plugins/kubernetes.io/gce-pd/mounts/gce-pd2",
+			},
+		},
+	}
+
+	for i, test := range tests {
+		if refs, err := getMountRefs(fm, test.mountPath); err != nil || !setEquivalent(test.expectedRefs, refs) {
+			t.Errorf("%d. getMountRefs(%q) = %v, %v; expected %v, nil", i, test.mountPath, refs, err, test.expectedRefs)
+		}
+	}
+}
+
+func setEquivalent(set1, set2 []string) bool {
+	map1 := make(map[string]bool)
+	map2 := make(map[string]bool)
+	for _, s := range set1 {
+		map1[s] = true
+	}
+	for _, s := range set2 {
+		map2[s] = true
+	}
+
+	for s := range map1 {
+		if !map2[s] {
+			return false
+		}
+	}
+	for s := range map2 {
+		if !map1[s] {
+			return false
+		}
+	}
+	return true
+}

test/e2e/pd.go

+/*
+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 e2e
+
+import (
+	"fmt"
+	"os/exec"
+	"strings"
+	"time"
+
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/client"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+)
+
+var _ = Describe("PD", func() {
+	var (
+		c         *client.Client
+		podClient client.PodInterface
+		diskName  string
+		host0Name string
+		host1Name string
+	)
+
+	BeforeEach(func() {
+		var err error
+		c, err = loadClient()
+		expectNoError(err)
+
+		podClient = c.Pods(api.NamespaceDefault)
+
+		nodes, err := c.Nodes().List()
+		expectNoError(err, "Failed to list nodes for e2e cluster.")
+		Expect(len(nodes.Items) >= 2)
+
+		diskName = fmt.Sprintf("e2e-%s", string(util.NewUUID()))
+		host0Name = nodes.Items[0].ObjectMeta.Name
+		host1Name = nodes.Items[1].ObjectMeta.Name
+	})
+
+	It("should schedule a pod w/ a RW PD, remove it, then schedule it on another host", func() {
+		host0Pod := testPDPod(diskName, host0Name, false)
+		host1Pod := testPDPod(diskName, host1Name, false)
+
+		By("creating PD")
+		expectNoError(createPD(diskName, testContext.gceConfig.Zone), "Error creating PD")
+
+		defer func() {
+			By("cleaning up PD-RW test environment")
+			// Teardown pods, PD. Ignore errors.
+			// Teardown should do nothing unless test failed.
+			podClient.Delete(host0Pod.Name)
+			podClient.Delete(host1Pod.Name)
+			detachPD(host0Name, diskName, testContext.gceConfig.Zone)
+			detachPD(host1Name, diskName, testContext.gceConfig.Zone)
+			deletePD(diskName, testContext.gceConfig.Zone)
+		}()
+
+		By("submitting host0Pod to kubernetes")
+		_, err := podClient.Create(host0Pod)
+		expectNoError(err, fmt.Sprintf("Failed to create host0Pod: %v", err))
+
+		By("waiting up to 60 seconds for host0Pod to start running")
+		expectNoError(waitForPodRunning(c, host0Pod.Name, 60*time.Second), "host0Pod not running after 60 seconds")
+
+		By("deleting host0Pod")
+		expectNoError(podClient.Delete(host0Pod.Name), "Failed to delete host0Pod")
+
+		By("submitting host1Pod to kubernetes")
+		_, err = podClient.Create(host1Pod)
+		expectNoError(err, "Failed to create host1Pod")
+
+		By("waiting up to 60 seconds for host1Pod to start running")
+		expectNoError(waitForPodRunning(c, host1Pod.Name, 60*time.Second), "host1Pod not running after 60 seconds")
+
+		By("deleting host1Pod")
+		expectNoError(podClient.Delete(host1Pod.Name), "Failed to delete host1Pod")
+
+		return
+	})
+
+	It("should schedule a pod w/ a readonly PD on two hosts, then remove both.", func() {
+		rwPod := testPDPod(diskName, host0Name, false)
+		host0ROPod := testPDPod(diskName, host0Name, true)
+		host1ROPod := testPDPod(diskName, host1Name, true)
+
+		defer func() {
+			By("cleaning up PD-RO test environment")
+			// Teardown pods, PD. Ignore errors.
+			// Teardown should do nothing unless test failed.
+			podClient.Delete(rwPod.Name)
+			podClient.Delete(host0ROPod.Name)
+			podClient.Delete(host1ROPod.Name)
+			detachPD(host0Name, diskName, testContext.gceConfig.Zone)
+			detachPD(host1Name, diskName, testContext.gceConfig.Zone)
+			deletePD(diskName, testContext.gceConfig.Zone)
+		}()
+
+		By("creating PD")
+		expectNoError(createPD(diskName, testContext.gceConfig.Zone), "Error creating PD")
+
+		By("submitting rwPod to ensure PD is formatted")
+		_, err := podClient.Create(rwPod)
+		expectNoError(err, "Failed to create rwPod")
+		expectNoError(waitForPodRunning(c, rwPod.Name, 60*time.Second), "rwPod not running after 60 seconds")
+		expectNoError(podClient.Delete(rwPod.Name), "Failed to delete host0Pod")
+
+		By("submitting host0ROPod to kubernetes")
+		_, err = podClient.Create(host0ROPod)
+		expectNoError(err, "Failed to create host0ROPod")
+
+		By("submitting host1ROPod to kubernetes")
+		_, err = podClient.Create(host1ROPod)
+		expectNoError(err, "Failed to create host1ROPod")
+
+		By("waiting up to 60 seconds for host0ROPod to start running")
+		expectNoError(waitForPodRunning(c, host0ROPod.Name, 60*time.Second), "host0ROPod not running after 60 seconds")
+
+		By("waiting up to 60 seconds for host1ROPod to start running")
+		expectNoError(waitForPodRunning(c, host1ROPod.Name, 60*time.Second), "host1ROPod not running after 60 seconds")
+
+		By("deleting host0ROPod")
+		expectNoError(podClient.Delete(host0ROPod.Name), "Failed to delete host0ROPod")
+
+		By("deleting host1ROPod")
+		expectNoError(podClient.Delete(host1ROPod.Name), "Failed to delete host1ROPod")
+	})
+})
+
+func createPD(pdName, zone string) error {
+	// TODO: make this hit the compute API directly instread of shelling out to gcloud.
+	return exec.Command("gcloud", "compute", "disks", "create", "--zone="+zone, "--size=10GB", pdName).Run()
+}
+
+func deletePD(pdName, zone string) error {
+	// TODO: make this hit the compute API directly.
+	return exec.Command("gcloud", "compute", "disks", "delete", "--zone="+zone, pdName).Run()
+}
+
+func detachPD(hostName, pdName, zone string) error {
+	instanceName := strings.Split(hostName, ".")[0]
+	// TODO: make this hit the compute API directly.
+	return exec.Command("gcloud", "compute", "instances", "detach-disk", "--zone="+zone, "--disk="+pdName, instanceName).Run()
+}
+
+func testPDPod(diskName, targetHost string, readOnly bool) *api.Pod {
+	return &api.Pod{
+		TypeMeta: api.TypeMeta{
+			Kind:       "Pod",
+			APIVersion: "v1beta1",
+		},
+		ObjectMeta: api.ObjectMeta{
+			Name: "pd-test-" + string(util.NewUUID()),
+		},
+		Spec: api.PodSpec{
+			Volumes: []api.Volume{
+				{
+					Name: "testpd",
+					Source: api.VolumeSource{
+						GCEPersistentDisk: &api.GCEPersistentDisk{
+							PDName:   diskName,
+							FSType:   "ext4",
+							ReadOnly: readOnly,
+						},
+					},
+				},
+			},
+			Containers: []api.Container{
+				{
+					Name:  "testpd",
+					Image: "kubernetes/pause",
+					VolumeMounts: []api.VolumeMount{
+						{
+							Name:      "testpd",
+							MountPath: "/testpd",
+						},
+					},
+				},
+			},
+			Host: targetHost,
+		},
+	}
+}