github
/
k3s
mirror of https://github.com/k3s-io/k3s
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Implement volumes as plugins. 

Break up the monolithic volumes code in kubelet into very small individual
modules with a well-defined interface.  Move them all into their own packages
and beef up testing along the way.
pull/6/head
Tim Hockin 2014-11-23 23:47:25 +08:00
parent f90ad573cf
commit 6cb275829f
31 changed files with 2059 additions and 985 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
cmd/integration/integration.go
View File

@ -41,6 +41,7 @@ import (
replicationControllerPkg "github.com/GoogleCloudPlatform/kubernetes/pkg/controller"
"github.com/GoogleCloudPlatform/kubernetes/pkg/health"
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/dockertools"
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/volume/empty_dir"
"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
"github.com/GoogleCloudPlatform/kubernetes/pkg/master"
"github.com/GoogleCloudPlatform/kubernetes/pkg/service"
@ -191,13 +192,13 @@ func startComponents(manifestURL string) (apiServerURL string) {
// Kubelet (localhost)
testRootDir := makeTempDirOrDie("kubelet_integ_1.")
glog.Infof("Using %s as root dir for kubelet #1", testRootDir)
standalone.SimpleRunKubelet(cl, nil, &fakeDocker1, machineList[0], testRootDir, manifestURL, "127.0.0.1", 10250, api.NamespaceDefault)
standalone.SimpleRunKubelet(cl, nil, &fakeDocker1, machineList[0], testRootDir, manifestURL, "127.0.0.1", 10250, api.NamespaceDefault, empty_dir.ProbeVolumePlugins())
// Kubelet (machine)
// Create a second kubelet so that the guestbook example's two redis slaves both
// have a place they can schedule.
testRootDir = makeTempDirOrDie("kubelet_integ_2.")
glog.Infof("Using %s as root dir for kubelet #2", testRootDir)
standalone.SimpleRunKubelet(cl, nil, &fakeDocker2, machineList[1], testRootDir, "", "127.0.0.1", 10251, api.NamespaceDefault)
standalone.SimpleRunKubelet(cl, nil, &fakeDocker2, machineList[1], testRootDir, "", "127.0.0.1", 10251, api.NamespaceDefault, empty_dir.ProbeVolumePlugins())
return apiServer.URL
}

43
cmd/kubelet/app/plugins.go Normal file
View File

@ -0,0 +1,43 @@
/*
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 app
// This file exists to force the desired plugin implementations to be linked.
import (
// Credential providers
_ "github.com/GoogleCloudPlatform/kubernetes/pkg/credentialprovider/gcp"
// Volume plugins
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/volume"
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/volume/empty_dir"
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/volume/gce_pd"
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/volume/git_repo"
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/volume/host_path"
)
func ProbeVolumePlugins() []volume.Plugin {
allPlugins := []volume.Plugin{}
// The list of plugins to probe is decided by the kubelet binary, not
// by dynamic linking or other "magic". Plugins will be analyzed and
// initialized later.
allPlugins = append(allPlugins, empty_dir.ProbeVolumePlugins()...)
allPlugins = append(allPlugins, gce_pd.ProbeVolumePlugins()...)
allPlugins = append(allPlugins, git_repo.ProbeVolumePlugins()...)
allPlugins = append(allPlugins, host_path.ProbeVolumePlugins()...)
return allPlugins
}

2
cmd/kubelet/kubelet.go
View File

@ -25,6 +25,7 @@ import (
"net"
"time"
"github.com/GoogleCloudPlatform/kubernetes/cmd/kubelet/app"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
_ "github.com/GoogleCloudPlatform/kubernetes/pkg/healthz"
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet"
@ -152,6 +153,7 @@ func main() {
KubeClient: client,
EtcdClient: kubelet.EtcdClientOrDie(etcdServerList, *etcdConfigFile),
MasterServiceNamespace: *masterServiceNamespace,
VolumePlugins: app.ProbeVolumePlugins(),
}
standalone.RunKubelet(&kcfg)

24
cmd/kubelet/plugins.go
View File

@ -1,24 +0,0 @@
/*
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 main
// This file exists to force the desired plugin implementations to be linked.
// This should probably be part of some configuration fed into the build for a
// given binary target.
import (
_ "github.com/GoogleCloudPlatform/kubernetes/pkg/credentialprovider/gcp"
)

3
cmd/kubernetes/kubernetes.go
View File

@ -24,6 +24,7 @@ import (
"fmt"
"time"
kubeletapp "github.com/GoogleCloudPlatform/kubernetes/cmd/kubelet/app"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api/testapi"
"github.com/GoogleCloudPlatform/kubernetes/pkg/client"
@ -54,7 +55,7 @@ func startComponents(etcdClient tools.EtcdClient, cl *client.Client, addr string
standalone.RunControllerManager(machineList, cl, *nodeMilliCPU, *nodeMemory)
dockerClient := util.ConnectToDockerOrDie(*dockerEndpoint)
standalone.SimpleRunKubelet(cl, nil, dockerClient, machineList[0], "/tmp/kubernetes", "", "127.0.0.1", 10250, *masterServiceNamespace)
standalone.SimpleRunKubelet(cl, nil, dockerClient, machineList[0], "/tmp/kubernetes", "", "127.0.0.1", 10250, *masterServiceNamespace, kubeletapp.ProbeVolumePlugins())
}
func newApiClient(addr string, port int) *client.Client {

124
pkg/kubelet/kubelet.go
View File

@ -39,12 +39,12 @@ import (
"github.com/GoogleCloudPlatform/kubernetes/pkg/health"
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/dockertools"
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/envvars"
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/volume"
"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
"github.com/GoogleCloudPlatform/kubernetes/pkg/tools"
"github.com/GoogleCloudPlatform/kubernetes/pkg/types"
"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
"github.com/GoogleCloudPlatform/kubernetes/pkg/util/errors"
"github.com/GoogleCloudPlatform/kubernetes/pkg/volume"
"github.com/fsouza/go-dockerclient"
"github.com/golang/glog"
)
@ -81,7 +81,8 @@ func NewMainKubelet(
sourceReady SourceReadyFn,
clusterDomain string,
clusterDNS net.IP,
masterServiceNamespace string) (*Kubelet, error) {
masterServiceNamespace string,
volumePlugins []volume.Plugin) (*Kubelet, error) {
if rootDirectory == "" {
return nil, fmt.Errorf("invalid root directory %q", rootDirectory)
}
@ -124,6 +125,9 @@ func NewMainKubelet(
if err := klet.setupDataDirs(); err != nil {
return nil, err
}
if err := klet.volumePluginMgr.InitPlugins(volumePlugins, &volumeHost{klet}); err != nil {
return nil, err
}
return klet, nil
}
@ -191,34 +195,52 @@ type Kubelet struct {
masterServiceNamespace string
serviceLister serviceLister
// Volume plugins.
volumePluginMgr volume.PluginMgr
}
// GetRootDir returns the full path to the directory under which kubelet can
// getRootDir returns the full path to the directory under which kubelet can
// store data. These functions are useful to pass interfaces to other modules
// that may need to know where to write data without getting a whole kubelet
// instance.
func (kl *Kubelet) GetRootDir() string {
func (kl *Kubelet) getRootDir() string {
return kl.rootDirectory
}
// GetPodsDir returns the full path to the directory under which pod
// getPodsDir returns the full path to the directory under which pod
// directories are created.
func (kl *Kubelet) GetPodsDir() string {
return path.Join(kl.GetRootDir(), "pods")
func (kl *Kubelet) getPodsDir() string {
return path.Join(kl.getRootDir(), "pods")
}
// GetPodDir returns the full path to the per-pod data directory for the
// getPluginsDir returns the full path to the directory under which plugin
// directories are created. Plugins can use these directories for data that
// they need to persist. Plugins should create subdirectories under this named
// after their own names.
func (kl *Kubelet) getPluginsDir() string {
return path.Join(kl.getRootDir(), "plugins")
}
// getPluginDir returns a data directory name for a given plugin name.
// Plugins can use these directories to store data that they need to persist.
// For per-pod plugin data, see getPodPluginDir.
func (kl *Kubelet) getPluginDir(pluginName string) string {
return path.Join(kl.getPluginsDir(), pluginName)
}
// getPodDir returns the full path to the per-pod data directory for the
// specified pod. This directory may not exist if the pod does not exist.
func (kl *Kubelet) GetPodDir(podUID types.UID) string {
func (kl *Kubelet) getPodDir(podUID types.UID) string {
// Backwards compat. The "old" stuff should be removed before 1.0
// release. The thinking here is this:
// !old && !new = use new
// !old && new = use new
// old && !new = use old
// old && new = use new (but warn)
oldPath := path.Join(kl.GetRootDir(), string(podUID))
oldPath := path.Join(kl.getRootDir(), string(podUID))
oldExists := dirExists(oldPath)
newPath := path.Join(kl.GetPodsDir(), string(podUID))
newPath := path.Join(kl.getPodsDir(), string(podUID))
newExists := dirExists(newPath)
if oldExists && !newExists {
return oldPath
@ -229,26 +251,47 @@ func (kl *Kubelet) GetPodDir(podUID types.UID) string {
return newPath
}
// GetPodVolumesDir returns the full path to the per-pod data directory under
// getPodVolumesDir returns the full path to the per-pod data directory under
// which volumes are created for the specified pod. This directory may not
// exist if the pod does not exist.
func (kl *Kubelet) GetPodVolumesDir(podUID types.UID) string {
return path.Join(kl.GetPodDir(podUID), "volumes")
func (kl *Kubelet) getPodVolumesDir(podUID types.UID) string {
return path.Join(kl.getPodDir(podUID), "volumes")
}
// GetPodContainerDir returns the full path to the per-pod data directory under
// getPodVolumeDir returns the full path to the directory which represents the
// named volume under the named plugin for specified pod. This directory may not
// exist if the pod does not exist.
func (kl *Kubelet) getPodVolumeDir(podUID types.UID, pluginName string, volumeName string) string {
return path.Join(kl.getPodVolumesDir(podUID), pluginName, volumeName)
}
// getPodPluginsDir returns the full path to the per-pod data directory under
// which plugins may store data for the specified pod. This directory may not
// exist if the pod does not exist.
func (kl *Kubelet) getPodPluginsDir(podUID types.UID) string {
return path.Join(kl.getPodDir(podUID), "plugins")
}
// getPodPluginDir returns a data directory name for a given plugin name for a
// given pod UID. Plugins can use these directories to store data that they
// need to persist. For non-per-pod plugin data, see getPluginDir.
func (kl *Kubelet) getPodPluginDir(podUID types.UID, pluginName string) string {
return path.Join(kl.getPodPluginsDir(podUID), pluginName)
}
// getPodContainerDir returns the full path to the per-pod data directory under
// which container data is held for the specified pod. This directory may not
// exist if the pod or container does not exist.
func (kl *Kubelet) GetPodContainerDir(podUID types.UID, ctrName string) string {
func (kl *Kubelet) getPodContainerDir(podUID types.UID, ctrName string) string {
// Backwards compat. The "old" stuff should be removed before 1.0
// release. The thinking here is this:
// !old && !new = use new
// !old && new = use new
// old && !new = use old
// old && new = use new (but warn)
oldPath := path.Join(kl.GetPodDir(podUID), ctrName)
oldPath := path.Join(kl.getPodDir(podUID), ctrName)
oldExists := dirExists(oldPath)
newPath := path.Join(kl.GetPodDir(podUID), "containers", ctrName)
newPath := path.Join(kl.getPodDir(podUID), "containers", ctrName)
newExists := dirExists(newPath)
if oldExists && !newExists {
return oldPath
@ -269,18 +312,21 @@ func dirExists(path string) bool {
func (kl *Kubelet) setupDataDirs() error {
kl.rootDirectory = path.Clean(kl.rootDirectory)
if err := os.MkdirAll(kl.GetRootDir(), 0750); err != nil {
if err := os.MkdirAll(kl.getRootDir(), 0750); err != nil {
return fmt.Errorf("error creating root directory: %v", err)
}
if err := os.MkdirAll(kl.GetPodsDir(), 0750); err != nil {
if err := os.MkdirAll(kl.getPodsDir(), 0750); err != nil {
return fmt.Errorf("error creating pods directory: %v", err)
}
if err := os.MkdirAll(kl.getPluginsDir(), 0750); err != nil {
return fmt.Errorf("error creating plugins directory: %v", err)
}
return nil
}
// Get a list of pods that have data directories.
func (kl *Kubelet) listPodsFromDisk() ([]types.UID, error) {
podInfos, err := ioutil.ReadDir(kl.GetPodsDir())
podInfos, err := ioutil.ReadDir(kl.getPodsDir())
if err != nil {
return nil, err
}
@ -508,27 +554,6 @@ func milliCPUToShares(milliCPU int64) int64 {
return shares
}
func (kl *Kubelet) mountExternalVolumes(pod *api.BoundPod) (volumeMap, error) {
podVolumes := make(volumeMap)
for _, vol := range pod.Spec.Volumes {
extVolume, err := volume.CreateVolumeBuilder(&vol, pod.Name, kl.rootDirectory)
if err != nil {
return nil, err
}
// TODO(jonesdl) When the default volume behavior is no longer supported, this case
// should never occur and an error should be thrown instead.
if extVolume == nil {
continue
}
podVolumes[vol.Name] = extVolume
err = extVolume.SetUp()
if err != nil {
return nil, err
}
}
return podVolumes, nil
}
// A basic interface that knows how to execute handlers
type actionHandler interface {
Run(podFullName string, uid types.UID, container *api.Container, handler *api.Handler) error
@ -657,7 +682,7 @@ func (kl *Kubelet) runContainer(pod *api.BoundPod, container *api.Container, pod
}
if len(container.TerminationMessagePath) != 0 {
p := kl.GetPodContainerDir(pod.UID, container.Name)
p := kl.getPodContainerDir(pod.UID, container.Name)
if err := os.MkdirAll(p, 0750); err != nil {
glog.Errorf("Error on creating %q: %v", p, err)
} else {
@ -974,10 +999,13 @@ func (kl *Kubelet) syncPod(pod *api.BoundPod, dockerContainers dockertools.Docke
glog.V(4).Infof("Syncing Pod, podFullName: %q, uid: %q", podFullName, uid)
// Make data dirs.
if err := os.Mkdir(kl.GetPodDir(uid), 0750); err != nil && !os.IsExist(err) {
if err := os.Mkdir(kl.getPodDir(uid), 0750); err != nil && !os.IsExist(err) {
return err
}
if err := os.Mkdir(kl.GetPodVolumesDir(uid), 0750); err != nil && !os.IsExist(err) {
if err := os.Mkdir(kl.getPodVolumesDir(uid), 0750); err != nil && !os.IsExist(err) {
return err
}
if err := os.Mkdir(kl.getPodPluginsDir(uid), 0750); err != nil && !os.IsExist(err) {
return err
}
@ -1148,7 +1176,7 @@ func getDesiredVolumes(pods []api.BoundPod) map[string]api.Volume {
desiredVolumes := make(map[string]api.Volume)
for _, pod := range pods {
for _, volume := range pod.Spec.Volumes {
identifier := path.Join(pod.Name, volume.Name)
identifier := path.Join(string(pod.UID), volume.Name)
desiredVolumes[identifier] = volume
}
}
@ -1168,7 +1196,7 @@ func (kl *Kubelet) cleanupOrphanedPods(pods []api.BoundPod) error {
for i := range found {
if !desired.Has(string(found[i])) {
glog.V(3).Infof("Orphaned pod %q found, removing", found[i])
if err := os.RemoveAll(kl.GetPodDir(found[i])); err != nil {
if err := os.RemoveAll(kl.getPodDir(found[i])); err != nil {
errlist = append(errlist, err)
}
}
@ -1180,7 +1208,7 @@ func (kl *Kubelet) cleanupOrphanedPods(pods []api.BoundPod) error {
// If an active volume does not have a respective desired volume, clean it up.
func (kl *Kubelet) cleanupOrphanedVolumes(pods []api.BoundPod) error {
desiredVolumes := getDesiredVolumes(pods)
currentVolumes := volume.GetCurrentVolumes(kl.rootDirectory)
currentVolumes := kl.getPodVolumesFromDisk()
for name, vol := range currentVolumes {
if _, ok := desiredVolumes[name]; !ok {
//TODO (jonesdl) We should somehow differentiate between volumes that are supposed

150
pkg/kubelet/kubelet_test.go
View File

@ -18,6 +18,7 @@ package kubelet
import (
"fmt"
"io/ioutil"
"net/http"
"os"
"path"
@ -32,9 +33,10 @@ import (
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
"github.com/GoogleCloudPlatform/kubernetes/pkg/health"
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/dockertools"
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/volume"
_ "github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/volume/host_path"
"github.com/GoogleCloudPlatform/kubernetes/pkg/types"
"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
"github.com/GoogleCloudPlatform/kubernetes/pkg/volume"
"github.com/fsouza/go-dockerclient"
"github.com/google/cadvisor/info"
"github.com/stretchr/testify/mock"
@ -53,11 +55,21 @@ func newTestKubelet(t *testing.T) (*Kubelet, *dockertools.FakeDockerClient) {
kubelet := &Kubelet{}
kubelet.dockerClient = fakeDocker
kubelet.dockerPuller = &dockertools.FakeDockerPuller{}
kubelet.rootDirectory = "/tmp/kubelet"
if tempDir, err := ioutil.TempDir("/tmp", "kubelet_test."); err != nil {
t.Fatalf("can't make a temp rootdir: %v", err)
} else {
kubelet.rootDirectory = tempDir
}
if err := os.MkdirAll(kubelet.rootDirectory, 0750); err != nil {
t.Fatalf("can't mkdir(%q): %v", kubelet.rootDirectory, err)
}
kubelet.podWorkers = newPodWorkers()
kubelet.sourceReady = func(source string) bool { return true }
kubelet.masterServiceNamespace = api.NamespaceDefault
kubelet.serviceLister = testServiceLister{}
if err := kubelet.setupDataDirs(); err != nil {
t.Fatalf("can't initialize kubelet data dirs: %v", err)
}
return kubelet, fakeDocker
}
@ -92,31 +104,58 @@ func verifyBoolean(t *testing.T, expected, value bool) {
func TestKubeletDirs(t *testing.T) {
kubelet, _ := newTestKubelet(t)
root := kubelet.rootDirectory
if err := os.MkdirAll(root, 0750); err != nil {
t.Fatalf("can't mkdir(%q): %s", root, err)
}
var exp, got string
got = kubelet.GetPodsDir()
got = kubelet.getPodsDir()
exp = path.Join(root, "pods")
if got != exp {
t.Errorf("expected %q', got %q", exp, got)
}
got = kubelet.GetPodDir("abc123")
got = kubelet.getPluginsDir()
exp = path.Join(root, "plugins")
if got != exp {
t.Errorf("expected %q', got %q", exp, got)
}
got = kubelet.getPluginDir("foobar")
exp = path.Join(root, "plugins/foobar")
if got != exp {
t.Errorf("expected %q', got %q", exp, got)
}
got = kubelet.getPodDir("abc123")
exp = path.Join(root, "pods/abc123")
if got != exp {
t.Errorf("expected %q', got %q", exp, got)
}
got = kubelet.GetPodVolumesDir("abc123")
got = kubelet.getPodVolumesDir("abc123")
exp = path.Join(root, "pods/abc123/volumes")
if got != exp {
t.Errorf("expected %q', got %q", exp, got)
}
got = kubelet.GetPodContainerDir("abc123", "def456")
got = kubelet.getPodVolumeDir("abc123", "plugin", "foobar")
exp = path.Join(root, "pods/abc123/volumes/plugin/foobar")
if got != exp {
t.Errorf("expected %q', got %q", exp, got)
}
got = kubelet.getPodPluginsDir("abc123")
exp = path.Join(root, "pods/abc123/plugins")
if got != exp {
t.Errorf("expected %q', got %q", exp, got)
}
got = kubelet.getPodPluginDir("abc123", "foobar")
exp = path.Join(root, "pods/abc123/plugins/foobar")
if got != exp {
t.Errorf("expected %q', got %q", exp, got)
}
got = kubelet.getPodContainerDir("abc123", "def456")
exp = path.Join(root, "pods/abc123/containers/def456")
if got != exp {
t.Errorf("expected %q', got %q", exp, got)
@ -148,31 +187,31 @@ func TestKubeletDirsCompat(t *testing.T) {
t.Fatalf("can't mkdir(%q): %s", root, err)
}
got = kubelet.GetPodDir("oldpod")
got = kubelet.getPodDir("oldpod")
exp = path.Join(root, "oldpod")
if got != exp {
t.Errorf("expected %q', got %q", exp, got)
}
got = kubelet.GetPodDir("newpod")
got = kubelet.getPodDir("newpod")
exp = path.Join(root, "pods/newpod")
if got != exp {
t.Errorf("expected %q', got %q", exp, got)
}
got = kubelet.GetPodDir("bothpod")
got = kubelet.getPodDir("bothpod")
exp = path.Join(root, "pods/bothpod")
if got != exp {
t.Errorf("expected %q', got %q", exp, got)
}
got = kubelet.GetPodDir("neitherpod")
got = kubelet.getPodDir("neitherpod")
exp = path.Join(root, "pods/neitherpod")
if got != exp {
t.Errorf("expected %q', got %q", exp, got)
}
root = kubelet.GetPodDir("newpod")
root = kubelet.getPodDir("newpod")
// Old-style container dir.
if err := os.MkdirAll(fmt.Sprintf("%s/oldctr", root), 0750); err != nil {
@ -190,25 +229,25 @@ func TestKubeletDirsCompat(t *testing.T) {
t.Fatalf("can't mkdir(%q): %s", root, err)
}
got = kubelet.GetPodContainerDir("newpod", "oldctr")
got = kubelet.getPodContainerDir("newpod", "oldctr")
exp = path.Join(root, "oldctr")
if got != exp {
t.Errorf("expected %q', got %q", exp, got)
}
got = kubelet.GetPodContainerDir("newpod", "newctr")
got = kubelet.getPodContainerDir("newpod", "newctr")
exp = path.Join(root, "containers/newctr")
if got != exp {
t.Errorf("expected %q', got %q", exp, got)
}
got = kubelet.GetPodContainerDir("newpod", "bothctr")
got = kubelet.getPodContainerDir("newpod", "bothctr")
exp = path.Join(root, "containers/bothctr")
if got != exp {
t.Errorf("expected %q', got %q", exp, got)
}
got = kubelet.GetPodContainerDir("newpod", "neitherctr")
got = kubelet.getPodContainerDir("newpod", "neitherctr")
exp = path.Join(root, "containers/neitherctr")
if got != exp {
t.Errorf("expected %q', got %q", exp, got)
@ -355,7 +394,7 @@ func TestSyncPodsWithTerminationLog(t *testing.T) {
fakeDocker.Lock()
parts := strings.Split(fakeDocker.Container.HostConfig.Binds[0], ":")
if !matchString(t, kubelet.GetPodContainerDir("12345678", "bar")+"/k8s_bar\\.[a-f0-9]", parts[0]) {
if !matchString(t, kubelet.getPodContainerDir("12345678", "bar")+"/k8s_bar\\.[a-f0-9]", parts[0]) {
t.Errorf("Unexpected host path: %s", parts[0])
}
if parts[1] != "/dev/somepath" {
@ -916,6 +955,8 @@ func TestSyncPodUnhealthy(t *testing.T) {
func TestMountExternalVolumes(t *testing.T) {
kubelet, _ := newTestKubelet(t)
kubelet.volumePluginMgr.InitPlugins([]volume.Plugin{&volume.FakePlugin{"fake", nil}}, &volumeHost{kubelet})
pod := api.BoundPod{
ObjectMeta: api.ObjectMeta{
UID: "12345678",
@ -925,27 +966,74 @@ func TestMountExternalVolumes(t *testing.T) {
Spec: api.PodSpec{
Volumes: []api.Volume{
{
Name: "host-dir",
Source: &api.VolumeSource{
HostDir: &api.HostDir{"/dir/path"},
},
Name: "vol1",
Source: &api.VolumeSource{},
},
},
},
}
podVolumes, _ := kubelet.mountExternalVolumes(&pod)
expectedPodVolumes := make(volumeMap)
expectedPodVolumes["host-dir"] = &volume.HostDir{"/dir/path"}
podVolumes, err := kubelet.mountExternalVolumes(&pod)
if err != nil {
t.Errorf("Expected sucess: %v", err)
}
expectedPodVolumes := []string{"vol1"}
if len(expectedPodVolumes) != len(podVolumes) {
t.Errorf("Unexpected volumes. Expected %#v got %#v. Manifest was: %#v", expectedPodVolumes, podVolumes, pod)
}
for name, expectedVolume := range expectedPodVolumes {
for _, name := range expectedPodVolumes {
if _, ok := podVolumes[name]; !ok {
t.Errorf("api.BoundPod volumes map is missing key: %s. %#v", expectedVolume, podVolumes)
t.Errorf("api.BoundPod volumes map is missing key: %s. %#v", name, podVolumes)
}
}
}
func TestGetPodVolumesFromDisk(t *testing.T) {
kubelet, _ := newTestKubelet(t)
plug := &volume.FakePlugin{"fake", nil}
kubelet.volumePluginMgr.InitPlugins([]volume.Plugin{plug}, &volumeHost{kubelet})
volsOnDisk := []struct {
podUID types.UID
volName string
}{
{"pod1", "vol1"},
{"pod1", "vol2"},
{"pod2", "vol1"},
}
expectedPaths := []string{}
for i := range volsOnDisk {
fv := volume.FakeVolume{volsOnDisk[i].podUID, volsOnDisk[i].volName, plug}
fv.SetUp()
expectedPaths = append(expectedPaths, fv.GetPath())
}
volumesFound := kubelet.getPodVolumesFromDisk()
if len(volumesFound) != len(expectedPaths) {
t.Errorf("Expected to find %d cleaners, got %d", len(expectedPaths), len(volumesFound))
}
for _, ep := range expectedPaths {
found := false
for _, cl := range volumesFound {
if ep == cl.GetPath() {
found = true
break
}
}
if !found {
t.Errorf("Could not find a volume with path %s", ep)
}
}
}
type stubVolume struct {
path string
}
func (f *stubVolume) GetPath() string {
return f.path
}
func TestMakeVolumesAndBinds(t *testing.T) {
container := api.Container{
VolumeMounts: []api.VolumeMount{
@ -981,9 +1069,9 @@ func TestMakeVolumesAndBinds(t *testing.T) {
}
podVolumes := volumeMap{
"disk": &volume.HostDir{"/mnt/disk"},
"disk4": &volume.HostDir{"/mnt/host"},
"disk5": &volume.EmptyDir{"disk5", "podID", "/var/lib/kubelet"},
"disk": &stubVolume{"/mnt/disk"},
"disk4": &stubVolume{"/mnt/host"},
"disk5": &stubVolume{"/var/lib/kubelet/podID/volumes/empty/disk5"},
}
binds := makeBinds(&pod, &container, podVolumes)

0
pkg/volume/doc.go → pkg/kubelet/volume/doc.go
View File

125
pkg/kubelet/volume/empty_dir/empty_dir.go Normal file
View File

@ -0,0 +1,125 @@
/*
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 empty_dir
import (
"fmt"
"os"
"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"
)
// This is the primary entrypoint for volume plugins.
func ProbeVolumePlugins() []volume.Plugin {
return []volume.Plugin{&emptyDirPlugin{nil, false}, &emptyDirPlugin{nil, true}}
}
type emptyDirPlugin struct {
host volume.Host
legacyMode bool // if set, plugin answers to the legacy name
}
var _ volume.Plugin = &emptyDirPlugin{}
const (
emptyDirPluginName = "kubernetes.io/empty-dir"
emptyDirPluginLegacyName = "empty"
)
func (plugin *emptyDirPlugin) Init(host volume.Host) {
plugin.host = host
}
func (plugin *emptyDirPlugin) Name() string {
if plugin.legacyMode {
return emptyDirPluginLegacyName
}
return emptyDirPluginName
}
func (plugin *emptyDirPlugin) CanSupport(spec *api.Volume) bool {
if plugin.legacyMode {
// Legacy mode instances can be cleaned up but not created anew.
return false
}
if spec.Source == nil || util.AllPtrFieldsNil(spec.Source) {
return true
}
if spec.Source.EmptyDir != nil {
return true
}
return false
}
func (plugin *emptyDirPlugin) NewBuilder(spec *api.Volume, podUID types.UID) (volume.Builder, error) {
if plugin.legacyMode {
// Legacy mode instances can be cleaned up but not created anew.
return nil, fmt.Errorf("legacy mode: can not create new instances")
}
return &emptyDir{podUID, spec.Name, plugin, false}, nil
}
func (plugin *emptyDirPlugin) NewCleaner(volName string, podUID types.UID) (volume.Cleaner, error) {
legacy := false
if plugin.legacyMode {
legacy = true
}
return &emptyDir{podUID, volName, plugin, legacy}, nil
}
// EmptyDir volumes are temporary directories exposed to the pod.
// These do not persist beyond the lifetime of a pod.
type emptyDir struct {
podUID types.UID
volName string
plugin *emptyDirPlugin
legacyMode bool
}
// SetUp creates new directory.
func (ed *emptyDir) SetUp() error {
if ed.legacyMode {
return fmt.Errorf("legacy mode: can not create new instances")
}
path := ed.GetPath()
return os.MkdirAll(path, 0750)
}
func (ed *emptyDir) GetPath() string {
name := emptyDirPluginName
if ed.legacyMode {
name = emptyDirPluginLegacyName
}
return ed.plugin.host.GetPodVolumeDir(ed.podUID, volume.EscapePluginName(name), ed.volName)
}
// TearDown simply deletes everything in the directory.
func (ed *emptyDir) TearDown() error {
tmpDir, err := volume.RenameDirectory(ed.GetPath(), ed.volName+".deleting~")
if err != nil {
return err
}
err = os.RemoveAll(tmpDir)
if err != nil {
return err
}
return nil
}

153
pkg/kubelet/volume/empty_dir/empty_dir_test.go Normal file
View File

@ -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 empty_dir
import (
"os"
"testing"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/volume"
"github.com/GoogleCloudPlatform/kubernetes/pkg/types"
)
func TestCanSupport(t *testing.T) {
plugMgr := volume.PluginMgr{}
plugMgr.InitPlugins(ProbeVolumePlugins(), &volume.FakeHost{"/tmp/fake"})
plug, err := plugMgr.FindPluginByName("kubernetes.io/empty-dir")
if err != nil {
t.Errorf("Can't find the plugin by name")
}
if plug.Name() != "kubernetes.io/empty-dir" {
t.Errorf("Wrong name: %s", plug.Name())
}
if !plug.CanSupport(&api.Volume{Source: &api.VolumeSource{EmptyDir: &api.EmptyDir{}}}) {
t.Errorf("Expected true")
}
if !plug.CanSupport(&api.Volume{Source: nil}) {
t.Errorf("Expected true")
}
}
func TestPlugin(t *testing.T) {
plugMgr := volume.PluginMgr{}
plugMgr.InitPlugins(ProbeVolumePlugins(), &volume.FakeHost{"/tmp/fake"})
plug, err := plugMgr.FindPluginByName("kubernetes.io/empty-dir")
if err != nil {
t.Errorf("Can't find the plugin by name")
}
spec := &api.Volume{
Name: "vol1",
Source: &api.VolumeSource{EmptyDir: &api.EmptyDir{}},
}
builder, err := plug.NewBuilder(spec, types.UID("poduid"))
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~empty-dir/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)
}
}
cleaner, err := plug.NewCleaner("vol1", types.UID("poduid"))
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)
}
}
func TestPluginBackCompat(t *testing.T) {
plugMgr := volume.PluginMgr{}
plugMgr.InitPlugins(ProbeVolumePlugins(), &volume.FakeHost{"/tmp/fake"})
plug, err := plugMgr.FindPluginByName("kubernetes.io/empty-dir")
if err != nil {
t.Errorf("Can't find the plugin by name")
}
spec := &api.Volume{
Name: "vol1",
Source: nil,
}
builder, err := plug.NewBuilder(spec, types.UID("poduid"))
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~empty-dir/vol1" {
t.Errorf("Got unexpected path: %s", path)
}
}
func TestPluginLegacy(t *testing.T) {
plugMgr := volume.PluginMgr{}
plugMgr.InitPlugins(ProbeVolumePlugins(), &volume.FakeHost{"/tmp/fake"})
plug, err := plugMgr.FindPluginByName("empty")
if err != nil {
t.Errorf("Can't find the plugin by name")
}
if plug.Name() != "empty" {
t.Errorf("Wrong name: %s", plug.Name())
}
if plug.CanSupport(&api.Volume{Source: &api.VolumeSource{EmptyDir: &api.EmptyDir{}}}) {
t.Errorf("Expected false")
}
if _, err := plug.NewBuilder(&api.Volume{Source: &api.VolumeSource{EmptyDir: &api.EmptyDir{}}}, types.UID("poduid")); err == nil {
t.Errorf("Expected failiure")
}
cleaner, err := plug.NewCleaner("vol1", types.UID("poduid"))
if err != nil {
t.Errorf("Failed to make a new Cleaner: %v", err)
}
if cleaner == nil {
t.Errorf("Got a nil Cleaner: %v")
}
}

237
pkg/kubelet/volume/gce_pd/gce_pd.go Normal file
View File

@ -0,0 +1,237 @@
/*
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 gce_pd
import (
"fmt"
"os"
"path"
"strconv"
"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"
"github.com/golang/glog"
)
// This is the primary entrypoint for volume plugins.
func ProbeVolumePlugins() []volume.Plugin {
return []volume.Plugin{&gcePersistentDiskPlugin{nil, false}, &gcePersistentDiskPlugin{nil, true}}
}
type gcePersistentDiskPlugin struct {
host volume.Host
legacyMode bool // if set, plugin answers to the legacy name
}
var _ volume.Plugin = &gcePersistentDiskPlugin{}
const (
gcePersistentDiskPluginName = "kubernetes.io/gce-pd"
gcePersistentDiskPluginLegacyName = "gce-pd"
)
func (plugin *gcePersistentDiskPlugin) Init(host volume.Host) {
plugin.host = host
}
func (plugin *gcePersistentDiskPlugin) Name() string {
if plugin.legacyMode {
return gcePersistentDiskPluginLegacyName
}
return gcePersistentDiskPluginName
}
func (plugin *gcePersistentDiskPlugin) CanSupport(spec *api.Volume) bool {
if plugin.legacyMode {
// Legacy mode instances can be cleaned up but not created anew.
return false
}
if spec.Source != nil && spec.Source.GCEPersistentDisk != nil {
return true
}
return false
}
func (plugin *gcePersistentDiskPlugin) NewBuilder(spec *api.Volume, podUID types.UID) (volume.Builder, error) {
// Inject real implementations here, test through the internal function.
return plugin.newBuilderInternal(spec, podUID, &GCEDiskUtil{}, mount.New())
}
func (plugin *gcePersistentDiskPlugin) newBuilderInternal(spec *api.Volume, podUID types.UID, manager pdManager, mounter mount.Interface) (volume.Builder, error) {
if plugin.legacyMode {
// Legacy mode instances can be cleaned up but not created anew.
return nil, fmt.Errorf("legacy mode: can not create new instances")
}
pdName := spec.Source.GCEPersistentDisk.PDName
fsType := spec.Source.GCEPersistentDisk.FSType
partition := ""
if spec.Source.GCEPersistentDisk.Partition != 0 {
partition = strconv.Itoa(spec.Source.GCEPersistentDisk.Partition)
}
readOnly := spec.Source.GCEPersistentDisk.ReadOnly
return &gcePersistentDisk{
podUID: podUID,
volName: spec.Name,
pdName: pdName,
fsType: fsType,
partition: partition,
readOnly: readOnly,
manager: manager,
mounter: mounter,
plugin: plugin,
legacyMode: false,
}, nil
}
func (plugin *gcePersistentDiskPlugin) NewCleaner(volName string, podUID types.UID) (volume.Cleaner, error) {
// Inject real implementations here, test through the internal function.
return plugin.newCleanerInternal(volName, podUID, &GCEDiskUtil{}, mount.New())
}
func (plugin *gcePersistentDiskPlugin) newCleanerInternal(volName string, podUID types.UID, manager pdManager, mounter mount.Interface) (volume.Cleaner, error) {
legacy := false
if plugin.legacyMode {
legacy = true
}
return &gcePersistentDisk{
podUID: podUID,
volName: volName,
manager: manager,
mounter: mounter,
plugin: plugin,
legacyMode: legacy,
}, nil
}
// Abstract interface to PD operations.
type pdManager interface {
// Attaches the disk to the kubelet's host machine.
AttachDisk(pd *gcePersistentDisk) error
// Detaches the disk from the kubelet's host machine.
DetachDisk(pd *gcePersistentDisk, devicePath string) error
}
// gcePersistentDisk volumes are disk resources provided by Google Compute Engine
// that are attached to the kubelet's host machine and exposed to the pod.
type gcePersistentDisk struct {
volName string
podUID types.UID
// Unique identifier 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 disks.
mounter mount.Interface
plugin *gcePersistentDiskPlugin
legacyMode bool
}
// SetUp attaches the disk and bind mounts to the volume path.
func (pd *gcePersistentDisk) SetUp() error {
if pd.legacyMode {
return fmt.Errorf("legacy mode: can not create new instances")
}
// TODO: handle failed mounts here.
mountpoint, err := isMountPoint(pd.GetPath())
glog.V(4).Infof("PersistentDisk set up: %s %v %v", pd.GetPath(), mountpoint, err)
if err != nil && !os.IsNotExist(err) {
return err
}
if mountpoint {
return nil
}
if err := pd.manager.AttachDisk(pd); err != nil {
return err
}
flags := uintptr(0)
if pd.readOnly {
flags = mount.FlagReadOnly
}
volPath := pd.GetPath()
if err := os.MkdirAll(volPath, 0750); err != nil {
return err
}
// 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 {
os.RemoveAll(pd.GetPath())
return err
}
return nil
}
func makeGlobalPDName(host volume.Host, devName string, readOnly bool) string {
return path.Join(host.GetPluginDir(gcePersistentDiskPluginName), "mounts", devName)
}
func (pd *gcePersistentDisk) GetPath() string {
name := gcePersistentDiskPluginName
if pd.legacyMode {
name = gcePersistentDiskPluginLegacyName
}
return pd.plugin.host.GetPodVolumeDir(pd.podUID, volume.EscapePluginName(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 *gcePersistentDisk) TearDown() error {
mountpoint, err := isMountPoint(pd.GetPath())
if err != nil {
return err
}
if !mountpoint {
return os.RemoveAll(pd.GetPath())
}
devicePath, refCount, err := getMountRefCount(pd.mounter, pd.GetPath())
if err != nil {
return err
}
if err := pd.mounter.Unmount(pd.GetPath(), 0); err != nil {
return err
}
refCount--
if err := os.RemoveAll(pd.GetPath()); err != nil {
return err
}
// If refCount 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 {
return err
}
}
return nil
}

173
pkg/kubelet/volume/gce_pd/gce_pd_test.go Normal file
View File

@ -0,0 +1,173 @@
/*
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 gce_pd
import (
"os"
"testing"
"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) {
plugMgr := volume.PluginMgr{}
plugMgr.InitPlugins(ProbeVolumePlugins(), &volume.FakeHost{"/tmp/fake"})
plug, err := plugMgr.FindPluginByName("kubernetes.io/gce-pd")
if err != nil {
t.Errorf("Can't find the plugin by name")
}
if plug.Name() != "kubernetes.io/gce-pd" {
t.Errorf("Wrong name: %s", plug.Name())
}
if !plug.CanSupport(&api.Volume{Source: &api.VolumeSource{GCEPersistentDisk: &api.GCEPersistentDisk{}}}) {
t.Errorf("Expected true")
}
}
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)
err := os.MkdirAll(globalPath, 0750)
if err != nil {
return err
}
return nil
}
func (fake *fakePDManager) DetachDisk(pd *gcePersistentDisk, devicePath string) error {
globalPath := makeGlobalPDName(pd.plugin.host, pd.pdName, pd.readOnly)
err := os.RemoveAll(globalPath)
if err != nil {
return err
}
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"})
plug, err := plugMgr.FindPluginByName("kubernetes.io/gce-pd")
if err != nil {
t.Errorf("Can't find the plugin by name")
}
spec := &api.Volume{
Name: "vol1",
Source: &api.VolumeSource{
GCEPersistentDisk: &api.GCEPersistentDisk{
PDName: "pd",
FSType: "ext4",
},
},
}
builder, err := plug.(*gcePersistentDiskPlugin).newBuilderInternal(spec, types.UID("poduid"), &fakePDManager{}, &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~gce-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.(*gcePersistentDiskPlugin).newCleanerInternal("vol1", types.UID("poduid"), &fakePDManager{}, &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)
}
}
func TestPluginLegacy(t *testing.T) {
plugMgr := volume.PluginMgr{}
plugMgr.InitPlugins(ProbeVolumePlugins(), &volume.FakeHost{"/tmp/fake"})
plug, err := plugMgr.FindPluginByName("gce-pd")
if err != nil {
t.Errorf("Can't find the plugin by name")
}
if plug.Name() != "gce-pd" {
t.Errorf("Wrong name: %s", plug.Name())
}
if plug.CanSupport(&api.Volume{Source: &api.VolumeSource{GCEPersistentDisk: &api.GCEPersistentDisk{}}}) {
t.Errorf("Expected false")
}
if _, err := plug.NewBuilder(&api.Volume{Source: &api.VolumeSource{GCEPersistentDisk: &api.GCEPersistentDisk{}}}, types.UID("poduid")); err == nil {
t.Errorf("Expected failiure")
}
cleaner, err := plug.NewCleaner("vol1", types.UID("poduid"))
if err != nil {
t.Errorf("Failed to make a new Cleaner: %v", err)
}
if cleaner == nil {
t.Errorf("Got a nil Cleaner: %v")
}
}

29
pkg/volume/gce_util.go → pkg/kubelet/volume/gce_pd/gce_util.go
View File

@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
limitations under the License.
*/
package volume
package gce_pd
import (
"errors"
@ -27,7 +27,8 @@ import (
"time"
"github.com/GoogleCloudPlatform/kubernetes/pkg/cloudprovider"
gce_cloud "github.com/GoogleCloudPlatform/kubernetes/pkg/cloudprovider/gce"
"github.com/GoogleCloudPlatform/kubernetes/pkg/cloudprovider/gce"
"github.com/GoogleCloudPlatform/kubernetes/pkg/util/mount"
)
const partitionRegex = "[a-z][a-z]*(?P<partition>[0-9][0-9]*)?"
@ -38,21 +39,21 @@ 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(GCEPD *GCEPersistentDisk) error {
func (util *GCEDiskUtil) AttachDisk(pd *gcePersistentDisk) error {
gce, err := cloudprovider.GetCloudProvider("gce", nil)
if err != nil {
return err
}
flags := uintptr(0)
if GCEPD.ReadOnly {
flags = MOUNT_MS_RDONLY
if pd.readOnly {
flags = mount.FlagReadOnly
}
if err := gce.(*gce_cloud.GCECloud).AttachDisk(GCEPD.PDName, GCEPD.ReadOnly); err != nil {
if err := gce.(*gce_cloud.GCECloud).AttachDisk(pd.pdName, pd.readOnly); err != nil {
return err
}
devicePath := path.Join("/dev/disk/by-id/", "google-"+GCEPD.PDName)
if GCEPD.Partition != "" {
devicePath = devicePath + "-part" + GCEPD.Partition
devicePath := path.Join("/dev/disk/by-id/", "google-"+pd.pdName)
if pd.partition != "" {
devicePath = devicePath + "-part" + pd.partition
}
//TODO(jonesdl) There should probably be better method than busy-waiting here.
numTries := 0
@ -70,7 +71,7 @@ func (util *GCEDiskUtil) AttachDisk(GCEPD *GCEPersistentDisk) error {
}
time.Sleep(time.Second)
}
globalPDPath := makeGlobalPDName(GCEPD.RootDir, GCEPD.PDName, GCEPD.ReadOnly)
globalPDPath := makeGlobalPDName(pd.plugin.host, pd.pdName, pd.readOnly)
// Only mount the PD globally once.
mountpoint, err := isMountPoint(globalPDPath)
if err != nil {
@ -84,7 +85,7 @@ func (util *GCEDiskUtil) AttachDisk(GCEPD *GCEPersistentDisk) error {
}
}
if !mountpoint {
err = GCEPD.mounter.Mount(devicePath, globalPDPath, GCEPD.FSType, flags, "")
err = pd.mounter.Mount(devicePath, globalPDPath, pd.fsType, flags, "")
if err != nil {
os.RemoveAll(globalPDPath)
return err
@ -112,7 +113,7 @@ func getDeviceName(devicePath, canonicalDevicePath string) (string, error) {
// 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(GCEPD *GCEPersistentDisk, devicePath string) error {
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 {
@ -122,8 +123,8 @@ func (util *GCEDiskUtil) DetachDisk(GCEPD *GCEPersistentDisk, devicePath string)
if err != nil {
return err
}
globalPDPath := makeGlobalPDName(GCEPD.RootDir, deviceName, GCEPD.ReadOnly)
if err := GCEPD.mounter.Unmount(globalPDPath, 0); err != nil {
globalPDPath := makeGlobalPDName(pd.plugin.host, deviceName, pd.readOnly)
if err := pd.mounter.Unmount(globalPDPath, 0); err != nil {
return err
}
if err := os.RemoveAll(globalPDPath); err != nil {

2
pkg/volume/gce_util_test.go → pkg/kubelet/volume/gce_pd/gce_util_test.go
View File

@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
limitations under the License.
*/
package volume
package gce_pd
import (
"testing"

53
pkg/kubelet/volume/gce_pd/mount_util.go Normal file
View File

@ -0,0 +1,53 @@
/*
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 gce_pd
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.
mps, err := mounter.List()
if err != nil {
return "", -1, err
}
// Find the device name.
deviceName := ""
for i := range mps {
if mps[i].Path == mountPath {
deviceName = mps[i].Device
break
}
}
// Find the number of references to the device.
refCount := 0
for i := range mps {
if mps[i].Device == deviceName {
refCount++
}
}
return deviceName, refCount, nil
}

2
pkg/volume/mount_utils.go → pkg/kubelet/volume/gce_pd/mount_util_linux.go
View File

@ -16,7 +16,7 @@ See the License for the specific language governing permissions and
limitations under the License.
*/
package volume
package gce_pd
import (
"os"

4
pkg/volume/mount_utils_windows.go → pkg/kubelet/volume/gce_pd/mount_util_unsupported.go
View File

@ -1,4 +1,4 @@
// +build windows
// +build !linux
/*
Copyright 2014 Google Inc. All rights reserved.
@ -16,7 +16,7 @@ See the License for the specific language governing permissions and
limitations under the License.
*/
package volume
package gce_pd
import (
"fmt"

214
pkg/kubelet/volume/git_repo/git_repo.go Normal file
View File

@ -0,0 +1,214 @@
/*
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 git_repo
import (
"fmt"
"io/ioutil"
"os"
"path"
"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/exec"
"github.com/golang/glog"
)
// This is the primary entrypoint for volume plugins.
func ProbeVolumePlugins() []volume.Plugin {
return []volume.Plugin{&gitRepoPlugin{nil, false}, &gitRepoPlugin{nil, true}}
}
type gitRepoPlugin struct {
host volume.Host
legacyMode bool // if set, plugin answers to the legacy name
}
var _ volume.Plugin = &gitRepoPlugin{}
const (
gitRepoPluginName = "kubernetes.io/git-repo"
gitRepoPluginLegacyName = "git"
)
func (plugin *gitRepoPlugin) Init(host volume.Host) {
plugin.host = host
}
func (plugin *gitRepoPlugin) Name() string {
if plugin.legacyMode {
return gitRepoPluginLegacyName
}
return gitRepoPluginName
}
func (plugin *gitRepoPlugin) CanSupport(spec *api.Volume) bool {
if plugin.legacyMode {
// Legacy mode instances can be cleaned up but not created anew.
return false
}
if spec.Source != nil && spec.Source.GitRepo != nil {
return true
}
return false
}
func (plugin *gitRepoPlugin) NewBuilder(spec *api.Volume, podUID types.UID) (volume.Builder, error) {
if plugin.legacyMode {
// Legacy mode instances can be cleaned up but not created anew.
return nil, fmt.Errorf("legacy mode: can not create new instances")
}
return &gitRepo{
podUID: podUID,
volName: spec.Name,
source: spec.Source.GitRepo.Repository,
revision: spec.Source.GitRepo.Revision,
exec: exec.New(),
plugin: plugin,
legacyMode: false,
}, nil
}
func (plugin *gitRepoPlugin) NewCleaner(volName string, podUID types.UID) (volume.Cleaner, error) {
legacy := false
if plugin.legacyMode {
legacy = true
}
return &gitRepo{
podUID: podUID,
volName: volName,
plugin: plugin,
legacyMode: legacy,
}, nil
}
// gitRepo volumes are directories which are pre-filled from a git repository.
// These do not persist beyond the lifetime of a pod.
type gitRepo struct {
volName string
podUID types.UID
source string
revision string
exec exec.Interface
plugin *gitRepoPlugin
legacyMode bool
}
// SetUp creates new directory and clones a git repo.
func (gr *gitRepo) SetUp() error {
if gr.isReady() {
return nil
}
if gr.legacyMode {
return fmt.Errorf("legacy mode: can not create new instances")
}
volPath := gr.GetPath()
if err := os.MkdirAll(volPath, 0750); err != nil {
return err
}
if output, err := gr.execCommand("git", []string{"clone", gr.source}, gr.GetPath()); err != nil {
return fmt.Errorf("failed to exec 'git clone %s': %s: %v", gr.source, output, err)
}
files, err := ioutil.ReadDir(gr.GetPath())
if err != nil {
return err
}
if len(files) != 1 {
return fmt.Errorf("unexpected directory contents: %v", files)
}
if len(gr.revision) == 0 {
// Done!
gr.setReady()
return nil
}
dir := path.Join(gr.GetPath(), files[0].Name())
if output, err := gr.execCommand("git", []string{"checkout", gr.revision}, dir); err != nil {
return fmt.Errorf("failed to exec 'git checkout %s': %s: %v", gr.revision, output, err)
}
if output, err := gr.execCommand("git", []string{"reset", "--hard"}, dir); err != nil {
return fmt.Errorf("failed to exec 'git reset --hard': %s: %v", output, err)
}
gr.setReady()
return nil
}
func (gr *gitRepo) getMetaDir() string {
return path.Join(gr.plugin.host.GetPodPluginDir(gr.podUID, volume.EscapePluginName(gitRepoPluginName)), gr.volName)
}
func (gr *gitRepo) isReady() bool {
metaDir := gr.getMetaDir()
readyFile := path.Join(metaDir, "ready")
s, err := os.Stat(readyFile)
if err != nil {
return false
}
if !s.Mode().IsRegular() {
glog.Errorf("GitRepo ready-file is not a file: %s", readyFile)
return false
}
return true
}
func (gr *gitRepo) setReady() {
metaDir := gr.getMetaDir()
if err := os.MkdirAll(metaDir, 0750); err != nil && !os.IsExist(err) {
glog.Errorf("Can't mkdir %s: %v", metaDir, err)
return
}
readyFile := path.Join(metaDir, "ready")
file, err := os.Create(readyFile)
if err != nil {
glog.Errorf("Can't touch %s: %v", readyFile, err)
return
}
file.Close()
}
func (gr *gitRepo) execCommand(command string, args []string, dir string) ([]byte, error) {
cmd := gr.exec.Command(command, args...)
cmd.SetDir(dir)
return cmd.CombinedOutput()
}
func (gr *gitRepo) GetPath() string {
name := gitRepoPluginName
if gr.legacyMode {
name = gitRepoPluginLegacyName
}
return gr.plugin.host.GetPodVolumeDir(gr.podUID, volume.EscapePluginName(name), gr.volName)
}
// TearDown simply deletes everything in the directory.
func (gr *gitRepo) TearDown() error {
tmpDir, err := volume.RenameDirectory(gr.GetPath(), gr.volName+".deleting~")
if err != nil {
return err
}
err = os.RemoveAll(tmpDir)
if err != nil {
return err
}
return nil
}

186
pkg/kubelet/volume/git_repo/git_repo_test.go Normal file
View File

@ -0,0 +1,186 @@
/*
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 git_repo
import (
"io/ioutil"
"os"
"path"
"reflect"
"strings"
"testing"
"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/exec"
)
func newTestHost(t *testing.T) volume.Host {
tempDir, err := ioutil.TempDir("/tmp", "git_repo_test.")
if err != nil {
t.Fatalf("can't make a temp rootdir: %v", err)
}
return &volume.FakeHost{tempDir}
}
func TestCanSupport(t *testing.T) {
plugMgr := volume.PluginMgr{}
plugMgr.InitPlugins(ProbeVolumePlugins(), newTestHost(t))
plug, err := plugMgr.FindPluginByName("kubernetes.io/git-repo")
if err != nil {
t.Errorf("Can't find the plugin by name")
}
if plug.Name() != "kubernetes.io/git-repo" {
t.Errorf("Wrong name: %s", plug.Name())
}
if !plug.CanSupport(&api.Volume{Source: &api.VolumeSource{GitRepo: &api.GitRepo{}}}) {
t.Errorf("Expected true")
}
}
func testSetUp(plug volume.Plugin, builder volume.Builder, t *testing.T) {
var fcmd exec.FakeCmd
fcmd = exec.FakeCmd{
CombinedOutputScript: []exec.FakeCombinedOutputAction{
// git clone
func() ([]byte, error) {
os.MkdirAll(path.Join(fcmd.Dirs[0], "kubernetes"), 0750)
return []byte{}, nil
},
// git checkout
func() ([]byte, error) { return []byte{}, nil },
// git reset
func() ([]byte, error) { return []byte{}, nil },
},
}
fake := exec.FakeExec{
CommandScript: []exec.FakeCommandAction{
func(cmd string, args ...string) exec.Cmd { return exec.InitFakeCmd(&fcmd, cmd, args...) },
func(cmd string, args ...string) exec.Cmd { return exec.InitFakeCmd(&fcmd, cmd, args...) },
func(cmd string, args ...string) exec.Cmd { return exec.InitFakeCmd(&fcmd, cmd, args...) },
},
}
g := builder.(*gitRepo)
g.exec = &fake
err := g.SetUp()
if err != nil {
t.Errorf("unexpected error: %v", err)
}
expectedCmds := [][]string{
{"git", "clone", g.source},
{"git", "checkout", g.revision},
{"git", "reset", "--hard"},
}
if fake.CommandCalls != len(expectedCmds) {
t.Errorf("unexpected command calls: expected 3, saw: %d", fake.CommandCalls)
}
if !reflect.DeepEqual(expectedCmds, fcmd.CombinedOutputLog) {
t.Errorf("unexpected commands: %v, expected: %v", fcmd.CombinedOutputLog, expectedCmds)
}
expectedDirs := []string{g.GetPath(), g.GetPath() + "/kubernetes", g.GetPath() + "/kubernetes"}
if len(fcmd.Dirs) != 3 || !reflect.DeepEqual(expectedDirs, fcmd.Dirs) {
t.Errorf("unexpected directories: %v, expected: %v", fcmd.Dirs, expectedDirs)
}
}
func TestPlugin(t *testing.T) {
plugMgr := volume.PluginMgr{}
plugMgr.InitPlugins(ProbeVolumePlugins(), newTestHost(t))
plug, err := plugMgr.FindPluginByName("kubernetes.io/git-repo")
if err != nil {
t.Errorf("Can't find the plugin by name")
}
spec := &api.Volume{
Name: "vol1",
Source: &api.VolumeSource{
GitRepo: &api.GitRepo{
Repository: "https://github.com/GoogleCloudPlatform/kubernetes.git",
Revision: "2a30ce65c5ab586b98916d83385c5983edd353a1",
},
},
}
builder, err := plug.NewBuilder(spec, types.UID("poduid"))
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 !strings.HasSuffix(path, "pods/poduid/volumes/kubernetes.io~git-repo/vol1") {
t.Errorf("Got unexpected path: %s", path)
}
testSetUp(plug, builder, t)
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.NewCleaner("vol1", types.UID("poduid"))
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)
}
}
func TestPluginLegacy(t *testing.T) {
plugMgr := volume.PluginMgr{}
plugMgr.InitPlugins(ProbeVolumePlugins(), newTestHost(t))
plug, err := plugMgr.FindPluginByName("git")
if err != nil {
t.Errorf("Can't find the plugin by name")
}
if plug.Name() != "git" {
t.Errorf("Wrong name: %s", plug.Name())
}
if plug.CanSupport(&api.Volume{Source: &api.VolumeSource{GitRepo: &api.GitRepo{}}}) {
t.Errorf("Expected false")
}
if _, err := plug.NewBuilder(&api.Volume{Source: &api.VolumeSource{GitRepo: &api.GitRepo{}}}, types.UID("poduid")); err == nil {
t.Errorf("Expected failiure")
}
cleaner, err := plug.NewCleaner("vol1", types.UID("poduid"))
if err != nil {
t.Errorf("Failed to make a new Cleaner: %v", err)
}
if cleaner == nil {
t.Errorf("Got a nil Cleaner: %v")
}
}

81
pkg/kubelet/volume/host_path/host_path.go Normal file
View File

@ -0,0 +1,81 @@
/*
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 host_path
import (
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/volume"
"github.com/GoogleCloudPlatform/kubernetes/pkg/types"
)
// This is the primary entrypoint for volume plugins.
func ProbeVolumePlugins() []volume.Plugin {
return []volume.Plugin{&hostPathPlugin{nil}}
}
type hostPathPlugin struct {
host volume.Host
}
var _ volume.Plugin = &hostPathPlugin{}
const (
hostPathPluginName = "kubernetes.io/host-path"
)
func (plugin *hostPathPlugin) Init(host volume.Host) {
plugin.host = host
}
func (plugin *hostPathPlugin) Name() string {
return hostPathPluginName
}
func (plugin *hostPathPlugin) CanSupport(spec *api.Volume) bool {
if spec.Source != nil && spec.Source.HostDir != nil {
return true
}
return false
}
func (plugin *hostPathPlugin) NewBuilder(spec *api.Volume, podUID types.UID) (volume.Builder, error) {
return &hostPath{spec.Source.HostDir.Path}, nil
}
func (plugin *hostPathPlugin) NewCleaner(volName string, podUID types.UID) (volume.Cleaner, error) {
return &hostPath{""}, nil
}
// HostPath volumes represent a bare host file or directory mount.
// The direct at the specified path will be directly exposed to the container.
type hostPath struct {
path string
}
// SetUp does nothing.
func (hp *hostPath) SetUp() error {
return nil
}
func (hp *hostPath) GetPath() string {
return hp.path
}
// TearDown does nothing.
func (hp *hostPath) TearDown() error {
return nil
}

86
pkg/kubelet/volume/host_path/host_path_test.go Normal file
View File

@ -0,0 +1,86 @@
/*
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 host_path
import (
"testing"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/volume"
"github.com/GoogleCloudPlatform/kubernetes/pkg/types"
)
func TestCanSupport(t *testing.T) {
plugMgr := volume.PluginMgr{}
plugMgr.InitPlugins(ProbeVolumePlugins(), &volume.FakeHost{"fake"})
plug, err := plugMgr.FindPluginByName("kubernetes.io/host-path")
if err != nil {
t.Errorf("Can't find the plugin by name")
}
if plug.Name() != "kubernetes.io/host-path" {
t.Errorf("Wrong name: %s", plug.Name())
}
if !plug.CanSupport(&api.Volume{Source: &api.VolumeSource{HostDir: &api.HostDir{}}}) {
t.Errorf("Expected true")
}
if plug.CanSupport(&api.Volume{Source: nil}) {
t.Errorf("Expected false")
}
}
func TestPlugin(t *testing.T) {
plugMgr := volume.PluginMgr{}
plugMgr.InitPlugins(ProbeVolumePlugins(), &volume.FakeHost{"fake"})
plug, err := plugMgr.FindPluginByName("kubernetes.io/host-path")
if err != nil {
t.Errorf("Can't find the plugin by name")
}
spec := &api.Volume{
Name: "vol1",
Source: &api.VolumeSource{HostDir: &api.HostDir{"/vol1"}},
}
builder, err := plug.NewBuilder(spec, types.UID("poduid"))
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 != "/vol1" {
t.Errorf("Got unexpected path: %s", path)
}
if err := builder.SetUp(); err != nil {
t.Errorf("Expected success, got: %v", err)
}
cleaner, err := plug.NewCleaner("vol1", types.UID("poduid"))
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)
}
}

174
pkg/kubelet/volume/plugins.go Normal file
View File

@ -0,0 +1,174 @@
/*
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 volume
import (
"fmt"
"strings"
"sync"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
"github.com/GoogleCloudPlatform/kubernetes/pkg/types"
"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
"github.com/GoogleCloudPlatform/kubernetes/pkg/util/errors"
"github.com/golang/glog"
)
// Plugin is an interface to volume plugins.
type Plugin interface {
// Init initializes the plugin. This will be called exactly once
// before any New* calls are made - implementations of plugins may
// depend on this.
Init(host Host)
// Name returns the plugin's name. Plugins should use namespaced names
// such as "example.com/volume". The "kubernetes.io" namespace is
// reserved for plugins which are bundled with kubernetes.
Name() string
// CanSupport tests whether the Plugin supports a given volume
// specification from the API. The spec pointer should be considered
// const.
CanSupport(spec *api.Volume) bool
// NewBuilder creates a new volume.Builder from an API specification.
// Ownership of the spec pointer in *not* transferred.
// - spec: The api.Volume spec
// - podUID: The UID of the enclosing pod
NewBuilder(spec *api.Volume, podUID types.UID) (Builder, error)
// NewCleaner creates a new volume.Cleaner from recoverable state.
// - name: The volume name, as per the api.Volume spec.
// - podUID: The UID of the enclosing pod
NewCleaner(name string, podUID types.UID) (Cleaner, error)
}
// Host is an interface that plugins can use to access the kubelet.
type Host interface {
// GetPluginDir returns the absolute path to a directory under which
// a given plugin may store data. This directory might not actually
// exist on disk yet. For plugin data that is per-pod, see
// GetPodPluginDir().
GetPluginDir(pluginName string) string
// GetPodVolumeDir returns the absolute path a directory which
// represents the named volume under the named plugin for the given
// pod. If the specified pod does not exist, the result of this call
// might not exist.
GetPodVolumeDir(podUID types.UID, pluginName string, volumeName string) string
// GetPodPluginDir returns the absolute path to a directory under which
// a given plugin may store data for a given pod. If the specified pod
// does not exist, the result of this call might not exist. This
// directory might not actually exist on disk yet.
GetPodPluginDir(podUID types.UID, pluginName string) string
}
// PluginMgr tracks registered plugins.
type PluginMgr struct {
mutex sync.Mutex
plugins map[string]Plugin
}
// InitPlugins initializes each plugin. All plugins must have unique names.
// This must be called exactly once before any New* methods are called on any
// plugins.
func (pm *PluginMgr) InitPlugins(plugins []Plugin, host Host) error {
pm.mutex.Lock()
defer pm.mutex.Unlock()
if pm.plugins == nil {
pm.plugins = map[string]Plugin{}
}
allErrs := []error{}
for _, plugin := range plugins {
name := plugin.Name()
if !util.IsQualifiedName(name) {
allErrs = append(allErrs, fmt.Errorf("volume plugin has invalid name: %#v", plugin))
continue
}
if _, found := pm.plugins[name]; found {
allErrs = append(allErrs, fmt.Errorf("volume plugin %q was registered more than once", name))
continue
}
plugin.Init(host)
pm.plugins[name] = plugin
glog.V(1).Infof("Loaded volume plugin %q", name)
}
return errors.NewAggregate(allErrs)
}
// FindPluginBySpec looks for a plugin that can support a given volume
// specification. If no plugins can support or more than one plugin can
// support it, return error.
func (pm *PluginMgr) FindPluginBySpec(spec *api.Volume) (Plugin, error) {
pm.mutex.Lock()
defer pm.mutex.Unlock()
matches := []string{}
for k, v := range pm.plugins {
if v.CanSupport(spec) {
matches = append(matches, k)
}
}
if len(matches) == 0 {
return nil, fmt.Errorf("no volume plugin matched")
}
if len(matches) > 1 {
return nil, fmt.Errorf("multiple volume plugins matched: %s", strings.Join(matches, ","))
}
return pm.plugins[matches[0]], nil
}
// FindPluginByName fetches a plugin by name or by legacy name. If no plugin
// is found, returns error.
func (pm *PluginMgr) FindPluginByName(name string) (Plugin, error) {
pm.mutex.Lock()
defer pm.mutex.Unlock()
// Once we can get rid of legacy names we can reduce this to a map lookup.
matches := []string{}
for k, v := range pm.plugins {
if v.Name() == name {
matches = append(matches, k)
}
}
if len(matches) == 0 {
return nil, fmt.Errorf("no volume plugin matched")
}
if len(matches) > 1 {
return nil, fmt.Errorf("multiple volume plugins matched: %s", strings.Join(matches, ","))
}
return pm.plugins[matches[0]], nil
}
// EscapePluginName converts a plugin name, which might contain a / into a
// string that is safe to use on-disk. This assumes that the input has already
// been validates as a qualified name. we use "~" rather than ":" here in case
// we ever use a filesystem that doesn't allow ":".
func EscapePluginName(in string) string {
return strings.Replace(in, "/", "~", -1)
}
// UnescapePluginName converts an escaped plugin name (as per EscapePluginName)
// back to its normal form. This assumes that the input has already been
// validates as a qualified name.
func UnescapePluginName(in string) string {
return strings.Replace(in, "~", "/", -1)
}

92
pkg/kubelet/volume/testing.go Normal file
View File

@ -0,0 +1,92 @@
/*
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 volume
import (
"os"
"path"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
"github.com/GoogleCloudPlatform/kubernetes/pkg/types"
)
// FakeHost is useful for testing volume plugins.
type FakeHost struct {
RootDir string
}
func (f *FakeHost) GetPluginDir(podUID string) string {
return path.Join(f.RootDir, "plugins", podUID)
}
func (f *FakeHost) GetPodVolumeDir(podUID types.UID, pluginName, volumeName string) string {
return path.Join(f.RootDir, "pods", string(podUID), "volumes", pluginName, volumeName)
}
func (f *FakeHost) GetPodPluginDir(podUID types.UID, pluginName string) string {
return path.Join(f.RootDir, "pods", string(podUID), "plugins", pluginName)
}
// FakePlugin is useful for for testing. It tries to be a fully compliant
// plugin, but all it does is make empty directories.
// Use as:
// volume.RegisterPlugin(&FakePlugin{"fake-name"})
type FakePlugin struct {
PluginName string
Host Host
}
var _ Plugin = &FakePlugin{}
func (plugin *FakePlugin) Init(host Host) {
plugin.Host = host
}
func (plugin *FakePlugin) Name() string {
return plugin.PluginName
}
func (plugin *FakePlugin) CanSupport(spec *api.Volume) bool {
// TODO: maybe pattern-match on spec.Name to decide?
return true
}
func (plugin *FakePlugin) NewBuilder(spec *api.Volume, podUID types.UID) (Builder, error) {
return &FakeVolume{podUID, spec.Name, plugin}, nil
}
func (plugin *FakePlugin) NewCleaner(volName string, podUID types.UID) (Cleaner, error) {
return &FakeVolume{podUID, volName, plugin}, nil
}
type FakeVolume struct {
PodUID types.UID
VolName string
Plugin *FakePlugin
}
func (fv *FakeVolume) SetUp() error {
return os.MkdirAll(fv.GetPath(), 0750)
}
func (fv *FakeVolume) GetPath() string {
return path.Join(fv.Plugin.Host.GetPodVolumeDir(fv.PodUID, EscapePluginName(fv.Plugin.PluginName), fv.VolName))
}
func (fv *FakeVolume) TearDown() error {
return os.RemoveAll(fv.GetPath())
}

59
pkg/kubelet/volume/volume.go Normal file
View File

@ -0,0 +1,59 @@
/*
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 volume
import (
"io/ioutil"
"os"
"path"
)
// Interface is a directory used by pods or hosts.
// All method implementations of methods in the volume interface must be idempotent.
type Interface interface {
// GetPath returns the directory path the volume is mounted to.
GetPath() string
}
// Builder interface provides method to set up/mount the volume.
type Builder interface {
// Uses Interface to provide the path for Docker binds.
Interface
// SetUp prepares and mounts/unpacks the volume to a directory path.
// This may be called more than once, so implementations must be
// idempotent.
SetUp() error
}
// Cleaner interface provides method to cleanup/unmount the volumes.
type Cleaner interface {
Interface
// TearDown unmounts the volume and removes traces of the SetUp procedure.
TearDown() error
}
func RenameDirectory(oldPath, newName string) (string, error) {
newPath, err := ioutil.TempDir(path.Dir(oldPath), newName)
if err != nil {
return "", err
}
err = os.Rename(oldPath, newPath)
if err != nil {
return "", err
}
return newPath, nil
}

154
pkg/kubelet/volumes.go Normal file
View File

@ -0,0 +1,154 @@
/*
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 kubelet
import (
"fmt"
"io/ioutil"
"path"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/volume"
"github.com/GoogleCloudPlatform/kubernetes/pkg/types"
"github.com/davecgh/go-spew/spew"
"github.com/golang/glog"
)
var errUnsupportedVolumeType = fmt.Errorf("unsupported volume type")
// This just exports required functions from kubelet proper, for use by volume
// plugins.
type volumeHost struct {
kubelet *Kubelet
}
func (vh *volumeHost) GetPluginDir(pluginName string) string {
return vh.kubelet.getPluginDir(pluginName)
}
func (vh *volumeHost) GetPodVolumeDir(podUID types.UID, pluginName string, volumeName string) string {
return vh.kubelet.getPodVolumeDir(podUID, pluginName, volumeName)
}
func (vh *volumeHost) GetPodPluginDir(podUID types.UID, pluginName string) string {
return vh.kubelet.getPodPluginDir(podUID, pluginName)
}
func (kl *Kubelet) newVolumeBuilderFromPlugins(spec *api.Volume, podUID types.UID) volume.Builder {
plugin, err := kl.volumePluginMgr.FindPluginBySpec(spec)
if err != nil {
glog.Warningf("Can't use volume plugins for %s: %v", spew.Sprintf("%#v", *spec), err)
return nil
}
if plugin == nil {
glog.Errorf("No error, but nil volume plugin for %s", spew.Sprintf("%#v", *spec))
return nil
}
builder, err := plugin.NewBuilder(spec, podUID)
if err != nil {
glog.Warningf("Error instantiating volume plugin for %s: %v", spew.Sprintf("%#v", *spec), err)
return nil
}
glog.V(3).Infof("Used volume plugin %q for %s", plugin.Name(), spew.Sprintf("%#v", *spec))
return builder
}
func (kl *Kubelet) mountExternalVolumes(pod *api.BoundPod) (volumeMap, error) {
podVolumes := make(volumeMap)
for i := range pod.Spec.Volumes {
volSpec := &pod.Spec.Volumes[i]
// Try to use a plugin for this volume.
builder := kl.newVolumeBuilderFromPlugins(volSpec, pod.UID)
if builder == nil {
return nil, errUnsupportedVolumeType
}
err := builder.SetUp()
if err != nil {
return nil, err
}
podVolumes[volSpec.Name] = builder
}
return podVolumes, nil
}
// getPodVolumesFromDisk examines directory structure to determine volumes that
// are presently active and mounted. Returns a map of volume.Cleaner types.
func (kl *Kubelet) getPodVolumesFromDisk() map[string]volume.Cleaner {
currentVolumes := make(map[string]volume.Cleaner)
podUIDs, err := kl.listPodsFromDisk()
if err != nil {
glog.Errorf("Could not get pods from disk: %v", err)
return map[string]volume.Cleaner{}
}
// Find the volumes for each on-disk pod.
for _, podUID := range podUIDs {
podVolDir := kl.getPodVolumesDir(podUID)
volumeKindDirs, err := ioutil.ReadDir(podVolDir)
if err != nil {
glog.Errorf("Could not read directory %s: %v", podVolDir, err)
}
for _, volumeKindDir := range volumeKindDirs {
volumeKind := volumeKindDir.Name()
volumeKindPath := path.Join(podVolDir, volumeKind)
volumeNameDirs, err := ioutil.ReadDir(volumeKindPath)
if err != nil {
glog.Errorf("Could not read directory %s: %v", volumeKindPath, err)
}
for _, volumeNameDir := range volumeNameDirs {
volumeName := volumeNameDir.Name()
identifier := fmt.Sprintf("%s/%s", podUID, volumeName)
glog.V(4).Infof("Making a volume.Cleaner for %s", volumeKindPath)
// TODO(thockin) This should instead return a reference to an extant
// volume object, except that we don't actually hold on to pod specs
// or volume objects.
// Try to use a plugin for this volume.
cleaner := kl.newVolumeCleanerFromPlugins(volumeKind, volumeName, podUID)
if cleaner == nil {
glog.Errorf("Could not create volume cleaner for %s: %v", volumeNameDir.Name(), errUnsupportedVolumeType)
continue
}
currentVolumes[identifier] = cleaner
}
}
}
return currentVolumes
}
func (kl *Kubelet) newVolumeCleanerFromPlugins(kind string, name string, podUID types.UID) volume.Cleaner {
plugName := volume.UnescapePluginName(kind)
plugin, err := kl.volumePluginMgr.FindPluginByName(plugName)
if err != nil {
// TODO: Maybe we should launch a cleanup of this dir?
glog.Warningf("Can't use volume plugins for %s/%s: %v", podUID, kind, err)
return nil
}
if plugin == nil {
glog.Errorf("No error, but nil volume plugin for %s/%s", podUID, kind)
return nil
}
cleaner, err := plugin.NewCleaner(name, podUID)
if err != nil {
glog.Warningf("Error instantiating volume plugin for %s/%s: %v", podUID, kind, err)
return nil
}
glog.V(3).Infof("Used volume plugin %q for %s/%s", plugin.Name(), podUID, kind)
return cleaner
}

9
pkg/standalone/standalone.go
View File

@ -34,6 +34,7 @@ import (
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet"
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/config"
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/dockertools"
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/volume"
"github.com/GoogleCloudPlatform/kubernetes/pkg/master"
"github.com/GoogleCloudPlatform/kubernetes/pkg/service"
"github.com/GoogleCloudPlatform/kubernetes/pkg/tools"
@ -150,7 +151,8 @@ func SimpleRunKubelet(client *client.Client,
dockerClient dockertools.DockerInterface,
hostname, rootDir, manifestURL, address string,
port uint,
masterServiceNamespace string) {
masterServiceNamespace string,
volumePlugins []volume.Plugin) {
kcfg := KubeletConfig{
KubeClient: client,
EtcdClient: etcdClient,
@ -167,6 +169,7 @@ func SimpleRunKubelet(client *client.Client,
MinimumGCAge: 10 * time.Second,
MaxContainerCount: 5,
MasterServiceNamespace: masterServiceNamespace,
VolumePlugins: volumePlugins,
}
RunKubelet(&kcfg)
}
@ -267,6 +270,7 @@ type KubeletConfig struct {
Port uint
Runonce bool
MasterServiceNamespace string
VolumePlugins []volume.Plugin
}
func createAndInitKubelet(kc *KubeletConfig, pc *config.PodConfig) (*kubelet.Kubelet, error) {
@ -288,7 +292,8 @@ func createAndInitKubelet(kc *KubeletConfig, pc *config.PodConfig) (*kubelet.Kub
pc.IsSourceSeen,
kc.ClusterDomain,
net.IP(kc.ClusterDNS),
kc.MasterServiceNamespace)
kc.MasterServiceNamespace,
kc.VolumePlugins)
if err != nil {
return nil, err

2
pkg/util/errors/errors.go
View File

@ -27,7 +27,7 @@ type Aggregate interface {
// NewAggregate converts a slice of errors into an Aggregate interface, which
// is itself an implementation of the error interface. If the slice is empty,
// this returs nil.
// this returns nil.
func NewAggregate(errlist []error) Aggregate {
if len(errlist) == 0 {
return nil

89
pkg/volume/mounter_linux.go
View File

@ -1,89 +0,0 @@
/*
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 volume
import (
"bufio"
"io"
"os"
"regexp"
"strings"
"syscall"
"github.com/golang/glog"
)
const MOUNT_MS_BIND = syscall.MS_BIND
const MOUNT_MS_RDONLY = syscall.MS_RDONLY
type DiskMounter struct{}
// Wraps syscall.Mount()
func (mounter *DiskMounter) Mount(source string, target string, fstype string, flags uintptr, data string) error {
glog.V(5).Infof("Mounting %s %s %s %d %s", source, target, fstype, flags, data)
return syscall.Mount(source, target, fstype, flags, data)
}
// Wraps syscall.Unmount()
func (mounter *DiskMounter) Unmount(target string, flags int) error {
return syscall.Unmount(target, flags)
}
// 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 (mounter *DiskMounter) RefCount(mount Interface) (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 volume's path is an active mount point.
file, err := os.Open("/proc/mounts")
if err != nil {
return "", -1, err
}
defer file.Close()
scanner := bufio.NewReader(file)
refCount := 0
var deviceName string
// Find the actual device path.
for {
line, err := scanner.ReadString('\n')
if err == io.EOF {
break
}
success, err := regexp.MatchString(mount.GetPath(), line)
if err != nil {
return "", -1, err
}
if success {
deviceName = strings.Split(line, " ")[0]
}
}
file.Close()
file, err = os.Open("/proc/mounts")
scanner.Reset(bufio.NewReader(file))
// Find the number of references to the device.
for {
line, err := scanner.ReadString('\n')
if err == io.EOF {
break
}
if strings.Split(line, " ")[0] == deviceName {
refCount++
}
}
return deviceName, refCount, nil
}

36
pkg/volume/mounter_unsupported.go
View File

@ -1,36 +0,0 @@
// +build !linux
/*
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 volume
const MOUNT_MS_BIND = 0
const MOUNT_MS_RDONLY = 0
type DiskMounter struct{}
func (mounter *DiskMounter) Mount(source string, target string, fstype string, flags uintptr, data string) error {
return nil
}
func (mounter *DiskMounter) Unmount(target string, flags int) error {
return nil
}
func (mounter *DiskMounter) RefCount(PD Interface) (string, int, error) {
return "", 0, nil
}

438
pkg/volume/volume.go
View File

@ -1,438 +0,0 @@
/*
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 volume
import (
"errors"
"fmt"
"io/ioutil"
"os"
"path"
"strconv"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
"github.com/GoogleCloudPlatform/kubernetes/pkg/util/exec"
"github.com/golang/glog"
)
var ErrUnsupportedVolumeType = errors.New("unsupported volume type")
// Interface is a directory used by pods or hosts.
// All method implementations of methods in the volume interface must be idempotent.
type Interface interface {
// GetPath returns the directory path the volume is mounted to.
GetPath() string
}
// Builder interface provides method to set up/mount the volume.
type Builder interface {
// Uses Interface to provide the path for Docker binds.
Interface
// SetUp prepares and mounts/unpacks the volume to a directory path.
SetUp() error
}
// Cleaner interface provides method to cleanup/unmount the volumes.
type Cleaner interface {
// TearDown unmounts the volume and removes traces of the SetUp procedure.
TearDown() error
}
type gcePersistentDiskUtil interface {
// Attaches the disk to the kubelet's host machine.
AttachDisk(PD *GCEPersistentDisk) error
// Detaches the disk from the kubelet's host machine.
DetachDisk(PD *GCEPersistentDisk, devicePath string) error
}
// Mounters wrap os/system specific calls to perform mounts.
type mounter interface {
Mount(source string, target string, fstype string, flags uintptr, data string) error
Unmount(target string, flags int) error
// RefCount returns the device path for the source disk of a volume, and
// the number of references to that target disk.
RefCount(vol Interface) (string, int, error)
}
// HostDir volumes represent a bare host directory mount.
// The directory in Path will be directly exposed to the container.
type HostDir struct {
Path string
}
// SetUp implements interface definitions, even though host directory
// mounts don't require any setup or cleanup.
func (hostVol *HostDir) SetUp() error {
return nil
}
func (hostVol *HostDir) GetPath() string {
return hostVol.Path
}
type execInterface interface {
ExecCommand(cmd []string, dir string) ([]byte, error)
}
type GitDir struct {
Source string
Revision string
PodID string
RootDir string
Name string
exec exec.Interface
}
func newGitRepo(volume *api.Volume, podID, rootDir string) *GitDir {
return &GitDir{
Source: volume.Source.GitRepo.Repository,
Revision: volume.Source.GitRepo.Revision,
PodID: podID,
RootDir: rootDir,
Name: volume.Name,
exec: exec.New(),
}
}
func (g *GitDir) ExecCommand(command string, args []string, dir string) ([]byte, error) {
cmd := g.exec.Command(command, args...)
cmd.SetDir(dir)
return cmd.CombinedOutput()
}
func (g *GitDir) SetUp() error {
volumePath := g.GetPath()
if err := os.MkdirAll(volumePath, 0750); err != nil {
return err
}
if _, err := g.ExecCommand("git", []string{"clone", g.Source}, g.GetPath()); err != nil {
return err
}
files, err := ioutil.ReadDir(g.GetPath())
if err != nil {
return err
}
if len(g.Revision) == 0 {
return nil
}
if len(files) != 1 {
return fmt.Errorf("unexpected directory contents: %v", files)
}
dir := path.Join(g.GetPath(), files[0].Name())
if _, err := g.ExecCommand("git", []string{"checkout", g.Revision}, dir); err != nil {
return err
}
if _, err := g.ExecCommand("git", []string{"reset", "--hard"}, dir); err != nil {
return err
}
return nil
}
func (g *GitDir) GetPath() string {
return path.Join(g.RootDir, g.PodID, "volumes", "git", g.Name)
}
// TearDown simply deletes everything in the directory.
func (g *GitDir) TearDown() error {
tmpDir, err := renameDirectory(g.GetPath(), g.Name+"~deleting")
if err != nil {
return err
}
err = os.RemoveAll(tmpDir)
if err != nil {
return err
}
return nil
}
// EmptyDir volumes are temporary directories exposed to the pod.
// These do not persist beyond the lifetime of a pod.
type EmptyDir struct {
Name string
PodID string
RootDir string
}
// SetUp creates new directory.
func (emptyDir *EmptyDir) SetUp() error {
path := emptyDir.GetPath()
return os.MkdirAll(path, 0750)
}
func (emptyDir *EmptyDir) GetPath() string {
return path.Join(emptyDir.RootDir, emptyDir.PodID, "volumes", "empty", emptyDir.Name)
}
func renameDirectory(oldPath, newName string) (string, error) {
newPath, err := ioutil.TempDir(path.Dir(oldPath), newName)
if err != nil {
return "", err
}
err = os.Rename(oldPath, newPath)
if err != nil {
return "", err
}
return newPath, nil
}
// TearDown simply deletes everything in the directory.
func (emptyDir *EmptyDir) TearDown() error {
tmpDir, err := renameDirectory(emptyDir.GetPath(), emptyDir.Name+".deleting~")
if err != nil {
return err
}
err = os.RemoveAll(tmpDir)
if err != nil {
return err
}
return nil
}
// createHostDir interprets API volume as a HostDir.
func createHostDir(volume *api.Volume) *HostDir {
return &HostDir{volume.Source.HostDir.Path}
}
// GCEPersistentDisk volumes are disk resources provided by Google Compute Engine
// that are attached to the kubelet's host machine and exposed to the pod.
type GCEPersistentDisk struct {
Name string
PodID string
RootDir string
// Unique identifier 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 ReadOnly.
ReadOnly bool
// Utility interface that provides API calls to the provider to attach/detach disks.
util gcePersistentDiskUtil
// Mounter interface that provides system calls to mount the disks.
mounter mounter
}
func (PD *GCEPersistentDisk) GetPath() string {
return path.Join(PD.RootDir, PD.PodID, "volumes", "gce-pd", PD.Name)
}
// Attaches the disk and bind mounts to the volume path.
func (PD *GCEPersistentDisk) SetUp() error {
// TODO: handle failed mounts here.
mountpoint, err := isMountPoint(PD.GetPath())
glog.V(4).Infof("PersistentDisk set up: %s %v %v", PD.GetPath(), mountpoint, err)
if err != nil && !os.IsNotExist(err) {
return err
}
if mountpoint {
return nil
}
if err := PD.util.AttachDisk(PD); err != nil {
return err
}
flags := uintptr(0)
if PD.ReadOnly {
flags = MOUNT_MS_RDONLY
}
//Perform a bind mount to the full path to allow duplicate mounts of the same PD.
if _, err = os.Stat(PD.GetPath()); os.IsNotExist(err) {
err = os.MkdirAll(PD.GetPath(), 0750)
if err != nil {
return err
}
globalPDPath := makeGlobalPDName(PD.RootDir, PD.PDName, PD.ReadOnly)
err = PD.mounter.Mount(globalPDPath, PD.GetPath(), "", MOUNT_MS_BIND|flags, "")
if err != nil {
os.RemoveAll(PD.GetPath())
return err
}
}
return nil
}
// 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 *GCEPersistentDisk) TearDown() error {
mountpoint, err := isMountPoint(PD.GetPath())
if err != nil {
return err
}
if !mountpoint {
return os.RemoveAll(PD.GetPath())
}
devicePath, refCount, err := PD.mounter.RefCount(PD)
if err != nil {
return err
}
if err := PD.mounter.Unmount(PD.GetPath(), 0); err != nil {
return err
}
refCount--
if err := os.RemoveAll(PD.GetPath()); err != nil {
return err
}
// If refCount 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.util.DetachDisk(PD, devicePath); err != nil {
return err
}
}
return nil
}
//TODO(jonesdl) prevent name collisions by using designated pod space as well.
// Ex. (ROOT_DIR)/pods/...
func makeGlobalPDName(rootDir, devName string, readOnly bool) string {
var mode string
if readOnly {
mode = "ro"
} else {
mode = "rw"
}
return path.Join(rootDir, "global", "pd", mode, devName)
}
// createEmptyDir interprets API volume as an EmptyDir.
func createEmptyDir(volume *api.Volume, podID string, rootDir string) *EmptyDir {
return &EmptyDir{volume.Name, podID, rootDir}
}
// Interprets API volume as a PersistentDisk
func createGCEPersistentDisk(volume *api.Volume, podID string, rootDir string) (*GCEPersistentDisk, error) {
PDName := volume.Source.GCEPersistentDisk.PDName
FSType := volume.Source.GCEPersistentDisk.FSType
partition := strconv.Itoa(volume.Source.GCEPersistentDisk.Partition)
if partition == "0" {
partition = ""
}
readOnly := volume.Source.GCEPersistentDisk.ReadOnly
// TODO: move these up into the Kubelet.
util := &GCEDiskUtil{}
mounter := &DiskMounter{}
return &GCEPersistentDisk{
Name: volume.Name,
PodID: podID,
RootDir: rootDir,
PDName: PDName,
FSType: FSType,
Partition: partition,
ReadOnly: readOnly,
util: util,
mounter: mounter}, nil
}
// CreateVolumeBuilder returns a Builder capable of mounting a volume described by an
// *api.Volume, or an error.
func CreateVolumeBuilder(volume *api.Volume, podID string, rootDir string) (Builder, error) {
source := volume.Source
// TODO(jonesdl) We will want to throw an error here when we no longer
// support the default behavior.
if source == nil {
return nil, nil
}
var vol Builder
var err error
// TODO(jonesdl) We should probably not check every pointer and directly
// resolve these types instead.
if source.HostDir != nil {
vol = createHostDir(volume)
} else if source.EmptyDir != nil {
vol = createEmptyDir(volume, podID, rootDir)
} else if source.GCEPersistentDisk != nil {
vol, err = createGCEPersistentDisk(volume, podID, rootDir)
if err != nil {
return nil, err
}
} else if source.GitRepo != nil {
vol = newGitRepo(volume, podID, rootDir)
} else {
return nil, ErrUnsupportedVolumeType
}
return vol, nil
}
// CreateVolumeCleaner returns a Cleaner capable of tearing down a volume.
func CreateVolumeCleaner(kind string, name string, podID string, rootDir string) (Cleaner, error) {
switch kind {
case "empty":
return &EmptyDir{name, podID, rootDir}, nil
case "gce-pd":
return &GCEPersistentDisk{
Name: name,
PodID: podID,
RootDir: rootDir,
util: &GCEDiskUtil{},
mounter: &DiskMounter{}}, nil
case "git":
return &GitDir{
Name: name,
PodID: podID,
RootDir: rootDir,
}, nil
default:
return nil, ErrUnsupportedVolumeType
}
}
// GetCurrentVolumes examines directory structure to determine volumes that are
// presently active and mounted. Returns a map of Cleaner types.
func GetCurrentVolumes(rootDirectory string) map[string]Cleaner {
currentVolumes := make(map[string]Cleaner)
podIDDirs, err := ioutil.ReadDir(rootDirectory)
if err != nil {
glog.Errorf("Could not read directory %s: %v", rootDirectory, err)
}
// Volume information is extracted from the directory structure:
// (ROOT_DIR)/(POD_ID)/volumes/(VOLUME_KIND)/(VOLUME_NAME)
for _, podIDDir := range podIDDirs {
if !podIDDir.IsDir() {
continue
}
podID := podIDDir.Name()
podIDPath := path.Join(rootDirectory, podID, "volumes")
if _, err := os.Stat(podIDPath); os.IsNotExist(err) {
continue
}
volumeKindDirs, err := ioutil.ReadDir(podIDPath)
if err != nil {
glog.Errorf("Could not read directory %s: %v", podIDPath, err)
}
for _, volumeKindDir := range volumeKindDirs {
volumeKind := volumeKindDir.Name()
volumeKindPath := path.Join(podIDPath, volumeKind)
volumeNameDirs, err := ioutil.ReadDir(volumeKindPath)
if err != nil {
glog.Errorf("Could not read directory %s: %v", volumeKindPath, err)
}
for _, volumeNameDir := range volumeNameDirs {
volumeName := volumeNameDir.Name()
identifier := path.Join(podID, volumeName)
// TODO(thockin) This should instead return a reference to an extant volume object
cleaner, err := CreateVolumeCleaner(volumeKind, volumeName, podID, rootDirectory)
if err != nil {
glog.Errorf("Could not create volume cleaner for %s: %v", volumeNameDir.Name(), err)
continue
}
currentVolumes[identifier] = cleaner
}
}
}
return currentVolumes
}

295
pkg/volume/volume_test.go
View File

@ -1,295 +0,0 @@
/*
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 volume
import (
"io/ioutil"
"os"
"path"
"reflect"
"testing"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
"github.com/GoogleCloudPlatform/kubernetes/pkg/util/exec"
)
type MockDiskUtil struct{}
// TODO(jonesdl) To fully test this, we could create a loopback device
// and mount that instead.
func (util *MockDiskUtil) AttachDisk(PD *GCEPersistentDisk) error {
err := os.MkdirAll(path.Join(PD.RootDir, "global", "pd", PD.PDName), 0750)
if err != nil {
return err
}
return nil
}
func (util *MockDiskUtil) DetachDisk(PD *GCEPersistentDisk, devicePath string) error {
err := os.RemoveAll(path.Join(PD.RootDir, "global", "pd", PD.PDName))
if err != nil {
return err
}
return nil
}
type MockMounter struct{}
func (mounter *MockMounter) Mount(source string, target string, fstype string, flags uintptr, data string) error {
return nil
}
func (mounter *MockMounter) Unmount(target string, flags int) error {
return nil
}
func (mounter *MockMounter) RefCount(vol Interface) (string, int, error) {
return "", 0, nil
}
func TestCreateVolumeBuilders(t *testing.T) {
tempDir := "CreateVolumes"
createVolumesTests := []struct {
volume api.Volume
path string
podID string
}{
{
api.Volume{
Name: "host-dir",
Source: &api.VolumeSource{
HostDir: &api.HostDir{"/dir/path"},
},
},
"/dir/path",
"",
},
{
api.Volume{
Name: "empty-dir",
Source: &api.VolumeSource{
EmptyDir: &api.EmptyDir{},
},
},
path.Join(tempDir, "/my-id/volumes/empty/empty-dir"),
"my-id",
},
{
api.Volume{
Name: "gce-pd",
Source: &api.VolumeSource{
GCEPersistentDisk: &api.GCEPersistentDisk{"my-disk", "ext4", 0, false},
},
},
path.Join(tempDir, "/my-id/volumes/gce-pd/gce-pd"),
"my-id",
},
{api.Volume{}, "", ""},
{
api.Volume{
Name: "empty-dir",
Source: &api.VolumeSource{},
},
"",
"",
},
}
for _, createVolumesTest := range createVolumesTests {
tt := createVolumesTest
vb, err := CreateVolumeBuilder(&tt.volume, tt.podID, tempDir)
if tt.volume.Source == nil {
if vb != nil {
t.Errorf("Expected volume to be nil")
}
continue
}
if tt.volume.Source.HostDir == nil && tt.volume.Source.EmptyDir == nil && tt.volume.Source.GCEPersistentDisk == nil {
if err != ErrUnsupportedVolumeType {
t.Errorf("Unexpected error: %v", err)
}
continue
}
if err != nil {
t.Errorf("Unexpected error: %v", err)
}
path := vb.GetPath()
if path != tt.path {
t.Errorf("Unexpected bind path. Expected %v, got %v", tt.path, path)
}
}
}
func TestCreateVolumeCleaners(t *testing.T) {
tempDir := "CreateVolumeCleaners"
createVolumeCleanerTests := []struct {
kind string
name string
podID string
}{
{"empty", "empty-vol", "my-id"},
{"", "", ""},
{"gce-pd", "gce-pd-vol", "my-id"},
}
for _, tt := range createVolumeCleanerTests {
vol, err := CreateVolumeCleaner(tt.kind, tt.name, tt.podID, tempDir)
if tt.kind == "" && err != nil && vol == nil {
continue
}
if err != nil {
t.Errorf("Unexpected error occurred: %v", err)
}
actualKind := reflect.TypeOf(vol).Elem().Name()
if tt.kind == "empty" && actualKind != "EmptyDir" {
t.Errorf("CreateVolumeCleaner returned invalid type. Expected EmptyDirectory, got %v, %v", tt.kind, actualKind)
}
if tt.kind == "gce-pd" && actualKind != "GCEPersistentDisk" {
t.Errorf("CreateVolumeCleaner returned invalid type. Expected PersistentDisk, got %v, %v", tt.kind, actualKind)
}
}
}
func TestSetUpAndTearDown(t *testing.T) {
tempDir, err := ioutil.TempDir("", "CreateVolumes")
if err != nil {
t.Errorf("Unexpected error: %v", err)
}
defer os.RemoveAll(tempDir)
fakeID := "my-id"
type VolumeTester interface {
Builder
Cleaner
}
volumes := []VolumeTester{
&EmptyDir{"empty", fakeID, tempDir},
&GCEPersistentDisk{"pd", fakeID, tempDir, "pd-disk", "ext4", "", false, &MockDiskUtil{}, &MockMounter{}},
}
for _, vol := range volumes {
err = vol.SetUp()
path := vol.GetPath()
if err != nil {
t.Errorf("Unexpected error: %v", err)
}
if _, err := os.Stat(path); os.IsNotExist(err) {
t.Errorf("SetUp() failed, volume path not created: %v", path)
}
err = vol.TearDown()
if err != nil {
t.Errorf("Unexpected error: %v", err)
}
if _, err := os.Stat(path); !os.IsNotExist(err) {
t.Errorf("TearDown() failed, original volume path not properly removed: %v", path)
}
}
}
func TestGetActiveVolumes(t *testing.T) {
tempDir, err := ioutil.TempDir("", "CreateVolumes")
if err != nil {
t.Errorf("Unexpected error: %v", err)
}
defer os.RemoveAll(tempDir)
getActiveVolumesTests := []struct {
name string
podID string
kind string
identifier string
}{
{"fakeName", "fakeID", "empty", "fakeID/fakeName"},
{"fakeName2", "fakeID2", "empty", "fakeID2/fakeName2"},
}
expectedIdentifiers := []string{}
for _, test := range getActiveVolumesTests {
volumeDir := path.Join(tempDir, test.podID, "volumes", test.kind, test.name)
os.MkdirAll(volumeDir, 0750)
expectedIdentifiers = append(expectedIdentifiers, test.identifier)
}
volumeMap := GetCurrentVolumes(tempDir)
for _, name := range expectedIdentifiers {
if _, ok := volumeMap[name]; !ok {
t.Errorf("Expected volume map entry not found: %v", name)
}
}
}
type fakeExec struct {
cmds [][]string
dirs []string
data []byte
err error
action func([]string, string)
}
func (f *fakeExec) ExecCommand(cmd []string, dir string) ([]byte, error) {
f.cmds = append(f.cmds, cmd)
f.dirs = append(f.dirs, dir)
f.action(cmd, dir)
return f.data, f.err
}
func TestGitVolume(t *testing.T) {
var fcmd exec.FakeCmd
fcmd = exec.FakeCmd{
CombinedOutputScript: []exec.FakeCombinedOutputAction{
func() ([]byte, error) {
os.MkdirAll(path.Join(fcmd.Dirs[0], "kubernetes"), 0750)
return []byte{}, nil
},
func() ([]byte, error) { return []byte{}, nil },
func() ([]byte, error) { return []byte{}, nil },
},
}
fake := exec.FakeExec{
CommandScript: []exec.FakeCommandAction{
func(cmd string, args ...string) exec.Cmd { return exec.InitFakeCmd(&fcmd, cmd, args...) },
func(cmd string, args ...string) exec.Cmd { return exec.InitFakeCmd(&fcmd, cmd, args...) },
func(cmd string, args ...string) exec.Cmd { return exec.InitFakeCmd(&fcmd, cmd, args...) },
},
}
dir := os.TempDir() + "/git"
g := GitDir{
Source: "https://github.com/GoogleCloudPlatform/kubernetes.git",
Revision: "2a30ce65c5ab586b98916d83385c5983edd353a1",
PodID: "foo",
RootDir: dir,
Name: "test-pod",
exec: &fake,
}
err := g.SetUp()
if err != nil {
t.Errorf("unexpected error: %v", err)
}
expectedCmds := [][]string{
{"git", "clone", g.Source},
{"git", "checkout", g.Revision},
{"git", "reset", "--hard"},
}
if fake.CommandCalls != len(expectedCmds) {
t.Errorf("unexpected command calls: expected 3, saw: %d", fake.CommandCalls)
}
if !reflect.DeepEqual(expectedCmds, fcmd.CombinedOutputLog) {
t.Errorf("unexpected commands: %v, expected: %v", fcmd.CombinedOutputLog, expectedCmds)
}
expectedDirs := []string{g.GetPath(), g.GetPath() + "/kubernetes", g.GetPath() + "/kubernetes"}
if len(fcmd.Dirs) != 3 || !reflect.DeepEqual(expectedDirs, fcmd.Dirs) {
t.Errorf("unexpected directories: %v, expected: %v", fcmd.Dirs, expectedDirs)
}
err = g.TearDown()
if err != nil {
t.Errorf("unexpected error: %v", err)
}
}