Flexvolume dynamic plugin discovery: Prober unit tests and basic e2e test.

pkg/kubelet/kubeletconfig/BUILD

@@ -21,9 +21,9 @@ go_library(
         "//pkg/kubelet/kubeletconfig/configfiles:go_default_library",
         "//pkg/kubelet/kubeletconfig/status:go_default_library",
         "//pkg/kubelet/kubeletconfig/util/equal:go_default_library",
-        "//pkg/kubelet/kubeletconfig/util/filesystem:go_default_library",
         "//pkg/kubelet/kubeletconfig/util/log:go_default_library",
         "//pkg/kubelet/kubeletconfig/util/panic:go_default_library",
+        "//pkg/util/filesystem: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/apimachinery/pkg/fields:go_default_library",
@@ -52,7 +52,6 @@ filegroup(
         "//pkg/kubelet/kubeletconfig/util/codec:all-srcs",
         "//pkg/kubelet/kubeletconfig/util/equal:all-srcs",
         "//pkg/kubelet/kubeletconfig/util/files:all-srcs",
-        "//pkg/kubelet/kubeletconfig/util/filesystem:all-srcs",
         "//pkg/kubelet/kubeletconfig/util/log:all-srcs",
         "//pkg/kubelet/kubeletconfig/util/panic:all-srcs",
         "//pkg/kubelet/kubeletconfig/util/test:all-srcs",

pkg/kubelet/kubeletconfig/checkpoint/store/BUILD

@@ -16,8 +16,8 @@ go_test(
     deps = [
         "//pkg/kubelet/kubeletconfig/checkpoint:go_default_library",
         "//pkg/kubelet/kubeletconfig/util/files:go_default_library",
-        "//pkg/kubelet/kubeletconfig/util/filesystem:go_default_library",
         "//pkg/kubelet/kubeletconfig/util/test:go_default_library",
+        "//pkg/util/filesystem:go_default_library",
         "//vendor/github.com/davecgh/go-spew/spew:go_default_library",
         "//vendor/k8s.io/api/core/v1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
@@ -35,8 +35,8 @@ go_library(
     deps = [
         "//pkg/kubelet/kubeletconfig/checkpoint:go_default_library",
         "//pkg/kubelet/kubeletconfig/util/files:go_default_library",
-        "//pkg/kubelet/kubeletconfig/util/filesystem:go_default_library",
         "//pkg/kubelet/kubeletconfig/util/log:go_default_library",
+        "//pkg/util/filesystem:go_default_library",
     ],
 )
 

pkg/kubelet/kubeletconfig/checkpoint/store/fsstore.go

@@ -23,8 +23,8 @@ import (
 
 	"k8s.io/kubernetes/pkg/kubelet/kubeletconfig/checkpoint"
 	utilfiles "k8s.io/kubernetes/pkg/kubelet/kubeletconfig/util/files"
-	utilfs "k8s.io/kubernetes/pkg/kubelet/kubeletconfig/util/filesystem"
 	utillog "k8s.io/kubernetes/pkg/kubelet/kubeletconfig/util/log"
+	utilfs "k8s.io/kubernetes/pkg/util/filesystem"
 )
 
 const (

pkg/kubelet/kubeletconfig/checkpoint/store/fsstore_test.go

@@ -29,8 +29,8 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/kubernetes/pkg/kubelet/kubeletconfig/checkpoint"
 	utilfiles "k8s.io/kubernetes/pkg/kubelet/kubeletconfig/util/files"
-	utilfs "k8s.io/kubernetes/pkg/kubelet/kubeletconfig/util/filesystem"
 	utiltest "k8s.io/kubernetes/pkg/kubelet/kubeletconfig/util/test"
+	utilfs "k8s.io/kubernetes/pkg/util/filesystem"
 )
 
 const testCheckpointsDir = "/test-checkpoints-dir"

pkg/kubelet/kubeletconfig/configfiles/BUILD

@@ -13,7 +13,7 @@ go_library(
         "//pkg/kubelet/apis/kubeletconfig:go_default_library",
         "//pkg/kubelet/apis/kubeletconfig/scheme:go_default_library",
         "//pkg/kubelet/kubeletconfig/util/codec:go_default_library",
-        "//pkg/kubelet/kubeletconfig/util/filesystem:go_default_library",
+        "//pkg/util/filesystem:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/runtime/serializer:go_default_library",
     ],
 )
@@ -40,8 +40,8 @@ go_test(
         "//pkg/kubelet/apis/kubeletconfig/scheme:go_default_library",
         "//pkg/kubelet/apis/kubeletconfig/v1alpha1:go_default_library",
         "//pkg/kubelet/kubeletconfig/util/files:go_default_library",
-        "//pkg/kubelet/kubeletconfig/util/filesystem:go_default_library",
         "//pkg/kubelet/kubeletconfig/util/test:go_default_library",
+        "//pkg/util/filesystem:go_default_library",
         "//vendor/github.com/davecgh/go-spew/spew:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/api/equality:go_default_library",
     ],

pkg/kubelet/kubeletconfig/configfiles/configfiles.go

@@ -24,7 +24,7 @@ import (
 	"k8s.io/kubernetes/pkg/kubelet/apis/kubeletconfig"
 	kubeletscheme "k8s.io/kubernetes/pkg/kubelet/apis/kubeletconfig/scheme"
 	utilcodec "k8s.io/kubernetes/pkg/kubelet/kubeletconfig/util/codec"
-	utilfs "k8s.io/kubernetes/pkg/kubelet/kubeletconfig/util/filesystem"
+	utilfs "k8s.io/kubernetes/pkg/util/filesystem"
 )
 
 const kubeletFile = "kubelet"

pkg/kubelet/kubeletconfig/configfiles/configfiles_test.go

@@ -29,8 +29,8 @@ import (
 	kubeletscheme "k8s.io/kubernetes/pkg/kubelet/apis/kubeletconfig/scheme"
 	kubeletconfigv1alpha1 "k8s.io/kubernetes/pkg/kubelet/apis/kubeletconfig/v1alpha1"
 	utilfiles "k8s.io/kubernetes/pkg/kubelet/kubeletconfig/util/files"
-	utilfs "k8s.io/kubernetes/pkg/kubelet/kubeletconfig/util/filesystem"
 	utiltest "k8s.io/kubernetes/pkg/kubelet/kubeletconfig/util/test"
+	utilfs "k8s.io/kubernetes/pkg/util/filesystem"
 )
 
 func addFile(fs utilfs.Filesystem, path string, file string) error {

pkg/kubelet/kubeletconfig/controller.go

@@ -31,9 +31,9 @@ import (
 	"k8s.io/kubernetes/pkg/kubelet/kubeletconfig/checkpoint/store"
 	"k8s.io/kubernetes/pkg/kubelet/kubeletconfig/configfiles"
 	"k8s.io/kubernetes/pkg/kubelet/kubeletconfig/status"
-	utilfs "k8s.io/kubernetes/pkg/kubelet/kubeletconfig/util/filesystem"
 	utillog "k8s.io/kubernetes/pkg/kubelet/kubeletconfig/util/log"
 	utilpanic "k8s.io/kubernetes/pkg/kubelet/kubeletconfig/util/panic"
+	utilfs "k8s.io/kubernetes/pkg/util/filesystem"
 )
 
 const (

pkg/kubelet/kubeletconfig/util/files/BUILD

@@ -8,7 +8,7 @@ load(
 go_library(
     name = "go_default_library",
     srcs = ["files.go"],
-    deps = ["//pkg/kubelet/kubeletconfig/util/filesystem:go_default_library"],
+    deps = ["//pkg/util/filesystem:go_default_library"],
 )
 
 filegroup(

pkg/kubelet/kubeletconfig/util/files/files.go

@@ -21,7 +21,7 @@ import (
 	"os"
 	"path/filepath"
 
-	utilfs "k8s.io/kubernetes/pkg/kubelet/kubeletconfig/util/filesystem"
+	utilfs "k8s.io/kubernetes/pkg/util/filesystem"
 )
 
 const defaultPerm = 0666

pkg/util/BUILD

@@ -19,6 +19,7 @@ filegroup(
         "//pkg/util/ebtables:all-srcs",
         "//pkg/util/env:all-srcs",
         "//pkg/util/file:all-srcs",
+        "//pkg/util/filesystem:all-srcs",
         "//pkg/util/flock:all-srcs",
         "//pkg/util/goroutinemap:all-srcs",
         "//pkg/util/hash:all-srcs",

pkg/kubelet/kubeletconfig/util/filesystem/BUILD → pkg/util/filesystem/BUILD

@@ -11,8 +11,12 @@ go_library(
         "defaultfs.go",
         "fakefs.go",
         "filesystem.go",
+        "watcher.go",
+    ],
+    deps = [
+        "//vendor/github.com/fsnotify/fsnotify:go_default_library",
+        "//vendor/github.com/spf13/afero:go_default_library",
     ],
-    deps = ["//vendor/github.com/spf13/afero:go_default_library"],
 )
 
 filegroup(

pkg/kubelet/kubeletconfig/util/filesystem/defaultfs.go → pkg/util/filesystem/defaultfs.go

@@ -19,12 +19,15 @@ package filesystem
 import (
 	"io/ioutil"
 	"os"
+	"path/filepath"
 	"time"
 )
 
 // DefaultFs implements Filesystem using same-named functions from "os" and "io/ioutil"
 type DefaultFs struct{}
 
+var _ Filesystem = DefaultFs{}
+
 // Stat via os.Stat
 func (DefaultFs) Stat(name string) (os.FileInfo, error) {
 	return os.Stat(name)
@@ -54,12 +57,17 @@ func (DefaultFs) Chtimes(name string, atime time.Time, mtime time.Time) error {
 	return os.Chtimes(name, atime, mtime)
 }
 
-// ReadFile via os.ReadFile
+// RemoveAll via os.RemoveAll
+func (DefaultFs) RemoveAll(path string) error {
+	return os.RemoveAll(path)
+}
+
+// ReadFile via ioutil.ReadFile
 func (DefaultFs) ReadFile(filename string) ([]byte, error) {
 	return ioutil.ReadFile(filename)
 }
 
-// TempFile via os.TempFile
+// TempFile via ioutil.TempFile
 func (DefaultFs) TempFile(dir, prefix string) (File, error) {
 	file, err := ioutil.TempFile(dir, prefix)
 	if err != nil {
@@ -68,6 +76,16 @@ func (DefaultFs) TempFile(dir, prefix string) (File, error) {
 	return &defaultFile{file}, nil
 }
 
+// ReadDir via ioutil.ReadDir
+func (DefaultFs) ReadDir(dirname string) ([]os.FileInfo, error) {
+	return ioutil.ReadDir(dirname)
+}
+
+// Walk via filepath.Walk
+func (DefaultFs) Walk(root string, walkFn filepath.WalkFunc) error {
+	return filepath.Walk(root, walkFn)
+}
+
 // defaultFile implements File using same-named functions from "os"
 type defaultFile struct {
 	file *os.File

pkg/kubelet/kubeletconfig/util/filesystem/fakefs.go → pkg/util/filesystem/fakefs.go

@@ -18,6 +18,7 @@ package filesystem
 
 import (
 	"os"
+	"path/filepath"
 	"time"
 
 	"github.com/spf13/afero"
@@ -62,12 +63,12 @@ func (fs *fakeFs) Chtimes(name string, atime time.Time, mtime time.Time) error {
 	return fs.a.Fs.Chtimes(name, atime, mtime)
 }
 
-// ReadFile via afero.Fs.ReadFile
+// ReadFile via afero.ReadFile
 func (fs *fakeFs) ReadFile(filename string) ([]byte, error) {
 	return fs.a.ReadFile(filename)
 }
 
-// TempFile via afero.Fs.TempFile
+// TempFile via afero.TempFile
 func (fs *fakeFs) TempFile(dir, prefix string) (File, error) {
 	file, err := fs.a.TempFile(dir, prefix)
 	if err != nil {
@@ -76,6 +77,21 @@ func (fs *fakeFs) TempFile(dir, prefix string) (File, error) {
 	return &fakeFile{file}, nil
 }
 
+// ReadDir via afero.ReadDir
+func (fs *fakeFs) ReadDir(dirname string) ([]os.FileInfo, error) {
+	return fs.a.ReadDir(dirname)
+}
+
+// Walk via afero.Walk
+func (fs *fakeFs) Walk(root string, walkFn filepath.WalkFunc) error {
+	return fs.a.Walk(root, walkFn)
+}
+
+// RemoveAll via afero.RemoveAll
+func (fs *fakeFs) RemoveAll(path string) error {
+	return fs.a.RemoveAll(path)
+}
+
 // fakeFile implements File; for use with fakeFs
 type fakeFile struct {
 	file afero.File

pkg/kubelet/kubeletconfig/util/filesystem/filesystem.go → pkg/util/filesystem/filesystem.go

@@ -18,6 +18,7 @@ package filesystem
 
 import (
 	"os"
+	"path/filepath"
 	"time"
 )
 
@@ -29,10 +30,13 @@ type Filesystem interface {
 	Rename(oldpath, newpath string) error
 	MkdirAll(path string, perm os.FileMode) error
 	Chtimes(name string, atime time.Time, mtime time.Time) error
+	RemoveAll(path string) error
 
 	// from "io/ioutil"
 	ReadFile(filename string) ([]byte, error)
 	TempFile(dir, prefix string) (File, error)
+	ReadDir(dirname string) ([]os.FileInfo, error)
+	Walk(root string, walkFn filepath.WalkFunc) error
 }
 
 // File is an interface that we can use to mock various filesystem operations typically

pkg/util/filesystem/watcher.go

@@ -0,0 +1,89 @@
+/*
+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 filesystem
+
+import (
+	"github.com/fsnotify/fsnotify"
+)
+
+// FSWatcher is a callback-based filesystem watcher abstraction for fsnotify.
+type FSWatcher interface {
+	// Initializes the watcher with the given watch handlers.
+	// Called before all other methods.
+	Init(FSEventHandler, FSErrorHandler) error
+
+	// Starts listening for events and errors.
+	// When an event or error occurs, the corresponding handler is called.
+	Run()
+
+	// Add a filesystem path to watch
+	AddWatch(path string) error
+}
+
+// FSEventHandler is called when a fsnotify event occurs.
+type FSEventHandler func(event fsnotify.Event)
+
+// FSErrorHandler is called when a fsnotify error occurs.
+type FSErrorHandler func(err error)
+
+type fsnotifyWatcher struct {
+	watcher      *fsnotify.Watcher
+	eventHandler FSEventHandler
+	errorHandler FSErrorHandler
+}
+
+var _ FSWatcher = &fsnotifyWatcher{}
+
+// NewFsnotifyWatcher returns an implementation of FSWatcher that continuously listens for
+// fsnotify events and calls the event handler as soon as an event is received.
+func NewFsnotifyWatcher() FSWatcher {
+	return &fsnotifyWatcher{}
+}
+
+func (w *fsnotifyWatcher) AddWatch(path string) error {
+	return w.watcher.Add(path)
+}
+
+func (w *fsnotifyWatcher) Init(eventHandler FSEventHandler, errorHandler FSErrorHandler) error {
+	var err error
+	w.watcher, err = fsnotify.NewWatcher()
+	if err != nil {
+		return err
+	}
+
+	w.eventHandler = eventHandler
+	w.errorHandler = errorHandler
+	return nil
+}
+
+func (w *fsnotifyWatcher) Run() {
+	go func() {
+		defer w.watcher.Close()
+		for {
+			select {
+			case event := <-w.watcher.Events:
+				if w.eventHandler != nil {
+					w.eventHandler(event)
+				}
+			case err := <-w.watcher.Errors:
+				if w.errorHandler != nil {
+					w.errorHandler(err)
+				}
+			}
+		}
+	}()
+}

pkg/volume/flexvolume/BUILD

@@ -14,6 +14,7 @@ go_library(
         "detacher.go",
         "detacher-defaults.go",
         "driver-call.go",
+        "fake_watcher.go",
         "mounter.go",
         "mounter-defaults.go",
         "plugin.go",
@@ -25,6 +26,7 @@ go_library(
         "volume.go",
     ],
     deps = [
+        "//pkg/util/filesystem:go_default_library",
         "//pkg/util/mount:go_default_library",
         "//pkg/util/strings:go_default_library",
         "//pkg/volume:go_default_library",
@@ -48,13 +50,17 @@ go_test(
         "flexvolume_test.go",
         "mounter_test.go",
         "plugin_test.go",
+        "probe_test.go",
         "unmounter_test.go",
     ],
     library = ":go_default_library",
     deps = [
+        "//pkg/util/filesystem:go_default_library",
         "//pkg/util/mount:go_default_library",
         "//pkg/volume:go_default_library",
         "//pkg/volume/testing:go_default_library",
+        "//vendor/github.com/fsnotify/fsnotify:go_default_library",
+        "//vendor/github.com/stretchr/testify/assert: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/apimachinery/pkg/types:go_default_library",