Support TCP type runtime endpoint for kubelet.

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

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

pkg/kubelet/dockershim/cm/container_manager_windows.go

+// +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

+// +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

+/*
+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

+// +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

+// +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)
+}