Add mount containers to kubelet

pkg/kubelet/BUILD

@@ -66,6 +66,7 @@ go_library(
         "//pkg/kubelet/kuberuntime:go_default_library",
         "//pkg/kubelet/lifecycle:go_default_library",
         "//pkg/kubelet/metrics:go_default_library",
+        "//pkg/kubelet/mountpod:go_default_library",
         "//pkg/kubelet/network:go_default_library",
         "//pkg/kubelet/pleg:go_default_library",
         "//pkg/kubelet/pod:go_default_library",
@@ -272,6 +273,7 @@ filegroup(
         "//pkg/kubelet/leaky:all-srcs",
         "//pkg/kubelet/lifecycle:all-srcs",
         "//pkg/kubelet/metrics:all-srcs",
+        "//pkg/kubelet/mountpod:all-srcs",
         "//pkg/kubelet/network:all-srcs",
         "//pkg/kubelet/pleg:all-srcs",
         "//pkg/kubelet/pod:all-srcs",

pkg/kubelet/config/defaults.go

@@ -17,8 +17,9 @@ limitations under the License.
 package config
 
 const (
-	DefaultKubeletPodsDirName       = "pods"
-	DefaultKubeletVolumesDirName    = "volumes"
-	DefaultKubeletPluginsDirName    = "plugins"
-	DefaultKubeletContainersDirName = "containers"
+	DefaultKubeletPodsDirName             = "pods"
+	DefaultKubeletVolumesDirName          = "volumes"
+	DefaultKubeletPluginsDirName          = "plugins"
+	DefaultKubeletContainersDirName       = "containers"
+	DefaultKubeletPluginContainersDirName = "plugin-containers"
 )

pkg/kubelet/mountpod/BUILD

+load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["mount_pod.go"],
+    importpath = "k8s.io/kubernetes/pkg/kubelet/mountpod",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//pkg/kubelet/config:go_default_library",
+        "//pkg/kubelet/pod:go_default_library",
+        "//pkg/util/strings:go_default_library",
+        "//vendor/k8s.io/api/core/v1:go_default_library",
+    ],
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = ["mount_pod_test.go"],
+    importpath = "k8s.io/kubernetes/pkg/kubelet/mountpod",
+    library = ":go_default_library",
+    deps = [
+        "//pkg/kubelet/configmap:go_default_library",
+        "//pkg/kubelet/pod:go_default_library",
+        "//pkg/kubelet/pod/testing:go_default_library",
+        "//pkg/kubelet/secret:go_default_library",
+        "//vendor/github.com/golang/glog:go_default_library",
+        "//vendor/k8s.io/api/core/v1:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
+        "//vendor/k8s.io/client-go/util/testing:go_default_library",
+    ],
+)
+
+filegroup(
+    name = "package-srcs",
+    srcs = glob(["**"]),
+    tags = ["automanaged"],
+    visibility = ["//visibility:private"],
+)
+
+filegroup(
+    name = "all-srcs",
+    srcs = [":package-srcs"],
+    tags = ["automanaged"],
+    visibility = ["//visibility:public"],
+)

pkg/kubelet/mountpod/mount_pod.go

+/*
+Copyright 2017 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package mountpod
+
+import (
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"os"
+	"path"
+
+	"k8s.io/api/core/v1"
+	"k8s.io/kubernetes/pkg/kubelet/config"
+	kubepod "k8s.io/kubernetes/pkg/kubelet/pod"
+	"k8s.io/kubernetes/pkg/util/strings"
+)
+
+// Manager is an interface that tracks pods with mount utilities for individual
+// volume plugins.
+type Manager interface {
+	GetMountPod(pluginName string) (pod *v1.Pod, container string, err error)
+}
+
+// basicManager is simple implementation of Manager. Pods with mount utilities
+// are registered by placing a JSON file into
+// /var/lib/kubelet/plugin-containers/<plugin name>.json and this manager just
+// finds them there.
+type basicManager struct {
+	registrationDirectory string
+	podManager            kubepod.Manager
+}
+
+// volumePluginRegistration specified format of the json files placed in
+// /var/lib/kubelet/plugin-containers/
+type volumePluginRegistration struct {
+	PodName       string `json:"podName"`
+	PodNamespace  string `json:"podNamespace"`
+	PodUID        string `json:"podUID"`
+	ContainerName string `json:"containerName"`
+}
+
+// NewManager returns a new mount pod manager.
+func NewManager(rootDirectory string, podManager kubepod.Manager) (Manager, error) {
+	regPath := path.Join(rootDirectory, config.DefaultKubeletPluginContainersDirName)
+
+	// Create the directory on startup
+	os.MkdirAll(regPath, 0700)
+
+	return &basicManager{
+		registrationDirectory: regPath,
+		podManager:            podManager,
+	}, nil
+}
+
+func (m *basicManager) getVolumePluginRegistrationPath(pluginName string) string {
+	// sanitize plugin name so it does not escape directory
+	safePluginName := strings.EscapePluginName(pluginName) + ".json"
+	return path.Join(m.registrationDirectory, safePluginName)
+}
+
+func (m *basicManager) GetMountPod(pluginName string) (pod *v1.Pod, containerName string, err error) {
+	// Read /var/lib/kubelet/plugin-containers/<plugin name>.json
+	regPath := m.getVolumePluginRegistrationPath(pluginName)
+	regBytes, err := ioutil.ReadFile(regPath)
+	if err != nil {
+		if os.IsNotExist(err) {
+			// No pod is registered for this plugin
+			return nil, "", nil
+		}
+		return nil, "", fmt.Errorf("cannot read %s: %v", regPath, err)
+	}
+
+	// Parse json
+	var reg volumePluginRegistration
+	if err := json.Unmarshal(regBytes, &reg); err != nil {
+		return nil, "", fmt.Errorf("unable to parse %s: %s", regPath, err)
+	}
+	if len(reg.ContainerName) == 0 {
+		return nil, "", fmt.Errorf("unable to parse %s: \"containerName\" is not set", regPath)
+	}
+	if len(reg.PodUID) == 0 {
+		return nil, "", fmt.Errorf("unable to parse %s: \"podUID\" is not set", regPath)
+	}
+	if len(reg.PodNamespace) == 0 {
+		return nil, "", fmt.Errorf("unable to parse %s: \"podNamespace\" is not set", regPath)
+	}
+	if len(reg.PodName) == 0 {
+		return nil, "", fmt.Errorf("unable to parse %s: \"podName\" is not set", regPath)
+	}
+
+	pod, ok := m.podManager.GetPodByName(reg.PodNamespace, reg.PodName)
+	if !ok {
+		return nil, "", fmt.Errorf("unable to process %s: pod %s/%s not found", regPath, reg.PodNamespace, reg.PodName)
+	}
+	if string(pod.UID) != reg.PodUID {
+		return nil, "", fmt.Errorf("unable to process %s: pod %s/%s has unexpected UID", regPath, reg.PodNamespace, reg.PodName)
+	}
+	// make sure that reg.ContainerName exists in the pod
+	for i := range pod.Spec.Containers {
+		if pod.Spec.Containers[i].Name == reg.ContainerName {
+			return pod, reg.ContainerName, nil
+		}
+	}
+	return nil, "", fmt.Errorf("unable to process %s: pod %s/%s has no container named %q", regPath, reg.PodNamespace, reg.PodName, reg.ContainerName)
+
+}

pkg/kubelet/mountpod/mount_pod_test.go

+/*
+Copyright 2017 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package mountpod
+
+import (
+	"io/ioutil"
+	"os"
+	"path"
+	"testing"
+
+	"github.com/golang/glog"
+
+	"k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	utiltesting "k8s.io/client-go/util/testing"
+	"k8s.io/kubernetes/pkg/kubelet/configmap"
+	kubepod "k8s.io/kubernetes/pkg/kubelet/pod"
+	podtest "k8s.io/kubernetes/pkg/kubelet/pod/testing"
+	"k8s.io/kubernetes/pkg/kubelet/secret"
+)
+
+func TestGetVolumeExec(t *testing.T) {
+	// prepare PodManager
+	pods := []*v1.Pod{
+		{
+			ObjectMeta: metav1.ObjectMeta{
+				UID:       "12345678",
+				Name:      "foo",
+				Namespace: "bar",
+			},
+			Spec: v1.PodSpec{
+				Containers: []v1.Container{
+					{Name: "baz"},
+				},
+			},
+		},
+	}
+	fakeSecretManager := secret.NewFakeManager()
+	fakeConfigMapManager := configmap.NewFakeManager()
+	podManager := kubepod.NewBasicPodManager(
+		podtest.NewFakeMirrorClient(), fakeSecretManager, fakeConfigMapManager)
+	podManager.SetPods(pods)
+
+	// Prepare fake /var/lib/kubelet
+	basePath, err := utiltesting.MkTmpdir("kubelet")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer os.RemoveAll(basePath)
+	regPath := path.Join(basePath, "plugin-containers")
+
+	mgr, err := NewManager(basePath, podManager)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	tests := []struct {
+		name        string
+		json        string
+		expectError bool
+	}{
+		{
+			"invalid json",
+			"{{{}",
+			true,
+		},
+		{
+			"missing json",
+			"", // this means no json file should be created
+			false,
+		},
+		{
+			"missing podNamespace",
+			`{"podName": "foo", "podUID": "87654321", "containerName": "baz"}`,
+			true,
+		},
+		{
+			"missing podName",
+			`{"podNamespace": "bar", "podUID": "87654321", "containerName": "baz"}`,
+			true,
+		},
+		{
+			"missing containerName",
+			`{"podNamespace": "bar", "podName": "foo", "podUID": "87654321"}`,
+			true,
+		},
+		{
+			"missing podUID",
+			`{"podNamespace": "bar", "podName": "foo", "containerName": "baz"}`,
+			true,
+		},
+		{
+			"missing pod",
+			`{"podNamespace": "bar", "podName": "non-existing-pod", "podUID": "12345678", "containerName": "baz"}`,
+			true,
+		},
+		{
+			"invalid uid",
+			`{"podNamespace": "bar", "podName": "foo", "podUID": "87654321", "containerName": "baz"}`,
+			true,
+		},
+		{
+			"invalid container",
+			`{"podNamespace": "bar", "podName": "foo", "podUID": "12345678", "containerName": "invalid"}`,
+			true,
+		},
+		{
+			"valid pod",
+			`{"podNamespace": "bar", "podName": "foo", "podUID": "12345678", "containerName": "baz"}`,
+			false,
+		},
+	}
+	for _, test := range tests {
+		p := path.Join(regPath, "kubernetes.io~glusterfs.json")
+		if len(test.json) > 0 {
+			if err := ioutil.WriteFile(p, []byte(test.json), 0600); err != nil {
+				t.Errorf("test %q: error writing %s: %v", test.name, p, err)
+				continue
+			}
+		} else {
+			// "" means no JSON file
+			os.Remove(p)
+		}
+		pod, container, err := mgr.GetMountPod("kubernetes.io/glusterfs")
+		if err != nil {
+			glog.V(5).Infof("test %q returned error %s", test.name, err)
+		}
+		if err == nil && test.expectError {
+			t.Errorf("test %q: expected error, got none", test.name)
+		}
+		if err != nil && !test.expectError {
+			t.Errorf("test %q: unexpected error: %v", test.name, err)
+		}
+
+		if err == nil {
+			// Pod must be returned when the json file was not empty
+			if pod == nil && len(test.json) != 0 {
+				t.Errorf("test %q: expected exec, got nil", test.name)
+			}
+			// Both pod and container must be returned
+			if pod != nil && len(container) == 0 {
+				t.Errorf("test %q: expected container name, got %q", test.name, container)
+			}
+		}
+	}
+}

pkg/kubelet/volume_host.go

@@ -20,11 +20,17 @@ import (
 	"fmt"
 	"net"
 
+	"github.com/golang/glog"
+
 	"k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/types"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	clientset "k8s.io/client-go/kubernetes"
 	"k8s.io/kubernetes/pkg/cloudprovider"
+	"k8s.io/kubernetes/pkg/features"
 	"k8s.io/kubernetes/pkg/kubelet/configmap"
+	"k8s.io/kubernetes/pkg/kubelet/container"
+	"k8s.io/kubernetes/pkg/kubelet/mountpod"
 	"k8s.io/kubernetes/pkg/kubelet/secret"
 	"k8s.io/kubernetes/pkg/util/io"
 	"k8s.io/kubernetes/pkg/util/mount"
@@ -43,11 +49,17 @@ func NewInitializedVolumePluginMgr(
 	configMapManager configmap.Manager,
 	plugins []volume.VolumePlugin,
 	prober volume.DynamicPluginProber) (*volume.VolumePluginMgr, error) {
+
+	mountPodManager, err := mountpod.NewManager(kubelet.getRootDir(), kubelet.podManager)
+	if err != nil {
+		return nil, err
+	}
 	kvh := &kubeletVolumeHost{
 		kubelet:          kubelet,
 		volumePluginMgr:  volume.VolumePluginMgr{},
 		secretManager:    secretManager,
 		configMapManager: configMapManager,
+		mountPodManager:  mountPodManager,
 	}
 
 	if err := kvh.volumePluginMgr.InitPlugins(plugins, prober, kvh); err != nil {
@@ -71,6 +83,7 @@ type kubeletVolumeHost struct {
 	volumePluginMgr  volume.VolumePluginMgr
 	secretManager    secret.Manager
 	configMapManager configmap.Manager
+	mountPodManager  mountpod.Manager
 }
 
 func (kvh *kubeletVolumeHost) GetPodVolumeDir(podUID types.UID, pluginName string, volumeName string) string {
@@ -119,7 +132,16 @@ func (kvh *kubeletVolumeHost) GetCloudProvider() cloudprovider.Interface {
 }
 
 func (kvh *kubeletVolumeHost) GetMounter(pluginName string) mount.Interface {
-	return kvh.kubelet.mounter
+	exec, err := kvh.getMountExec(pluginName)
+	if err != nil {
+		glog.V(2).Info("Error finding mount pod for plugin %s: %s", pluginName, err.Error())
+		// Use the default mounter
+		exec = nil
+	}
+	if exec == nil {
+		return kvh.kubelet.mounter
+	}
+	return mount.NewExecMounter(exec, kvh.kubelet.mounter)
 }
 
 func (kvh *kubeletVolumeHost) GetWriter() io.Writer {
@@ -159,5 +181,56 @@ func (kvh *kubeletVolumeHost) GetNodeLabels() (map[string]string, error) {
 }
 
 func (kvh *kubeletVolumeHost) GetExec(pluginName string) mount.Exec {
-	return mount.NewOsExec()
+	exec, err := kvh.getMountExec(pluginName)
+	if err != nil {
+		glog.V(2).Info("Error finding mount pod for plugin %s: %s", pluginName, err.Error())
+		// Use the default exec
+		exec = nil
+	}
+	if exec == nil {
+		return mount.NewOsExec()
+	}
+	return exec
+}
+
+// getMountExec returns mount.Exec implementation that leads to pod with mount
+// utilities. It returns nil,nil when there is no such pod and default mounter /
+// os.Exec should be used.
+func (kvh *kubeletVolumeHost) getMountExec(pluginName string) (mount.Exec, error) {
+	if !utilfeature.DefaultFeatureGate.Enabled(features.MountContainers) {
+		glog.V(5).Infof("using default mounter/exec for %s", pluginName)
+		return nil, nil
+	}
+
+	pod, container, err := kvh.mountPodManager.GetMountPod(pluginName)
+	if err != nil {
+		return nil, err
+	}
+	if pod == nil {
+		// Use default mounter/exec for this plugin
+		glog.V(5).Infof("using default mounter/exec for %s", pluginName)
+		return nil, nil
+	}
+	glog.V(5).Infof("using container %s/%s/%s to execute mount utilites for %s", pod.Namespace, pod.Name, container, pluginName)
+	return &containerExec{
+		pod:           pod,
+		containerName: container,
+		kl:            kvh.kubelet,
+	}, nil
+}
+
+// containerExec is implementation of mount.Exec that executes commands in given
+// container in given pod.
+type containerExec struct {
+	pod           *v1.Pod
+	containerName string
+	kl            *Kubelet
+}
+
+var _ mount.Exec = &containerExec{}
+
+func (e *containerExec) Run(cmd string, args ...string) ([]byte, error) {
+	cmdline := append([]string{cmd}, args...)
+	glog.V(5).Infof("Exec mounter running in pod %s/%s/%s: %v", e.pod.Namespace, e.pod.Name, e.containerName, cmdline)
+	return e.kl.RunInContainer(container.GetPodFullName(e.pod), e.pod.UID, e.containerName, cmdline)
 }