Merge pull request #36079 from apprenda/windows_kube_proxy

cmd/kube-proxy/app/BUILD

@@ -29,6 +29,7 @@ go_library(
         "//pkg/proxy/config:go_default_library",
         "//pkg/proxy/iptables:go_default_library",
         "//pkg/proxy/userspace:go_default_library",
+        "//pkg/proxy/winuserspace:go_default_library",
         "//pkg/types:go_default_library",
         "//pkg/util/configz:go_default_library",
         "//pkg/util/dbus:go_default_library",
@@ -36,6 +37,7 @@ go_library(
         "//pkg/util/iptables:go_default_library",
         "//pkg/util/mount:go_default_library",
         "//pkg/util/net:go_default_library",
+        "//pkg/util/netsh:go_default_library",
         "//pkg/util/node:go_default_library",
         "//pkg/util/oom:go_default_library",
         "//pkg/util/resourcecontainer:go_default_library",

cmd/kube-proxy/app/server.go

@@ -39,12 +39,14 @@ import (
 	proxyconfig "k8s.io/kubernetes/pkg/proxy/config"
 	"k8s.io/kubernetes/pkg/proxy/iptables"
 	"k8s.io/kubernetes/pkg/proxy/userspace"
+	"k8s.io/kubernetes/pkg/proxy/winuserspace"
 	"k8s.io/kubernetes/pkg/types"
 	"k8s.io/kubernetes/pkg/util/configz"
 	utildbus "k8s.io/kubernetes/pkg/util/dbus"
 	"k8s.io/kubernetes/pkg/util/exec"
 	utiliptables "k8s.io/kubernetes/pkg/util/iptables"
 	utilnet "k8s.io/kubernetes/pkg/util/net"
+	utilnetsh "k8s.io/kubernetes/pkg/util/netsh"
 	nodeutil "k8s.io/kubernetes/pkg/util/node"
 	"k8s.io/kubernetes/pkg/util/oom"
 	"k8s.io/kubernetes/pkg/util/resourcecontainer"
@@ -136,10 +138,19 @@ func NewProxyServerDefault(config *options.ProxyServerConfig) (*ProxyServer, err
 		protocol = utiliptables.ProtocolIpv6
 	}
 
+	var netshInterface utilnetsh.Interface
+	var iptInterface utiliptables.Interface
+	var dbus utildbus.Interface
+
 	// Create a iptables utils.
 	execer := exec.New()
-	dbus := utildbus.New()
-	iptInterface := utiliptables.New(execer, dbus, protocol)
+
+	if runtime.GOOS == "windows" {
+		netshInterface = utilnetsh.New(execer)
+	} else {
+		dbus = utildbus.New()
+		iptInterface = utiliptables.New(execer, dbus, protocol)
+	}
 
 	// We omit creation of pretty much everything if we run in cleanup mode
 	if config.CleanupAndExit {
@@ -223,24 +234,44 @@ func NewProxyServerDefault(config *options.ProxyServerConfig) (*ProxyServer, err
 		// set EndpointsConfigHandler to our loadBalancer
 		endpointsHandler = loadBalancer
 
-		proxierUserspace, err := userspace.NewProxier(
-			loadBalancer,
-			net.ParseIP(config.BindAddress),
-			iptInterface,
-			*utilnet.ParsePortRangeOrDie(config.PortRange),
-			config.IPTablesSyncPeriod.Duration,
-			config.IPTablesMinSyncPeriod.Duration,
-			config.UDPIdleTimeout.Duration,
-		)
+		var proxierUserspace proxy.ProxyProvider
+
+		if runtime.GOOS == "windows" {
+			proxierUserspace, err = winuserspace.NewProxier(
+				loadBalancer,
+				net.ParseIP(config.BindAddress),
+				netshInterface,
+				*utilnet.ParsePortRangeOrDie(config.PortRange),
+				// TODO @pires replace below with default values, if applicable
+				config.IPTablesSyncPeriod.Duration,
+				config.UDPIdleTimeout.Duration,
+			)
+		} else {
+			proxierUserspace, err = userspace.NewProxier(
+				loadBalancer,
+				net.ParseIP(config.BindAddress),
+				iptInterface,
+				*utilnet.ParsePortRangeOrDie(config.PortRange),
+				config.IPTablesSyncPeriod.Duration,
+				config.IPTablesMinSyncPeriod.Duration,
+				config.UDPIdleTimeout.Duration,
+			)
+		}
 		if err != nil {
 			glog.Fatalf("Unable to create proxier: %v", err)
 		}
 		proxier = proxierUserspace
-		// Remove artifacts from the pure-iptables Proxier.
-		glog.V(0).Info("Tearing down pure-iptables proxy rules.")
-		iptables.CleanupLeftovers(iptInterface)
+		// Remove artifacts from the pure-iptables Proxier, if not on Windows.
+		if runtime.GOOS != "windows" {
+			glog.V(0).Info("Tearing down pure-iptables proxy rules.")
+			iptables.CleanupLeftovers(iptInterface)
+		}
+	}
+
+	// Add iptables reload function, if not on Windows.
+	if runtime.GOOS != "windows" {
+		iptInterface.AddReloadFunc(proxier.Sync)
 	}
-	iptInterface.AddReloadFunc(proxier.Sync)
 
 	// Create configs (i.e. Watches for Services and Endpoints)
 	// Note: RegisterHandler() calls need to happen before creation of Sources because sources
@@ -300,7 +331,7 @@ func (s *ProxyServer) Run() error {
 	}
 
 	// Tune conntrack, if requested
-	if s.Conntracker != nil {
+	if s.Conntracker != nil && runtime.GOOS != "windows" {
 		max, err := getConntrackMax(s.Config)
 		if err != nil {
 			return err

hack/.linted_packages

@@ -220,6 +220,7 @@ pkg/util/limitwriter
 pkg/util/logs
 pkg/util/maps
 pkg/util/metrics
+pkg/util/netsh
 pkg/util/ratelimit
 pkg/util/replicaset
 pkg/util/validation/field

pkg/proxy/winuserspace/BUILD

+package(default_visibility = ["//visibility:public"])
+
+licenses(["notice"])
+
+load(
+    "@io_bazel_rules_go//go:def.bzl",
+    "go_binary",
+    "go_library",
+    "go_test",
+    "cgo_library",
+)
+
+go_library(
+    name = "go_default_library",
+    srcs = [
+        "loadbalancer.go",
+        "port_allocator.go",
+        "proxier.go",
+        "proxysocket.go",
+        "roundrobin.go",
+        "udp_server.go",
+    ],
+    tags = ["automanaged"],
+    deps = [
+        "//pkg/api:go_default_library",
+        "//pkg/proxy:go_default_library",
+        "//pkg/types:go_default_library",
+        "//pkg/util/errors:go_default_library",
+        "//pkg/util/net:go_default_library",
+        "//pkg/util/netsh:go_default_library",
+        "//pkg/util/runtime:go_default_library",
+        "//pkg/util/slice:go_default_library",
+        "//pkg/util/wait:go_default_library",
+        "//vendor:github.com/golang/glog",
+    ],
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = [
+        "port_allocator_test.go",
+        "proxier_test.go",
+        "roundrobin_test.go",
+    ],
+    library = "go_default_library",
+    tags = ["automanaged"],
+    deps = [
+        "//pkg/api:go_default_library",
+        "//pkg/proxy:go_default_library",
+        "//pkg/types:go_default_library",
+        "//pkg/util/net:go_default_library",
+        "//pkg/util/netsh/testing:go_default_library",
+        "//pkg/util/runtime:go_default_library",
+    ],
+)

pkg/proxy/winuserspace/loadbalancer.go

+/*
+Copyright 2016 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 winuserspace
+
+import (
+	"net"
+
+	"k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/proxy"
+)
+
+// LoadBalancer is an interface for distributing incoming requests to service endpoints.
+type LoadBalancer interface {
+	// NextEndpoint returns the endpoint to handle a request for the given
+	// service-port and source address.
+	NextEndpoint(service proxy.ServicePortName, srcAddr net.Addr, sessionAffinityReset bool) (string, error)
+	NewService(service proxy.ServicePortName, sessionAffinityType api.ServiceAffinity, stickyMaxAgeMinutes int) error
+	DeleteService(service proxy.ServicePortName)
+	CleanupStaleStickySessions(service proxy.ServicePortName)
+}

pkg/proxy/winuserspace/port_allocator.go

+/*
+Copyright 2016 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 winuserspace
+
+import (
+	"errors"
+	"math/big"
+	"math/rand"
+	"sync"
+	"time"
+
+	"k8s.io/kubernetes/pkg/util/net"
+	"k8s.io/kubernetes/pkg/util/wait"
+)
+
+var (
+	errPortRangeNoPortsRemaining = errors.New("port allocation failed; there are no remaining ports left to allocate in the accepted range")
+)
+
+type PortAllocator interface {
+	AllocateNext() (int, error)
+	Release(int)
+}
+
+// randomAllocator is a PortAllocator implementation that allocates random ports, yielding
+// a port value of 0 for every call to AllocateNext().
+type randomAllocator struct{}
+
+// AllocateNext always returns 0
+func (r *randomAllocator) AllocateNext() (int, error) {
+	return 0, nil
+}
+
+// Release is a noop
+func (r *randomAllocator) Release(_ int) {
+	// noop
+}
+
+// newPortAllocator builds PortAllocator for a given PortRange. If the PortRange is empty
+// then a random port allocator is returned; otherwise, a new range-based allocator
+// is returned.
+func newPortAllocator(r net.PortRange) PortAllocator {
+	if r.Base == 0 {
+		return &randomAllocator{}
+	}
+	return newPortRangeAllocator(r)
+}
+
+const (
+	portsBufSize         = 16
+	nextFreePortCooldown = 500 * time.Millisecond
+	allocateNextTimeout  = 1 * time.Second
+)
+
+type rangeAllocator struct {
+	net.PortRange
+	ports chan int
+	used  big.Int
+	lock  sync.Mutex
+	rand  *rand.Rand
+}
+
+func newPortRangeAllocator(r net.PortRange) PortAllocator {
+	if r.Base == 0 || r.Size == 0 {
+		panic("illegal argument: may not specify an empty port range")
+	}
+	ra := &rangeAllocator{
+		PortRange: r,
+		ports:     make(chan int, portsBufSize),
+		rand:      rand.New(rand.NewSource(time.Now().UnixNano())),
+	}
+	go wait.Until(func() { ra.fillPorts(wait.NeverStop) }, nextFreePortCooldown, wait.NeverStop)
+	return ra
+}
+
+// fillPorts loops, always searching for the next free port and, if found, fills the ports buffer with it.
+// this func blocks until either there are no remaining free ports, or else the stopCh chan is closed.
+func (r *rangeAllocator) fillPorts(stopCh <-chan struct{}) {
+	for {
+		port := r.nextFreePort()
+		if port == -1 {
+			return
+		}
+		select {
+		case <-stopCh:
+			return
+		case r.ports <- port:
+		}
+	}
+}
+
+// nextFreePort finds a free port, first picking a random port. if that port is already in use
+// then the port range is scanned sequentially until either a port is found or the scan completes
+// unsuccessfully. an unsuccessful scan returns a port of -1.
+func (r *rangeAllocator) nextFreePort() int {
+	r.lock.Lock()
+	defer r.lock.Unlock()
+
+	// choose random port
+	j := r.rand.Intn(r.Size)
+	if b := r.used.Bit(j); b == 0 {
+		r.used.SetBit(&r.used, j, 1)
+		return j + r.Base
+	}
+
+	// search sequentially
+	for i := j + 1; i < r.Size; i++ {
+		if b := r.used.Bit(i); b == 0 {
+			r.used.SetBit(&r.used, i, 1)
+			return i + r.Base
+		}
+	}
+	for i := 0; i < j; i++ {
+		if b := r.used.Bit(i); b == 0 {
+			r.used.SetBit(&r.used, i, 1)
+			return i + r.Base
+		}
+	}
+	return -1
+}
+
+func (r *rangeAllocator) AllocateNext() (port int, err error) {
+	select {
+	case port = <-r.ports:
+	case <-time.After(allocateNextTimeout):
+		err = errPortRangeNoPortsRemaining
+	}
+	return
+}
+
+func (r *rangeAllocator) Release(port int) {
+	port -= r.Base
+	if port < 0 || port >= r.Size {
+		return
+	}
+	r.lock.Lock()
+	defer r.lock.Unlock()
+	r.used.SetBit(&r.used, port, 0)
+}

pkg/proxy/winuserspace/port_allocator_test.go

+/*
+Copyright 2016 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 winuserspace
+
+import (
+	"reflect"
+	"testing"
+
+	"k8s.io/kubernetes/pkg/util/net"
+)
+
+func TestRangeAllocatorEmpty(t *testing.T) {
+	r := &net.PortRange{}
+	r.Set("0-0")
+	defer func() {
+		if rv := recover(); rv == nil {
+			t.Fatalf("expected panic because of empty port range: %#v", r)
+		}
+	}()
+	_ = newPortRangeAllocator(*r)
+}
+
+func TestRangeAllocatorFullyAllocated(t *testing.T) {
+	r := &net.PortRange{}
+	r.Set("1-1")
+	a := newPortRangeAllocator(*r)
+	p, err := a.AllocateNext()
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if p != 1 {
+		t.Fatalf("unexpected allocated port: %d", p)
+	}
+
+	_, err = a.AllocateNext()
+	if err == nil {
+		t.Fatalf("expected error because of fully-allocated range")
+	}
+
+	a.Release(p)
+	p, err = a.AllocateNext()
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if p != 1 {
+		t.Fatalf("unexpected allocated port: %d", p)
+	}
+
+	_, err = a.AllocateNext()
+	if err == nil {
+		t.Fatalf("expected error because of fully-allocated range")
+	}
+}
+
+func TestRangeAllocator_RandomishAllocation(t *testing.T) {
+	r := &net.PortRange{}
+	r.Set("1-100")
+	a := newPortRangeAllocator(*r)
+
+	// allocate all the ports
+	var err error
+	ports := make([]int, 100, 100)
+	for i := 0; i < 100; i++ {
+		ports[i], err = a.AllocateNext()
+		if err != nil {
+			t.Fatalf("unexpected error: %v", err)
+		}
+	}
+
+	// release them all
+	for i := 0; i < 100; i++ {
+		a.Release(ports[i])
+	}
+
+	// allocate the ports again
+	rports := make([]int, 100, 100)
+	for i := 0; i < 100; i++ {
+		rports[i], err = a.AllocateNext()
+		if err != nil {
+			t.Fatalf("unexpected error: %v", err)
+		}
+	}
+
+	if reflect.DeepEqual(ports, rports) {
+		t.Fatalf("expected re-allocated ports to be in a somewhat random order")
+	}
+}

pkg/proxy/winuserspace/proxysocket.go

+/*
+Copyright 2016 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 winuserspace
+
+import (
+	"fmt"
+	"io"
+	"net"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/golang/glog"
+	"k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/proxy"
+	"k8s.io/kubernetes/pkg/util/runtime"
+)
+
+// Abstraction over TCP/UDP sockets which are proxied.
+type proxySocket interface {
+	// Addr gets the net.Addr for a proxySocket.
+	Addr() net.Addr
+	// Close stops the proxySocket from accepting incoming connections.
+	// Each implementation should comment on the impact of calling Close
+	// while sessions are active.
+	Close() error
+	// ProxyLoop proxies incoming connections for the specified service to the service endpoints.
+	ProxyLoop(service proxy.ServicePortName, info *serviceInfo, proxier *Proxier)
+	// ListenPort returns the host port that the proxySocket is listening on
+	ListenPort() int
+}
+
+func newProxySocket(protocol api.Protocol, ip net.IP, port int) (proxySocket, error) {
+	host := ""
+	if ip != nil {
+		host = ip.String()
+	}
+
+	switch strings.ToUpper(string(protocol)) {
+	case "TCP":
+		listener, err := net.Listen("tcp", net.JoinHostPort(host, strconv.Itoa(port)))
+		if err != nil {
+			return nil, err
+		}
+		return &tcpProxySocket{Listener: listener, port: port}, nil
+	case "UDP":
+		addr, err := net.ResolveUDPAddr("udp", net.JoinHostPort(host, strconv.Itoa(port)))
+		if err != nil {
+			return nil, err
+		}
+		conn, err := net.ListenUDP("udp", addr)
+		if err != nil {
+			return nil, err
+		}
+		return &udpProxySocket{UDPConn: conn, port: port}, nil
+	}
+	return nil, fmt.Errorf("unknown protocol %q", protocol)
+}
+
+// How long we wait for a connection to a backend in seconds
+var endpointDialTimeout = []time.Duration{250 * time.Millisecond, 500 * time.Millisecond, 1 * time.Second, 2 * time.Second}
+
+// tcpProxySocket implements proxySocket.  Close() is implemented by net.Listener.  When Close() is called,
+// no new connections are allowed but existing connections are left untouched.
+type tcpProxySocket struct {
+	net.Listener
+	port int
+}
+
+func (tcp *tcpProxySocket) ListenPort() int {
+	return tcp.port
+}
+
+func tryConnect(service proxy.ServicePortName, srcAddr net.Addr, protocol string, proxier *Proxier) (out net.Conn, err error) {
+	sessionAffinityReset := false
+	for _, dialTimeout := range endpointDialTimeout {
+		endpoint, err := proxier.loadBalancer.NextEndpoint(service, srcAddr, sessionAffinityReset)
+		if err != nil {
+			glog.Errorf("Couldn't find an endpoint for %s: %v", service, err)
+			return nil, err
+		}
+		glog.V(3).Infof("Mapped service %q to endpoint %s", service, endpoint)
+		// TODO: This could spin up a new goroutine to make the outbound connection,
+		// and keep accepting inbound traffic.
+		outConn, err := net.DialTimeout(protocol, endpoint, dialTimeout)
+		if err != nil {
+			if isTooManyFDsError(err) {
+				panic("Dial failed: " + err.Error())
+			}
+			glog.Errorf("Dial failed: %v", err)
+			sessionAffinityReset = true
+			continue
+		}
+		return outConn, nil
+	}
+	return nil, fmt.Errorf("failed to connect to an endpoint.")
+}
+
+func (tcp *tcpProxySocket) ProxyLoop(service proxy.ServicePortName, myInfo *serviceInfo, proxier *Proxier) {
+	for {
+		if !myInfo.isAlive() {
+			// The service port was closed or replaced.
+			return
+		}
+		// Block until a connection is made.
+		inConn, err := tcp.Accept()
+		if err != nil {
+			if isTooManyFDsError(err) {
+				panic("Accept failed: " + err.Error())
+			}
+
+			if isClosedError(err) {
+				return
+			}
+			if !myInfo.isAlive() {
+				// Then the service port was just closed so the accept failure is to be expected.
+				return
+			}
+			glog.Errorf("Accept failed: %v", err)
+			continue
+		}
+		glog.V(3).Infof("Accepted TCP connection from %v to %v", inConn.RemoteAddr(), inConn.LocalAddr())
+		outConn, err := tryConnect(service, inConn.(*net.TCPConn).RemoteAddr(), "tcp", proxier)
+		if err != nil {
+			glog.Errorf("Failed to connect to balancer: %v", err)
+			inConn.Close()
+			continue
+		}
+		// Spin up an async copy loop.
+		go proxyTCP(inConn.(*net.TCPConn), outConn.(*net.TCPConn))
+	}
+}
+
+// proxyTCP proxies data bi-directionally between in and out.
+func proxyTCP(in, out *net.TCPConn) {
+	var wg sync.WaitGroup
+	wg.Add(2)
+	glog.V(4).Infof("Creating proxy between %v <-> %v <-> %v <-> %v",
+		in.RemoteAddr(), in.LocalAddr(), out.LocalAddr(), out.RemoteAddr())
+	go copyBytes("from backend", in, out, &wg)
+	go copyBytes("to backend", out, in, &wg)
+	wg.Wait()
+}
+
+func copyBytes(direction string, dest, src *net.TCPConn, wg *sync.WaitGroup) {
+	defer wg.Done()
+	glog.V(4).Infof("Copying %s: %s -> %s", direction, src.RemoteAddr(), dest.RemoteAddr())
+	n, err := io.Copy(dest, src)
+	if err != nil {
+		if !isClosedError(err) {
+			glog.Errorf("I/O error: %v", err)
+		}
+	}
+	glog.V(4).Infof("Copied %d bytes %s: %s -> %s", n, direction, src.RemoteAddr(), dest.RemoteAddr())
+	dest.Close()
+	src.Close()
+}
+
+// udpProxySocket implements proxySocket.  Close() is implemented by net.UDPConn.  When Close() is called,
+// no new connections are allowed and existing connections are broken.
+// TODO: We could lame-duck this ourselves, if it becomes important.
+type udpProxySocket struct {
+	*net.UDPConn
+	port int
+}
+
+func (udp *udpProxySocket) ListenPort() int {
+	return udp.port
+}
+
+func (udp *udpProxySocket) Addr() net.Addr {
+	return udp.LocalAddr()
+}
+
+// Holds all the known UDP clients that have not timed out.
+type clientCache struct {
+	mu      sync.Mutex
+	clients map[string]net.Conn // addr string -> connection
+}
+
+func newClientCache() *clientCache {
+	return &clientCache{clients: map[string]net.Conn{}}
+}
+
+func (udp *udpProxySocket) ProxyLoop(service proxy.ServicePortName, myInfo *serviceInfo, proxier *Proxier) {
+	var buffer [4096]byte // 4KiB should be enough for most whole-packets
+	for {
+		if !myInfo.isAlive() {
+			// The service port was closed or replaced.
+			break
+		}
+
+		// Block until data arrives.
+		// TODO: Accumulate a histogram of n or something, to fine tune the buffer size.
+		n, cliAddr, err := udp.ReadFrom(buffer[0:])
+		if err != nil {
+			if e, ok := err.(net.Error); ok {
+				if e.Temporary() {
+					glog.V(1).Infof("ReadFrom had a temporary failure: %v", err)
+					continue
+				}
+			}
+			glog.Errorf("ReadFrom failed, exiting ProxyLoop: %v", err)
+			break
+		}
+		// If this is a client we know already, reuse the connection and goroutine.
+		svrConn, err := udp.getBackendConn(myInfo.activeClients, cliAddr, proxier, service, myInfo.timeout)
+		if err != nil {
+			continue
+		}
+		// TODO: It would be nice to let the goroutine handle this write, but we don't
+		// really want to copy the buffer.  We could do a pool of buffers or something.
+		_, err = svrConn.Write(buffer[0:n])
+		if err != nil {
+			if !logTimeout(err) {
+				glog.Errorf("Write failed: %v", err)
+				// TODO: Maybe tear down the goroutine for this client/server pair?
+			}
+			continue
+		}
+		err = svrConn.SetDeadline(time.Now().Add(myInfo.timeout))
+		if err != nil {
+			glog.Errorf("SetDeadline failed: %v", err)
+			continue
+		}
+	}
+}
+
+func (udp *udpProxySocket) getBackendConn(activeClients *clientCache, cliAddr net.Addr, proxier *Proxier, service proxy.ServicePortName, timeout time.Duration) (net.Conn, error) {
+	activeClients.mu.Lock()
+	defer activeClients.mu.Unlock()
+
+	svrConn, found := activeClients.clients[cliAddr.String()]
+	if !found {
+		// TODO: This could spin up a new goroutine to make the outbound connection,
+		// and keep accepting inbound traffic.
+		glog.V(3).Infof("New UDP connection from %s", cliAddr)
+		var err error
+		svrConn, err = tryConnect(service, cliAddr, "udp", proxier)
+		if err != nil {
+			return nil, err
+		}
+		if err = svrConn.SetDeadline(time.Now().Add(timeout)); err != nil {
+			glog.Errorf("SetDeadline failed: %v", err)
+			return nil, err
+		}
+		activeClients.clients[cliAddr.String()] = svrConn
+		go func(cliAddr net.Addr, svrConn net.Conn, activeClients *clientCache, timeout time.Duration) {
+			defer runtime.HandleCrash()
+			udp.proxyClient(cliAddr, svrConn, activeClients, timeout)
+		}(cliAddr, svrConn, activeClients, timeout)
+	}
+	return svrConn, nil
+}
+
+// This function is expected to be called as a goroutine.
+// TODO: Track and log bytes copied, like TCP
+func (udp *udpProxySocket) proxyClient(cliAddr net.Addr, svrConn net.Conn, activeClients *clientCache, timeout time.Duration) {
+	defer svrConn.Close()
+	var buffer [4096]byte
+	for {
+		n, err := svrConn.Read(buffer[0:])
+		if err != nil {
+			if !logTimeout(err) {
+				glog.Errorf("Read failed: %v", err)
+			}
+			break
+		}
+		err = svrConn.SetDeadline(time.Now().Add(timeout))
+		if err != nil {
+			glog.Errorf("SetDeadline failed: %v", err)
+			break
+		}
+		n, err = udp.WriteTo(buffer[0:n], cliAddr)
+		if err != nil {
+			if !logTimeout(err) {
+				glog.Errorf("WriteTo failed: %v", err)
+			}
+			break
+		}
+	}
+	activeClients.mu.Lock()
+	delete(activeClients.clients, cliAddr.String())
+	activeClients.mu.Unlock()
+}

pkg/proxy/winuserspace/udp_server.go

+/*
+Copyright 2016 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 winuserspace
+
+import (
+	"fmt"
+	"net"
+)
+
+// udpEchoServer is a simple echo server in UDP, intended for testing the proxy.
+type udpEchoServer struct {
+	net.PacketConn
+}
+
+func (r *udpEchoServer) Loop() {
+	var buffer [4096]byte
+	for {
+		n, cliAddr, err := r.ReadFrom(buffer[0:])
+		if err != nil {
+			fmt.Printf("ReadFrom failed: %v\n", err)
+			continue
+		}
+		r.WriteTo(buffer[0:n], cliAddr)
+	}
+}
+
+func newUDPEchoServer() (*udpEchoServer, error) {
+	packetconn, err := net.ListenPacket("udp", ":0")
+	if err != nil {
+		return nil, err
+	}
+	return &udpEchoServer{packetconn}, nil
+}

pkg/util/netsh/BUILD

+package(default_visibility = ["//visibility:public"])
+
+licenses(["notice"])
+
+load(
+    "@io_bazel_rules_go//go:def.bzl",
+    "go_binary",
+    "go_library",
+    "go_test",
+    "cgo_library",
+)
+
+go_library(
+    name = "go_default_library",
+    srcs = [
+        "doc.go",
+        "netsh.go",
+    ],
+    tags = ["automanaged"],
+    deps = [
+        "//pkg/util/exec:go_default_library",
+        "//vendor:github.com/golang/glog",
+    ],
+)

pkg/util/netsh/doc.go

+/*
+Copyright 2016 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 netsh provides an interface and implementations for running Windows netsh commands.
+package netsh // import "k8s.io/kubernetes/pkg/util/netsh"

pkg/util/netsh/netsh.go

+/*
+Copyright 2016 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 netsh
+
+import (
+	"fmt"
+	"net"
+	"os"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/golang/glog"
+	utilexec "k8s.io/kubernetes/pkg/util/exec"
+)
+
+// Interface is an injectable interface for running netsh commands.  Implementations must be goroutine-safe.
+type Interface interface {
+	// EnsurePortProxyRule checks if the specified redirect exists, if not creates it
+	EnsurePortProxyRule(args []string) (bool, error)
+	// DeletePortProxyRule deletes the specified portproxy rule.  If the rule did not exist, return error.
+	DeletePortProxyRule(args []string) error
+	// EnsureIPAddress checks if the specified IP Address is added to vEthernet (HNSTransparent) interface, if not, add it.  If the address existed, return true.
+	EnsureIPAddress(args []string, ip net.IP) (bool, error)
+	// DeleteIPAddress checks if the specified IP address is present and, if so, deletes it.
+	DeleteIPAddress(args []string) error
+	// Restore runs `netsh exec` to restore portproxy or addresses using a file.
+	// TODO Check if this is required, most likely not
+	Restore(args []string) error
+
+	// GetInterfaceToAddIP returns the interface name where Service IP needs to be added
+	// IP Address needs to be added for netsh portproxy to redirect traffic
+	// Reads Environment variable INTERFACE_TO_ADD_SERVICE_IP, if it is not defined then "vEthernet (HNSTransparent)" is returned
+	GetInterfaceToAddIP() string
+}
+
+const (
+	cmdNetsh string = "netsh"
+)
+
+// runner implements Interface in terms of exec("netsh").
+type runner struct {
+	mu   sync.Mutex
+	exec utilexec.Interface
+}
+
+// New returns a new Interface which will exec netsh.
+func New(exec utilexec.Interface) Interface {
+	runner := &runner{
+		exec: exec,
+	}
+	return runner
+}
+
+// EnsurePortProxyRule checks if the specified redirect exists, if not creates it.
+func (runner *runner) EnsurePortProxyRule(args []string) (bool, error) {
+	glog.V(4).Infof("running netsh interface portproxy add v4tov4 %v", args)
+	out, err := runner.exec.Command(cmdNetsh, args...).CombinedOutput()
+
+	if err == nil {
+		return true, nil
+	}
+	if ee, ok := err.(utilexec.ExitError); ok {
+		// netsh uses exit(0) to indicate a success of the operation,
+		// as compared to a malformed commandline, for example.
+		if ee.Exited() && ee.ExitStatus() != 0 {
+			return false, nil
+		}
+	}
+	return false, fmt.Errorf("error checking portproxy rule: %v: %s", err, out)
+
+}
+
+// DeletePortProxyRule deletes the specified portproxy rule.  If the rule did not exist, return error.
+func (runner *runner) DeletePortProxyRule(args []string) error {
+	glog.V(4).Infof("running netsh interface portproxy delete v4tov4 %v", args)
+	out, err := runner.exec.Command(cmdNetsh, args...).CombinedOutput()
+
+	if err == nil {
+		return nil
+	}
+	if ee, ok := err.(utilexec.ExitError); ok {
+		// netsh uses exit(0) to indicate a success of the operation,
+		// as compared to a malformed commandline, for example.
+		if ee.Exited() && ee.ExitStatus() == 0 {
+			return nil
+		}
+	}
+	return fmt.Errorf("error deleting portproxy rule: %v: %s", err, out)
+}
+
+// EnsureIPAddress checks if the specified IP Address is added to interface identified by Environment variable INTERFACE_TO_ADD_SERVICE_IP, if not, add it.  If the address existed, return true.
+func (runner *runner) EnsureIPAddress(args []string, ip net.IP) (bool, error) {
+	// Check if the ip address exists
+	intName := runner.GetInterfaceToAddIP()
+	argsShowAddress := []string{
+		"interface", "ipv4", "show", "address",
+		"name=" + intName,
+	}
+
+	ipToCheck := ip.String()
+
+	exists, _ := checkIPExists(ipToCheck, argsShowAddress, runner)
+	if exists == true {
+		glog.V(4).Infof("not adding IP address %q as it already exists", ipToCheck)
+		return true, nil
+	}
+
+	// IP Address is not already added, add it now
+	glog.V(4).Infof("running netsh interface ipv4 add address %v", args)
+	out, err := runner.exec.Command(cmdNetsh, args...).CombinedOutput()
+
+	if err == nil {
+		// Once the IP Address is added, it takes a bit to initialize and show up when querying for it
+		// Query all the IP addresses and see if the one we added is present
+		// PS: We are using netsh interface ipv4 show address here to query all the IP addresses, instead of
+		// querying net.InterfaceAddrs() as it returns the IP address as soon as it is added even though it is uninitialized
+		glog.V(3).Infof("Waiting until IP: %v is added to the network adapter", ipToCheck)
+		for {
+			if exists, _ := checkIPExists(ipToCheck, argsShowAddress, runner); exists {
+				return true, nil
+			}
+			time.Sleep(500 * time.Millisecond)
+		}
+	}
+	if ee, ok := err.(utilexec.ExitError); ok {
+		// netsh uses exit(0) to indicate a success of the operation,
+		// as compared to a malformed commandline, for example.
+		if ee.Exited() && ee.ExitStatus() != 0 {
+			return false, nil
+		}
+	}
+	return false, fmt.Errorf("error adding ipv4 address: %v: %s", err, out)
+}
+
+// DeleteIPAddress checks if the specified IP address is present and, if so, deletes it.
+func (runner *runner) DeleteIPAddress(args []string) error {
+	glog.V(4).Infof("running netsh interface ipv4 delete address %v", args)
+	out, err := runner.exec.Command(cmdNetsh, args...).CombinedOutput()
+
+	if err == nil {
+		return nil
+	}
+	if ee, ok := err.(utilexec.ExitError); ok {
+		// netsh uses exit(0) to indicate a success of the operation,
+		// as compared to a malformed commandline, for example.
+		if ee.Exited() && ee.ExitStatus() == 0 {
+			return nil
+		}
+	}
+	return fmt.Errorf("error deleting ipv4 address: %v: %s", err, out)
+}
+
+// GetInterfaceToAddIP returns the interface name where Service IP needs to be added
+// IP Address needs to be added for netsh portproxy to redirect traffic
+// Reads Environment variable INTERFACE_TO_ADD_SERVICE_IP, if it is not defined then "vEthernet (HNS Internal NIC)" is returned
+func (runner *runner) GetInterfaceToAddIP() string {
+	if iface := os.Getenv("INTERFACE_TO_ADD_SERVICE_IP"); len(iface) > 0 {
+		return iface
+	}
+	return "vEthernet (HNS Internal NIC)"
+}
+
+// Restore is part of Interface.
+func (runner *runner) Restore(args []string) error {
+	return nil
+}
+
+// checkIPExists checks if an IP address exists in 'netsh interface ipv4 show address' output
+func checkIPExists(ipToCheck string, args []string, runner *runner) (bool, error) {
+	ipAddress, err := runner.exec.Command(cmdNetsh, args...).CombinedOutput()
+	if err != nil {
+		return false, err
+	}
+	ipAddressString := string(ipAddress[:])
+	glog.V(3).Infof("Searching for IP: %v in IP dump: %v", ipToCheck, ipAddressString)
+	showAddressArray := strings.Split(ipAddressString, "\n")
+	for _, showAddress := range showAddressArray {
+		if strings.Contains(showAddress, "IP Address:") {
+			ipFromNetsh := strings.TrimLeft(showAddress, "IP Address:")
+			ipFromNetsh = strings.TrimSpace(ipFromNetsh)
+			if ipFromNetsh == ipToCheck {
+				return true, nil
+			}
+		}
+	}
+
+	return false, nil
+}

pkg/util/netsh/testing/BUILD

+package(default_visibility = ["//visibility:public"])
+
+licenses(["notice"])
+
+load(
+    "@io_bazel_rules_go//go:def.bzl",
+    "go_binary",
+    "go_library",
+    "go_test",
+    "cgo_library",
+)
+
+go_library(
+    name = "go_default_library",
+    srcs = ["fake.go"],
+    tags = ["automanaged"],
+    deps = ["//pkg/util/netsh:go_default_library"],
+)

pkg/util/netsh/testing/fake.go

+/*
+Copyright 2016 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 testing
+
+import (
+	"net"
+
+	"k8s.io/kubernetes/pkg/util/netsh"
+)
+
+// no-op implementation of netsh Interface
+type FakeNetsh struct {
+}
+
+func NewFake() *FakeNetsh {
+	return &FakeNetsh{}
+}
+
+func (*FakeNetsh) EnsurePortProxyRule(args []string) (bool, error) {
+	return true, nil
+}
+
+// DeletePortProxyRule deletes the specified portproxy rule.  If the rule did not exist, return error.
+func (*FakeNetsh) DeletePortProxyRule(args []string) error {
+	// Do Nothing
+	return nil
+}
+
+// EnsureIPAddress checks if the specified IP Address is added to vEthernet (HNSTransparent) interface, if not, add it.  If the address existed, return true.
+func (*FakeNetsh) EnsureIPAddress(args []string, ip net.IP) (bool, error) {
+	return true, nil
+}
+
+// DeleteIPAddress checks if the specified IP address is present and, if so, deletes it.
+func (*FakeNetsh) DeleteIPAddress(args []string) error {
+	// Do Nothing
+	return nil
+}
+
+// Restore runs `netsh exec` to restore portproxy or addresses using a file.
+// TODO Check if this is required, most likely not
+func (*FakeNetsh) Restore(args []string) error {
+	// Do Nothing
+	return nil
+}
+
+// GetInterfaceToAddIP returns the interface name where Service IP needs to be added
+// IP Address needs to be added for netsh portproxy to redirect traffic
+// Reads Environment variable INTERFACE_TO_ADD_SERVICE_IP, if it is not defined then "vEthernet (HNSTransparent)" is returned
+func (*FakeNetsh) GetInterfaceToAddIP() string {
+	return "Interface 1"
+}
+
+var _ = netsh.Interface(&FakeNetsh{})

test/test_owners.csv

@@ -676,6 +676,7 @@ k8s.io/kubernetes/pkg/proxy/config,ixdy,1
 k8s.io/kubernetes/pkg/proxy/healthcheck,ghodss,1
 k8s.io/kubernetes/pkg/proxy/iptables,freehan,0
 k8s.io/kubernetes/pkg/proxy/userspace,luxas,1
+k8s.io/kubernetes/pkg/proxy/winuserspace,jbhurat,0
 k8s.io/kubernetes/pkg/quota,sttts,1
 k8s.io/kubernetes/pkg/quota/evaluator/core,yifan-gu,1
 k8s.io/kubernetes/pkg/registry/apps/petset,kevin-wangzefeng,1