Merge pull request #46089 from karataliu/wincri1

Automatic merge from submit-queue (batch tested with PRs 46124, 46434, 46089, 45589, 46045)

Support TCP type runtime endpoint for kubelet

**What this PR does / why we need it**:
Currently the grpc server for kubelet and dockershim has a hardcoded endpoint: unix socket '/var/run/dockershim.sock', which is not applicable on non-unix OS.

This PR is to support TCP endpoint type besides unix socket.

**Which issue this PR fixes** 
This is a first attempt to address issue https://github.com/kubernetes/kubernetes/issues/45927

**Special notes for your reviewer**:
Before this change, running on Windows node results in:
```
Container Manager is unsupported in this build
```

After adding the cm stub, error becomes:
```
listen unix /var/run/dockershim.sock: socket: An address incompatible with the requested protocol was used.
```

This PR is to fix those two issues.

After this change, still meets 'seccomp' related issue when running on Windows node, needs more updates later.

**Release note**:

cmd/kubelet/app/options/options.go

@@ -302,8 +302,8 @@ func (c *kubeletConfiguration) addFlags(fs *pflag.FlagSet) {
 	// CRI flags.
 	fs.BoolVar(&c.ExperimentalDockershim, "experimental-dockershim", c.ExperimentalDockershim, "Enable dockershim only mode. In this mode, kubelet will only start dockershim without any other functionalities. This flag only serves test purpose, please do not use it unless you are conscious of what you are doing. [default=false]")
 	fs.MarkHidden("experimental-dockershim")
-	fs.StringVar(&c.RemoteRuntimeEndpoint, "container-runtime-endpoint", c.RemoteRuntimeEndpoint, "[Experimental] The unix socket endpoint of remote runtime service.")
-	fs.StringVar(&c.RemoteImageEndpoint, "image-service-endpoint", c.RemoteImageEndpoint, "[Experimental] The unix socket endpoint of remote image service. If not specified, it will be the same with container-runtime-endpoint by default.")
+	fs.StringVar(&c.RemoteRuntimeEndpoint, "container-runtime-endpoint", c.RemoteRuntimeEndpoint, "[Experimental] The endpoint of remote runtime service. Currently unix socket is supported on Linux, and tcp is supported on windows.  Examples:'unix:///var/run/dockershim.sock', 'tcp://localhost:3735'")
+	fs.StringVar(&c.RemoteImageEndpoint, "image-service-endpoint", c.RemoteImageEndpoint, "[Experimental] The endpoint of remote image service. If not specified, it will be the same with container-runtime-endpoint by default. Currently unix socket is supported on Linux, and tcp is supported on windows.  Examples:'unix:///var/run/dockershim.sock', 'tcp://localhost:3735'")
 	fs.BoolVar(&c.DockerDisableSharedPID, "docker-disable-shared-pid", c.DockerDisableSharedPID, "The Container Runtime Interface (CRI) defaults to using a shared PID namespace for containers in a pod when running with Docker 1.13.1 or higher. Setting this flag reverts to the previous behavior of isolated PID namespaces. This ability will be removed in a future Kubernetes release.")
 
 	fs.BoolVar(&c.ExperimentalCheckNodeCapabilitiesBeforeMount, "experimental-check-node-capabilities-before-mount", c.ExperimentalCheckNodeCapabilitiesBeforeMount, "[Experimental] if set true, the kubelet will check the underlying node for required componenets (binaries, etc.) before performing the mount")

cmd/kubelet/app/server.go

@@ -975,14 +975,8 @@ func RunDockershim(c *componentconfig.KubeletConfiguration, dockershimRootDir st
 		return err
 	}
 
-	// The unix socket for kubelet <-> dockershim communication.
-	ep := c.RemoteRuntimeEndpoint
-	if len(ep) == 0 {
-		ep = "/var/run/dockershim.sock"
-	}
-
 	glog.V(2).Infof("Starting the GRPC server for the docker CRI shim.")
-	server := dockerremote.NewDockerServer(ep, ds)
+	server := dockerremote.NewDockerServer(c.RemoteRuntimeEndpoint, ds)
 	if err := server.Start(); err != nil {
 		return err
 	}

pkg/apis/componentconfig/v1alpha1/defaults.go

@@ -438,6 +438,13 @@ func SetDefaults_KubeletConfiguration(obj *KubeletConfiguration) {
 	if obj.ExperimentalDockershim == nil {
 		obj.ExperimentalDockershim = boolVar(false)
 	}
+	if obj.RemoteRuntimeEndpoint == "" {
+		if runtime.GOOS == "linux" {
+			obj.RemoteRuntimeEndpoint = "unix:///var/run/dockershim.sock"
+		} else if runtime.GOOS == "windows" {
+			obj.RemoteRuntimeEndpoint = "tcp://localhost:3735"
+		}
+	}
 }
 
 func boolVar(b bool) *bool {

pkg/kubelet/dockershim/cm/container_manager_unsupported.go

@@ -1,4 +1,4 @@
-// +build !linux
+// +build !linux,!windows
 
 /*
 Copyright 2016 The Kubernetes Authors.

pkg/kubelet/dockershim/cm/container_manager_windows.go

@@ -0,0 +1,35 @@
+// +build windows
+
+/*
+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 cm
+
+import (
+	"k8s.io/kubernetes/pkg/kubelet/dockershim/libdocker"
+)
+
+// no-op
+type containerManager struct {
+}
+
+func NewContainerManager(_ string, _ libdocker.Interface) ContainerManager {
+	return &containerManager{}
+}
+
+func (m *containerManager) Start() error {
+	return nil
+}

pkg/kubelet/dockershim/remote/BUILD

@@ -18,6 +18,7 @@ go_library(
         "//pkg/kubelet/apis/cri:go_default_library",
         "//pkg/kubelet/apis/cri/v1alpha1:go_default_library",
         "//pkg/kubelet/dockershim:go_default_library",
+        "//pkg/kubelet/util:go_default_library",
         "//pkg/util/exec:go_default_library",
         "//pkg/util/interrupt:go_default_library",
         "//vendor/github.com/golang/glog:go_default_library",

pkg/kubelet/dockershim/remote/docker_server.go

@@ -18,29 +18,20 @@ package remote
 
 import (
 	"fmt"
-	"net"
-	"os"
-	"syscall"
 
 	"github.com/golang/glog"
 	"google.golang.org/grpc"
 
 	runtimeapi "k8s.io/kubernetes/pkg/kubelet/apis/cri/v1alpha1"
 	"k8s.io/kubernetes/pkg/kubelet/dockershim"
+	"k8s.io/kubernetes/pkg/kubelet/util"
 	"k8s.io/kubernetes/pkg/util/interrupt"
 )
 
-const (
-	// defaultEndpoint is the default address of dockershim grpc server socket.
-	defaultAddress = "/var/run/dockershim.sock"
-	// unixProtocol is the network protocol of unix socket.
-	unixProtocol = "unix"
-)
-
 // DockerServer is the grpc server of dockershim.
 type DockerServer struct {
-	// addr is the address to serve on.
-	addr string
+	// endpoint is the endpoint to serve on.
+	endpoint string
 	// service is the docker service which implements runtime and image services.
 	service DockerService
 	// server is the grpc server.
@@ -48,24 +39,19 @@ type DockerServer struct {
 }
 
 // NewDockerServer creates the dockershim grpc server.
-func NewDockerServer(addr string, s dockershim.DockerService) *DockerServer {
+func NewDockerServer(endpoint string, s dockershim.DockerService) *DockerServer {
 	return &DockerServer{
-		addr:    addr,
-		service: NewDockerService(s),
+		endpoint: endpoint,
+		service:  NewDockerService(s),
 	}
 }
 
 // Start starts the dockershim grpc server.
 func (s *DockerServer) Start() error {
 	glog.V(2).Infof("Start dockershim grpc server")
-	// Unlink to cleanup the previous socket file.
-	err := syscall.Unlink(s.addr)
-	if err != nil && !os.IsNotExist(err) {
-		return fmt.Errorf("failed to unlink socket file %q: %v", s.addr, err)
-	}
-	l, err := net.Listen(unixProtocol, s.addr)
+	l, err := util.CreateListener(s.endpoint)
 	if err != nil {
-		return fmt.Errorf("failed to listen on %q: %v", s.addr, err)
+		return fmt.Errorf("failed to listen on %q: %v", s.endpoint, err)
 	}
 	// Create the grpc server and register runtime and image services.
 	s.server = grpc.NewServer()

pkg/kubelet/kubelet.go

@@ -545,17 +545,10 @@ func NewMainKubelet(kubeCfg *componentconfig.KubeletConfiguration, kubeDeps *Kub
 			// kubelet, which handles the requests using DockerService..
 			klet.criHandler = ds
 
-			const (
-				// The unix socket for kubelet <-> dockershim communication.
-				ep = "/var/run/dockershim.sock"
-			)
-			if len(kubeCfg.RemoteRuntimeEndpoint) == 0 {
-				kubeCfg.RemoteRuntimeEndpoint = ep
-			}
-			if len(kubeCfg.RemoteImageEndpoint) == 0 {
-				kubeCfg.RemoteImageEndpoint = ep
-			}
-
+			// The unix socket for kubelet <-> dockershim communication.
+			glog.V(5).Infof("RemoteRuntimeEndpoint: %q, RemoteImageEndpoint: %q",
+				kubeCfg.RemoteRuntimeEndpoint,
+				kubeCfg.RemoteImageEndpoint)
 			glog.V(2).Infof("Starting the GRPC server for the docker CRI shim.")
 			server := dockerremote.NewDockerServer(kubeCfg.RemoteRuntimeEndpoint, ds)
 			if err := server.Start(); err != nil {
@@ -577,6 +570,9 @@ func NewMainKubelet(kubeCfg *componentconfig.KubeletConfiguration, kubeDeps *Kub
 			return nil, fmt.Errorf("unsupported CRI runtime: %q", kubeCfg.ContainerRuntime)
 		}
 		runtimeService, imageService, err := getRuntimeAndImageServices(kubeCfg)
+		if err != nil {
+			return nil, err
+		}
 		runtime, err := kuberuntime.NewKubeGenericRuntimeManager(
 			kubecontainer.FilterEventRecorder(kubeDeps.Recorder),
 			klet.livenessManager,

pkg/kubelet/remote/BUILD

@@ -19,6 +19,7 @@ go_library(
     deps = [
         "//pkg/kubelet/apis/cri:go_default_library",
         "//pkg/kubelet/apis/cri/v1alpha1:go_default_library",
+        "//pkg/kubelet/util:go_default_library",
         "//pkg/util/exec:go_default_library",
         "//vendor/github.com/golang/glog:go_default_library",
         "//vendor/golang.org/x/net/context:go_default_library",

pkg/kubelet/remote/remote_image.go

@@ -26,6 +26,7 @@ import (
 
 	internalapi "k8s.io/kubernetes/pkg/kubelet/apis/cri"
 	runtimeapi "k8s.io/kubernetes/pkg/kubelet/apis/cri/v1alpha1"
+	"k8s.io/kubernetes/pkg/kubelet/util"
 )
 
 // RemoteImageService is a gRPC implementation of internalapi.ImageManagerService.
@@ -35,9 +36,14 @@ type RemoteImageService struct {
 }
 
 // NewRemoteImageService creates a new internalapi.ImageManagerService.
-func NewRemoteImageService(addr string, connectionTimeout time.Duration) (internalapi.ImageManagerService, error) {
-	glog.V(3).Infof("Connecting to image service %s", addr)
-	conn, err := grpc.Dial(addr, grpc.WithInsecure(), grpc.WithTimeout(connectionTimeout), grpc.WithDialer(dial))
+func NewRemoteImageService(endpoint string, connectionTimeout time.Duration) (internalapi.ImageManagerService, error) {
+	glog.V(3).Infof("Connecting to image service %s", endpoint)
+	addr, dailer, err := util.GetAddressAndDialer(endpoint)
+	if err != nil {
+		return nil, err
+	}
+
+	conn, err := grpc.Dial(addr, grpc.WithInsecure(), grpc.WithTimeout(connectionTimeout), grpc.WithDialer(dailer))
 	if err != nil {
 		glog.Errorf("Connect remote image service %s failed: %v", addr, err)
 		return nil, err

pkg/kubelet/remote/remote_runtime.go

@@ -27,6 +27,7 @@ import (
 
 	internalapi "k8s.io/kubernetes/pkg/kubelet/apis/cri"
 	runtimeapi "k8s.io/kubernetes/pkg/kubelet/apis/cri/v1alpha1"
+	"k8s.io/kubernetes/pkg/kubelet/util"
 	utilexec "k8s.io/kubernetes/pkg/util/exec"
 )
 
@@ -37,9 +38,13 @@ type RemoteRuntimeService struct {
 }
 
 // NewRemoteRuntimeService creates a new internalapi.RuntimeService.
-func NewRemoteRuntimeService(addr string, connectionTimout time.Duration) (internalapi.RuntimeService, error) {
-	glog.Infof("Connecting to runtime service %s", addr)
-	conn, err := grpc.Dial(addr, grpc.WithInsecure(), grpc.WithTimeout(connectionTimout), grpc.WithDialer(dial))
+func NewRemoteRuntimeService(endpoint string, connectionTimout time.Duration) (internalapi.RuntimeService, error) {
+	glog.Infof("Connecting to runtime service %s", endpoint)
+	addr, dailer, err := util.GetAddressAndDialer(endpoint)
+	if err != nil {
+		return nil, err
+	}
+	conn, err := grpc.Dial(addr, grpc.WithInsecure(), grpc.WithTimeout(connectionTimout), grpc.WithDialer(dailer))
 	if err != nil {
 		glog.Errorf("Connect remote runtime %s failed: %v", addr, err)
 		return nil, err

pkg/kubelet/remote/utils.go

@@ -18,7 +18,6 @@ package remote
 
 import (
 	"fmt"
-	"net"
 	"time"
 
 	"golang.org/x/net/context"
@@ -26,11 +25,6 @@ import (
 	runtimeapi "k8s.io/kubernetes/pkg/kubelet/apis/cri/v1alpha1"
 )
 
-// dial creates a net.Conn by unix socket addr.
-func dial(addr string, timeout time.Duration) (net.Conn, error) {
-	return net.DialTimeout("unix", addr, timeout)
-}
-
 // getContextWithTimeout returns a context with timeout.
 func getContextWithTimeout(timeout time.Duration) (context.Context, context.CancelFunc) {
 	return context.WithTimeout(context.Background(), timeout)

pkg/kubelet/util/BUILD

@@ -5,6 +5,7 @@ licenses(["notice"])
 load(
     "@io_bazel_rules_go//go:def.bzl",
     "go_library",
+    "go_test",
 )
 
 go_library(
@@ -12,9 +13,21 @@ go_library(
     srcs = [
         "doc.go",
         "util.go",
+        "util_linux.go",
     ],
     tags = ["automanaged"],
-    deps = ["//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library"],
+    deps = [
+        "//vendor/github.com/golang/glog:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
+    ],
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = ["util_test.go"],
+    library = ":go_default_library",
+    tags = ["automanaged"],
+    deps = ["//vendor/github.com/stretchr/testify/assert:go_default_library"],
 )
 
 filegroup(

pkg/kubelet/util/util.go

@@ -17,6 +17,9 @@ limitations under the License.
 package util
 
 import (
+	"fmt"
+	"net/url"
+
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
@@ -25,3 +28,18 @@ import (
 func FromApiserverCache(opts *metav1.GetOptions) {
 	opts.ResourceVersion = "0"
 }
+
+func parseEndpoint(endpoint string) (string, string, error) {
+	u, err := url.Parse(endpoint)
+	if err != nil {
+		return "", "", err
+	}
+
+	if u.Scheme == "tcp" {
+		return "tcp", u.Host, nil
+	} else if u.Scheme == "unix" {
+		return "unix", u.Path, nil
+	} else {
+		return "", "", fmt.Errorf("protocol %q not supported", u.Scheme)
+	}
+}

pkg/kubelet/util/util_linux.go

@@ -0,0 +1,79 @@
+// +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 util
+
+import (
+	"fmt"
+	"net"
+	"os"
+	"syscall"
+	"time"
+
+	"github.com/golang/glog"
+)
+
+const (
+	// unixProtocol is the network protocol of unix socket.
+	unixProtocol = "unix"
+)
+
+func CreateListener(endpoint string) (net.Listener, error) {
+	protocol, addr, err := parseEndpointWithFallbackProtocol(endpoint, unixProtocol)
+	if err != nil {
+		return nil, err
+	}
+	if protocol != unixProtocol {
+		return nil, fmt.Errorf("only support unix socket endpoint")
+	}
+
+	// Unlink to cleanup the previous socket file.
+	err = syscall.Unlink(addr)
+	if err != nil && !os.IsNotExist(err) {
+		return nil, fmt.Errorf("failed to unlink socket file %q: %v", addr, err)
+	}
+
+	return net.Listen(protocol, addr)
+}
+
+func GetAddressAndDialer(endpoint string) (string, func(addr string, timeout time.Duration) (net.Conn, error), error) {
+	protocol, addr, err := parseEndpointWithFallbackProtocol(endpoint, unixProtocol)
+	if err != nil {
+		return "", nil, err
+	}
+	if protocol != unixProtocol {
+		return "", nil, fmt.Errorf("only support unix socket endpoint")
+	}
+
+	return addr, dial, nil
+}
+
+func dial(addr string, timeout time.Duration) (net.Conn, error) {
+	return net.DialTimeout(unixProtocol, addr, timeout)
+}
+
+func parseEndpointWithFallbackProtocol(endpoint string, fallbackProtocol string) (protocol string, addr string, err error) {
+	if protocol, addr, err = parseEndpoint(endpoint); err != nil {
+		fallbackEndpoint := fallbackProtocol + "://" + endpoint
+		protocol, addr, err = parseEndpoint(fallbackEndpoint)
+		if err == nil {
+			glog.Warningf("Using %q as endpoint is depercated, please consider using full url format %q.", endpoint, fallbackEndpoint)
+		}
+	}
+	return
+}

pkg/kubelet/util/util_test.go

@@ -0,0 +1,63 @@
+/*
+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 util
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestParseEndpoint(t *testing.T) {
+	tests := []struct {
+		endpoint         string
+		expectError      bool
+		expectedProtocol string
+		expectedAddr     string
+	}{
+		{
+			endpoint:         "unix:///tmp/s1.sock",
+			expectedProtocol: "unix",
+			expectedAddr:     "/tmp/s1.sock",
+		},
+		{
+			endpoint:         "tcp://localhost:15880",
+			expectedProtocol: "tcp",
+			expectedAddr:     "localhost:15880",
+		},
+		{
+			endpoint:    "tcp1://abc",
+			expectError: true,
+		},
+		{
+			endpoint:    "a b c",
+			expectError: true,
+		},
+	}
+
+	for _, test := range tests {
+		protocol, addr, err := parseEndpoint(test.endpoint)
+		if test.expectError {
+			assert.NotNil(t, err, "Expect error during parsing %q", test.endpoint)
+			continue
+		}
+		assert.Nil(t, err, "Expect no error during parsing %q", test.endpoint)
+		assert.Equal(t, test.expectedProtocol, protocol)
+		assert.Equal(t, test.expectedAddr, addr)
+	}
+
+}

pkg/kubelet/util/util_unsupported.go

@@ -0,0 +1,33 @@
+// +build !linux,!windows
+
+/*
+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 util
+
+import (
+	"fmt"
+	"net"
+	"time"
+)
+
+func CreateListener(endpoint string) (net.Listener, error) {
+	return nil, fmt.Errorf("CreateListener is unsupported in this build")
+}
+
+func GetAddressAndDialer(endpoint string) (string, func(addr string, timeout time.Duration) (net.Conn, error), error) {
+	return "", nil, fmt.Errorf("GetAddressAndDialer is unsupported in this build")
+}

pkg/kubelet/util/util_windows.go

@@ -0,0 +1,57 @@
+// +build windows
+
+/*
+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 util
+
+import (
+	"fmt"
+	"net"
+	"time"
+)
+
+const (
+	tcpProtocol = "tcp"
+)
+
+func CreateListener(endpoint string) (net.Listener, error) {
+	protocol, addr, err := parseEndpoint(endpoint)
+	if err != nil {
+		return nil, err
+	}
+	if protocol != tcpProtocol {
+		return nil, fmt.Errorf("only support tcp endpoint")
+	}
+
+	return net.Listen(protocol, addr)
+}
+
+func GetAddressAndDialer(endpoint string) (string, func(addr string, timeout time.Duration) (net.Conn, error), error) {
+	protocol, addr, err := parseEndpoint(endpoint)
+	if err != nil {
+		return "", nil, err
+	}
+	if protocol != tcpProtocol {
+		return "", nil, fmt.Errorf("only support tcp endpoint")
+	}
+
+	return addr, dial, nil
+}
+
+func dial(addr string, timeout time.Duration) (net.Conn, error) {
+	return net.DialTimeout(tcpProtocol, addr, timeout)
+}