Merge pull request #81397 from ddebroy/win-socket

Support Kubelet PluginWatcher in Windows

pkg/kubelet/pluginmanager/pluginwatcher/BUILD

@@ -14,6 +14,7 @@ go_library(
         "//pkg/kubelet/pluginmanager/cache:go_default_library",
         "//pkg/kubelet/pluginmanager/pluginwatcher/example_plugin_apis/v1beta1:go_default_library",
         "//pkg/kubelet/pluginmanager/pluginwatcher/example_plugin_apis/v1beta2:go_default_library",
+        "//pkg/kubelet/util:go_default_library",
         "//pkg/util/filesystem:go_default_library",
         "//vendor/github.com/fsnotify/fsnotify:go_default_library",
         "//vendor/golang.org/x/net/context:go_default_library",

pkg/kubelet/pluginmanager/pluginwatcher/plugin_watcher.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"runtime"
 	"strings"
 	"time"
 
@@ -27,6 +28,7 @@ import (
 	"k8s.io/klog"
 
 	"k8s.io/kubernetes/pkg/kubelet/pluginmanager/cache"
+	"k8s.io/kubernetes/pkg/kubelet/util"
 	utilfs "k8s.io/kubernetes/pkg/util/filesystem"
 )
 
@@ -190,7 +192,11 @@ func (w *Watcher) handleCreateEvent(event fsnotify.Event) error {
 	}
 
 	if !fi.IsDir() {
-		if fi.Mode()&os.ModeSocket == 0 {
+		isSocket, err := util.IsUnixDomainSocket(util.NormalizePath(event.Name))
+		if err != nil {
+			return fmt.Errorf("failed to determine if file: %s is a unix domain socket: %v", event.Name, err)
+		}
+		if !isSocket {
 			klog.V(5).Infof("Ignoring non socket file %s", fi.Name())
 			return nil
 		}
@@ -202,6 +208,9 @@ func (w *Watcher) handleCreateEvent(event fsnotify.Event) error {
 }
 
 func (w *Watcher) handlePluginRegistration(socketPath string) error {
+	if runtime.GOOS == "windows" {
+		socketPath = util.NormalizePath(socketPath)
+	}
 	//TODO: Implement rate limiting to mitigate any DOS kind of attacks.
 	// Update desired state of world list of plugins
 	// If the socket path does exist in the desired world cache, there's still

pkg/kubelet/util/BUILD

@@ -16,14 +16,18 @@ go_test(
     deps = select({
         "@io_bazel_rules_go//go/platform:darwin": [
             "//vendor/github.com/stretchr/testify/assert:go_default_library",
+            "//vendor/github.com/stretchr/testify/require:go_default_library",
         ],
         "@io_bazel_rules_go//go/platform:freebsd": [
             "//vendor/github.com/stretchr/testify/assert:go_default_library",
+            "//vendor/github.com/stretchr/testify/require:go_default_library",
         ],
         "@io_bazel_rules_go//go/platform:linux": [
             "//vendor/github.com/stretchr/testify/assert:go_default_library",
+            "//vendor/github.com/stretchr/testify/require:go_default_library",
         ],
         "@io_bazel_rules_go//go/platform:windows": [
+            "//vendor/github.com/Microsoft/go-winio:go_default_library",
             "//vendor/github.com/stretchr/testify/assert:go_default_library",
             "//vendor/github.com/stretchr/testify/require:go_default_library",
         ],

pkg/kubelet/util/util_unix.go

@@ -135,3 +135,20 @@ func LocalEndpoint(path, file string) (string, error) {
 	}
 	return filepath.Join(u.String(), file+".sock"), nil
 }
+
+// IsUnixDomainSocket returns whether a given file is a AF_UNIX socket file
+func IsUnixDomainSocket(filePath string) (bool, error) {
+	fi, err := os.Stat(filePath)
+	if err != nil {
+		return false, fmt.Errorf("stat file %s failed: %v", filePath, err)
+	}
+	if fi.Mode()&os.ModeSocket == 0 {
+		return false, nil
+	}
+	return true, nil
+}
+
+// NormalizePath is a no-op for Linux for now
+func NormalizePath(path string) string {
+	return path
+}

pkg/kubelet/util/util_unix_test.go

@@ -19,9 +19,13 @@ limitations under the License.
 package util
 
 import (
+	"io/ioutil"
+	"net"
+	"os"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestParseEndpoint(t *testing.T) {
@@ -69,3 +73,63 @@ func TestParseEndpoint(t *testing.T) {
 	}
 
 }
+
+func TestIsUnixDomainSocket(t *testing.T) {
+	tests := []struct {
+		label          string
+		listenOnSocket bool
+		expectSocket   bool
+		expectError    bool
+		invalidFile    bool
+	}{
+		{
+			label:          "Domain Socket file",
+			listenOnSocket: true,
+			expectSocket:   true,
+			expectError:    false,
+		},
+		{
+			label:       "Non Existent file",
+			invalidFile: true,
+			expectError: true,
+		},
+		{
+			label:          "Regular file",
+			listenOnSocket: false,
+			expectSocket:   false,
+			expectError:    false,
+		},
+	}
+	for _, test := range tests {
+		f, err := ioutil.TempFile("", "test-domain-socket")
+		require.NoErrorf(t, err, "Failed to create file for test purposes: %v while setting up: %s", err, test.label)
+		addr := f.Name()
+		f.Close()
+		var ln *net.UnixListener
+		if test.listenOnSocket {
+			os.Remove(addr)
+			ta, err := net.ResolveUnixAddr("unix", addr)
+			require.NoErrorf(t, err, "Failed to ResolveUnixAddr: %v while setting up: %s", err, test.label)
+			ln, err = net.ListenUnix("unix", ta)
+			require.NoErrorf(t, err, "Failed to ListenUnix: %v while setting up: %s", err, test.label)
+		}
+		fileToTest := addr
+		if test.invalidFile {
+			fileToTest = fileToTest + ".invalid"
+		}
+		result, err := IsUnixDomainSocket(fileToTest)
+		if test.listenOnSocket {
+			// this takes care of removing the file associated with the domain socket
+			ln.Close()
+		} else {
+			// explicitly remove regular file
+			os.Remove(addr)
+		}
+		if test.expectError {
+			assert.NotNil(t, err, "Unexpected nil error from IsUnixDomainSocket for %s", test.label)
+		} else {
+			assert.Nil(t, err, "Unexpected error invoking IsUnixDomainSocket for %s", test.label)
+		}
+		assert.Equal(t, result, test.expectSocket, "Unexpected result from IsUnixDomainSocket: %v for %s", result, test.label)
+	}
+}

pkg/kubelet/util/util_windows.go

@@ -123,3 +123,31 @@ func GetBootTime() (time.Time, error) {
 	}
 	return currentTime.Add(-time.Duration(output) * time.Millisecond).Truncate(time.Second), nil
 }
+
+// IsUnixDomainSocket returns whether a given file is a AF_UNIX socket file
+func IsUnixDomainSocket(filePath string) (bool, error) {
+	// Due to the absence of golang support for os.ModeSocket in Windows (https://github.com/golang/go/issues/33357)
+	// we need to dial the file and check if we receive an error to determine if a file is Unix Domain Socket file.
+
+	// Note that querrying for the Reparse Points (https://docs.microsoft.com/en-us/windows/win32/fileio/reparse-points)
+	// for the file (using FSCTL_GET_REPARSE_POINT) and checking for reparse tag: reparseTagSocket
+	// does NOT work in 1809 if the socket file is created within a bind mounted directory by a container
+	// and the FSCTL is issued in the host by the kubelet.
+
+	c, err := net.Dial("unix", filePath)
+	if err == nil {
+		c.Close()
+		return true, nil
+	}
+	return false, nil
+}
+
+// NormalizePath converts FS paths returned by certain go frameworks (like fsnotify)
+// to native Windows paths that can be passed to Windows specific code
+func NormalizePath(path string) string {
+	path = strings.ReplaceAll(path, "/", "\\")
+	if strings.HasPrefix(path, "\\") {
+		path = "c:" + path
+	}
+	return path
+}

pkg/kubelet/util/util_windows_test.go

@@ -19,8 +19,14 @@ limitations under the License.
 package util
 
 import (
+	"io/ioutil"
+	"math/rand"
+	"net"
+	"os"
 	"testing"
+	"time"
 
+	winio "github.com/Microsoft/go-winio"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -90,3 +96,100 @@ func TestParseEndpoint(t *testing.T) {
 	}
 
 }
+
+func testPipe(t *testing.T, label string) {
+	generatePipeName := func(suffixlen int) string {
+		rand.Seed(time.Now().UnixNano())
+		letter := []rune("abcdef0123456789")
+		b := make([]rune, suffixlen)
+		for i := range b {
+			b[i] = letter[rand.Intn(len(letter))]
+		}
+		return "\\\\.\\pipe\\test-pipe" + string(b)
+	}
+	testFile := generatePipeName(4)
+	pipeln, err := winio.ListenPipe(testFile, &winio.PipeConfig{SecurityDescriptor: "D:P(A;;GA;;;BA)(A;;GA;;;SY)"})
+	defer pipeln.Close()
+
+	require.NoErrorf(t, err, "Failed to listen on named pipe for test purposes: %v while setting up: %s", err, label)
+	result, err := IsUnixDomainSocket(testFile)
+	assert.Nil(t, err, "Unexpected error: %v from IsUnixDomainSocket for %s", err, label)
+	assert.False(t, result, "Unexpected result: true from IsUnixDomainSocket: %v for %s", result, label)
+}
+
+func testRegularFile(t *testing.T, label string, exists bool) {
+	f, err := ioutil.TempFile("", "test-file")
+	require.NoErrorf(t, err, "Failed to create file for test purposes: %v while setting up: %s", err, label)
+	testFile := f.Name()
+	if !exists {
+		testFile = testFile + ".absent"
+	}
+	f.Close()
+	result, err := IsUnixDomainSocket(testFile)
+	os.Remove(f.Name())
+	assert.Nil(t, err, "Unexpected error: %v from IsUnixDomainSocket for %s", err, label)
+	assert.False(t, result, "Unexpected result: true from IsUnixDomainSocket: %v for %s", result, label)
+}
+
+func testUnixDomainSocket(t *testing.T, label string) {
+	f, err := ioutil.TempFile("", "test-domain-socket")
+	require.NoErrorf(t, err, "Failed to create file for test purposes: %v while setting up: %s", err, label)
+	testFile := f.Name()
+	f.Close()
+	os.Remove(testFile)
+	ta, err := net.ResolveUnixAddr("unix", testFile)
+	require.NoErrorf(t, err, "Failed to ResolveUnixAddr: %v while setting up: %s", err, label)
+	unixln, err := net.ListenUnix("unix", ta)
+	require.NoErrorf(t, err, "Failed to ListenUnix: %v while setting up: %s", err, label)
+	result, err := IsUnixDomainSocket(testFile)
+	unixln.Close()
+	assert.Nil(t, err, "Unexpected error: %v from IsUnixDomainSocket for %s", err, label)
+	assert.True(t, result, "Unexpected result: false from IsUnixDomainSocket: %v for %s", result, label)
+}
+
+func TestIsUnixDomainSocket(t *testing.T) {
+	testPipe(t, "Named Pipe")
+	testRegularFile(t, "Regular File that Exists", true)
+	testRegularFile(t, "Regular File that Does Not Exist", false)
+	testUnixDomainSocket(t, "Unix Domain Socket File")
+}
+
+func TestNormalizePath(t *testing.T) {
+	tests := []struct {
+		originalpath   string
+		normalizedPath string
+	}{
+		{
+			originalpath:   "\\path\\to\\file",
+			normalizedPath: "c:\\path\\to\\file",
+		},
+		{
+			originalpath:   "/path/to/file",
+			normalizedPath: "c:\\path\\to\\file",
+		},
+		{
+			originalpath:   "/path/to/dir/",
+			normalizedPath: "c:\\path\\to\\dir\\",
+		},
+		{
+			originalpath:   "\\path\\to\\dir\\",
+			normalizedPath: "c:\\path\\to\\dir\\",
+		},
+		{
+			originalpath:   "/file",
+			normalizedPath: "c:\\file",
+		},
+		{
+			originalpath:   "\\file",
+			normalizedPath: "c:\\file",
+		},
+		{
+			originalpath:   "fileonly",
+			normalizedPath: "fileonly",
+		},
+	}
+
+	for _, test := range tests {
+		assert.Equal(t, test.normalizedPath, NormalizePath(test.originalpath))
+	}
+}