Use userspace proxier when namespace is without network

Author: feiskyer

cmd/kube-proxy/app/server.go

@@ -217,7 +217,7 @@ func NewProxyServerDefault(config *options.ProxyServerConfig) (*ProxyServer, err
 		userspace.CleanupLeftovers(iptInterface)
 	case proxyModeHaproxy:
 		glog.V(2).Info("Using pod-buildin-haproxy proxy.")
-		proxierBuildin, err := haproxy.NewProxier(config.SyncPeriod)
+		proxierBuildin, err := haproxy.NewProxier(config.SyncPeriod, client)
 		if err != nil {
 			glog.Fatalf("Unable to create proxier: %v", err)
 		}
@@ -227,7 +227,7 @@ func NewProxyServerDefault(config *options.ProxyServerConfig) (*ProxyServer, err
 		glog.V(2).Info("Using userspace Proxier.")
 		// This is a proxy.LoadBalancer which NewProxier needs but has methods we don't need for
 		// our config.EndpointsConfigHandler.
-		loadBalancer := userspace.NewLoadBalancerRR()
+		loadBalancer := userspace.NewLoadBalancerRR(client, false)
 		// set EndpointsConfigHandler to our loadBalancer
 		endpointsHandler = loadBalancer
 
@@ -238,6 +238,8 @@ func NewProxyServerDefault(config *options.ProxyServerConfig) (*ProxyServer, err
 			*utilnet.ParsePortRangeOrDie(config.PortRange),
 			config.IPTablesSyncPeriod.Duration,
 			config.UDPIdleTimeout.Duration,
+			client,
+			false,
 		)
 		if err != nil {
 			glog.Fatalf("Unable to create proxier: %v", err)
@@ -266,7 +268,16 @@ func NewProxyServerDefault(config *options.ProxyServerConfig) (*ProxyServer, err
 		loadBalancer := userspace.NewLoadBalancerRR(client, true)
 		endpointsConfig.RegisterHandler(loadBalancer)
 
-		proxierUserspace, err := userspace.NewProxier(loadBalancer, config.BindAddress, iptInterface, config.PortRange, config.IptablesSyncPeriod, config.UDPIdleTimeout, client, true)
+		proxierUserspace, err := userspace.NewProxier(
+			loadBalancer,
+			net.ParseIP(config.BindAddress),
+			iptInterface,
+			*utilnet.ParsePortRangeOrDie(config.PortRange),
+			config.IPTablesSyncPeriod.Duration,
+			config.UDPIdleTimeout.Duration,
+			client,
+			true,
+		)
 		if err != nil {
 			glog.Fatalf("Unable to create proxier: %v", err)
 		}

pkg/proxy/haproxy/proxier.go

@@ -25,6 +25,7 @@ import (
 	"github.com/davecgh/go-spew/spew"
 	"github.com/golang/glog"
 	"k8s.io/kubernetes/pkg/api"
+	kubeclient "k8s.io/kubernetes/pkg/client/unversioned"
 	"k8s.io/kubernetes/pkg/kubelet/hyper"
 	"k8s.io/kubernetes/pkg/proxy"
 	"k8s.io/kubernetes/pkg/types"
@@ -58,6 +59,7 @@ type Proxier struct {
 	serviceMap                  map[proxy.ServicePortName]*serviceInfo
 	portsMap                    map[localPort]closeable
 	hyperClient                 *hyper.HyperClient
+	kubeClient                  *kubeclient.Client
 	haveReceivedServiceUpdate   bool // true once we've seen an OnServiceUpdate event
 	haveReceivedEndpointsUpdate bool // true once we've seen an OnEndpointsUpdate event
 
@@ -85,7 +87,7 @@ type closeable interface {
 var _ proxy.ProxyProvider = &Proxier{}
 
 // NewProxier returns a new Proxier given an pod-buildin-haproxy Interface instance.
-func NewProxier(syncPeriod time.Duration) (*Proxier, error) {
+func NewProxier(syncPeriod time.Duration, kubeClient *kubeclient.Client) (*Proxier, error) {
 	client := hyper.NewHyperClient()
 	_, err := client.Version()
 	if err != nil {
@@ -98,6 +100,7 @@ func NewProxier(syncPeriod time.Duration) (*Proxier, error) {
 		portsMap:    make(map[localPort]closeable),
 		syncPeriod:  syncPeriod,
 		hyperClient: client,
+		kubeClient:  kubeClient,
 	}, nil
 }
 
@@ -162,6 +165,19 @@ func (proxier *Proxier) OnServiceUpdate(allServices []api.Service) {
 
 	for i := range allServices {
 		service := &allServices[i]
+
+		// Check if namespace is configured with network
+		namespace, err := proxier.kubeClient.Namespaces().Get(service.Namespace)
+		if err != nil {
+			glog.Warningf("Get namespace error: %v", err)
+			continue
+		}
+		if namespace.Spec.Network == "" {
+			// Only process namespaces with network
+			// Namespaces without network will be processed by userspace proxier
+			continue
+		}
+
 		svcName := types.NamespacedName{
 			Namespace: service.Namespace,
 			Name:      service.Name,
@@ -233,6 +249,18 @@ func (proxier *Proxier) OnEndpointsUpdate(allEndpoints []api.Endpoints) {
 	for i := range allEndpoints {
 		svcEndpoints := &allEndpoints[i]
 
+		// Check is namespace is configured with network
+		namespace, err := proxier.kubeClient.Namespaces().Get(svcEndpoints.Namespace)
+		if err != nil {
+			glog.Warningf("Get namespace error: %v", err)
+			continue
+		}
+		if namespace.Spec.Network == "" {
+			// Only process namespaces with network
+			// Namespaces without network will be processed by userspace proxier
+			continue
+		}
+
 		// We need to build a map of portname -> all ip:ports for that
 		// portname.  Explode Endpoints.Subsets[*] into this structure.
 		portsToEndpoints := map[string][]hostPortPair{}

pkg/proxy/userspace/proxier.go

@@ -27,6 +27,7 @@ import (
 
 	"github.com/golang/glog"
 	"k8s.io/kubernetes/pkg/api"
+	kubeclient "k8s.io/kubernetes/pkg/client/unversioned"
 	"k8s.io/kubernetes/pkg/proxy"
 	"k8s.io/kubernetes/pkg/types"
 	utilnet "k8s.io/kubernetes/pkg/util/net"
@@ -95,6 +96,8 @@ type Proxier struct {
 	iptables       iptables.Interface
 	hostIP         net.IP
 	proxyPorts     PortAllocator
+	kubeClient     *kubeclient.Client
+	withHaproxier  bool
 }
 
 // assert Proxier is a ProxyProvider
@@ -139,7 +142,7 @@ func IsProxyLocked(err error) bool {
 // if iptables fails to update or acquire the initial lock. Once a proxier is
 // created, it will keep iptables up to date in the background and will not
 // terminate if a particular iptables call fails.
-func NewProxier(loadBalancer LoadBalancer, listenIP net.IP, iptables iptables.Interface, pr utilnet.PortRange, syncPeriod, udpIdleTimeout time.Duration) (*Proxier, error) {
+func NewProxier(loadBalancer LoadBalancer, listenIP net.IP, iptables iptables.Interface, pr utilnet.PortRange, syncPeriod, udpIdleTimeout time.Duration, kubeClient *kubeclient.Client, withHaproxier bool) (*Proxier, error) {
 	if listenIP.Equal(localhostIPv4) || listenIP.Equal(localhostIPv6) {
 		return nil, ErrProxyOnLocalhost
 	}
@@ -157,10 +160,14 @@ func NewProxier(loadBalancer LoadBalancer, listenIP net.IP, iptables iptables.In
 	proxyPorts := newPortAllocator(pr)
 
 	glog.V(2).Infof("Setting proxy IP to %v and initializing iptables", hostIP)
-	return createProxier(loadBalancer, listenIP, iptables, hostIP, proxyPorts, syncPeriod, udpIdleTimeout)
+	return createProxier(loadBalancer, listenIP, iptables, hostIP, proxyPorts, syncPeriod, udpIdleTimeout, kubeClient, withHaproxier)
 }
 
-func createProxier(loadBalancer LoadBalancer, listenIP net.IP, iptables iptables.Interface, hostIP net.IP, proxyPorts PortAllocator, syncPeriod, udpIdleTimeout time.Duration) (*Proxier, error) {
+func setRLimit(limit uint64) error {
+	return syscall.Setrlimit(syscall.RLIMIT_NOFILE, &syscall.Rlimit{Max: limit, Cur: limit})
+}
+
+func createProxier(loadBalancer LoadBalancer, listenIP net.IP, iptables iptables.Interface, hostIP net.IP, proxyPorts PortAllocator, syncPeriod, udpIdleTimeout time.Duration, kubeClient *kubeclient.Client, withHaproxier bool) (*Proxier, error) {
 	// convenient to pass nil for tests..
 	if proxyPorts == nil {
 		proxyPorts = newPortAllocator(utilnet.PortRange{})
@@ -184,6 +191,8 @@ func createProxier(loadBalancer LoadBalancer, listenIP net.IP, iptables iptables
 		iptables:       iptables,
 		hostIP:         hostIP,
 		proxyPorts:     proxyPorts,
+		kubeClient:     kubeClient,
+		withHaproxier:  withHaproxier,
 	}, nil
 }
 
@@ -377,6 +386,20 @@ func (proxier *Proxier) OnServiceUpdate(services []api.Service) {
 	for i := range services {
 		service := &services[i]
 
+		// Check if namespace is configured with network
+		if proxier.withHaproxier {
+			namespace, err := proxier.kubeClient.Namespaces().Get(service.Namespace)
+			if err != nil {
+				glog.Warningf("Get namespace error: %v", err)
+				continue
+			}
+			if namespace.Spec.Network != "" {
+				// Only process namespaces without network
+				// Namespaces with network will be processed by haproxy proxier
+				continue
+			}
+		}
+
 		// if ClusterIP is "None" or empty, skip proxying
 		if !api.IsServiceIPSet(service) {
 			glog.V(3).Infof("Skipping service %s due to clusterIP = %q", types.NamespacedName{Namespace: service.Namespace, Name: service.Name}, service.Spec.ClusterIP)