Merge pull request #54219 from m1093782566/ipset

cmd/kube-proxy/app/BUILD

@@ -41,6 +41,7 @@ go_library(
         "//pkg/proxy/userspace:go_default_library",
         "//pkg/util/configz:go_default_library",
         "//pkg/util/dbus:go_default_library",
+        "//pkg/util/ipset:go_default_library",
         "//pkg/util/iptables:go_default_library",
         "//pkg/util/ipvs:go_default_library",
         "//pkg/util/mount:go_default_library",

cmd/kube-proxy/app/server.go

@@ -63,6 +63,7 @@ import (
 	"k8s.io/kubernetes/pkg/proxy/ipvs"
 	"k8s.io/kubernetes/pkg/proxy/userspace"
 	"k8s.io/kubernetes/pkg/util/configz"
+	utilipset "k8s.io/kubernetes/pkg/util/ipset"
 	utiliptables "k8s.io/kubernetes/pkg/util/iptables"
 	utilipvs "k8s.io/kubernetes/pkg/util/ipvs"
 	utilnode "k8s.io/kubernetes/pkg/util/node"
@@ -356,6 +357,7 @@ type ProxyServer struct {
 	EventClient            v1core.EventsGetter
 	IptInterface           utiliptables.Interface
 	IpvsInterface          utilipvs.Interface
+	IpsetInterface         utilipset.Interface
 	execer                 exec.Interface
 	Proxier                proxy.ProxyProvider
 	Broadcaster            record.EventBroadcaster
@@ -422,7 +424,7 @@ func (s *ProxyServer) Run() error {
 	if s.CleanupAndExit {
 		encounteredError := userspace.CleanupLeftovers(s.IptInterface)
 		encounteredError = iptables.CleanupLeftovers(s.IptInterface) || encounteredError
-		encounteredError = ipvs.CleanupLeftovers(s.IpvsInterface, s.IptInterface) || encounteredError
+		encounteredError = ipvs.CleanupLeftovers(s.IpvsInterface, s.IptInterface, s.IpsetInterface) || encounteredError
 		if encounteredError {
 			return errors.New("encountered an error while tearing down rules.")
 		}

cmd/kube-proxy/app/server_others.go

@@ -43,6 +43,7 @@ import (
 	"k8s.io/kubernetes/pkg/proxy/userspace"
 	"k8s.io/kubernetes/pkg/util/configz"
 	utildbus "k8s.io/kubernetes/pkg/util/dbus"
+	utilipset "k8s.io/kubernetes/pkg/util/ipset"
 	utiliptables "k8s.io/kubernetes/pkg/util/iptables"
 	utilipvs "k8s.io/kubernetes/pkg/util/ipvs"
 	utilnode "k8s.io/kubernetes/pkg/util/node"
@@ -72,6 +73,7 @@ func NewProxyServer(config *proxyconfigapi.KubeProxyConfiguration, cleanupAndExi
 
 	var iptInterface utiliptables.Interface
 	var ipvsInterface utilipvs.Interface
+	var ipsetInterface utilipset.Interface
 	var dbus utildbus.Interface
 
 	// Create a iptables utils.
@@ -80,6 +82,7 @@ func NewProxyServer(config *proxyconfigapi.KubeProxyConfiguration, cleanupAndExi
 	dbus = utildbus.New()
 	iptInterface = utiliptables.New(execer, dbus, protocol)
 	ipvsInterface = utilipvs.New(execer)
+	ipsetInterface = utilipset.New(execer)
 
 	// We omit creation of pretty much everything if we run in cleanup mode
 	if cleanupAndExit {
@@ -87,6 +90,7 @@ func NewProxyServer(config *proxyconfigapi.KubeProxyConfiguration, cleanupAndExi
 			execer:         execer,
 			IptInterface:   iptInterface,
 			IpvsInterface:  ipvsInterface,
+			IpsetInterface: ipsetInterface,
 			CleanupAndExit: cleanupAndExit,
 		}, nil
 	}
@@ -119,7 +123,7 @@ func NewProxyServer(config *proxyconfigapi.KubeProxyConfiguration, cleanupAndExi
 	var serviceEventHandler proxyconfig.ServiceHandler
 	var endpointsEventHandler proxyconfig.EndpointsHandler
 
-	proxyMode := getProxyMode(string(config.Mode), iptInterface, iptables.LinuxKernelCompatTester{})
+	proxyMode := getProxyMode(string(config.Mode), iptInterface, ipsetInterface, iptables.LinuxKernelCompatTester{})
 	if proxyMode == proxyModeIPTables {
 		glog.V(0).Info("Using iptables Proxier.")
 		nodeIP := net.ParseIP(config.BindAddress)
@@ -159,12 +163,13 @@ func NewProxyServer(config *proxyconfigapi.KubeProxyConfiguration, cleanupAndExi
 		userspace.CleanupLeftovers(iptInterface)
 		// IPVS Proxier will generate some iptables rules,
 		// need to clean them before switching to other proxy mode.
-		ipvs.CleanupLeftovers(ipvsInterface, iptInterface)
+		ipvs.CleanupLeftovers(ipvsInterface, iptInterface, ipsetInterface)
 	} else if proxyMode == proxyModeIPVS {
 		glog.V(0).Info("Using ipvs Proxier.")
 		proxierIPVS, err := ipvs.NewProxier(
 			iptInterface,
 			ipvsInterface,
+			ipsetInterface,
 			utilsysctl.New(),
 			execer,
 			config.IPVS.SyncPeriod.Duration,
@@ -220,7 +225,7 @@ func NewProxyServer(config *proxyconfigapi.KubeProxyConfiguration, cleanupAndExi
 		iptables.CleanupLeftovers(iptInterface)
 		// IPVS Proxier will generate some iptables rules,
 		// need to clean them before switching to other proxy mode.
-		ipvs.CleanupLeftovers(ipvsInterface, iptInterface)
+		ipvs.CleanupLeftovers(ipvsInterface, iptInterface, ipsetInterface)
 	}
 
 	iptInterface.AddReloadFunc(proxier.Sync)
@@ -230,6 +235,7 @@ func NewProxyServer(config *proxyconfigapi.KubeProxyConfiguration, cleanupAndExi
 		EventClient:            eventClient,
 		IptInterface:           iptInterface,
 		IpvsInterface:          ipvsInterface,
+		IpsetInterface:         ipsetInterface,
 		execer:                 execer,
 		Proxier:                proxier,
 		Broadcaster:            eventBroadcaster,
@@ -249,7 +255,7 @@ func NewProxyServer(config *proxyconfigapi.KubeProxyConfiguration, cleanupAndExi
 	}, nil
 }
 
-func getProxyMode(proxyMode string, iptver iptables.IPTablesVersioner, kcompat iptables.KernelCompatTester) string {
+func getProxyMode(proxyMode string, iptver iptables.IPTablesVersioner, ipsetver ipvs.IPSetVersioner, kcompat iptables.KernelCompatTester) string {
 	if proxyMode == proxyModeUserspace {
 		return proxyModeUserspace
 	}
@@ -260,7 +266,7 @@ func getProxyMode(proxyMode string, iptver iptables.IPTablesVersioner, kcompat i
 
 	if utilfeature.DefaultFeatureGate.Enabled(features.SupportIPVSProxyMode) {
 		if proxyMode == proxyModeIPVS {
-			return tryIPVSProxy(iptver, kcompat)
+			return tryIPVSProxy(iptver, ipsetver, kcompat)
 		} else {
 			glog.Warningf("Can't use ipvs proxier, trying iptables proxier")
 			return tryIPTablesProxy(iptver, kcompat)
@@ -270,10 +276,10 @@ func getProxyMode(proxyMode string, iptver iptables.IPTablesVersioner, kcompat i
 	return tryIPTablesProxy(iptver, kcompat)
 }
 
-func tryIPVSProxy(iptver iptables.IPTablesVersioner, kcompat iptables.KernelCompatTester) string {
+func tryIPVSProxy(iptver iptables.IPTablesVersioner, ipsetver ipvs.IPSetVersioner, kcompat iptables.KernelCompatTester) string {
 	// guaranteed false on error, error only necessary for debugging
-	// IPVS Proxier relies on iptables
-	useIPVSProxy, err := ipvs.CanUseIPVSProxier()
+	// IPVS Proxier relies on ipset
+	useIPVSProxy, err := ipvs.CanUseIPVSProxier(ipsetver)
 	if err != nil {
 		// Try to fallback to iptables before falling back to userspace
 		utilruntime.HandleError(fmt.Errorf("can't determine whether to use ipvs proxy, error: %v", err))
@@ -282,8 +288,6 @@ func tryIPVSProxy(iptver iptables.IPTablesVersioner, kcompat iptables.KernelComp
 		return proxyModeIPVS
 	}
 
-	// TODO: Check ipvs version
-
 	// Try to fallback to iptables before falling back to userspace
 	glog.V(1).Infof("Can't use ipvs proxier, trying iptables proxier")
 	return tryIPTablesProxy(iptver, kcompat)

cmd/kube-proxy/app/server_test.go

@@ -52,6 +52,15 @@ func (fake *fakeIPTablesVersioner) GetVersion() (string, error) {
 	return fake.version, fake.err
 }
 
+type fakeIPSetVersioner struct {
+	version string // what to return
+	err     error  // what to return
+}
+
+func (fake *fakeIPSetVersioner) GetVersion() (string, error) {
+	return fake.version, fake.err
+}
+
 type fakeKernelCompatTester struct {
 	ok bool
 }
@@ -72,8 +81,10 @@ func Test_getProxyMode(t *testing.T) {
 		annotationKey   string
 		annotationVal   string
 		iptablesVersion string
+		ipsetVersion    string
 		kernelCompat    bool
 		iptablesError   error
+		ipsetError      error
 		expected        string
 	}{
 		{ // flag says userspace
@@ -128,7 +139,8 @@ func Test_getProxyMode(t *testing.T) {
 	for i, c := range cases {
 		versioner := &fakeIPTablesVersioner{c.iptablesVersion, c.iptablesError}
 		kcompater := &fakeKernelCompatTester{c.kernelCompat}
-		r := getProxyMode(c.flag, versioner, kcompater)
+		ipsetver := &fakeIPSetVersioner{c.ipsetVersion, c.ipsetError}
+		r := getProxyMode(c.flag, versioner, ipsetver, kcompater)
 		if r != c.expected {
 			t.Errorf("Case[%d] Expected %q, got %q", i, c.expected, r)
 		}

pkg/proxy/ipvs/BUILD

@@ -9,6 +9,7 @@ load(
 go_test(
     name = "go_default_test",
     srcs = [
+        "ipset_test.go",
         "proxier_test.go",
     ],
     importpath = "k8s.io/kubernetes/pkg/proxy/ipvs",
@@ -18,6 +19,8 @@ go_test(
         "//pkg/proxy:go_default_library",
         "//pkg/proxy/ipvs/testing:go_default_library",
         "//pkg/proxy/util:go_default_library",
+        "//pkg/util/ipset:go_default_library",
+        "//pkg/util/ipset/testing:go_default_library",
         "//pkg/util/iptables:go_default_library",
         "//pkg/util/iptables/testing:go_default_library",
         "//pkg/util/ipvs:go_default_library",
@@ -35,6 +38,7 @@ go_test(
 go_library(
     name = "go_default_library",
     srcs = [
+        "ipset.go",
         "netlink.go",
         "netlink_unsupported.go",
         "proxier.go",
@@ -55,9 +59,11 @@ go_library(
         "//pkg/proxy/metrics:go_default_library",
         "//pkg/proxy/util:go_default_library",
         "//pkg/util/async:go_default_library",
+        "//pkg/util/ipset:go_default_library",
         "//pkg/util/iptables:go_default_library",
         "//pkg/util/ipvs:go_default_library",
         "//pkg/util/sysctl:go_default_library",
+        "//pkg/util/version:go_default_library",
         "//vendor/github.com/golang/glog:go_default_library",
         "//vendor/k8s.io/api/core/v1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/types:go_default_library",

pkg/proxy/ipvs/ipset.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 ipvs
+
+import (
+	"k8s.io/apimachinery/pkg/util/sets"
+	utilipset "k8s.io/kubernetes/pkg/util/ipset"
+	utilversion "k8s.io/kubernetes/pkg/util/version"
+
+	"github.com/golang/glog"
+)
+
+const (
+	// MinIPSetCheckVersion is the min ipset version we need.  IPv6 is supported in ipset 6.x
+	MinIPSetCheckVersion = "6.0"
+
+	// KubeLoopBackIPSet is used to store endpoints dst ip:port, source ip for solving hairpin purpose.
+	KubeLoopBackIPSet = "KUBE-LOOP-BACK"
+
+	// KubeClusterIPSet is used to store service cluster ip + port for masquerade purpose.
+	KubeClusterIPSet = "KUBE-CLUSTER-IP"
+
+	// KubeExternalIPSet is used to store service external ip + port for masquerade and filter purpose.
+	KubeExternalIPSet = "KUBE-EXTERNAL-IP"
+
+	// KubeLoadBalancerSet is used to store service load balancer ingress ip + port, it is the service lb portal.
+	KubeLoadBalancerSet = "KUBE-LOAD-BALANCER"
+
+	// KubeLoadBalancerMasqSet is used to store service load balancer ingress ip + port for masquerade purpose.
+	KubeLoadBalancerMasqSet = "KUBE-LOAD-BALANCER-MASQ"
+
+	// KubeLoadBalancerSourceIPSet is used to store service load balancer ingress ip + port + source IP for packet filter purpose.
+	KubeLoadBalancerSourceIPSet = "KUBE-LOAD-BALANCER-SOURCE-IP"
+
+	// KubeLoadBalancerSourceCIDRSet is used to store service load balancer ingress ip + port + source cidr for packet filter purpose.
+	KubeLoadBalancerSourceCIDRSet = "KUBE-LOAD-BALANCER-SOURCE-CIDR"
+
+	// KubeNodePortSetTCP is used to store nodeport TCP port for masquerade purpose.
+	KubeNodePortSetTCP = "KUBE-NODE-PORT-TCP"
+
+	// KubeNodePortSetUDP is used to store nodeport UDP port for masquerade purpose.
+	KubeNodePortSetUDP = "KUBE-NODE-PORT-UDP"
+)
+
+// IPSetVersioner can query the current ipset version.
+type IPSetVersioner interface {
+	// returns "X.Y"
+	GetVersion() (string, error)
+}
+
+// IPSet wraps util/ipset which is used by IPVS proxier.
+type IPSet struct {
+	utilipset.IPSet
+	// activeEntries is the current active entries of the ipset.
+	activeEntries sets.String
+	// handle is the util ipset interface handle.
+	handle utilipset.Interface
+}
+
+// NewIPSet initialize a new IPSet struct
+func NewIPSet(handle utilipset.Interface, name string, setType utilipset.Type, isIPv6 bool) *IPSet {
+	hashFamily := utilipset.ProtocolFamilyIPV4
+	if isIPv6 {
+		hashFamily = utilipset.ProtocolFamilyIPV6
+	}
+	set := &IPSet{
+		IPSet: utilipset.IPSet{
+			Name:       name,
+			SetType:    setType,
+			HashFamily: hashFamily,
+		},
+		activeEntries: sets.NewString(),
+		handle:        handle,
+	}
+	return set
+}
+
+func (set *IPSet) isEmpty() bool {
+	return len(set.activeEntries.UnsortedList()) == 0
+}
+
+func (set *IPSet) resetEntries() {
+	set.activeEntries = sets.NewString()
+}
+
+func (set *IPSet) syncIPSetEntries() {
+	appliedEntries, err := set.handle.ListEntries(set.Name)
+	if err != nil {
+		glog.Errorf("Failed to list ip set entries, error: %v", err)
+		return
+	}
+
+	// currentIPSetEntries represents Endpoints watched from API Server.
+	currentIPSetEntries := sets.NewString()
+	for _, appliedEntry := range appliedEntries {
+		currentIPSetEntries.Insert(appliedEntry)
+	}
+
+	if !set.activeEntries.Equal(currentIPSetEntries) {
+		// Clean legacy entries
+		for _, entry := range currentIPSetEntries.Difference(set.activeEntries).List() {
+			if err := set.handle.DelEntry(entry, set.Name); err != nil {
+				glog.Errorf("Failed to delete ip set entry: %s from ip set: %s, error: %v", entry, set.Name, err)
+			} else {
+				glog.V(3).Infof("Successfully delete legacy ip set entry: %s from ip set: %s", entry, set.Name)
+			}
+		}
+		// Create active entries
+		for _, entry := range set.activeEntries.Difference(currentIPSetEntries).List() {
+			if err := set.handle.AddEntry(entry, set.Name, true); err != nil {
+				glog.Errorf("Failed to add entry: %v to ip set: %s, error: %v", entry, set.Name, err)
+			} else {
+				glog.V(3).Infof("Successfully add entry: %v to ip set: %s", entry, set.Name)
+			}
+		}
+	}
+}
+
+func ensureIPSets(ipSets ...*IPSet) error {
+	for _, set := range ipSets {
+		if err := set.handle.CreateSet(&set.IPSet, true); err != nil {
+			glog.Errorf("Failed to make sure ip set: %v exist, error: %v", set, err)
+			return err
+		}
+	}
+	return nil
+}
+
+// checkMinVersion checks if ipset current version satisfies required min version
+func checkMinVersion(vstring string) bool {
+	version, err := utilversion.ParseGeneric(vstring)
+	if err != nil {
+		glog.Errorf("vstring (%s) is not a valid version string: %v", vstring, err)
+		return false
+	}
+
+	minVersion, err := utilversion.ParseGeneric(MinIPSetCheckVersion)
+	if err != nil {
+		glog.Errorf("MinCheckVersion (%s) is not a valid version string: %v", MinIPSetCheckVersion, err)
+		return false
+	}
+	return !version.LessThan(minVersion)
+}

pkg/proxy/ipvs/ipset_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 ipvs
+
+import (
+	"testing"
+)
+
+func TestCheckIPSetVersion(t *testing.T) {
+	testCases := []struct {
+		vstring string
+		valid   bool
+	}{
+		// version less than "6.0" is not valid.
+		{"4.0", false},
+		{"5.1", false},
+		{"5.1.2", false},
+		// "7" is not a valid version string.
+		{"7", false},
+		{"6.0", true},
+		{"6.1", true},
+		{"6.19", true},
+		{"7.0", true},
+		{"8.1.2", true},
+		{"9.3.4.0", true},
+		{"total junk", false},
+	}
+
+	for i := range testCases {
+		valid := checkMinVersion(testCases[i].vstring)
+		if testCases[i].valid != valid {
+			t.Errorf("Expected result: %v, Got result: %v", testCases[i].valid, valid)
+		}
+	}
+}

pkg/proxy/ipvs/proxier_test.go

@@ -33,6 +33,8 @@ import (
 	"k8s.io/apimachinery/pkg/util/sets"
 	netlinktest "k8s.io/kubernetes/pkg/proxy/ipvs/testing"
 	proxyutil "k8s.io/kubernetes/pkg/proxy/util"
+	utilipset "k8s.io/kubernetes/pkg/util/ipset"
+	ipsettest "k8s.io/kubernetes/pkg/util/ipset/testing"
 	utiliptables "k8s.io/kubernetes/pkg/util/iptables"
 	iptablestest "k8s.io/kubernetes/pkg/util/iptables/testing"
 	utilipvs "k8s.io/kubernetes/pkg/util/ipvs"
@@ -85,7 +87,7 @@ func (fake *fakeHealthChecker) SyncEndpoints(newEndpoints map[types.NamespacedNa
 	return nil
 }
 
-func NewFakeProxier(ipt utiliptables.Interface, ipvs utilipvs.Interface, nodeIPs []net.IP) *Proxier {
+func NewFakeProxier(ipt utiliptables.Interface, ipvs utilipvs.Interface, ipset utilipset.Interface, nodeIPs []net.IP) *Proxier {
 	fcmd := fakeexec.FakeCmd{
 		CombinedOutputScript: []fakeexec.FakeCombinedOutputAction{
 			func() ([]byte, error) { return []byte("dummy device have been created"), nil },
@@ -98,24 +100,34 @@ func NewFakeProxier(ipt utiliptables.Interface, ipvs utilipvs.Interface, nodeIPs
 		LookPathFunc: func(cmd string) (string, error) { return cmd, nil },
 	}
 	return &Proxier{
-		exec:             fexec,
-		serviceMap:       make(proxyServiceMap),
-		serviceChanges:   newServiceChangeMap(),
-		endpointsMap:     make(proxyEndpointsMap),
-		endpointsChanges: newEndpointsChangeMap(testHostname),
-		iptables:         ipt,
-		ipvs:             ipvs,
-		clusterCIDR:      "10.0.0.0/24",
-		hostname:         testHostname,
-		portsMap:         make(map[proxyutil.LocalPort]proxyutil.Closeable),
-		portMapper:       &fakePortOpener{[]*proxyutil.LocalPort{}},
-		healthChecker:    newFakeHealthChecker(),
-		ipvsScheduler:    DefaultScheduler,
-		ipGetter:         &fakeIPGetter{nodeIPs: nodeIPs},
-		iptablesData:     bytes.NewBuffer(nil),
-		natChains:        bytes.NewBuffer(nil),
-		natRules:         bytes.NewBuffer(nil),
-		netlinkHandle:    netlinktest.NewFakeNetlinkHandle(),
+		exec:               fexec,
+		serviceMap:         make(proxyServiceMap),
+		serviceChanges:     newServiceChangeMap(),
+		endpointsMap:       make(proxyEndpointsMap),
+		endpointsChanges:   newEndpointsChangeMap(testHostname),
+		iptables:           ipt,
+		ipvs:               ipvs,
+		ipset:              ipset,
+		clusterCIDR:        "10.0.0.0/24",
+		hostname:           testHostname,
+		portsMap:           make(map[proxyutil.LocalPort]proxyutil.Closeable),
+		portMapper:         &fakePortOpener{[]*proxyutil.LocalPort{}},
+		healthChecker:      newFakeHealthChecker(),
+		ipvsScheduler:      DefaultScheduler,
+		ipGetter:           &fakeIPGetter{nodeIPs: nodeIPs},
+		iptablesData:       bytes.NewBuffer(nil),
+		natChains:          bytes.NewBuffer(nil),
+		natRules:           bytes.NewBuffer(nil),
+		netlinkHandle:      netlinktest.NewFakeNetlinkHandle(),
+		loopbackSet:        NewIPSet(ipset, KubeLoopBackIPSet, utilipset.HashIPPortIP, false),
+		clusterIPSet:       NewIPSet(ipset, KubeClusterIPSet, utilipset.HashIPPort, false),
+		externalIPSet:      NewIPSet(ipset, KubeExternalIPSet, utilipset.HashIPPort, false),
+		lbIngressSet:       NewIPSet(ipset, KubeLoadBalancerSet, utilipset.HashIPPort, false),
+		lbMasqSet:          NewIPSet(ipset, KubeLoadBalancerMasqSet, utilipset.HashIPPort, false),
+		lbWhiteListIPSet:   NewIPSet(ipset, KubeLoadBalancerSourceIPSet, utilipset.HashIPPortIP, false),
+		lbWhiteListCIDRSet: NewIPSet(ipset, KubeLoadBalancerSourceCIDRSet, utilipset.HashIPPortNet, false),
+		nodePortSetTCP:     NewIPSet(ipset, KubeNodePortSetTCP, utilipset.BitmapPort, false),
+		nodePortSetUDP:     NewIPSet(ipset, KubeNodePortSetUDP, utilipset.BitmapPort, false),
 	}
 }
 
@@ -171,10 +183,11 @@ func makeTestEndpoints(namespace, name string, eptFunc func(*api.Endpoints)) *ap
 func TestNodePort(t *testing.T) {
 	ipt := iptablestest.NewFake()
 	ipvs := ipvstest.NewFake()
+	ipset := ipsettest.NewFake()
 	nodeIPv4 := net.ParseIP("100.101.102.103")
 	nodeIPv6 := net.ParseIP("2001:db8::1:1")
 	nodeIPs := sets.NewString(nodeIPv4.String(), nodeIPv6.String())
-	fp := NewFakeProxier(ipt, ipvs, []net.IP{nodeIPv4, nodeIPv6})
+	fp := NewFakeProxier(ipt, ipvs, ipset, []net.IP{nodeIPv4, nodeIPv6})
 	svcIP := "10.20.30.41"
 	svcPort := 80
 	svcNodePort := 3001
@@ -248,8 +261,9 @@ func TestNodePort(t *testing.T) {
 func TestNodePortNoEndpoint(t *testing.T) {
 	ipt := iptablestest.NewFake()
 	ipvs := ipvstest.NewFake()
+	ipset := ipsettest.NewFake()
 	nodeIP := net.ParseIP("100.101.102.103")
-	fp := NewFakeProxier(ipt, ipvs, []net.IP{nodeIP})
+	fp := NewFakeProxier(ipt, ipvs, ipset, []net.IP{nodeIP})
 	svcIP := "10.20.30.41"
 	svcPort := 80
 	svcNodePort := 3001
@@ -301,7 +315,8 @@ func TestNodePortNoEndpoint(t *testing.T) {
 func TestClusterIPNoEndpoint(t *testing.T) {
 	ipt := iptablestest.NewFake()
 	ipvs := ipvstest.NewFake()
-	fp := NewFakeProxier(ipt, ipvs, nil)
+	ipset := ipsettest.NewFake()
+	fp := NewFakeProxier(ipt, ipvs, ipset, nil)
 	svcIP := "10.20.30.41"
 	svcPort := 80
 	svcPortName := proxy.ServicePortName{
@@ -344,7 +359,8 @@ func TestClusterIPNoEndpoint(t *testing.T) {
 func TestClusterIP(t *testing.T) {
 	ipt := iptablestest.NewFake()
 	ipvs := ipvstest.NewFake()
-	fp := NewFakeProxier(ipt, ipvs, nil)
+	ipset := ipsettest.NewFake()
+	fp := NewFakeProxier(ipt, ipvs, ipset, nil)
 
 	svcIPv4 := "10.20.30.41"
 	svcPortV4 := 80
@@ -450,7 +466,8 @@ func TestClusterIP(t *testing.T) {
 func TestExternalIPsNoEndpoint(t *testing.T) {
 	ipt := iptablestest.NewFake()
 	ipvs := ipvstest.NewFake()
-	fp := NewFakeProxier(ipt, ipvs, nil)
+	ipset := ipsettest.NewFake()
+	fp := NewFakeProxier(ipt, ipvs, ipset, nil)
 	svcIP := "10.20.30.41"
 	svcPort := 80
 	svcExternalIPs := "50.60.70.81"
@@ -504,7 +521,8 @@ func TestExternalIPsNoEndpoint(t *testing.T) {
 func TestExternalIPs(t *testing.T) {
 	ipt := iptablestest.NewFake()
 	ipvs := ipvstest.NewFake()
-	fp := NewFakeProxier(ipt, ipvs, nil)
+	ipset := ipsettest.NewFake()
+	fp := NewFakeProxier(ipt, ipvs, ipset, nil)
 	svcIP := "10.20.30.41"
 	svcPort := 80
 	svcExternalIPs := sets.NewString("50.60.70.81", "2012::51")
@@ -573,7 +591,8 @@ func TestExternalIPs(t *testing.T) {
 func TestLoadBalancer(t *testing.T) {
 	ipt := iptablestest.NewFake()
 	ipvs := ipvstest.NewFake()
-	fp := NewFakeProxier(ipt, ipvs, nil)
+	ipset := ipsettest.NewFake()
+	fp := NewFakeProxier(ipt, ipvs, ipset, nil)
 	svcIP := "10.20.30.41"
 	svcPort := 80
 	svcNodePort := 3001
@@ -624,8 +643,9 @@ func strPtr(s string) *string {
 func TestOnlyLocalNodePorts(t *testing.T) {
 	ipt := iptablestest.NewFake()
 	ipvs := ipvstest.NewFake()
+	ipset := ipsettest.NewFake()
 	nodeIP := net.ParseIP("100.101.102.103")
-	fp := NewFakeProxier(ipt, ipvs, []net.IP{nodeIP})
+	fp := NewFakeProxier(ipt, ipvs, ipset, []net.IP{nodeIP})
 	svcIP := "10.20.30.41"
 	svcPort := 80
 	svcNodePort := 3001
@@ -705,7 +725,8 @@ func TestOnlyLocalNodePorts(t *testing.T) {
 func TestOnlyLocalLoadBalancing(t *testing.T) {
 	ipt := iptablestest.NewFake()
 	ipvs := ipvstest.NewFake()
-	fp := NewFakeProxier(ipt, ipvs, nil)
+	ipset := ipsettest.NewFake()
+	fp := NewFakeProxier(ipt, ipvs, ipset, nil)
 	svcIP := "10.20.30.41"
 	svcPort := 80
 	svcNodePort := 3001
@@ -769,7 +790,8 @@ func addTestPort(array []api.ServicePort, name string, protocol api.Protocol, po
 func TestBuildServiceMapAddRemove(t *testing.T) {
 	ipt := iptablestest.NewFake()
 	ipvs := ipvstest.NewFake()
-	fp := NewFakeProxier(ipt, ipvs, nil)
+	ipset := ipsettest.NewFake()
+	fp := NewFakeProxier(ipt, ipvs, ipset, nil)
 
 	services := []*api.Service{
 		makeTestService("somewhere-else", "cluster-ip", func(svc *api.Service) {
@@ -874,7 +896,8 @@ func TestBuildServiceMapAddRemove(t *testing.T) {
 func TestBuildServiceMapServiceHeadless(t *testing.T) {
 	ipt := iptablestest.NewFake()
 	ipvs := ipvstest.NewFake()
-	fp := NewFakeProxier(ipt, ipvs, nil)
+	ipset := ipsettest.NewFake()
+	fp := NewFakeProxier(ipt, ipvs, ipset, nil)
 
 	makeServiceMap(fp,
 		makeTestService("somewhere-else", "headless", func(svc *api.Service) {
@@ -907,7 +930,8 @@ func TestBuildServiceMapServiceHeadless(t *testing.T) {
 func TestBuildServiceMapServiceTypeExternalName(t *testing.T) {
 	ipt := iptablestest.NewFake()
 	ipvs := ipvstest.NewFake()
-	fp := NewFakeProxier(ipt, ipvs, nil)
+	ipset := ipsettest.NewFake()
+	fp := NewFakeProxier(ipt, ipvs, ipset, nil)
 
 	makeServiceMap(fp,
 		makeTestService("somewhere-else", "external-name", func(svc *api.Service) {
@@ -934,7 +958,8 @@ func TestBuildServiceMapServiceTypeExternalName(t *testing.T) {
 func TestBuildServiceMapServiceUpdate(t *testing.T) {
 	ipt := iptablestest.NewFake()
 	ipvs := ipvstest.NewFake()
-	fp := NewFakeProxier(ipt, ipvs, nil)
+	ipset := ipsettest.NewFake()
+	fp := NewFakeProxier(ipt, ipvs, ipset, nil)
 
 	servicev1 := makeTestService("somewhere", "some-service", func(svc *api.Service) {
 		svc.Spec.Type = api.ServiceTypeClusterIP
@@ -1016,8 +1041,9 @@ func TestBuildServiceMapServiceUpdate(t *testing.T) {
 func TestSessionAffinity(t *testing.T) {
 	ipt := iptablestest.NewFake()
 	ipvs := ipvstest.NewFake()
+	ipset := ipsettest.NewFake()
 	nodeIP := net.ParseIP("100.101.102.103")
-	fp := NewFakeProxier(ipt, ipvs, []net.IP{nodeIP})
+	fp := NewFakeProxier(ipt, ipvs, ipset, []net.IP{nodeIP})
 	svcIP := "10.20.30.41"
 	svcPort := 80
 	svcNodePort := 3001
@@ -1879,7 +1905,8 @@ func Test_updateEndpointsMap(t *testing.T) {
 	for tci, tc := range testCases {
 		ipt := iptablestest.NewFake()
 		ipvs := ipvstest.NewFake()
-		fp := NewFakeProxier(ipt, ipvs, nil)
+		ipset := ipsettest.NewFake()
+		fp := NewFakeProxier(ipt, ipvs, ipset, nil)
 		fp.hostname = nodeName
 
 		// First check that after adding all previous versions of endpoints,

pkg/proxy/util/endpoints.go

@@ -19,6 +19,7 @@ package util
 import (
 	"fmt"
 	"net"
+	"strconv"
 
 	"github.com/golang/glog"
 )
@@ -46,6 +47,21 @@ func IPPart(s string) string {
 	return ""
 }
 
+func PortPart(s string) (int, error) {
+	// Must be IP:port
+	_, port, err := net.SplitHostPort(s)
+	if err != nil {
+		glog.Errorf("Error parsing '%s': %v", s, err)
+		return -1, err
+	}
+	portNumber, err := strconv.Atoi(port)
+	if err != nil {
+		glog.Errorf("Error parsing '%s': %v", port, err)
+		return -1, err
+	}
+	return portNumber, nil
+}
+
 // ToCIDR returns a host address of the form <ip-address>/32 for
 // IPv4 and <ip-address>/128 for IPv6
 func ToCIDR(ip net.IP) string {

pkg/util/BUILD

@@ -27,6 +27,7 @@ filegroup(
         "//pkg/util/interrupt:all-srcs",
         "//pkg/util/io:all-srcs",
         "//pkg/util/ipconfig:all-srcs",
+        "//pkg/util/ipset:all-srcs",
         "//pkg/util/iptables:all-srcs",
         "//pkg/util/ipvs:all-srcs",
         "//pkg/util/keymutex:all-srcs",

pkg/util/ipset/BUILD

+load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "go_default_library",
+    srcs = [
+        "ipset.go",
+        "types.go",
+    ],
+    importpath = "k8s.io/kubernetes/pkg/util/ipset",
+    visibility = ["//visibility:public"],
+    deps = ["//vendor/k8s.io/utils/exec:go_default_library"],
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = ["ipset_test.go"],
+    importpath = "k8s.io/kubernetes/pkg/util/ipset",
+    library = ":go_default_library",
+    deps = [
+        "//vendor/k8s.io/apimachinery/pkg/util/sets:go_default_library",
+        "//vendor/k8s.io/utils/exec:go_default_library",
+        "//vendor/k8s.io/utils/exec/testing:go_default_library",
+    ],
+)
+
+filegroup(
+    name = "package-srcs",
+    srcs = glob(["**"]),
+    tags = ["automanaged"],
+    visibility = ["//visibility:private"],
+)
+
+filegroup(
+    name = "all-srcs",
+    srcs = [
+        ":package-srcs",
+        "//pkg/util/ipset/testing:all-srcs",
+    ],
+    tags = ["automanaged"],
+    visibility = ["//visibility:public"],
+)

pkg/util/ipset/testing/BUILD

+load("@io_bazel_rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["fake.go"],
+    importpath = "k8s.io/kubernetes/pkg/util/ipset/testing",
+    visibility = ["//visibility:public"],
+    deps = ["//pkg/util/ipset:go_default_library"],
+)
+
+filegroup(
+    name = "package-srcs",
+    srcs = glob(["**"]),
+    tags = ["automanaged"],
+    visibility = ["//visibility:private"],
+)
+
+filegroup(
+    name = "all-srcs",
+    srcs = [":package-srcs"],
+    tags = ["automanaged"],
+    visibility = ["//visibility:public"],
+)

pkg/util/ipset/testing/fake.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 testing
+
+import (
+	"k8s.io/kubernetes/pkg/util/ipset"
+)
+
+// FakeIPSet is a no-op implementation of ipset Interface
+type FakeIPSet struct {
+	Lines []byte
+}
+
+// NewFake create a new fake ipset interface.
+func NewFake() *FakeIPSet {
+	return &FakeIPSet{}
+}
+
+// GetVersion is part of interface.
+func (*FakeIPSet) GetVersion() (string, error) {
+	return "0.0", nil
+}
+
+// FlushSet is part of interface.
+func (*FakeIPSet) FlushSet(set string) error {
+	return nil
+}
+
+// DestroySet is part of interface.
+func (*FakeIPSet) DestroySet(set string) error {
+	return nil
+}
+
+// DestroyAllSets is part of interface.
+func (*FakeIPSet) DestroyAllSets() error {
+	return nil
+}
+
+// CreateSet is part of interface.
+func (*FakeIPSet) CreateSet(set *ipset.IPSet, ignoreExistErr bool) error {
+	return nil
+}
+
+// AddEntry is part of interface.
+func (*FakeIPSet) AddEntry(entry string, set string, ignoreExistErr bool) error {
+	return nil
+}
+
+// DelEntry is part of interface.
+func (*FakeIPSet) DelEntry(entry string, set string) error {
+	return nil
+}
+
+// TestEntry is part of interface.
+func (*FakeIPSet) TestEntry(entry string, set string) (bool, error) {
+	return true, nil
+}
+
+// ListEntries is part of interface.
+func (*FakeIPSet) ListEntries(set string) ([]string, error) {
+	return nil, nil
+}
+
+// ListSets is part of interface.
+func (*FakeIPSet) ListSets() ([]string, error) {
+	return nil, nil
+}
+
+var _ = ipset.Interface(&FakeIPSet{})

pkg/util/ipset/types.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 ipset
+
+// Type represents the ipset type
+type Type string
+
+const (
+	// HashIPPort represents the `hash:ip,port` type ipset.  The hash:ip,port is similar to hash:ip but
+	// you can store IP address and protocol-port pairs in it.  TCP, SCTP, UDP, UDPLITE, ICMP and ICMPv6 are supported
+	// with port numbers/ICMP(v6) types and other protocol numbers without port information.
+	HashIPPort Type = "hash:ip,port"
+	// HashIPPortIP represents the `hash:ip,port,ip` type ipset.  The hash:ip,port,ip set type uses a hash to store
+	// IP address, port number and a second IP address triples.  The port number is interpreted together with a
+	// protocol (default TCP) and zero protocol number cannot be used.
+	HashIPPortIP Type = "hash:ip,port,ip"
+	// HashIPPortNet represents the `hash:ip,port,net` type ipset.  The hash:ip,port,net set type uses a hash to store IP address, port number and IP network address triples.  The port
+	// number is interpreted together with a protocol (default TCP) and zero protocol number cannot be used.   Network address
+	// with zero prefix size cannot be stored either.
+	HashIPPortNet Type = "hash:ip,port,net"
+	// BitmapPort represents the `bitmap:port` type ipset.  The bitmap:port set type uses a memory range, where each bit
+	// represents one TCP/UDP port.  A bitmap:port type of set can store up to 65535 ports.
+	BitmapPort Type = "bitmap:port"
+)
+
+// DefaultPortRange defines the default bitmap:port valid port range.
+const DefaultPortRange string = "0-65535"
+
+const (
+	// ProtocolFamilyIPV4 represents IPv4 protocol.
+	ProtocolFamilyIPV4 = "inet"
+	// ProtocolFamilyIPV6 represents IPv6 protocol.
+	ProtocolFamilyIPV6 = "inet6"
+	// ProtocolTCP represents TCP protocol.
+	ProtocolTCP = "tcp"
+	// ProtocolUDP represents UDP protocol.
+	ProtocolUDP = "udp"
+)
+
+// ValidIPSetTypes defines the supported ip set type.
+var ValidIPSetTypes = []Type{
+	HashIPPort,
+	HashIPPortIP,
+	BitmapPort,
+	HashIPPortNet,
+}
+
+// IsValidIPSetType checks if the given ipset type is valid.
+func IsValidIPSetType(set Type) bool {
+	for _, valid := range ValidIPSetTypes {
+		if set == valid {
+			return true
+		}
+	}
+	return false
+}