Pluginwatcher backport sanity patches on 1.12

pkg/kubelet/util/pluginwatcher/plugin_watcher.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"net"
 	"os"
+	"path/filepath"
 	"strings"
 	"sync"
 	"time"
@@ -183,6 +184,10 @@ func (w *Watcher) traversePluginDir(dir string) error {
 
 		switch mode := info.Mode(); {
 		case mode.IsDir():
+			if w.containsBlacklistedDir(path) {
+				return filepath.SkipDir
+			}
+
 			if err := w.fsWatcher.Add(path); err != nil {
 				return fmt.Errorf("failed to watch %s, err: %v", path, err)
 			}
@@ -205,6 +210,10 @@ func (w *Watcher) traversePluginDir(dir string) error {
 func (w *Watcher) handleCreateEvent(event fsnotify.Event) error {
 	glog.V(6).Infof("Handling create event: %v", event)
 
+	if w.containsBlacklistedDir(event.Name) {
+		return nil
+	}
+
 	fi, err := os.Stat(event.Name)
 	if err != nil {
 		return fmt.Errorf("stat file %s failed: %v", event.Name, err)
@@ -400,3 +409,10 @@ func dial(unixSocketPath string, timeout time.Duration) (registerapi.Registratio
 
 	return registerapi.NewRegistrationClient(c), c, nil
 }
+
+// To prevent 1.12 from filling the fd limit we will explicitly blacklist
+// kubernetes.io directory.
+func (w *Watcher) containsBlacklistedDir(path string) bool {
+	return strings.HasPrefix(path, w.path+"/kubernetes.io/") ||
+		path == w.path+"/kubernetes.io"
+}

pkg/kubelet/util/pluginwatcher/plugin_watcher_test.go

@@ -236,3 +236,49 @@ func newWatcherWithHandler(t *testing.T, hdlr PluginHandler) *Watcher {
 
 	return w
 }
+
+func TestContainsBlacklistedDir(t *testing.T) {
+	testCases := []struct {
+		sockDir  string
+		path     string
+		expected bool
+	}{
+		{
+			sockDir:  "/var/lib/kubelet/plugins_registry",
+			path:     "/var/lib/kubelet/plugins_registry/mydriver.foo/csi.sock",
+			expected: false,
+		},
+		{
+			sockDir:  "/var/lib/kubelet/plugins_registry",
+			path:     "/var/lib/kubelet/plugins_registry/my.driver.com",
+			expected: false,
+		},
+		{
+			sockDir:  "/var/lib/kubelet/plugins_registry",
+			path:     "/var/lib/kubelet/plugins_registry/kubernetes.io/",
+			expected: true,
+		},
+		{
+			sockDir:  "/var/lib/kubelet/plugins_registry",
+			path:     "/var/lib/kubelet/plugins_registry/my-kubernetes.io-plugin",
+			expected: false,
+		},
+		{
+			sockDir:  "/var/lib/kubelet/plugins_registry",
+			path:     "/var/lib/kubelet/plugins_registry",
+			expected: false,
+		},
+	}
+
+	for _, tc := range testCases {
+		// Arrange & Act
+		watcher := NewWatcher(tc.sockDir)
+
+		actual := watcher.containsBlacklistedDir(tc.path)
+
+		// Assert
+		if tc.expected != actual {
+			t.Fatalf("expecting %v but got %v for testcase: %#v", tc.expected, actual, tc)
+		}
+	}
+}