Merge pull request #58052 from m1093782566/nodeip-config

Automatic merge from submit-queue (batch tested with PRs 60430, 60115, 58052, 60355, 60116). 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>.

Make nodeport ip configurable

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

By default, kube-proxy accepts everything from NodePort without any filter. It can be a problem for nodes which has both public and private NICs, and people only want to provide a service in private network and avoid exposing any internal service on the public IPs.

This PR makes nodeport ip configurable.

**Which issue(s) this PR fixes**:
Closes: #21070

**Special notes for your reviewer**:

Design proposal see: https://github.com/kubernetes/community/pull/1547

Issue in feature repo: https://github.com/kubernetes/features/issues/539

**Release note**:

```release-note
Make NodePort IP addresses configurable
```
This commit is contained in:
Kubernetes Submit Queue 2018-02-27 09:38:44 -08:00 committed by GitHub
commit 42378eab40
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
25 changed files with 686 additions and 40 deletions

View File

@ -169,6 +169,8 @@ func (o *Options) AddFlags(fs *pflag.FlagSet) {
"NAT timeout for TCP connections in the CLOSE_WAIT state")
fs.BoolVar(&o.config.EnableProfiling, "profiling", o.config.EnableProfiling, "If true enables profiling via web interface on /debug/pprof handler.")
fs.StringVar(&o.config.IPVS.Scheduler, "ipvs-scheduler", o.config.IPVS.Scheduler, "The ipvs scheduler type when proxy mode is ipvs")
fs.StringSliceVar(&o.config.NodePortAddresses, "nodeport-addresses", o.config.NodePortAddresses,
"A string slice of values which specify the addresses to use for NodePorts. Values may be valid IP blocks (e.g. 1.2.3.0/24, 1.2.3.4/32). The default empty string slice ([]) means to use all local addresses.")
fs.Var(flag.NewMapStringBool(&o.config.FeatureGates), "feature-gates", "A set of key=value pairs that describe feature gates for alpha/experimental features. "+
"Options are:\n"+strings.Join(utilfeature.DefaultFeatureGate.KnownFeatures(), "\n"))
}

View File

@ -161,6 +161,7 @@ func newProxyServer(
nodeIP,
recorder,
healthzUpdater,
config.NodePortAddresses,
)
if err != nil {
return nil, fmt.Errorf("unable to create proxier: %v", err)
@ -196,6 +197,7 @@ func newProxyServer(
recorder,
healthzServer,
config.IPVS.Scheduler,
config.NodePortAddresses,
)
if err != nil {
return nil, fmt.Errorf("unable to create proxier: %v", err)
@ -226,6 +228,7 @@ func newProxyServer(
config.IPTables.SyncPeriod.Duration,
config.IPTables.MinSyncPeriod.Duration,
config.UDPIdleTimeout.Duration,
config.NodePortAddresses,
)
if err != nil {
return nil, fmt.Errorf("unable to create proxier: %v", err)

View File

@ -433,6 +433,9 @@ oomScoreAdj: 17
portRange: "2-7"
resourceContainer: /foo
udpIdleTimeout: 123ms
nodePortAddresses:
- "10.20.30.40/16"
- "fd00:1::0/64"
`
testCases := []struct {
@ -545,6 +548,7 @@ udpIdleTimeout: 123ms
PortRange: "2-7",
ResourceContainer: "/foo",
UDPIdleTimeout: metav1.Duration{Duration: 123 * time.Millisecond},
NodePortAddresses: []string{"10.20.30.40/16", "fd00:1::0/64"},
}
options := NewOptions()

View File

@ -92,6 +92,7 @@ func NewHollowProxyOrDie(
getNodeIP(client, nodeName),
recorder,
nil,
[]string{},
)
if err != nil {
return nil, fmt.Errorf("unable to create proxier: %v", err)

View File

@ -44,7 +44,9 @@ func Funcs(codecs runtimeserializer.CodecFactory) []interface{} {
obj.IPTables.MasqueradeBit = utilpointer.Int32Ptr(c.Int31())
obj.MetricsBindAddress = fmt.Sprintf("%d.%d.%d.%d:%d", c.Intn(256), c.Intn(256), c.Intn(256), c.Intn(256), c.Intn(65536))
obj.OOMScoreAdj = utilpointer.Int32Ptr(c.Int31())
obj.ResourceContainer = c.RandString()
obj.ResourceContainer = "foo"
obj.ClientConnection.ContentType = "bar"
obj.NodePortAddresses = []string{"1.2.3.0/24"}
},
}
}

View File

@ -144,6 +144,14 @@ type KubeProxyConfiguration struct {
// configSyncPeriod is how often configuration from the apiserver is refreshed. Must be greater
// than 0.
ConfigSyncPeriod metav1.Duration
// nodePortAddresses is the --nodeport-addresses value for kube-proxy process. Values must be valid
// IP blocks. These values are as a parameter to select the interfaces where nodeport works.
// In case someone would like to expose a service on localhost for local visit and some other interfaces for
// particular purpose, a list of IP blocks would do that.
// If set it to "127.0.0.0/8", kube-proxy will only select the loopback interface for NodePort.
// If set it to a non-zero IP block, kube-proxy will filter that down to just the IPs that applied to the node.
// An empty string slice is meant to select all network interfaces.
NodePortAddresses []string
}
// Currently, three modes of proxy are available in Linux platform: 'userspace' (older, going to be EOL), 'iptables'

View File

@ -140,6 +140,14 @@ type KubeProxyConfiguration struct {
// configSyncPeriod is how often configuration from the apiserver is refreshed. Must be greater
// than 0.
ConfigSyncPeriod metav1.Duration `json:"configSyncPeriod"`
// nodePortAddresses is the --nodeport-addresses value for kube-proxy process. Values must be valid
// IP blocks. These values are as a parameter to select the interfaces where nodeport works.
// In case someone would like to expose a service on localhost for local visit and some other interfaces for
// particular purpose, a list of IP blocks would do that.
// If set it to "127.0.0.0/8", kube-proxy will only select the loopback interface for NodePort.
// If set it to a non-zero IP block, kube-proxy will filter that down to just the IPs that applied to the node.
// An empty string slice is meant to select all network interfaces.
NodePortAddresses []string `json:"nodePortAddresses"`
}
// Currently, three modes of proxy are available in Linux platform: 'userspace' (older, going to be EOL), 'iptables'

View File

@ -104,6 +104,7 @@ func autoConvert_v1alpha1_KubeProxyConfiguration_To_kubeproxyconfig_KubeProxyCon
return err
}
out.ConfigSyncPeriod = in.ConfigSyncPeriod
out.NodePortAddresses = *(*[]string)(unsafe.Pointer(&in.NodePortAddresses))
return nil
}
@ -138,6 +139,7 @@ func autoConvert_kubeproxyconfig_KubeProxyConfiguration_To_v1alpha1_KubeProxyCon
return err
}
out.ConfigSyncPeriod = in.ConfigSyncPeriod
out.NodePortAddresses = *(*[]string)(unsafe.Pointer(&in.NodePortAddresses))
return nil
}

View File

@ -67,6 +67,11 @@ func (in *KubeProxyConfiguration) DeepCopyInto(out *KubeProxyConfiguration) {
out.UDPIdleTimeout = in.UDPIdleTimeout
in.Conntrack.DeepCopyInto(&out.Conntrack)
out.ConfigSyncPeriod = in.ConfigSyncPeriod
if in.NodePortAddresses != nil {
in, out := &in.NodePortAddresses, &out.NodePortAddresses
*out = make([]string, len(*in))
copy(*out, *in)
}
return
}

View File

@ -73,6 +73,8 @@ func Validate(config *kubeproxyconfig.KubeProxyConfiguration) field.ErrorList {
allErrs = append(allErrs, field.Invalid(newPath.Child("PortRange"), config.PortRange, "must be a valid port range (e.g. 300-2000)"))
}
allErrs = append(allErrs, validateKubeProxyNodePortAddress(config.NodePortAddresses, newPath.Child("NodePortAddresses"))...)
return allErrs
}
@ -238,3 +240,16 @@ func validateIPVSSchedulerMethod(scheduler kubeproxyconfig.IPVSSchedulerMethod,
}
return allErrs
}
func validateKubeProxyNodePortAddress(nodePortAddresses []string, fldPath *field.Path) field.ErrorList {
allErrs := field.ErrorList{}
for i := range nodePortAddresses {
if _, _, err := net.ParseCIDR(nodePortAddresses[i]); err != nil {
allErrs = append(allErrs, field.Invalid(fldPath, nodePortAddresses, "must be a valid IP block"))
break
}
}
return allErrs
}

View File

@ -679,3 +679,73 @@ func TestValidateIPVSSchedulerMethod(t *testing.T) {
}
}
}
func TestValidateKubeProxyNodePortAddress(t *testing.T) {
newPath := field.NewPath("KubeProxyConfiguration")
successCases := []struct {
addresses []string
}{
{[]string{}},
{[]string{"127.0.0.0/8"}},
{[]string{"0.0.0.0/0"}},
{[]string{"::/0"}},
{[]string{"127.0.0.1/32", "1.2.3.0/24"}},
{[]string{"127.0.0.0/8"}},
{[]string{"127.0.0.1/32"}},
{[]string{"::1/128"}},
{[]string{"1.2.3.4/32"}},
{[]string{"10.20.30.0/24"}},
{[]string{"10.20.0.0/16", "100.200.0.0/16"}},
{[]string{"10.0.0.0/8"}},
{[]string{"2001:db8::/32"}},
}
for _, successCase := range successCases {
if errs := validateKubeProxyNodePortAddress(successCase.addresses, newPath.Child("NodePortAddresses")); len(errs) != 0 {
t.Errorf("expected success: %v", errs)
}
}
errorCases := []struct {
addresses []string
msg string
}{
{
addresses: []string{"foo"},
msg: "must be a valid IP block",
},
{
addresses: []string{"1.2.3"},
msg: "must be a valid IP block",
},
{
addresses: []string{""},
msg: "must be a valid IP block",
},
{
addresses: []string{"10.20.30.40"},
msg: "must be a valid IP block",
},
{
addresses: []string{"::1"},
msg: "must be a valid IP block",
},
{
addresses: []string{"2001:db8:1"},
msg: "must be a valid IP block",
},
{
addresses: []string{"2001:db8:xyz/64"},
msg: "must be a valid IP block",
},
}
for _, errorCase := range errorCases {
if errs := validateKubeProxyNodePortAddress(errorCase.addresses, newPath.Child("NodePortAddresses")); len(errs) == 0 {
t.Errorf("expected failure for %s", errorCase.msg)
} else if !strings.Contains(errs[0].Error(), errorCase.msg) {
t.Errorf("unexpected error: %v, expected: %s", errs[0], errorCase.msg)
}
}
}

View File

@ -89,6 +89,11 @@ func (in *KubeProxyConfiguration) DeepCopyInto(out *KubeProxyConfiguration) {
out.UDPIdleTimeout = in.UDPIdleTimeout
in.Conntrack.DeepCopyInto(&out.Conntrack)
out.ConfigSyncPeriod = in.ConfigSyncPeriod
if in.NodePortAddresses != nil {
in, out := &in.NodePortAddresses, &out.NodePortAddresses
*out = make([]string, len(*in))
copy(*out, *in)
}
return
}

View File

@ -42,6 +42,7 @@ go_test(
"//pkg/apis/core:go_default_library",
"//pkg/proxy:go_default_library",
"//pkg/proxy/util:go_default_library",
"//pkg/proxy/util/testing:go_default_library",
"//pkg/util/async:go_default_library",
"//pkg/util/conntrack:go_default_library",
"//pkg/util/iptables:go_default_library",

View File

@ -341,6 +341,12 @@ type Proxier struct {
filterRules *bytes.Buffer
natChains *bytes.Buffer
natRules *bytes.Buffer
// Values are as a parameter to select the interfaces where nodeport works.
nodePortAddresses []string
// networkInterfacer defines an interface for several net library functions.
// Inject for test purpose.
networkInterfacer utilproxy.NetworkInterfacer
}
// listenPortOpener opens ports by calling bind() and listen().
@ -371,6 +377,7 @@ func NewProxier(ipt utiliptables.Interface,
nodeIP net.IP,
recorder record.EventRecorder,
healthzServer healthcheck.HealthzUpdater,
nodePortAddresses []string,
) (*Proxier, error) {
// Set the route_localnet sysctl we need for
if err := sysctl.SetSysctl(sysctlRouteLocalnet, 1); err != nil {
@ -422,6 +429,8 @@ func NewProxier(ipt utiliptables.Interface,
filterRules: bytes.NewBuffer(nil),
natChains: bytes.NewBuffer(nil),
natRules: bytes.NewBuffer(nil),
nodePortAddresses: nodePortAddresses,
networkInterfacer: utilproxy.RealNetwork{},
}
burstSyncs := 2
glog.V(3).Infof("minSyncPeriod: %v, syncPeriod: %v, burstSyncs: %d", minSyncPeriod, syncPeriod, burstSyncs)
@ -1289,11 +1298,37 @@ func (proxier *Proxier) syncProxyRules() {
// Finally, tail-call to the nodeports chain. This needs to be after all
// other service portal rules.
writeLine(proxier.natRules,
"-A", string(kubeServicesChain),
"-m", "comment", "--comment", `"kubernetes service nodeports; NOTE: this must be the last rule in this chain"`,
"-m", "addrtype", "--dst-type", "LOCAL",
"-j", string(kubeNodePortsChain))
addresses, err := utilproxy.GetNodeAddresses(proxier.nodePortAddresses, proxier.networkInterfacer)
if err != nil {
glog.Errorf("Failed to get node ip address matching nodeport cidr")
} else {
isIPv6 := proxier.iptables.IsIpv6()
for address := range addresses {
// TODO(thockin, m1093782566): If/when we have dual-stack support we will want to distinguish v4 from v6 zero-CIDRs.
if utilproxy.IsZeroCIDR(address) {
args = append(args[:0],
"-A", string(kubeServicesChain),
"-m", "comment", "--comment", `"kubernetes service nodeports; NOTE: this must be the last rule in this chain"`,
"-m", "addrtype", "--dst-type", "LOCAL",
"-j", string(kubeNodePortsChain))
writeLine(proxier.natRules, args...)
// Nothing else matters after the zero CIDR.
break
}
// Ignore IP addresses with incorrect version
if isIPv6 && !conntrack.IsIPv6String(address) || !isIPv6 && conntrack.IsIPv6String(address) {
glog.Errorf("IP address %s has incorrect IP version", address)
continue
}
// create nodeport rules for each IP one by one
args = append(args[:0],
"-A", string(kubeServicesChain),
"-m", "comment", "--comment", `"kubernetes service nodeports; NOTE: this must be the last rule in this chain"`,
"-d", address,
"-j", string(kubeNodePortsChain))
writeLine(proxier.natRules, args...)
}
}
// If the masqueradeMark has been added then we want to forward that same
// traffic, this allows NodePort traffic to be forwarded even if the default

View File

@ -34,6 +34,7 @@ import (
api "k8s.io/kubernetes/pkg/apis/core"
"k8s.io/kubernetes/pkg/proxy"
utilproxy "k8s.io/kubernetes/pkg/proxy/util"
utilproxytest "k8s.io/kubernetes/pkg/proxy/util/testing"
"k8s.io/kubernetes/pkg/util/async"
"k8s.io/kubernetes/pkg/util/conntrack"
utiliptables "k8s.io/kubernetes/pkg/util/iptables"
@ -405,6 +406,8 @@ func NewFakeProxier(ipt utiliptables.Interface) *Proxier {
filterRules: bytes.NewBuffer(nil),
natChains: bytes.NewBuffer(nil),
natRules: bytes.NewBuffer(nil),
nodePortAddresses: make([]string, 0),
networkInterfacer: utilproxytest.NewFakeNetwork(),
}
p.syncRunner = async.NewBoundedFrequencyRunner("test-sync-runner", p.syncProxyRules, 0, time.Minute, 1)
return p
@ -769,6 +772,14 @@ func TestNodePort(t *testing.T) {
}),
)
itf := net.Interface{Index: 0, MTU: 0, Name: "lo", HardwareAddr: nil, Flags: 0}
addrs := []net.Addr{utilproxytest.AddrStruct{Val: "127.0.0.1/16"}}
itf1 := net.Interface{Index: 1, MTU: 0, Name: "eth1", HardwareAddr: nil, Flags: 0}
addrs1 := []net.Addr{utilproxytest.AddrStruct{Val: "::1/128"}}
fp.networkInterfacer.(*utilproxytest.FakeNetwork).AddInterfaceAddr(&itf, addrs)
fp.networkInterfacer.(*utilproxytest.FakeNetwork).AddInterfaceAddr(&itf1, addrs1)
fp.nodePortAddresses = []string{}
fp.syncProxyRules()
proto := strings.ToLower(string(api.ProtocolTCP))
@ -1001,6 +1012,11 @@ func onlyLocalNodePorts(t *testing.T, fp *Proxier, ipt *iptablestest.FakeIPTable
}),
)
itf := net.Interface{Index: 0, MTU: 0, Name: "eth0", HardwareAddr: nil, Flags: 0}
addrs := []net.Addr{utilproxytest.AddrStruct{Val: "10.20.30.51/24"}}
fp.networkInterfacer.(*utilproxytest.FakeNetwork).AddInterfaceAddr(&itf, addrs)
fp.nodePortAddresses = []string{"10.20.30.0/24"}
fp.syncProxyRules()
proto := strings.ToLower(string(api.ProtocolTCP))
@ -1013,11 +1029,15 @@ func onlyLocalNodePorts(t *testing.T, fp *Proxier, ipt *iptablestest.FakeIPTable
if !hasJump(kubeNodePortRules, lbChain, "", svcNodePort) {
errorf(fmt.Sprintf("Failed to find jump to lb chain %v", lbChain), kubeNodePortRules, t)
}
if !hasJump(kubeNodePortRules, string(KubeMarkMasqChain), "", svcNodePort) {
errorf(fmt.Sprintf("Failed to find jump to %s chain for destination IP %d", KubeMarkMasqChain, svcNodePort), kubeNodePortRules, t)
}
kubeServiceRules := ipt.GetRules(string(kubeServicesChain))
if !hasJump(kubeServiceRules, string(kubeNodePortsChain), "10.20.30.51", 0) {
errorf(fmt.Sprintf("Failed to find jump to KUBE-NODEPORTS chain %v", string(kubeNodePortsChain)), kubeServiceRules, t)
}
svcChain := string(servicePortChainName(svcPortName.String(), proto))
lbRules := ipt.GetRules(lbChain)
if hasJump(lbRules, nonLocalEpChain, "", 0) {

View File

@ -18,6 +18,7 @@ go_test(
"//pkg/proxy:go_default_library",
"//pkg/proxy/ipvs/testing:go_default_library",
"//pkg/proxy/util:go_default_library",
"//pkg/proxy/util/testing:go_default_library",
"//pkg/util/ipset:go_default_library",
"//pkg/util/ipset/testing:go_default_library",
"//pkg/util/iptables:go_default_library",

View File

@ -170,6 +170,11 @@ type Proxier struct {
// lbWhiteListIPSet is the hash:ip,port,net type ipset where stores all service load balancer ingress IP:Port,sourceCIDR pair, any packets
// from the source CIDR visit ingress IP:Port can pass through.
lbWhiteListCIDRSet *IPSet
// Values are as a parameter to select the interfaces where nodeport works.
nodePortAddresses []string
// networkInterfacer defines an interface for several net library functions.
// Inject for test purpose.
networkInterfacer utilproxy.NetworkInterfacer
}
// IPGetter helps get node network interface IP
@ -253,6 +258,7 @@ func NewProxier(ipt utiliptables.Interface,
recorder record.EventRecorder,
healthzServer healthcheck.HealthzUpdater,
scheduler string,
nodePortAddresses []string,
) (*Proxier, error) {
// Set the route_localnet sysctl we need for
if err := sysctl.SetSysctl(sysctlRouteLocalnet, 1); err != nil {
@ -336,6 +342,8 @@ func NewProxier(ipt utiliptables.Interface,
lbWhiteListCIDRSet: NewIPSet(ipset, KubeLoadBalancerSourceCIDRSet, utilipset.HashIPPortNet, isIPv6),
nodePortSetTCP: NewIPSet(ipset, KubeNodePortSetTCP, utilipset.BitmapPort, false),
nodePortSetUDP: NewIPSet(ipset, KubeNodePortSetUDP, utilipset.BitmapPort, false),
nodePortAddresses: nodePortAddresses,
networkInterfacer: utilproxy.RealNetwork{},
}
burstSyncs := 2
glog.V(3).Infof("minSyncPeriod: %v, syncPeriod: %v, burstSyncs: %d", minSyncPeriod, syncPeriod, burstSyncs)
@ -1134,6 +1142,7 @@ func (proxier *Proxier) syncProxyRules() {
}
var nodePortSet *IPSet
switch protocol {
case "tcp":
nodePortSet = proxier.nodePortSetTCP
case "udp":
@ -1152,31 +1161,43 @@ func (proxier *Proxier) syncProxyRules() {
}
// Build ipvs kernel routes for each node ip address
nodeIPs, err := proxier.ipGetter.NodeIPs()
nodeIPs := make([]net.IP, 0)
addresses, err := utilproxy.GetNodeAddresses(proxier.nodePortAddresses, proxier.networkInterfacer)
if err != nil {
glog.Errorf("Failed to get node IP, err: %v", err)
} else {
for _, nodeIP := range nodeIPs {
// ipvs call
serv := &utilipvs.VirtualServer{
Address: nodeIP,
Port: uint16(svcInfo.nodePort),
Protocol: string(svcInfo.protocol),
Scheduler: proxier.ipvsScheduler,
}
if svcInfo.sessionAffinityType == api.ServiceAffinityClientIP {
serv.Flags |= utilipvs.FlagPersistent
serv.Timeout = uint32(svcInfo.stickyMaxAgeSeconds)
}
// There is no need to bind Node IP to dummy interface, so set parameter `bindAddr` to `false`.
if err := proxier.syncService(svcNameString, serv, false); err == nil {
activeIPVSServices[serv.String()] = true
if err := proxier.syncEndpoint(svcName, svcInfo.onlyNodeLocalEndpoints, serv); err != nil {
glog.Errorf("Failed to sync endpoint for service: %v, err: %v", serv, err)
}
} else {
glog.Errorf("Failed to sync service: %v, err: %v", serv, err)
glog.Errorf("Failed to get node ip address matching nodeport cidr")
continue
}
for address := range addresses {
if !utilproxy.IsZeroCIDR(address) {
nodeIPs = append(nodeIPs, net.ParseIP(address))
continue
}
// zero cidr
nodeIPs, err = proxier.ipGetter.NodeIPs()
if err != nil {
glog.Errorf("Failed to list all node IPs from host, err: %v", err)
}
}
for _, nodeIP := range nodeIPs {
// ipvs call
serv := &utilipvs.VirtualServer{
Address: nodeIP,
Port: uint16(svcInfo.nodePort),
Protocol: string(svcInfo.protocol),
Scheduler: proxier.ipvsScheduler,
}
if svcInfo.sessionAffinityType == api.ServiceAffinityClientIP {
serv.Flags |= utilipvs.FlagPersistent
serv.Timeout = uint32(svcInfo.stickyMaxAgeSeconds)
}
// There is no need to bind Node IP to dummy interface, so set parameter `bindAddr` to `false`.
if err := proxier.syncService(svcNameString, serv, false); err == nil {
activeIPVSServices[serv.String()] = true
if err := proxier.syncEndpoint(svcName, svcInfo.onlyNodeLocalEndpoints, serv); err != nil {
glog.Errorf("Failed to sync endpoint for service: %v, err: %v", serv, err)
}
} else {
glog.Errorf("Failed to sync service: %v, err: %v", serv, err)
}
}
}

View File

@ -34,6 +34,7 @@ import (
"k8s.io/apimachinery/pkg/util/sets"
netlinktest "k8s.io/kubernetes/pkg/proxy/ipvs/testing"
proxyutil "k8s.io/kubernetes/pkg/proxy/util"
proxyutiltest "k8s.io/kubernetes/pkg/proxy/util/testing"
utilipset "k8s.io/kubernetes/pkg/util/ipset"
ipsettest "k8s.io/kubernetes/pkg/util/ipset/testing"
utiliptables "k8s.io/kubernetes/pkg/util/iptables"
@ -146,6 +147,8 @@ func NewFakeProxier(ipt utiliptables.Interface, ipvs utilipvs.Interface, ipset u
lbWhiteListCIDRSet: NewIPSet(ipset, KubeLoadBalancerSourceCIDRSet, utilipset.HashIPPortNet, false),
nodePortSetTCP: NewIPSet(ipset, KubeNodePortSetTCP, utilipset.BitmapPort, false),
nodePortSetUDP: NewIPSet(ipset, KubeNodePortSetUDP, utilipset.BitmapPort, false),
nodePortAddresses: make([]string, 0),
networkInterfacer: proxyutiltest.NewFakeNetwork(),
}
}
@ -386,6 +389,8 @@ func TestNodePort(t *testing.T) {
}),
)
fp.nodePortAddresses = []string{"0.0.0.0/0"}
fp.syncProxyRules()
// Check ipvs service and destinations
@ -445,6 +450,8 @@ func TestNodePortNoEndpoint(t *testing.T) {
)
makeEndpointsMap(fp)
fp.nodePortAddresses = []string{"0.0.0.0/0"}
fp.syncProxyRules()
// Check ipvs service and destinations
@ -847,15 +854,23 @@ func TestOnlyLocalNodePorts(t *testing.T) {
}),
)
itf := net.Interface{Index: 0, MTU: 0, Name: "eth0", HardwareAddr: nil, Flags: 0}
addrs := []net.Addr{proxyutiltest.AddrStruct{Val: "100.101.102.103/24"}}
itf1 := net.Interface{Index: 1, MTU: 0, Name: "eth1", HardwareAddr: nil, Flags: 0}
addrs1 := []net.Addr{proxyutiltest.AddrStruct{Val: "2001:db8::0/64"}}
fp.networkInterfacer.(*proxyutiltest.FakeNetwork).AddInterfaceAddr(&itf, addrs)
fp.networkInterfacer.(*proxyutiltest.FakeNetwork).AddInterfaceAddr(&itf1, addrs1)
fp.nodePortAddresses = []string{"100.101.102.0/24", "2001:db8::0/64"}
fp.syncProxyRules()
// Expect 2 services and 1 destination
// Expect 3 services and 1 destination
services, err := ipvs.GetVirtualServers()
if err != nil {
t.Errorf("Failed to get ipvs services, err: %v", err)
}
if len(services) != 2 {
t.Errorf("Expect 2 ipvs services, got %d", len(services))
if len(services) != 3 {
t.Errorf("Expect 3 ipvs services, got %d", len(services))
}
found := false
for _, svc := range services {

View File

@ -155,18 +155,15 @@ 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, exec utilexec.Interface, pr utilnet.PortRange, syncPeriod, minSyncPeriod, udpIdleTimeout time.Duration) (*Proxier, error) {
return NewCustomProxier(loadBalancer, listenIP, iptables, exec, pr, syncPeriod, minSyncPeriod, udpIdleTimeout, newProxySocket)
func NewProxier(loadBalancer LoadBalancer, listenIP net.IP, iptables iptables.Interface, exec utilexec.Interface, pr utilnet.PortRange, syncPeriod, minSyncPeriod, udpIdleTimeout time.Duration, nodePortAddresses []string) (*Proxier, error) {
return NewCustomProxier(loadBalancer, listenIP, iptables, exec, pr, syncPeriod, minSyncPeriod, udpIdleTimeout, nodePortAddresses, newProxySocket)
}
// NewCustomProxier functions similarly to NewProxier, returning a new Proxier
// for the given LoadBalancer and address. The new proxier is constructed using
// the ProxySocket constructor provided, however, instead of constructing the
// default ProxySockets.
func NewCustomProxier(loadBalancer LoadBalancer, listenIP net.IP, iptables iptables.Interface, exec utilexec.Interface, pr utilnet.PortRange, syncPeriod, minSyncPeriod, udpIdleTimeout time.Duration, makeProxySocket ProxySocketFunc) (*Proxier, error) {
if listenIP == nil {
return nil, fmt.Errorf("invalid listen ip for kube-proxy")
}
func NewCustomProxier(loadBalancer LoadBalancer, listenIP net.IP, iptables iptables.Interface, exec utilexec.Interface, pr utilnet.PortRange, syncPeriod, minSyncPeriod, udpIdleTimeout time.Duration, nodePortAddresses []string, makeProxySocket ProxySocketFunc) (*Proxier, error) {
if listenIP.Equal(localhostIPv4) || listenIP.Equal(localhostIPv6) {
return nil, ErrProxyOnLocalhost
}

View File

@ -4,6 +4,7 @@ go_library(
name = "go_default_library",
srcs = [
"endpoints.go",
"network.go",
"port.go",
"utils.go",
],
@ -12,8 +13,10 @@ go_library(
deps = [
"//pkg/apis/core:go_default_library",
"//pkg/apis/core/helper:go_default_library",
"//pkg/util/conntrack:go_default_library",
"//vendor/github.com/golang/glog:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/types:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/util/sets:go_default_library",
],
)
@ -27,8 +30,10 @@ go_test(
embed = [":go_default_library"],
deps = [
"//pkg/apis/core:go_default_library",
"//pkg/proxy/util/testing:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/types:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/util/sets:go_default_library",
],
)
@ -41,7 +46,10 @@ filegroup(
filegroup(
name = "all-srcs",
srcs = [":package-srcs"],
srcs = [
":package-srcs",
"//pkg/proxy/util/testing:all-srcs",
],
tags = ["automanaged"],
visibility = ["//visibility:public"],
)

45
pkg/proxy/util/network.go Normal file
View File

@ -0,0 +1,45 @@
/*
Copyright 2018 The Kubernetes Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package util
import (
"net"
)
// NetworkInterfacer defines an interface for several net library functions. Production
// code will forward to net library functions, and unit tests will override the methods
// for testing purposes.
type NetworkInterfacer interface {
Addrs(intf *net.Interface) ([]net.Addr, error)
Interfaces() ([]net.Interface, error)
}
// RealNetwork implements the NetworkInterfacer interface for production code, just
// wrapping the underlying net library function calls.
type RealNetwork struct{}
// Addrs wraps net.Interface.Addrs(), it's a part of NetworkInterfacer interface.
func (_ RealNetwork) Addrs(intf *net.Interface) ([]net.Addr, error) {
return intf.Addrs()
}
// Interfaces wraps net.Interfaces(), it's a part of NetworkInterfacer interface.
func (_ RealNetwork) Interfaces() ([]net.Interface, error) {
return net.Interfaces()
}
var _ NetworkInterfacer = &RealNetwork{}

View File

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

View File

@ -0,0 +1,65 @@
/*
Copyright 2018 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"
// FakeNetwork implements the NetworkInterfacer interface for test purpose.
type FakeNetwork struct {
NetworkInterfaces []net.Interface
// The key of map Addrs is the network interface name
Address map[string][]net.Addr
}
// NewFakeNetwork initializes a FakeNetwork.
func NewFakeNetwork() *FakeNetwork {
return &FakeNetwork{
NetworkInterfaces: make([]net.Interface, 0),
Address: make(map[string][]net.Addr),
}
}
// AddInterfaceAddr create an interface and its associated addresses for FakeNetwork implementation.
func (f *FakeNetwork) AddInterfaceAddr(intf *net.Interface, addrs []net.Addr) {
f.NetworkInterfaces = append(f.NetworkInterfaces, *intf)
f.Address[intf.Name] = addrs
}
// Addrs is part of NetworkInterfacer interface.
func (f *FakeNetwork) Addrs(intf *net.Interface) ([]net.Addr, error) {
return f.Address[intf.Name], nil
}
// Interfaces is part of NetworkInterfacer interface.
func (f *FakeNetwork) Interfaces() ([]net.Interface, error) {
return f.NetworkInterfaces, nil
}
// AddrStruct implements the net.Addr for test purpose.
type AddrStruct struct{ Val string }
// Network is part of net.Addr interface.
func (a AddrStruct) Network() string {
return a.Val
}
// String is part of net.Addr interface.
func (a AddrStruct) String() string {
return a.Val
}
var _ net.Addr = &AddrStruct{}

View File

@ -17,15 +17,30 @@ limitations under the License.
package util
import (
"fmt"
"net"
"k8s.io/apimachinery/pkg/types"
"k8s.io/apimachinery/pkg/util/sets"
api "k8s.io/kubernetes/pkg/apis/core"
"k8s.io/kubernetes/pkg/apis/core/helper"
"k8s.io/kubernetes/pkg/util/conntrack"
"github.com/golang/glog"
)
const (
IPv4ZeroCIDR = "0.0.0.0/0"
IPv6ZeroCIDR = "::/0"
)
func IsZeroCIDR(cidr string) bool {
if cidr == IPv4ZeroCIDR || cidr == IPv6ZeroCIDR {
return true
}
return false
}
func IsLocalIP(ip string) (bool, error) {
addrs, err := net.InterfaceAddrs()
if err != nil {
@ -56,3 +71,61 @@ func ShouldSkipService(svcName types.NamespacedName, service *api.Service) bool
}
return false
}
// GetNodeAddresses return all matched node IP addresses based on given cidr slice.
// Some callers, e.g. IPVS proxier, need concrete IPs, not ranges, which is why this exists.
// NetworkInterfacer is injected for test purpose.
// We expect the cidrs passed in is already validated.
// Given an empty input `[]`, it will return `0.0.0.0/0` and `::/0` directly.
// If multiple cidrs is given, it will return the minimal IP sets, e.g. given input `[1.2.0.0/16, 0.0.0.0/0]`, it will
// only return `0.0.0.0/0`.
// NOTE: GetNodeAddresses only accepts CIDRs, if you want concrete IPs, e.g. 1.2.3.4, then the input should be 1.2.3.4/32.
func GetNodeAddresses(cidrs []string, nw NetworkInterfacer) (sets.String, error) {
uniqueAddressList := sets.NewString()
if len(cidrs) == 0 {
uniqueAddressList.Insert(IPv4ZeroCIDR)
uniqueAddressList.Insert(IPv6ZeroCIDR)
return uniqueAddressList, nil
}
// First round of iteration to pick out `0.0.0.0/0` or `::/0` for the sake of excluding non-zero IPs.
for _, cidr := range cidrs {
if IsZeroCIDR(cidr) {
uniqueAddressList.Insert(cidr)
}
}
// Second round of iteration to parse IPs based on cidr.
for _, cidr := range cidrs {
if IsZeroCIDR(cidr) {
continue
}
_, ipNet, _ := net.ParseCIDR(cidr)
itfs, err := nw.Interfaces()
if err != nil {
return nil, fmt.Errorf("error listing all interfaces from host, error: %v", err)
}
for _, itf := range itfs {
addrs, err := nw.Addrs(&itf)
if err != nil {
return nil, fmt.Errorf("error getting address from interface %s, error: %v", itf.Name, err)
}
for _, addr := range addrs {
if addr == nil {
continue
}
ip, _, err := net.ParseCIDR(addr.String())
if err != nil {
return nil, fmt.Errorf("error parsing CIDR for interface %s, error: %v", itf.Name, err)
}
if ipNet.Contains(ip) {
if conntrack.IsIPv6(ip) && !uniqueAddressList.Has(IPv6ZeroCIDR) {
uniqueAddressList.Insert(ip.String())
}
if !conntrack.IsIPv6(ip) && !uniqueAddressList.Has(IPv4ZeroCIDR) {
uniqueAddressList.Insert(ip.String())
}
}
}
}
}
return uniqueAddressList, nil
}

View File

@ -17,11 +17,14 @@ limitations under the License.
package util
import (
"net"
"testing"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/types"
"k8s.io/apimachinery/pkg/util/sets"
api "k8s.io/kubernetes/pkg/apis/core"
fake "k8s.io/kubernetes/pkg/proxy/util/testing"
)
func TestShouldSkipService(t *testing.T) {
@ -109,3 +112,218 @@ func TestShouldSkipService(t *testing.T) {
}
}
}
type InterfaceAddrsPair struct {
itf net.Interface
addrs []net.Addr
}
func TestGetNodeAddressses(t *testing.T) {
testCases := []struct {
cidrs []string
nw *fake.FakeNetwork
itfAddrsPairs []InterfaceAddrsPair
expected sets.String
}{
{ // case 0
cidrs: []string{"10.20.30.0/24"},
nw: fake.NewFakeNetwork(),
itfAddrsPairs: []InterfaceAddrsPair{
{
itf: net.Interface{Index: 0, MTU: 0, Name: "eth0", HardwareAddr: nil, Flags: 0},
addrs: []net.Addr{fake.AddrStruct{Val: "10.20.30.51/24"}},
},
{
itf: net.Interface{Index: 2, MTU: 0, Name: "eth1", HardwareAddr: nil, Flags: 0},
addrs: []net.Addr{fake.AddrStruct{Val: "100.200.201.1/24"}},
},
},
expected: sets.NewString("10.20.30.51"),
},
{ // case 1
cidrs: []string{"0.0.0.0/0"},
nw: fake.NewFakeNetwork(),
itfAddrsPairs: []InterfaceAddrsPair{
{
itf: net.Interface{Index: 0, MTU: 0, Name: "eth0", HardwareAddr: nil, Flags: 0},
addrs: []net.Addr{fake.AddrStruct{Val: "10.20.30.51/24"}},
},
{
itf: net.Interface{Index: 1, MTU: 0, Name: "lo", HardwareAddr: nil, Flags: 0},
addrs: []net.Addr{fake.AddrStruct{Val: "127.0.0.1/8"}},
},
},
expected: sets.NewString("0.0.0.0/0"),
},
{ // case 2
cidrs: []string{"2001:db8::/32", "::1/128"},
nw: fake.NewFakeNetwork(),
itfAddrsPairs: []InterfaceAddrsPair{
{
itf: net.Interface{Index: 0, MTU: 0, Name: "eth0", HardwareAddr: nil, Flags: 0},
addrs: []net.Addr{fake.AddrStruct{Val: "2001:db8::1/32"}},
},
{
itf: net.Interface{Index: 1, MTU: 0, Name: "lo", HardwareAddr: nil, Flags: 0},
addrs: []net.Addr{fake.AddrStruct{Val: "::1/128"}},
},
},
expected: sets.NewString("2001:db8::1", "::1"),
},
{ // case 3
cidrs: []string{"::/0"},
nw: fake.NewFakeNetwork(),
itfAddrsPairs: []InterfaceAddrsPair{
{
itf: net.Interface{Index: 0, MTU: 0, Name: "eth0", HardwareAddr: nil, Flags: 0},
addrs: []net.Addr{fake.AddrStruct{Val: "2001:db8::1/32"}},
},
{
itf: net.Interface{Index: 1, MTU: 0, Name: "lo", HardwareAddr: nil, Flags: 0},
addrs: []net.Addr{fake.AddrStruct{Val: "::1/128"}},
},
},
expected: sets.NewString("::/0"),
},
{ // case 4
cidrs: []string{"127.0.0.1/32"},
nw: fake.NewFakeNetwork(),
itfAddrsPairs: []InterfaceAddrsPair{
{
itf: net.Interface{Index: 0, MTU: 0, Name: "eth0", HardwareAddr: nil, Flags: 0},
addrs: []net.Addr{fake.AddrStruct{Val: "10.20.30.51/24"}},
},
{
itf: net.Interface{Index: 1, MTU: 0, Name: "lo", HardwareAddr: nil, Flags: 0},
addrs: []net.Addr{fake.AddrStruct{Val: "127.0.0.1/8"}},
},
},
expected: sets.NewString("127.0.0.1"),
},
{ // case 5
cidrs: []string{"127.0.0.0/8"},
nw: fake.NewFakeNetwork(),
itfAddrsPairs: []InterfaceAddrsPair{
{
itf: net.Interface{Index: 1, MTU: 0, Name: "lo", HardwareAddr: nil, Flags: 0},
addrs: []net.Addr{fake.AddrStruct{Val: "127.0.1.1/8"}},
},
},
expected: sets.NewString("127.0.1.1"),
},
{ // case 6
cidrs: []string{"10.20.30.0/24", "100.200.201.0/24"},
nw: fake.NewFakeNetwork(),
itfAddrsPairs: []InterfaceAddrsPair{
{
itf: net.Interface{Index: 0, MTU: 0, Name: "eth0", HardwareAddr: nil, Flags: 0},
addrs: []net.Addr{fake.AddrStruct{Val: "10.20.30.51/24"}},
},
{
itf: net.Interface{Index: 2, MTU: 0, Name: "eth1", HardwareAddr: nil, Flags: 0},
addrs: []net.Addr{fake.AddrStruct{Val: "100.200.201.1/24"}},
},
},
expected: sets.NewString("10.20.30.51", "100.200.201.1"),
},
{ // case 7
cidrs: []string{"10.20.30.0/24", "100.200.201.0/24"},
nw: fake.NewFakeNetwork(),
itfAddrsPairs: []InterfaceAddrsPair{
{
itf: net.Interface{Index: 0, MTU: 0, Name: "eth0", HardwareAddr: nil, Flags: 0},
addrs: []net.Addr{fake.AddrStruct{Val: "192.168.1.2/24"}},
},
{
itf: net.Interface{Index: 1, MTU: 0, Name: "lo", HardwareAddr: nil, Flags: 0},
addrs: []net.Addr{fake.AddrStruct{Val: "127.0.0.1/8"}},
},
},
expected: sets.NewString(),
},
{ // case 8
cidrs: []string{},
nw: fake.NewFakeNetwork(),
itfAddrsPairs: []InterfaceAddrsPair{
{
itf: net.Interface{Index: 0, MTU: 0, Name: "eth0", HardwareAddr: nil, Flags: 0},
addrs: []net.Addr{fake.AddrStruct{Val: "192.168.1.2/24"}},
},
{
itf: net.Interface{Index: 1, MTU: 0, Name: "lo", HardwareAddr: nil, Flags: 0},
addrs: []net.Addr{fake.AddrStruct{Val: "127.0.0.1/8"}},
},
},
expected: sets.NewString("0.0.0.0/0", "::/0"),
},
{ // case 9
cidrs: []string{},
nw: fake.NewFakeNetwork(),
itfAddrsPairs: []InterfaceAddrsPair{
{
itf: net.Interface{Index: 0, MTU: 0, Name: "eth0", HardwareAddr: nil, Flags: 0},
addrs: []net.Addr{fake.AddrStruct{Val: "2001:db8::1/32"}},
},
{
itf: net.Interface{Index: 1, MTU: 0, Name: "lo", HardwareAddr: nil, Flags: 0},
addrs: []net.Addr{fake.AddrStruct{Val: "::1/128"}},
},
},
expected: sets.NewString("0.0.0.0/0", "::/0"),
},
{ // case 9
cidrs: []string{"1.2.3.0/24", "0.0.0.0/0"},
nw: fake.NewFakeNetwork(),
itfAddrsPairs: []InterfaceAddrsPair{
{
itf: net.Interface{Index: 0, MTU: 0, Name: "eth0", HardwareAddr: nil, Flags: 0},
addrs: []net.Addr{fake.AddrStruct{Val: "1.2.3.4/30"}},
},
},
expected: sets.NewString("0.0.0.0/0"),
},
{ // case 10
cidrs: []string{"0.0.0.0/0", "1.2.3.0/24", "::1/128"},
nw: fake.NewFakeNetwork(),
itfAddrsPairs: []InterfaceAddrsPair{
{
itf: net.Interface{Index: 0, MTU: 0, Name: "eth0", HardwareAddr: nil, Flags: 0},
addrs: []net.Addr{fake.AddrStruct{Val: "1.2.3.4/30"}},
},
{
itf: net.Interface{Index: 1, MTU: 0, Name: "lo", HardwareAddr: nil, Flags: 0},
addrs: []net.Addr{fake.AddrStruct{Val: "::1/128"}},
},
},
expected: sets.NewString("0.0.0.0/0", "::1"),
},
{ // case 11
cidrs: []string{"::/0", "1.2.3.0/24", "::1/128"},
nw: fake.NewFakeNetwork(),
itfAddrsPairs: []InterfaceAddrsPair{
{
itf: net.Interface{Index: 0, MTU: 0, Name: "eth0", HardwareAddr: nil, Flags: 0},
addrs: []net.Addr{fake.AddrStruct{Val: "1.2.3.4/30"}},
},
{
itf: net.Interface{Index: 1, MTU: 0, Name: "lo", HardwareAddr: nil, Flags: 0},
addrs: []net.Addr{fake.AddrStruct{Val: "::1/128"}},
},
},
expected: sets.NewString("::/0", "1.2.3.4"),
},
}
for i := range testCases {
for _, pair := range testCases[i].itfAddrsPairs {
testCases[i].nw.AddInterfaceAddr(&pair.itf, pair.addrs)
}
addrList, err := GetNodeAddresses(testCases[i].cidrs, testCases[i].nw)
if err != nil {
t.Errorf("case [%d], unexpected error: %v", i, err)
}
if !addrList.Equal(testCases[i].expected) {
t.Errorf("case [%d], unexpected mismatch, expected: %v, got: %v", i, testCases[i].expected, addrList)
}
}
}