Merge pull request #56685 from m1093782566/fix-nodeport

Automatic merge from submit-queue (batch tested with PRs 54379, 56593, 56685, 54174, 57309). If you want to cherry-pick this change to another branch, please follow the instructions <a href="https://github.com/kubernetes/community/blob/master/contributors/devel/cherry-picks.md">here</a>.

Fix ipvs proxier nodeport

**What this PR does / why we need it**:

Fix ipvs proxier nodeport.

**Which issue(s) this PR fixes**:
Fixes #55923 

**Special notes for your reviewer**:

We bump the netwlink version in the 1st commit because:

* We call `netlink.RouteListFiltered()` to filter LOCAL type addresses from kernel route table. `netlink.RouteListFiltered()` exists in newer-version netlink package


* newer-version netlink package migrate `syscall` to `golang.org/x/sys/unix`, k8s cross-build can benefit from it as well. The Go doc for syscall says:

> NOTE: This package is locked down. Code outside the standard Go repository should be migrated to use the corresponding package in the golang.org/x/sys repository. That is also where updates required by new systems or versions should be applied. See https://golang.org/s/go1.4-syscall for more information.



**Release note**:

```release-note
Fix ipvs proxier nodeport eth* assumption
```

pkg/proxy/ipvs/BUILD

@@ -67,7 +67,6 @@ go_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",
-        "//vendor/k8s.io/apimachinery/pkg/util/runtime:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/sets:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/wait:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/util/feature:go_default_library",

pkg/proxy/ipvs/netlink.go

@@ -16,14 +16,21 @@ limitations under the License.
 
 package ipvs
 
+import (
+	"k8s.io/apimachinery/pkg/util/sets"
+)
+
 // NetLinkHandle for revoke netlink interface
 type NetLinkHandle interface {
-	// EnsureAddressBind checks if address is bound to the interface and, if not, binds it. If the address is already bound, return true.
+	// EnsureAddressBind checks if address is bound to the interface and, if not, binds it.  If the address is already bound, return true.
 	EnsureAddressBind(address, devName string) (exist bool, err error)
 	// UnbindAddress unbind address from the interface
 	UnbindAddress(address, devName string) error
-	// EnsureDummyDevice checks if dummy device is exist and, if not, create one. If the dummy device is already exist, return true.
+	// EnsureDummyDevice checks if dummy device is exist and, if not, create one.  If the dummy device is already exist, return true.
 	EnsureDummyDevice(devName string) (exist bool, err error)
 	// DeleteDummyDevice deletes the given dummy device by name.
 	DeleteDummyDevice(devName string) error
+	// GetLocalAddresses returns all unique local type IP addresses based on filter device interface.  If filter device is not given,
+	// it will list all unique local type addresses.
+	GetLocalAddresses(filterDev string) (sets.String, error)
 }

pkg/proxy/ipvs/netlink_linux.go

@@ -22,6 +22,14 @@ import (
 	"fmt"
 	"net"
 	"syscall"
+	// TODO: replace syscall with golang.org/x/sys/unix?
+	// The Go doc for syscall says:
+	// NOTE: This package is locked down.
+	// Code outside the standard Go repository should be migrated to use the corresponding package in the golang.org/x/sys repository.
+	// That is also where updates required by new systems or versions should be applied.
+	// See https://golang.org/s/go1.4-syscall for more information.
+
+	"k8s.io/apimachinery/pkg/util/sets"
 
 	"github.com/vishvananda/netlink"
 )
@@ -30,7 +38,7 @@ type netlinkHandle struct {
 	netlink.Handle
 }
 
-// NewNetLinkHandle will crate a new netlinkHandle
+// NewNetLinkHandle will crate a new NetLinkHandle
 func NewNetLinkHandle() NetLinkHandle {
 	return &netlinkHandle{netlink.Handle{}}
 }
@@ -96,3 +104,59 @@ func (h *netlinkHandle) DeleteDummyDevice(devName string) error {
 	}
 	return h.LinkDel(dummy)
 }
+
+// GetLocalAddresses lists all LOCAL type IP addresses from host based on filter device.
+// If filter device is not specified, it's equivalent to exec:
+// $ ip route show table local type local proto kernel
+// 10.0.0.1 dev kube-ipvs0  scope host  src 10.0.0.1
+// 10.0.0.10 dev kube-ipvs0  scope host  src 10.0.0.10
+// 10.0.0.252 dev kube-ipvs0  scope host  src 10.0.0.252
+// 100.106.89.164 dev eth0  scope host  src 100.106.89.164
+// 127.0.0.0/8 dev lo  scope host  src 127.0.0.1
+// 127.0.0.1 dev lo  scope host  src 127.0.0.1
+// 172.17.0.1 dev docker0  scope host  src 172.17.0.1
+// 192.168.122.1 dev virbr0  scope host  src 192.168.122.1
+// Then cut the unique src IP fields,
+// --> result set: [10.0.0.1, 10.0.0.10, 10.0.0.252, 100.106.89.164, 127.0.0.1, 192.168.122.1]
+
+// If filter device is specified, it's equivalent to exec:
+// $ ip route show table local type local proto kernel dev kube-ipvs0
+// 10.0.0.1  scope host  src 10.0.0.1
+// 10.0.0.10  scope host  src 10.0.0.10
+// Then cut the unique src IP fields,
+// --> result set: [10.0.0.1, 10.0.0.10]
+func (h *netlinkHandle) GetLocalAddresses(filterDev string) (sets.String, error) {
+	linkIndex := -1
+	if len(filterDev) != 0 {
+		link, err := h.LinkByName(filterDev)
+		if err != nil {
+			return nil, fmt.Errorf("error get filter device %s, err: %v", filterDev, err)
+		}
+		linkIndex = link.Attrs().Index
+	}
+
+	routeFilter := &netlink.Route{
+		Table:    syscall.RT_TABLE_LOCAL,
+		Type:     syscall.RTN_LOCAL,
+		Protocol: syscall.RTPROT_KERNEL,
+	}
+	filterMask := netlink.RT_FILTER_TABLE | netlink.RT_FILTER_TYPE | netlink.RT_FILTER_PROTOCOL
+
+	// find filter device
+	if linkIndex != -1 {
+		routeFilter.LinkIndex = linkIndex
+		filterMask |= netlink.RT_FILTER_OIF
+	}
+
+	routes, err := h.RouteListFiltered(netlink.FAMILY_ALL, routeFilter, filterMask)
+	if err != nil {
+		return nil, fmt.Errorf("error list route table, err: %v", err)
+	}
+	res := sets.NewString()
+	for _, route := range routes {
+		if route.Src != nil {
+			res.Insert(route.Src.String())
+		}
+	}
+	return res, nil
+}

pkg/proxy/ipvs/netlink_unsupported.go

@@ -20,6 +20,8 @@ package ipvs
 
 import (
 	"fmt"
+
+	"k8s.io/apimachinery/pkg/util/sets"
 )
 
 type emptyHandle struct {
@@ -49,3 +51,8 @@ func (h *emptyHandle) EnsureDummyDevice(devName string) (bool, error) {
 func (h *emptyHandle) DeleteDummyDevice(devName string) error {
 	return fmt.Errorf("netlink is not supported in this platform")
 }
+
+// GetLocalAddresses is part of interface.
+func (h *emptyHandle) GetLocalAddresses(filterDev string) (sets.String, error) {
+	return nil, fmt.Errorf("netlink is not supported in this platform")
+}

pkg/proxy/ipvs/proxier.go

@@ -35,7 +35,6 @@ import (
 
 	clientv1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/types"
-	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/wait"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
@@ -171,36 +170,57 @@ type IPGetter interface {
 	NodeIPs() ([]net.IP, error)
 }
 
-type realIPGetter struct{}
+// realIPGetter is a real NodeIP handler, it implements IPGetter.
+type realIPGetter struct {
+	// nl is a handle for revoking netlink interface
+	nl NetLinkHandle
+}
 
+// NodeIPs returns all LOCAL type IP addresses from host which are taken as the Node IPs of NodePort service.
+// Firstly, it will list source IP exists in local route table with `kernel` protocol type.  For example,
+// $ ip route show table local type local proto kernel
+// 10.0.0.1 dev kube-ipvs0  scope host  src 10.0.0.1
+// 10.0.0.10 dev kube-ipvs0  scope host  src 10.0.0.10
+// 10.0.0.252 dev kube-ipvs0  scope host  src 10.0.0.252
+// 100.106.89.164 dev eth0  scope host  src 100.106.89.164
+// 127.0.0.0/8 dev lo  scope host  src 127.0.0.1
+// 127.0.0.1 dev lo  scope host  src 127.0.0.1
+// 172.17.0.1 dev docker0  scope host  src 172.17.0.1
+// 192.168.122.1 dev virbr0  scope host  src 192.168.122.1
+// Then cut the unique src IP fields,
+// --> result set1: [10.0.0.1, 10.0.0.10, 10.0.0.252, 100.106.89.164, 127.0.0.1, 192.168.122.1]
+
+// NOTE: For cases where an LB acts as a VIP (e.g. Google cloud), the VIP IP is considered LOCAL, but the protocol
+// of the entry is 66, e.g. `10.128.0.6 dev ens4  proto 66  scope host`.  Therefore, the rule mentioned above will
+// filter these entries out.
+
+// Secondly, as we bind Cluster IPs to the dummy interface in IPVS proxier, we need to filter the them out so that
+// we can eventually get the Node IPs.  Fortunately, the dummy interface created by IPVS proxier is known as `kube-ipvs0`,
+// so we just need to specify the `dev kube-ipvs0` argument in ip route command, for example,
+// $ ip route show table local type local proto kernel dev kube-ipvs0
+// 10.0.0.1  scope host  src 10.0.0.1
+// 10.0.0.10  scope host  src 10.0.0.10
+// Then cut the unique src IP fields,
+// --> result set2: [10.0.0.1, 10.0.0.10]
+
+// Finally, Node IP set = set1 - set2
 func (r *realIPGetter) NodeIPs() (ips []net.IP, err error) {
-	interfaces, err := net.Interfaces()
+	// Pass in empty filter device name for list all LOCAL type addresses.
+	allAddress, err := r.nl.GetLocalAddresses("")
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("error listing LOCAL type addresses from host, error: %v", err)
 	}
-	for i := range interfaces {
-		name := interfaces[i].Name
-		// We assume node ip bind to eth{x}
-		if !strings.HasPrefix(name, "eth") {
-			continue
-		}
-		intf, err := net.InterfaceByName(name)
-		if err != nil {
-			utilruntime.HandleError(fmt.Errorf("Failed to get interface by name: %s, error: %v", name, err))
-			continue
-		}
-		addrs, err := intf.Addrs()
-		if err != nil {
-			utilruntime.HandleError(fmt.Errorf("Failed to get addresses from interface: %s, error: %v", name, err))
-			continue
-		}
-		for _, a := range addrs {
-			if ipnet, ok := a.(*net.IPNet); ok {
-				ips = append(ips, ipnet.IP)
-			}
-		}
+	dummyAddress, err := r.nl.GetLocalAddresses(DefaultDummyDevice)
+	if err != nil {
+		return nil, fmt.Errorf("error listing LOCAL type addresses from device: %s, error: %v", DefaultDummyDevice, err)
 	}
-	return
+	// exclude ip address from dummy interface created by IPVS proxier - they are all Cluster IPs.
+	nodeAddress := allAddress.Difference(dummyAddress)
+	// translate ip string to IP
+	for _, ipStr := range nodeAddress.UnsortedList() {
+		ips = append(ips, net.ParseIP(ipStr))
+	}
+	return ips, nil
 }
 
 // Proxier implements ProxyProvider
@@ -294,7 +314,7 @@ func NewProxier(ipt utiliptables.Interface,
 		healthzServer:      healthzServer,
 		ipvs:               ipvs,
 		ipvsScheduler:      scheduler,
-		ipGetter:           &realIPGetter{},
+		ipGetter:           &realIPGetter{nl: NewNetLinkHandle()},
 		iptablesData:       bytes.NewBuffer(nil),
 		natChains:          bytes.NewBuffer(nil),
 		natRules:           bytes.NewBuffer(nil),

pkg/proxy/ipvs/proxier_test.go

@@ -264,6 +264,83 @@ func TestCanUseIPVSProxier(t *testing.T) {
 	}
 }
 
+func TestGetNodeIPs(t *testing.T) {
+	testCases := []struct {
+		devAddresses map[string][]string
+		expectIPs    []string
+	}{
+		// case 0
+		{
+			devAddresses: map[string][]string{"eth0": {"1.2.3.4"}, "lo": {"127.0.0.1"}},
+			expectIPs:    []string{"1.2.3.4", "127.0.0.1"},
+		},
+		// case 1
+		{
+			devAddresses: map[string][]string{"lo": {"127.0.0.1"}},
+			expectIPs:    []string{"127.0.0.1"},
+		},
+		// case 2
+		{
+			devAddresses: map[string][]string{},
+			expectIPs:    []string{},
+		},
+		// case 3
+		{
+			devAddresses: map[string][]string{"encap0": {"10.20.30.40"}, "lo": {"127.0.0.1"}, "docker0": {"172.17.0.1"}},
+			expectIPs:    []string{"10.20.30.40", "127.0.0.1", "172.17.0.1"},
+		},
+		// case 4
+		{
+			devAddresses: map[string][]string{"encaps9": {"10.20.30.40"}, "lo": {"127.0.0.1"}, "encap7": {"10.20.30.31"}},
+			expectIPs:    []string{"10.20.30.40", "127.0.0.1", "10.20.30.31"},
+		},
+		// case 5
+		{
+			devAddresses: map[string][]string{"kube-ipvs0": {"1.2.3.4"}, "lo": {"127.0.0.1"}, "encap7": {"10.20.30.31"}},
+			expectIPs:    []string{"127.0.0.1", "10.20.30.31"},
+		},
+		// case 6
+		{
+			devAddresses: map[string][]string{"kube-ipvs0": {"1.2.3.4", "2.3.4.5"}, "lo": {"127.0.0.1"}},
+			expectIPs:    []string{"127.0.0.1"},
+		},
+		// case 7
+		{
+			devAddresses: map[string][]string{"kube-ipvs0": {"1.2.3.4", "2.3.4.5"}},
+			expectIPs:    []string{},
+		},
+		// case 8
+		{
+			devAddresses: map[string][]string{"kube-ipvs0": {"1.2.3.4", "2.3.4.5"}, "eth5": {"3.4.5.6"}, "lo": {"127.0.0.1"}},
+			expectIPs:    []string{"127.0.0.1", "3.4.5.6"},
+		},
+		// case 9
+		{
+			devAddresses: map[string][]string{"ipvs0": {"1.2.3.4"}, "lo": {"127.0.0.1"}, "encap7": {"10.20.30.31"}},
+			expectIPs:    []string{"127.0.0.1", "10.20.30.31", "1.2.3.4"},
+		},
+	}
+
+	for i := range testCases {
+		fake := netlinktest.NewFakeNetlinkHandle()
+		for dev, addresses := range testCases[i].devAddresses {
+			fake.SetLocalAddresses(dev, addresses...)
+		}
+		r := realIPGetter{nl: fake}
+		ips, err := r.NodeIPs()
+		if err != nil {
+			t.Errorf("Unexpected error: %v", err)
+		}
+		ipStrs := sets.NewString()
+		for _, ip := range ips {
+			ipStrs.Insert(ip.String())
+		}
+		if !ipStrs.Equal(sets.NewString(testCases[i].expectIPs...)) {
+			t.Errorf("case[%d], unexpected mismatch, expected: %v, got: %v", i, testCases[i].expectIPs, ips)
+		}
+	}
+}
+
 func TestNodePort(t *testing.T) {
 	ipt := iptablestest.NewFake()
 	ipvs := ipvstest.NewFake()

pkg/proxy/ipvs/testing/BUILD

@@ -5,6 +5,7 @@ licenses(["notice"])
 load(
     "@io_bazel_rules_go//go:def.bzl",
     "go_library",
+    "go_test",
 )
 
 go_library(
@@ -12,6 +13,7 @@ go_library(
     srcs = ["fake.go"],
     importpath = "k8s.io/kubernetes/pkg/proxy/ipvs/testing",
     tags = ["automanaged"],
+    deps = ["//vendor/k8s.io/apimachinery/pkg/util/sets:go_default_library"],
 )
 
 filegroup(
@@ -26,3 +28,11 @@ filegroup(
     srcs = [":package-srcs"],
     tags = ["automanaged"],
 )
+
+go_test(
+    name = "go_default_test",
+    srcs = ["fake_test.go"],
+    importpath = "k8s.io/kubernetes/pkg/proxy/ipvs/testing",
+    library = ":go_default_library",
+    deps = ["//vendor/k8s.io/apimachinery/pkg/util/sets:go_default_library"],
+)

pkg/proxy/ipvs/testing/fake.go

@@ -16,21 +16,33 @@ limitations under the License.
 
 package testing
 
-//FakeNetlinkHandle mock implementation of proxy NetlinkHandle
+import (
+	"fmt"
+
+	"k8s.io/apimachinery/pkg/util/sets"
+)
+
+// FakeNetlinkHandle mock implementation of proxy NetlinkHandle
 type FakeNetlinkHandle struct {
+	// localAddresses is a network interface name to all of its IP addresses map, e.g.
+	// eth0 -> [1.2.3.4, 10.20.30.40]
+	localAddresses map[string][]string
 }
 
-//NewFakeNetlinkHandle will create a new FakeNetlinkHandle
+// NewFakeNetlinkHandle will create a new FakeNetlinkHandle
 func NewFakeNetlinkHandle() *FakeNetlinkHandle {
-	return &FakeNetlinkHandle{}
+	fake := &FakeNetlinkHandle{
+		localAddresses: make(map[string][]string),
+	}
+	return fake
 }
 
-//EnsureAddressBind is a mock implementation
+// EnsureAddressBind is a mock implementation
 func (h *FakeNetlinkHandle) EnsureAddressBind(address, devName string) (exist bool, err error) {
 	return false, nil
 }
 
-//UnbindAddress is a mock implementation
+// UnbindAddress is a mock implementation
 func (h *FakeNetlinkHandle) UnbindAddress(address, devName string) error {
 	return nil
 }
@@ -44,3 +56,36 @@ func (h *FakeNetlinkHandle) EnsureDummyDevice(devName string) (bool, error) {
 func (h *FakeNetlinkHandle) DeleteDummyDevice(devName string) error {
 	return nil
 }
+
+// GetLocalAddresses is a mock implementation
+func (h *FakeNetlinkHandle) GetLocalAddresses(filterDev string) (sets.String, error) {
+	res := sets.NewString()
+	if len(filterDev) != 0 {
+		// list all addresses from a given network interface.
+		for _, addr := range h.localAddresses[filterDev] {
+			res.Insert(addr)
+		}
+		return res, nil
+	}
+	// If filterDev is not given, will list all addresses from all available network interface.
+	for linkName := range h.localAddresses {
+		// list all addresses from a given network interface.
+		for _, addr := range h.localAddresses[linkName] {
+			res.Insert(addr)
+		}
+	}
+	return res, nil
+}
+
+// SetLocalAddresses set IP addresses to the given interface device.  It's not part of interface.
+func (h *FakeNetlinkHandle) SetLocalAddresses(dev string, ips ...string) error {
+	if h.localAddresses == nil {
+		h.localAddresses = make(map[string][]string)
+	}
+	if len(dev) == 0 {
+		return fmt.Errorf("device name can't be empty")
+	}
+	h.localAddresses[dev] = make([]string, 0)
+	h.localAddresses[dev] = append(h.localAddresses[dev], ips...)
+	return nil
+}

pkg/proxy/ipvs/testing/fake_test.go

@@ -0,0 +1,49 @@
+/*
+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 (
+	"reflect"
+	"testing"
+
+	"k8s.io/apimachinery/pkg/util/sets"
+)
+
+func TestSetGetLocalAddresses(t *testing.T) {
+	fake := NewFakeNetlinkHandle()
+	fake.SetLocalAddresses("eth0", "1.2.3.4")
+	expected := sets.NewString("1.2.3.4")
+	addr, _ := fake.GetLocalAddresses("eth0")
+	if !reflect.DeepEqual(expected, addr) {
+		t.Errorf("Unexpected mismatch, expected: %v, got: %v", expected, addr)
+	}
+	list, _ := fake.GetLocalAddresses("")
+	if !reflect.DeepEqual(expected, list) {
+		t.Errorf("Unexpected mismatch, expected: %v, got: %v", expected, list)
+	}
+	fake.SetLocalAddresses("lo", "127.0.0.1")
+	expected = sets.NewString("127.0.0.1")
+	addr, _ = fake.GetLocalAddresses("lo")
+	if !reflect.DeepEqual(expected, addr) {
+		t.Errorf("Unexpected mismatch, expected: %v, got: %v", expected, addr)
+	}
+	list, _ = fake.GetLocalAddresses("")
+	expected = sets.NewString("1.2.3.4", "127.0.0.1")
+	if !reflect.DeepEqual(expected, list) {
+		t.Errorf("Unexpected mismatch, expected: %v, got: %v", expected, list)
+	}
+}