Implement fixes for flexvolume when kubelet is contanerized

Fix bug with nsenter root path

cluster/get-kube-local.sh

@@ -91,6 +91,7 @@ function create_cluster {
   --volume=/sys:/sys:ro \
   --volume=/var/lib/docker/:/var/lib/docker:rw \
   --volume=/var/lib/kubelet/:/var/lib/kubelet:rw \
+  --volume=/usr/libexec/kubernetes/kubelet-plugins/volume/exec:/usr/libexec/kubernetes/kubelet-plugins/volume/exec:rw \
   --volume=/var/run:/var/run:rw \
   --volume=/run/xtables.lock:/run/xtables.lock:rw \
   --net=host \

cmd/kube-controller-manager/app/BUILD

@@ -127,6 +127,7 @@ go_library(
         "//staging/src/k8s.io/metrics/pkg/client/external_metrics:go_default_library",
         "//vendor/github.com/golang/glog:go_default_library",
         "//vendor/github.com/spf13/cobra:go_default_library",
+        "//vendor/k8s.io/utils/exec:go_default_library",
     ],
 )
 

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

@@ -26,6 +26,7 @@ import (
 	// Cloud providers
 	"k8s.io/kubernetes/pkg/apis/componentconfig"
 	_ "k8s.io/kubernetes/pkg/cloudprovider/providers"
+	"k8s.io/utils/exec"
 
 	// Volume plugins
 	"github.com/golang/glog"
@@ -87,7 +88,7 @@ func ProbeAttachableVolumePlugins() []volume.VolumePlugin {
 // for the attach/detach controller.
 // Currently only Flexvolume plugins are dynamically discoverable.
 func GetDynamicPluginProber(config componentconfig.VolumeConfiguration) volume.DynamicPluginProber {
-	return flexvolume.GetDynamicPluginProber(config.FlexVolumePluginDir)
+	return flexvolume.GetDynamicPluginProber(config.FlexVolumePluginDir, exec.New() /*exec.Interface*/)
 }
 
 // ProbeExpandableVolumePlugins returns volume plugins which are expandable

cmd/kubelet/app/plugins.go

@@ -23,6 +23,7 @@ import (
 	_ "k8s.io/kubernetes/pkg/credentialprovider/azure"
 	_ "k8s.io/kubernetes/pkg/credentialprovider/gcp"
 	_ "k8s.io/kubernetes/pkg/credentialprovider/rancher"
+	"k8s.io/utils/exec"
 	// Volume plugins
 	"k8s.io/kubernetes/pkg/volume"
 	"k8s.io/kubernetes/pkg/volume/aws_ebs"
@@ -105,6 +106,6 @@ func ProbeVolumePlugins() []volume.VolumePlugin {
 // GetDynamicPluginProber gets the probers of dynamically discoverable plugins
 // for kubelet.
 // Currently only Flexvolume plugins are dynamically discoverable.
-func GetDynamicPluginProber(pluginDir string) volume.DynamicPluginProber {
+func GetDynamicPluginProber(pluginDir string, runner exec.Interface) volume.DynamicPluginProber {
-	return flexvolume.GetDynamicPluginProber(pluginDir)
+	return flexvolume.GetDynamicPluginProber(pluginDir, runner)
 }

cmd/kubelet/app/server.go

@@ -360,6 +360,7 @@ func UnsecuredDependencies(s *options.KubeletServer) (*kubelet.Dependencies, err
 
 	mounter := mount.New(s.ExperimentalMounterPath)
 	var writer kubeio.Writer = &kubeio.StdWriter{}
+	var pluginRunner = exec.New()
 	if s.Containerized {
 		glog.V(2).Info("Running kubelet in containerized mode")
 		ne, err := nsenter.NewNsenter(nsenter.DefaultHostRootFsPath, exec.New())
@@ -368,6 +369,8 @@ func UnsecuredDependencies(s *options.KubeletServer) (*kubelet.Dependencies, err
 		}
 		mounter = mount.NewNsenterMounter(s.RootDirectory, ne)
 		writer = kubeio.NewNsenterWriter(ne)
+		// an exec interface which can use nsenter for flex plugin calls
+		pluginRunner = nsenter.NewNsenterExecutor(nsenter.DefaultHostRootFsPath, exec.New())
 	}
 
 	var dockerClientConfig *dockershim.ClientConfig
@@ -394,7 +397,7 @@ func UnsecuredDependencies(s *options.KubeletServer) (*kubelet.Dependencies, err
 		OSInterface:         kubecontainer.RealOS{},
 		Writer:              writer,
 		VolumePlugins:       ProbeVolumePlugins(),
-		DynamicPluginProber: GetDynamicPluginProber(s.VolumePluginDir),
+		DynamicPluginProber: GetDynamicPluginProber(s.VolumePluginDir, pluginRunner),
 		TLSOptions:          tlsOptions}, nil
 }
 

hack/local-up-cluster.sh

@@ -826,6 +826,7 @@ function start_kubelet {
         --volume=/:/rootfs:ro,rslave \
         --volume=/var/run:/var/run:rw \
         --volume=/sys:/sys:ro \
+        --volume=/usr/libexec/kubernetes/kubelet-plugins/volume/exec:/usr/libexec/kubernetes/kubelet-plugins/volume/exec:rw \
         --volume=/var/lib/docker/:/var/lib/docker:rslave \
         --volume=/var/lib/kubelet/:/var/lib/kubelet:rslave \
         --volume=/dev:/dev \

pkg/util/nsenter/BUILD

@@ -3,6 +3,8 @@ load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
 go_library(
     name = "go_default_library",
     srcs = [
+        "exec.go",
+        "exec_unsupported.go",
         "nsenter.go",
         "nsenter_unsupported.go",
     ],

pkg/util/nsenter/exec.go

@@ -0,0 +1,67 @@
+// +build linux
+
+/*
+Copyright 2018 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 nsenter
+
+import (
+	"context"
+	"fmt"
+	"path/filepath"
+
+	"github.com/golang/glog"
+	"k8s.io/utils/exec"
+)
+
+// Executor wraps executor interface to be executed via nsenter
+type Executor struct {
+	// Exec implementation
+	executor exec.Interface
+	// Path to the host's root proc path
+	hostProcMountNsPath string
+}
+
+// NewNsenterExecutor returns new nsenter based executor
+func NewNsenterExecutor(hostRootFsPath string, executor exec.Interface) *Executor {
+	hostProcMountNsPath := filepath.Join(hostRootFsPath, mountNsPath)
+	nsExecutor := &Executor{
+		hostProcMountNsPath: hostProcMountNsPath,
+		executor:            executor,
+	}
+	return nsExecutor
+}
+
+// Command returns a command wrapped with nenter
+func (nsExecutor *Executor) Command(cmd string, args ...string) exec.Cmd {
+	fullArgs := append([]string{fmt.Sprintf("--mount=%s", nsExecutor.hostProcMountNsPath), "--"},
+		append([]string{cmd}, args...)...)
+	glog.V(5).Infof("Running nsenter command: %v %v", nsenterPath, fullArgs)
+	return nsExecutor.executor.Command(nsenterPath, fullArgs...)
+}
+
+// CommandContext returns a CommandContext wrapped with nsenter
+func (nsExecutor *Executor) CommandContext(ctx context.Context, cmd string, args ...string) exec.Cmd {
+	fullArgs := append([]string{fmt.Sprintf("--mount=%s", nsExecutor.hostProcMountNsPath), "--"},
+		append([]string{cmd}, args...)...)
+	glog.V(5).Infof("Running nsenter command: %v %v", nsenterPath, fullArgs)
+	return nsExecutor.executor.CommandContext(ctx, nsenterPath, fullArgs...)
+}
+
+// LookPath returns a LookPath wrapped with nsenter
+func (nsExecutor *Executor) LookPath(file string) (string, error) {
+	return "", fmt.Errorf("not implemented, error looking up : %s", file)
+}

pkg/util/nsenter/exec_unsupported.go

@@ -0,0 +1,58 @@
+// +build !linux
+
+/*
+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 nsenter
+
+import (
+	"context"
+	"fmt"
+
+	"k8s.io/utils/exec"
+)
+
+// Executor wraps executor interface to be executed via nsenter
+type Executor struct {
+	// Exec implementation
+	executor exec.Interface
+	// Path to the host's root proc path
+	hostProcMountNsPath string
+}
+
+// NewNsenterExecutor returns new nsenter based executor
+func NewNsenterExecutor(hostRootFsPath string, executor exec.Interface) *Executor {
+	nsExecutor := &Executor{
+		hostProcMountNsPath: hostRootFsPath,
+		executor:            executor,
+	}
+	return nsExecutor
+}
+
+// Command returns a command wrapped with nenter
+func (nsExecutor *Executor) Command(cmd string, args ...string) exec.Cmd {
+	return nil
+}
+
+// CommandContext returns a CommandContext wrapped with nsenter
+func (nsExecutor *Executor) CommandContext(ctx context.Context, cmd string, args ...string) exec.Cmd {
+	return nil
+}
+
+// LookPath returns a LookPath wrapped with nsenter
+func (nsExecutor *Executor) LookPath(file string) (string, error) {
+	return "", fmt.Errorf("not implemented, error looking up : %s", file)
+}

pkg/volume/flexvolume/flexvolume_test.go

@@ -28,6 +28,7 @@ import (
 	utiltesting "k8s.io/client-go/util/testing"
 	"k8s.io/kubernetes/pkg/volume"
 	volumetest "k8s.io/kubernetes/pkg/volume/testing"
+	"k8s.io/utils/exec"
 )
 
 const execScriptTempl1 = `#!/usr/bin/env bash
@@ -173,8 +174,9 @@ func TestCanSupport(t *testing.T) {
 	defer os.RemoveAll(tmpDir)
 
 	plugMgr := volume.VolumePluginMgr{}
+	runner := exec.New()
 	installPluginUnderTest(t, "kubernetes.io", "fakeAttacher", tmpDir, execScriptTempl1, nil)
-	plugMgr.InitPlugins(nil, GetDynamicPluginProber(tmpDir), volumetest.NewFakeVolumeHost("fake", nil, nil))
+	plugMgr.InitPlugins(nil, GetDynamicPluginProber(tmpDir, runner), volumetest.NewFakeVolumeHost("fake", nil, nil))
 	plugin, err := plugMgr.FindPluginByName("flexvolume-kubernetes.io/fakeAttacher")
 	if err != nil {
 		t.Fatalf("Can't find the plugin by name")
@@ -201,8 +203,9 @@ func TestGetAccessModes(t *testing.T) {
 	defer os.RemoveAll(tmpDir)
 
 	plugMgr := volume.VolumePluginMgr{}
+	runner := exec.New()
 	installPluginUnderTest(t, "kubernetes.io", "fakeAttacher", tmpDir, execScriptTempl1, nil)
-	plugMgr.InitPlugins(nil, GetDynamicPluginProber(tmpDir), volumetest.NewFakeVolumeHost(tmpDir, nil, nil))
+	plugMgr.InitPlugins(nil, GetDynamicPluginProber(tmpDir, runner), volumetest.NewFakeVolumeHost(tmpDir, nil, nil))
 
 	plugin, err := plugMgr.FindPersistentPluginByName("flexvolume-kubernetes.io/fakeAttacher")
 	if err != nil {

pkg/volume/flexvolume/plugin.go

@@ -59,12 +59,12 @@ var _ volume.AttachableVolumePlugin = &flexVolumeAttachablePlugin{}
 var _ volume.PersistentVolumePlugin = &flexVolumePlugin{}
 
 type PluginFactory interface {
-	NewFlexVolumePlugin(pluginDir, driverName string) (volume.VolumePlugin, error)
+	NewFlexVolumePlugin(pluginDir, driverName string, runner exec.Interface) (volume.VolumePlugin, error)
 }
 
 type pluginFactory struct{}
 
-func (pluginFactory) NewFlexVolumePlugin(pluginDir, name string) (volume.VolumePlugin, error) {
+func (pluginFactory) NewFlexVolumePlugin(pluginDir, name string, runner exec.Interface) (volume.VolumePlugin, error) {
 	execPath := path.Join(pluginDir, name)
 
 	driverName := utilstrings.UnescapePluginName(name)
@@ -72,7 +72,7 @@ func (pluginFactory) NewFlexVolumePlugin(pluginDir, name string) (volume.VolumeP
 	flexPlugin := &flexVolumePlugin{
 		driverName:          driverName,
 		execPath:            execPath,
-		runner:              exec.New(),
+		runner:              runner,
 		unsupportedCommands: []string{},
 	}
 

pkg/volume/flexvolume/probe.go

@@ -19,6 +19,7 @@ package flexvolume
 import (
 	"github.com/golang/glog"
 	"k8s.io/kubernetes/pkg/volume"
+	"k8s.io/utils/exec"
 
 	"os"
 
@@ -26,16 +27,18 @@ import (
 	"path/filepath"
 	"sync"
 
+	"strings"
+
 	"github.com/fsnotify/fsnotify"
 	"k8s.io/apimachinery/pkg/util/errors"
 	utilfs "k8s.io/kubernetes/pkg/util/filesystem"
 	utilstrings "k8s.io/kubernetes/pkg/util/strings"
-	"strings"
 )
 
 type flexVolumeProber struct {
 	mutex          sync.Mutex
-	pluginDir      string // Flexvolume driver directory
+	pluginDir      string         // Flexvolume driver directory
+	runner         exec.Interface // Interface to use for execing flex calls
 	watcher        utilfs.FSWatcher
 	factory        PluginFactory
 	fs             utilfs.Filesystem
@@ -43,11 +46,12 @@ type flexVolumeProber struct {
 	eventsMap      map[string]volume.ProbeOperation // the key is the driver directory path, the value is the coresponding operation
 }
 
-func GetDynamicPluginProber(pluginDir string) volume.DynamicPluginProber {
+func GetDynamicPluginProber(pluginDir string, runner exec.Interface) volume.DynamicPluginProber {
 	return &flexVolumeProber{
 		pluginDir: pluginDir,
 		watcher:   utilfs.NewFsnotifyWatcher(),
 		factory:   pluginFactory{},
+		runner:    runner,
 		fs:        &utilfs.DefaultFs{},
 	}
 }
@@ -127,7 +131,7 @@ func (prober *flexVolumeProber) newProbeEvent(driverDirName string, op volume.Pr
 		Op: op,
 	}
 	if op == volume.ProbeAddOrUpdate {
-		plugin, pluginErr := prober.factory.NewFlexVolumePlugin(prober.pluginDir, driverDirName)
+		plugin, pluginErr := prober.factory.NewFlexVolumePlugin(prober.pluginDir, driverDirName, prober.runner)
 		if pluginErr != nil {
 			pluginErr = fmt.Errorf(
 				"Error creating Flexvolume plugin from directory %s, skipping. Error: %s",

pkg/volume/flexvolume/probe_test.go

@@ -25,6 +25,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	utilfs "k8s.io/kubernetes/pkg/util/filesystem"
 	"k8s.io/kubernetes/pkg/volume"
+	"k8s.io/utils/exec"
 )
 
 const (
@@ -318,7 +319,7 @@ type fakePluginFactory struct {
 
 var _ PluginFactory = fakePluginFactory{}
 
-func (m fakePluginFactory) NewFlexVolumePlugin(_, driverName string) (volume.VolumePlugin, error) {
+func (m fakePluginFactory) NewFlexVolumePlugin(_, driverName string, _ exec.Interface) (volume.VolumePlugin, error) {
 	if m.error {
 		return nil, fmt.Errorf("Flexvolume plugin error")
 	}