github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2025-08-05 10:19:50 +00:00
Code Issues Projects Releases Wiki Activity

Merge pull request #122630 from uablrek/kube-proxy-ip-types

Browse Source 
Kube-proxy: use type net.IP for addresses
This commit is contained in:
Kubernetes Prow Robot 2024-01-09 21:18:25 +01:00 committed by GitHub
commit 13fad117aa
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
10 changed files with 171 additions and 152 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
cmd/kube-proxy/app/server_linux.go
View File

@ -508,7 +508,7 @@ func getLocalDetector(logger klog.Logger, ipFamily v1.IPFamily, mode proxyconfig
cidrsByFamily := proxyutil.MapCIDRsByIPFamily(strings.Split(clusterCIDRs, ","))
if len(cidrsByFamily[ipFamily]) != 0 {
return proxyutiliptables.NewDetectLocalByCIDR(cidrsByFamily[ipFamily][0])
return proxyutiliptables.NewDetectLocalByCIDR(cidrsByFamily[ipFamily][0].String())
}
logger.Info("Detect-local-mode set to ClusterCIDR, but no cluster CIDR for family", "ipFamily", ipFamily)
@ -516,7 +516,7 @@ func getLocalDetector(logger klog.Logger, ipFamily v1.IPFamily, mode proxyconfig
case proxyconfigapi.LocalModeNodeCIDR:
cidrsByFamily := proxyutil.MapCIDRsByIPFamily(nodePodCIDRs)
if len(cidrsByFamily[ipFamily]) != 0 {
return proxyutiliptables.NewDetectLocalByCIDR(cidrsByFamily[ipFamily][0])
return proxyutiliptables.NewDetectLocalByCIDR(cidrsByFamily[ipFamily][0].String())
}
logger.Info("Detect-local-mode set to NodeCIDR, but no PodCIDR defined at node for family", "ipFamily", ipFamily)

16
pkg/proxy/conntrack/cleanup.go
View File

@ -50,11 +50,11 @@ func deleteStaleServiceConntrackEntries(isIPv6 bool, exec utilexec.Interface, sv
if svcInfo, ok := svcPortMap[svcPortName]; ok {
klog.V(4).InfoS("Newly-active UDP service may have stale conntrack entries", "servicePortName", svcPortName)
conntrackCleanupServiceIPs.Insert(svcInfo.ClusterIP().String())
for _, extIP := range svcInfo.ExternalIPStrings() {
conntrackCleanupServiceIPs.Insert(extIP)
for _, extIP := range svcInfo.ExternalIPs() {
conntrackCleanupServiceIPs.Insert(extIP.String())
}
for _, lbIP := range svcInfo.LoadBalancerVIPStrings() {
conntrackCleanupServiceIPs.Insert(lbIP)
for _, lbIP := range svcInfo.LoadBalancerVIPs() {
conntrackCleanupServiceIPs.Insert(lbIP.String())
}
nodePort := svcInfo.NodePort()
if svcInfo.Protocol() == v1.ProtocolUDP && nodePort != 0 {
@ -97,14 +97,14 @@ func deleteStaleEndpointConntrackEntries(exec utilexec.Interface, svcPortMap pro
if err != nil {
klog.ErrorS(err, "Failed to delete endpoint connections", "servicePortName", epSvcPair.ServicePortName)
}
for _, extIP := range svcInfo.ExternalIPStrings() {
err := ClearEntriesForNAT(exec, extIP, endpointIP, v1.ProtocolUDP)
for _, extIP := range svcInfo.ExternalIPs() {
err := ClearEntriesForNAT(exec, extIP.String(), endpointIP, v1.ProtocolUDP)
if err != nil {
klog.ErrorS(err, "Failed to delete endpoint connections for externalIP", "servicePortName", epSvcPair.ServicePortName, "externalIP", extIP)
}
}
for _, lbIP := range svcInfo.LoadBalancerVIPStrings() {
err := ClearEntriesForNAT(exec, lbIP, endpointIP, v1.ProtocolUDP)
for _, lbIP := range svcInfo.LoadBalancerVIPs() {
err := ClearEntriesForNAT(exec, lbIP.String(), endpointIP, v1.ProtocolUDP)
if err != nil {
klog.ErrorS(err, "Failed to delete endpoint connections for LoadBalancerIP", "servicePortName", epSvcPair.ServicePortName, "loadBalancerIP", lbIP)
}

32
pkg/proxy/iptables/proxier.go
View File

@ -54,7 +54,6 @@ import (
"k8s.io/kubernetes/pkg/util/async"
utiliptables "k8s.io/kubernetes/pkg/util/iptables"
utilexec "k8s.io/utils/exec"
netutils "k8s.io/utils/net"
)
const (
@ -1014,7 +1013,7 @@ func (proxier *Proxier) syncProxyRules() {
// create a firewall chain.
loadBalancerTrafficChain := externalTrafficChain
fwChain := svcInfo.firewallChainName
usesFWChain := hasEndpoints && len(svcInfo.LoadBalancerVIPStrings()) > 0 && len(svcInfo.LoadBalancerSourceRanges()) > 0
usesFWChain := hasEndpoints && len(svcInfo.LoadBalancerVIPs()) > 0 && len(svcInfo.LoadBalancerSourceRanges()) > 0
if usesFWChain {
loadBalancerTrafficChain = fwChain
}
@ -1077,7 +1076,7 @@ func (proxier *Proxier) syncProxyRules() {
}
// Capture externalIPs.
for _, externalIP := range svcInfo.ExternalIPStrings() {
for _, externalIP := range svcInfo.ExternalIPs() {
if hasEndpoints {
// Send traffic bound for external IPs to the "external
// destinations" chain.
@ -1085,7 +1084,7 @@ func (proxier *Proxier) syncProxyRules() {
"-A", string(kubeServicesChain),
"-m", "comment", "--comment", fmt.Sprintf(`"%s external IP"`, svcPortNameString),
"-m", protocol, "-p", protocol,
"-d", externalIP,
"-d", externalIP.String(),
"--dport", strconv.Itoa(svcInfo.Port()),
"-j", string(externalTrafficChain))
}
@ -1097,7 +1096,7 @@ func (proxier *Proxier) syncProxyRules() {
"-A", string(kubeExternalServicesChain),
"-m", "comment", "--comment", externalTrafficFilterComment,
"-m", protocol, "-p", protocol,
"-d", externalIP,
"-d", externalIP.String(),
"--dport", strconv.Itoa(svcInfo.Port()),
"-j", externalTrafficFilterTarget,
)
@ -1105,13 +1104,13 @@ func (proxier *Proxier) syncProxyRules() {
}
// Capture load-balancer ingress.
for _, lbip := range svcInfo.LoadBalancerVIPStrings() {
for _, lbip := range svcInfo.LoadBalancerVIPs() {
if hasEndpoints {
natRules.Write(
"-A", string(kubeServicesChain),
"-m", "comment", "--comment", fmt.Sprintf(`"%s loadbalancer IP"`, svcPortNameString),
"-m", protocol, "-p", protocol,
"-d", lbip,
"-d", lbip.String(),
"--dport", strconv.Itoa(svcInfo.Port()),
"-j", string(loadBalancerTrafficChain))
@ -1121,7 +1120,7 @@ func (proxier *Proxier) syncProxyRules() {
"-A", string(kubeProxyFirewallChain),
"-m", "comment", "--comment", fmt.Sprintf(`"%s traffic not accepted by %s"`, svcPortNameString, svcInfo.firewallChainName),
"-m", protocol, "-p", protocol,
"-d", lbip,
"-d", lbip.String(),
"--dport", strconv.Itoa(svcInfo.Port()),
"-j", "DROP")
}
@ -1130,12 +1129,12 @@ func (proxier *Proxier) syncProxyRules() {
// Either no endpoints at all (REJECT) or no endpoints for
// external traffic (DROP anything that didn't get short-circuited
// by the EXT chain.)
for _, lbip := range svcInfo.LoadBalancerVIPStrings() {
for _, lbip := range svcInfo.LoadBalancerVIPs() {
filterRules.Write(
"-A", string(kubeExternalServicesChain),
"-m", "comment", "--comment", externalTrafficFilterComment,
"-m", protocol, "-p", protocol,
"-d", lbip,
"-d", lbip.String(),
"--dport", strconv.Itoa(svcInfo.Port()),
"-j", externalTrafficFilterTarget,
)
@ -1294,12 +1293,9 @@ func (proxier *Proxier) syncProxyRules() {
// firewall filter based on each source range
allowFromNode := false
for _, src := range svcInfo.LoadBalancerSourceRanges() {
natRules.Write(args, "-s", src, "-j", string(externalTrafficChain))
_, cidr, err := netutils.ParseCIDRSloppy(src)
if err != nil {
klog.ErrorS(err, "Error parsing CIDR in LoadBalancerSourceRanges, dropping it", "cidr", cidr)
} else if cidr.Contains(proxier.nodeIP) {
for _, cidr := range svcInfo.LoadBalancerSourceRanges() {
natRules.Write(args, "-s", cidr.String(), "-j", string(externalTrafficChain))
if cidr.Contains(proxier.nodeIP) {
allowFromNode = true
}
}
@ -1309,10 +1305,10 @@ func (proxier *Proxier) syncProxyRules() {
// will loop back with the source IP set to the VIP. We
// need the following rules to allow requests from this node.
if allowFromNode {
for _, lbip := range svcInfo.LoadBalancerVIPStrings() {
for _, lbip := range svcInfo.LoadBalancerVIPs() {
natRules.Write(
args,
"-s", lbip,
"-s", lbip.String(),
"-j", string(externalTrafficChain))
}
}

24
pkg/proxy/ipvs/proxier.go
View File

@ -1097,10 +1097,10 @@ func (proxier *Proxier) syncProxyRules() {
}
// Capture externalIPs.
for _, externalIP := range svcInfo.ExternalIPStrings() {
for _, externalIP := range svcInfo.ExternalIPs() {
// ipset call
entry := &utilipset.Entry{
IP: externalIP,
IP: externalIP.String(),
Port: svcInfo.Port(),
Protocol: protocol,
SetType: utilipset.HashIPPort,
@ -1123,7 +1123,7 @@ func (proxier *Proxier) syncProxyRules() {
// ipvs call
serv := &utilipvs.VirtualServer{
Address: netutils.ParseIPSloppy(externalIP),
Address: externalIP,
Port: uint16(svcInfo.Port()),
Protocol: string(svcInfo.Protocol()),
Scheduler: proxier.ipvsScheduler,
@ -1152,10 +1152,10 @@ func (proxier *Proxier) syncProxyRules() {
}
// Capture load-balancer ingress.
for _, ingress := range svcInfo.LoadBalancerVIPStrings() {
for _, ingress := range svcInfo.LoadBalancerVIPs() {
// ipset call
entry = &utilipset.Entry{
IP: ingress,
IP: ingress.String(),
Port: svcInfo.Port(),
Protocol: protocol,
SetType: utilipset.HashIPPort,
@ -1187,13 +1187,13 @@ func (proxier *Proxier) syncProxyRules() {
}
proxier.ipsetList[kubeLoadBalancerFWSet].activeEntries.Insert(entry.String())
allowFromNode := false
for _, src := range svcInfo.LoadBalancerSourceRanges() {
for _, cidr := range svcInfo.LoadBalancerSourceRanges() {
// ipset call
entry = &utilipset.Entry{
IP: ingress,
IP: ingress.String(),
Port: svcInfo.Port(),
Protocol: protocol,
Net: src,
Net: cidr.String(),
SetType: utilipset.HashIPPortNet,
}
// enumerate all white list source cidr
@ -1203,8 +1203,6 @@ func (proxier *Proxier) syncProxyRules() {
}
proxier.ipsetList[kubeLoadBalancerSourceCIDRSet].activeEntries.Insert(entry.String())
// ignore error because it has been validated
_, cidr, _ := netutils.ParseCIDRSloppy(src)
if cidr.Contains(proxier.nodeIP) {
allowFromNode = true
}
@ -1214,10 +1212,10 @@ func (proxier *Proxier) syncProxyRules() {
// Need to add the following rule to allow request on host.
if allowFromNode {
entry = &utilipset.Entry{
IP: ingress,
IP: ingress.String(),
Port: svcInfo.Port(),
Protocol: protocol,
IP2: ingress,
IP2: ingress.String(),
SetType: utilipset.HashIPPortIP,
}
// enumerate all white list source ip
@ -1234,7 +1232,7 @@ func (proxier *Proxier) syncProxyRules() {
}
// ipvs call
serv := &utilipvs.VirtualServer{
Address: netutils.ParseIPSloppy(ingress),
Address: ingress,
Port: uint16(svcInfo.Port()),
Protocol: string(svcInfo.Protocol()),
Scheduler: proxier.ipvsScheduler,

31
pkg/proxy/nftables/proxier.go
View File

@ -55,7 +55,6 @@ import (
proxyutiliptables "k8s.io/kubernetes/pkg/proxy/util/iptables"
"k8s.io/kubernetes/pkg/util/async"
utilexec "k8s.io/utils/exec"
netutils "k8s.io/utils/net"
"k8s.io/utils/ptr"
)
@ -1153,14 +1152,14 @@ func (proxier *Proxier) syncProxyRules() {
}
// Capture externalIPs.
for _, externalIP := range svcInfo.ExternalIPStrings() {
for _, externalIP := range svcInfo.ExternalIPs() {
if hasEndpoints {
// Send traffic bound for external IPs to the "external
// destinations" chain.
tx.Add(&knftables.Element{
Map: kubeServiceIPsMap,
Key: []string{
externalIP,
externalIP.String(),
protocol,
strconv.Itoa(svcInfo.Port()),
},
@ -1176,7 +1175,7 @@ func (proxier *Proxier) syncProxyRules() {
tx.Add(&knftables.Element{
Map: kubeNoEndpointServicesMap,
Key: []string{
externalIP,
externalIP.String(),
protocol,
strconv.Itoa(svcInfo.Port()),
},
@ -1188,21 +1187,17 @@ func (proxier *Proxier) syncProxyRules() {
}
}
usesFWChain := len(svcInfo.LoadBalancerVIPStrings()) > 0 && len(svcInfo.LoadBalancerSourceRanges()) > 0
usesFWChain := len(svcInfo.LoadBalancerVIPs()) > 0 && len(svcInfo.LoadBalancerSourceRanges()) > 0
fwChain := svcInfo.firewallChainName
if usesFWChain {
ensureChain(fwChain, tx, activeChains)
var sources []string
allowFromNode := false
for _, src := range svcInfo.LoadBalancerSourceRanges() {
_, cidr, _ := netutils.ParseCIDRSloppy(src)
if cidr == nil {
continue
}
for _, cidr := range svcInfo.LoadBalancerSourceRanges() {
if len(sources) > 0 {
sources = append(sources, ",")
}
sources = append(sources, src)
sources = append(sources, cidr.String())
if cidr.Contains(proxier.nodeIP) {
allowFromNode = true
}
@ -1213,8 +1208,8 @@ func (proxier *Proxier) syncProxyRules() {
// will loop back with the source IP set to the VIP. We
// need the following rules to allow requests from this node.
if allowFromNode {
for _, lbip := range svcInfo.LoadBalancerVIPStrings() {
sources = append(sources, ",", lbip)
for _, lbip := range svcInfo.LoadBalancerVIPs() {
sources = append(sources, ",", lbip.String())
}
}
tx.Add(&knftables.Rule{
@ -1227,12 +1222,12 @@ func (proxier *Proxier) syncProxyRules() {
}
// Capture load-balancer ingress.
for _, lbip := range svcInfo.LoadBalancerVIPStrings() {
for _, lbip := range svcInfo.LoadBalancerVIPs() {
if hasEndpoints {
tx.Add(&knftables.Element{
Map: kubeServiceIPsMap,
Key: []string{
lbip,
lbip.String(),
protocol,
strconv.Itoa(svcInfo.Port()),
},
@ -1246,7 +1241,7 @@ func (proxier *Proxier) syncProxyRules() {
tx.Add(&knftables.Element{
Map: kubeFirewallIPsMap,
Key: []string{
lbip,
lbip.String(),
protocol,
strconv.Itoa(svcInfo.Port()),
},
@ -1261,11 +1256,11 @@ func (proxier *Proxier) syncProxyRules() {
// Either no endpoints at all (REJECT) or no endpoints for
// external traffic (DROP anything that didn't get short-circuited
// by the EXT chain.)
for _, lbip := range svcInfo.LoadBalancerVIPStrings() {
for _, lbip := range svcInfo.LoadBalancerVIPs() {
tx.Add(&knftables.Element{
Map: kubeNoEndpointServicesMap,
Key: []string{
lbip,
lbip.String(),
protocol,
strconv.Itoa(svcInfo.Port()),
},

2
pkg/proxy/nftables/proxier_test.go
View File

@ -175,7 +175,7 @@ func TestDeleteEndpointConnections(t *testing.T) {
}
endpointIP := proxyutil.IPPart(tc.endpoint)
_, fp := NewFakeProxier(proxyutil.GetIPFamilyFromIP(endpointIP))
_, fp := NewFakeProxier(proxyutil.GetIPFamilyFromIP(netutils.ParseIPSloppy(endpointIP)))
fp.exec = fexec
makeServiceMap(fp,

74
pkg/proxy/servicechangetracker_test.go
View File

@ -17,6 +17,7 @@ limitations under the License.
package proxy
import (
"net"
"reflect"
"testing"
"time"
@ -26,7 +27,6 @@ import (
"k8s.io/apimachinery/pkg/types"
"k8s.io/apimachinery/pkg/util/dump"
"k8s.io/apimachinery/pkg/util/intstr"
"k8s.io/apimachinery/pkg/util/sets"
utilfeature "k8s.io/apiserver/pkg/util/feature"
featuregatetesting "k8s.io/component-base/featuregate/testing"
"k8s.io/kubernetes/pkg/features"
@ -86,6 +86,24 @@ func makeServicePortName(ns, name, port string, protocol v1.Protocol) ServicePor
Protocol: protocol,
}
}
func makeIPs(ipStr ...string) []net.IP {
var ips []net.IP
for _, s := range ipStr {
ips = append(ips, netutils.ParseIPSloppy(s))
}
return ips
}
func mustMakeCIDRs(cidrStr ...string) []*net.IPNet {
var cidrs []*net.IPNet
for _, s := range cidrStr {
if _, n, err := netutils.ParseCIDRSloppy(s); err == nil {
cidrs = append(cidrs, n)
} else {
panic(err)
}
}
return cidrs
}
func TestServiceToServiceMap(t *testing.T) {
testClusterIPv4 := "10.0.0.1"
@ -187,10 +205,10 @@ func TestServiceToServiceMap(t *testing.T) {
}),
expected: map[ServicePortName]*BaseServicePortInfo{
makeServicePortName("ns1", "load-balancer", "port3", v1.ProtocolUDP): makeTestServiceInfo("172.16.55.11", 8675, "UDP", 0, func(bsvcPortInfo *BaseServicePortInfo) {
bsvcPortInfo.loadBalancerVIPs = []string{"10.1.2.4"}
bsvcPortInfo.loadBalancerVIPs = makeIPs("10.1.2.4")
}),
makeServicePortName("ns1", "load-balancer", "port4", v1.ProtocolUDP): makeTestServiceInfo("172.16.55.11", 8676, "UDP", 0, func(bsvcPortInfo *BaseServicePortInfo) {
bsvcPortInfo.loadBalancerVIPs = []string{"10.1.2.4"}
bsvcPortInfo.loadBalancerVIPs = makeIPs("10.1.2.4")
}),
},
},
@ -208,10 +226,10 @@ func TestServiceToServiceMap(t *testing.T) {
}),
expected: map[ServicePortName]*BaseServicePortInfo{
makeServicePortName("ns1", "load-balancer", "port3", v1.ProtocolUDP): makeTestServiceInfo("172.16.55.11", 8675, "UDP", 0, func(bsvcPortInfo *BaseServicePortInfo) {
bsvcPortInfo.loadBalancerVIPs = []string{"10.1.2.4"}
bsvcPortInfo.loadBalancerVIPs = makeIPs("10.1.2.4")
}),
makeServicePortName("ns1", "load-balancer", "port4", v1.ProtocolUDP): makeTestServiceInfo("172.16.55.11", 8676, "UDP", 0, func(bsvcPortInfo *BaseServicePortInfo) {
bsvcPortInfo.loadBalancerVIPs = []string{"10.1.2.4"}
bsvcPortInfo.loadBalancerVIPs = makeIPs("10.1.2.4")
}),
},
},
@ -229,10 +247,10 @@ func TestServiceToServiceMap(t *testing.T) {
}),
expected: map[ServicePortName]*BaseServicePortInfo{
makeServicePortName("ns1", "load-balancer", "port3", v1.ProtocolUDP): makeTestServiceInfo("172.16.55.11", 8675, "UDP", 0, func(bsvcPortInfo *BaseServicePortInfo) {
bsvcPortInfo.loadBalancerVIPs = []string{"10.1.2.4"}
bsvcPortInfo.loadBalancerVIPs = makeIPs("10.1.2.4")
}),
makeServicePortName("ns1", "load-balancer", "port4", v1.ProtocolUDP): makeTestServiceInfo("172.16.55.11", 8676, "UDP", 0, func(bsvcPortInfo *BaseServicePortInfo) {
bsvcPortInfo.loadBalancerVIPs = []string{"10.1.2.4"}
bsvcPortInfo.loadBalancerVIPs = makeIPs("10.1.2.4")
}),
},
},
@ -251,10 +269,10 @@ func TestServiceToServiceMap(t *testing.T) {
}),
expected: map[ServicePortName]*BaseServicePortInfo{
makeServicePortName("ns1", "load-balancer", "port3", v1.ProtocolUDP): makeTestServiceInfo("172.16.55.11", 8675, "UDP", 0, func(bsvcPortInfo *BaseServicePortInfo) {
bsvcPortInfo.loadBalancerVIPs = []string{"10.1.2.4"}
bsvcPortInfo.loadBalancerVIPs = makeIPs("10.1.2.4")
}),
makeServicePortName("ns1", "load-balancer", "port4", v1.ProtocolUDP): makeTestServiceInfo("172.16.55.11", 8676, "UDP", 0, func(bsvcPortInfo *BaseServicePortInfo) {
bsvcPortInfo.loadBalancerVIPs = []string{"10.1.2.4"}
bsvcPortInfo.loadBalancerVIPs = makeIPs("10.1.2.4")
}),
},
},
@ -294,10 +312,10 @@ func TestServiceToServiceMap(t *testing.T) {
}),
expected: map[ServicePortName]*BaseServicePortInfo{
makeServicePortName("ns1", "only-local-load-balancer", "portx", v1.ProtocolUDP): makeTestServiceInfo("172.16.55.12", 8677, "UDP", 345, func(bsvcPortInfo *BaseServicePortInfo) {
bsvcPortInfo.loadBalancerVIPs = []string{"10.1.2.3"}
bsvcPortInfo.loadBalancerVIPs = makeIPs("10.1.2.3")
}),
makeServicePortName("ns1", "only-local-load-balancer", "porty", v1.ProtocolUDP): makeTestServiceInfo("172.16.55.12", 8678, "UDP", 345, func(bsvcPortInfo *BaseServicePortInfo) {
bsvcPortInfo.loadBalancerVIPs = []string{"10.1.2.3"}
bsvcPortInfo.loadBalancerVIPs = makeIPs("10.1.2.3")
}),
},
},
@ -403,9 +421,9 @@ func TestServiceToServiceMap(t *testing.T) {
},
expected: map[ServicePortName]*BaseServicePortInfo{
makeServicePortName("test", "validIPv4", "testPort", v1.ProtocolTCP): makeTestServiceInfo(testClusterIPv4, 12345, "TCP", 0, func(bsvcPortInfo *BaseServicePortInfo) {
bsvcPortInfo.externalIPs = []string{testExternalIPv4}
bsvcPortInfo.loadBalancerSourceRanges = []string{testSourceRangeIPv4}
bsvcPortInfo.loadBalancerVIPs = []string{testExternalIPv4}
bsvcPortInfo.externalIPs = makeIPs(testExternalIPv4)
bsvcPortInfo.loadBalancerSourceRanges = mustMakeCIDRs(testSourceRangeIPv4)
bsvcPortInfo.loadBalancerVIPs = makeIPs(testExternalIPv4)
}),
},
},
@ -441,9 +459,9 @@ func TestServiceToServiceMap(t *testing.T) {
},
expected: map[ServicePortName]*BaseServicePortInfo{
makeServicePortName("test", "validIPv6", "testPort", v1.ProtocolTCP): makeTestServiceInfo(testClusterIPv6, 12345, "TCP", 0, func(bsvcPortInfo *BaseServicePortInfo) {
bsvcPortInfo.externalIPs = []string{testExternalIPv6}
bsvcPortInfo.loadBalancerSourceRanges = []string{testSourceRangeIPv6}
bsvcPortInfo.loadBalancerVIPs = []string{testExternalIPv6}
bsvcPortInfo.externalIPs = makeIPs(testExternalIPv6)
bsvcPortInfo.loadBalancerSourceRanges = mustMakeCIDRs(testSourceRangeIPv6)
bsvcPortInfo.loadBalancerVIPs = makeIPs(testExternalIPv6)
}),
},
},
@ -479,9 +497,9 @@ func TestServiceToServiceMap(t *testing.T) {
},
expected: map[ServicePortName]*BaseServicePortInfo{
makeServicePortName("test", "filterIPv6InIPV4Mode", "testPort", v1.ProtocolTCP): makeTestServiceInfo(testClusterIPv4, 12345, "TCP", 0, func(bsvcPortInfo *BaseServicePortInfo) {
bsvcPortInfo.externalIPs = []string{testExternalIPv4}
bsvcPortInfo.loadBalancerSourceRanges = []string{testSourceRangeIPv4}
bsvcPortInfo.loadBalancerVIPs = []string{testExternalIPv4}
bsvcPortInfo.externalIPs = makeIPs(testExternalIPv4)
bsvcPortInfo.loadBalancerSourceRanges = mustMakeCIDRs(testSourceRangeIPv4)
bsvcPortInfo.loadBalancerVIPs = makeIPs(testExternalIPv4)
}),
},
},
@ -517,9 +535,9 @@ func TestServiceToServiceMap(t *testing.T) {
},
expected: map[ServicePortName]*BaseServicePortInfo{
makeServicePortName("test", "filterIPv4InIPV6Mode", "testPort", v1.ProtocolTCP): makeTestServiceInfo(testClusterIPv6, 12345, "TCP", 0, func(bsvcPortInfo *BaseServicePortInfo) {
bsvcPortInfo.externalIPs = []string{testExternalIPv6}
bsvcPortInfo.loadBalancerSourceRanges = []string{testSourceRangeIPv6}
bsvcPortInfo.loadBalancerVIPs = []string{testExternalIPv6}
bsvcPortInfo.externalIPs = makeIPs(testExternalIPv6)
bsvcPortInfo.loadBalancerSourceRanges = mustMakeCIDRs(testSourceRangeIPv6)
bsvcPortInfo.loadBalancerVIPs = makeIPs(testExternalIPv6)
}),
},
},
@ -546,7 +564,7 @@ func TestServiceToServiceMap(t *testing.T) {
},
expected: map[ServicePortName]*BaseServicePortInfo{
makeServicePortName("test", "extra-space", "testPort", v1.ProtocolTCP): makeTestServiceInfo(testClusterIPv4, 12345, "TCP", 0, func(bsvcPortInfo *BaseServicePortInfo) {
bsvcPortInfo.loadBalancerSourceRanges = []string{"10.1.2.0/28"}
bsvcPortInfo.loadBalancerSourceRanges = mustMakeCIDRs("10.1.2.0/28")
}),
},
},
@ -572,8 +590,8 @@ func TestServiceToServiceMap(t *testing.T) {
svcInfo.port != expectedInfo.port ||
svcInfo.protocol != expectedInfo.protocol ||
svcInfo.healthCheckNodePort != expectedInfo.healthCheckNodePort ||
!sets.New[string](svcInfo.externalIPs...).Equal(sets.New[string](expectedInfo.externalIPs...)) ||
!sets.New[string](svcInfo.loadBalancerSourceRanges...).Equal(sets.New[string](expectedInfo.loadBalancerSourceRanges...)) ||
!reflect.DeepEqual(svcInfo.externalIPs, expectedInfo.externalIPs) ||
!reflect.DeepEqual(svcInfo.loadBalancerSourceRanges, expectedInfo.loadBalancerSourceRanges) ||
!reflect.DeepEqual(svcInfo.loadBalancerVIPs, expectedInfo.loadBalancerVIPs) {
t.Errorf("[%s] expected new[%v]to be %v, got %v", tc.desc, svcKey, expectedInfo, *svcInfo)
}
@ -583,8 +601,8 @@ func TestServiceToServiceMap(t *testing.T) {
svcInfo.port != expectedInfo.port ||
svcInfo.protocol != expectedInfo.protocol ||
svcInfo.healthCheckNodePort != expectedInfo.healthCheckNodePort ||
!sets.New[string](svcInfo.externalIPs...).Equal(sets.New[string](expectedInfo.externalIPs...)) ||
!sets.New[string](svcInfo.loadBalancerSourceRanges...).Equal(sets.New[string](expectedInfo.loadBalancerSourceRanges...)) ||
!reflect.DeepEqual(svcInfo.externalIPs, expectedInfo.externalIPs) ||
!reflect.DeepEqual(svcInfo.loadBalancerSourceRanges, expectedInfo.loadBalancerSourceRanges) ||
!reflect.DeepEqual(svcInfo.loadBalancerVIPs, expectedInfo.loadBalancerVIPs) {
t.Errorf("expected new[%v]to be %v, got %v", svcKey, expectedInfo, *svcInfo)
}

53
pkg/proxy/serviceport.go
View File

@ -19,7 +19,6 @@ package proxy
import (
"fmt"
"net"
"strings"
v1 "k8s.io/api/core/v1"
"k8s.io/klog/v2"
@ -40,14 +39,14 @@ type ServicePort interface {
SessionAffinityType() v1.ServiceAffinity
// StickyMaxAgeSeconds returns service max connection age
StickyMaxAgeSeconds() int
// ExternalIPStrings returns service ExternalIPs as a string array.
ExternalIPStrings() []string
// LoadBalancerVIPStrings returns service LoadBalancerIPs which are VIP mode as a string array.
LoadBalancerVIPStrings() []string
// ExternalIPs returns service ExternalIPs
ExternalIPs() []net.IP
// LoadBalancerVIPs returns service LoadBalancerIPs which are VIP mode
LoadBalancerVIPs() []net.IP
// Protocol returns service protocol.
Protocol() v1.Protocol
// LoadBalancerSourceRanges returns service LoadBalancerSourceRanges if present empty array if not
LoadBalancerSourceRanges() []string
LoadBalancerSourceRanges() []*net.IPNet
// HealthCheckNodePort returns service health check node port if present. If return 0, it means not present.
HealthCheckNodePort() int
// NodePort returns a service Node port if present. If return 0, it means not present.
@ -78,11 +77,11 @@ type BaseServicePortInfo struct {
port int
protocol v1.Protocol
nodePort int
loadBalancerVIPs []string
loadBalancerVIPs []net.IP
sessionAffinityType v1.ServiceAffinity
stickyMaxAgeSeconds int
externalIPs []string
loadBalancerSourceRanges []string
externalIPs []net.IP
loadBalancerSourceRanges []*net.IPNet
healthCheckNodePort int
externalPolicyLocal bool
internalPolicyLocal bool
@ -122,7 +121,7 @@ func (bsvcPortInfo *BaseServicePortInfo) Protocol() v1.Protocol {
}
// LoadBalancerSourceRanges is part of ServicePort interface
func (bsvcPortInfo *BaseServicePortInfo) LoadBalancerSourceRanges() []string {
func (bsvcPortInfo *BaseServicePortInfo) LoadBalancerSourceRanges() []*net.IPNet {
return bsvcPortInfo.loadBalancerSourceRanges
}
@ -136,13 +135,13 @@ func (bsvcPortInfo *BaseServicePortInfo) NodePort() int {
return bsvcPortInfo.nodePort
}
// ExternalIPStrings is part of ServicePort interface.
func (bsvcPortInfo *BaseServicePortInfo) ExternalIPStrings() []string {
// ExternalIPs is part of ServicePort interface.
func (bsvcPortInfo *BaseServicePortInfo) ExternalIPs() []net.IP {
return bsvcPortInfo.externalIPs
}
// LoadBalancerVIPStrings is part of ServicePort interface.
func (bsvcPortInfo *BaseServicePortInfo) LoadBalancerVIPStrings() []string {
// LoadBalancerVIPs is part of ServicePort interface.
func (bsvcPortInfo *BaseServicePortInfo) LoadBalancerVIPs() []net.IP {
return bsvcPortInfo.loadBalancerVIPs
}
@ -208,34 +207,29 @@ func newBaseServiceInfo(service *v1.Service, ipFamily v1.IPFamily, port *v1.Serv
info.hintsAnnotation = service.Annotations[v1.AnnotationTopologyMode]
}
loadBalancerSourceRanges := make([]string, len(service.Spec.LoadBalancerSourceRanges))
for i, sourceRange := range service.Spec.LoadBalancerSourceRanges {
loadBalancerSourceRanges[i] = strings.TrimSpace(sourceRange)
}
// filter external ips, source ranges and ingress ips
// prior to dual stack services, this was considered an error, but with dual stack
// services, this is actually expected. Hence we downgraded from reporting by events
// to just log lines with high verbosity
ipFamilyMap := proxyutil.MapIPsByIPFamily(service.Spec.ExternalIPs)
info.externalIPs = ipFamilyMap[ipFamily]
// Log the IPs not matching the ipFamily
if ips, ok := ipFamilyMap[proxyutil.OtherIPFamily(ipFamily)]; ok && len(ips) > 0 {
klog.V(4).InfoS("Service change tracker ignored the following external IPs for given service as they don't match IP Family",
"ipFamily", ipFamily, "externalIPs", strings.Join(ips, ", "), "service", klog.KObj(service))
"ipFamily", ipFamily, "externalIPs", ips, "service", klog.KObj(service))
}
ipFamilyMap = proxyutil.MapCIDRsByIPFamily(loadBalancerSourceRanges)
info.loadBalancerSourceRanges = ipFamilyMap[ipFamily]
cidrFamilyMap := proxyutil.MapCIDRsByIPFamily(service.Spec.LoadBalancerSourceRanges)
info.loadBalancerSourceRanges = cidrFamilyMap[ipFamily]
// Log the CIDRs not matching the ipFamily
if cidrs, ok := ipFamilyMap[proxyutil.OtherIPFamily(ipFamily)]; ok && len(cidrs) > 0 {
if cidrs, ok := cidrFamilyMap[proxyutil.OtherIPFamily(ipFamily)]; ok && len(cidrs) > 0 {
klog.V(4).InfoS("Service change tracker ignored the following load balancer source ranges for given Service as they don't match IP Family",
"ipFamily", ipFamily, "loadBalancerSourceRanges", strings.Join(cidrs, ", "), "service", klog.KObj(service))
"ipFamily", ipFamily, "loadBalancerSourceRanges", cidrs, "service", klog.KObj(service))
}
// Obtain Load Balancer Ingress
var invalidIPs []string
var invalidIPs []net.IP
for _, ing := range service.Status.LoadBalancer.Ingress {
if ing.IP == "" {
continue
@ -252,15 +246,16 @@ func newBaseServiceInfo(service *v1.Service, ipFamily v1.IPFamily, port *v1.Serv
// kube-proxy does not implement IP family translation, skip addresses with
// different IP family
if ingFamily := proxyutil.GetIPFamilyFromIP(ing.IP); ingFamily == ipFamily {
info.loadBalancerVIPs = append(info.loadBalancerVIPs, ing.IP)
ip := netutils.ParseIPSloppy(ing.IP) // (already verified as an IP-address)
if ingFamily := proxyutil.GetIPFamilyFromIP(ip); ingFamily == ipFamily {
info.loadBalancerVIPs = append(info.loadBalancerVIPs, ip)
} else {
invalidIPs = append(invalidIPs, ing.IP)
invalidIPs = append(invalidIPs, ip)
}
}
if len(invalidIPs) > 0 {
klog.V(4).InfoS("Service change tracker ignored the following load balancer ingress IPs for given Service as they don't match the IP Family",
"ipFamily", ipFamily, "loadBalancerIngressIPs", strings.Join(invalidIPs, ", "), "service", klog.KObj(service))
"ipFamily", ipFamily, "loadBalancerIngressIPs", invalidIPs, "service", klog.KObj(service))
}
if apiservice.NeedsHealthCheck(service) {

58
pkg/proxy/util/utils.go
View File

@ -180,49 +180,49 @@ func LogAndEmitIncorrectIPVersionEvent(recorder events.EventRecorder, fieldName,
}
// MapIPsByIPFamily maps a slice of IPs to their respective IP families (v4 or v6)
func MapIPsByIPFamily(ipStrings []string) map[v1.IPFamily][]string {
ipFamilyMap := map[v1.IPFamily][]string{}
for _, ip := range ipStrings {
// Handle only the valid IPs
if ipFamily := GetIPFamilyFromIP(ip); ipFamily != v1.IPFamilyUnknown {
func MapIPsByIPFamily(ipStrings []string) map[v1.IPFamily][]net.IP {
ipFamilyMap := map[v1.IPFamily][]net.IP{}
for _, ipStr := range ipStrings {
ip := netutils.ParseIPSloppy(ipStr)
if ip != nil {
// Since ip is parsed ok, GetIPFamilyFromIP will never return v1.IPFamilyUnknown
ipFamily := GetIPFamilyFromIP(ip)
ipFamilyMap[ipFamily] = append(ipFamilyMap[ipFamily], ip)
} else {
// this function is called in multiple places. All of which
// have sanitized data. Except the case of ExternalIPs which is
// not validated by api-server. Specifically empty strings
// validation. Which yields into a lot of bad error logs.
// check for empty string
if len(strings.TrimSpace(ip)) != 0 {
klog.ErrorS(nil, "Skipping invalid IP", "ip", ip)
// ExternalIPs may not be validated by the api-server.
// Specifically empty strings validation, which yields into a lot
// of bad error logs.
if len(strings.TrimSpace(ipStr)) != 0 {
klog.ErrorS(nil, "Skipping invalid IP", "ip", ipStr)
}
}
}
return ipFamilyMap
}
// MapCIDRsByIPFamily maps a slice of IPs to their respective IP families (v4 or v6)
func MapCIDRsByIPFamily(cidrStrings []string) map[v1.IPFamily][]string {
ipFamilyMap := map[v1.IPFamily][]string{}
for _, cidr := range cidrStrings {
// Handle only the valid CIDRs
if ipFamily := getIPFamilyFromCIDR(cidr); ipFamily != v1.IPFamilyUnknown {
ipFamilyMap[ipFamily] = append(ipFamilyMap[ipFamily], cidr)
} else {
klog.ErrorS(nil, "Skipping invalid CIDR", "cidr", cidr)
// MapCIDRsByIPFamily maps a slice of CIDRs to their respective IP families (v4 or v6)
func MapCIDRsByIPFamily(cidrsStrings []string) map[v1.IPFamily][]*net.IPNet {
ipFamilyMap := map[v1.IPFamily][]*net.IPNet{}
for _, cidrStrUntrimmed := range cidrsStrings {
cidrStr := strings.TrimSpace(cidrStrUntrimmed)
_, cidr, err := netutils.ParseCIDRSloppy(cidrStr)
if err != nil {
// Ignore empty strings. Same as in MapIPsByIPFamily
if len(cidrStr) != 0 {
klog.ErrorS(err, "Invalid CIDR ignored", "CIDR", cidrStr)
}
continue
}
// since we just succefully parsed the CIDR, IPFamilyOfCIDR will never return "IPFamilyUnknown"
ipFamily := convertToV1IPFamily(netutils.IPFamilyOfCIDR(cidr))
ipFamilyMap[ipFamily] = append(ipFamilyMap[ipFamily], cidr)
}
return ipFamilyMap
}
// GetIPFamilyFromIP Returns the IP family of ipStr, or IPFamilyUnknown if ipStr can't be parsed as an IP
func GetIPFamilyFromIP(ipStr string) v1.IPFamily {
return convertToV1IPFamily(netutils.IPFamilyOfString(ipStr))
}
// Returns the IP family of cidrStr, or IPFamilyUnknown if cidrStr can't be parsed as a CIDR
func getIPFamilyFromCIDR(cidrStr string) v1.IPFamily {
return convertToV1IPFamily(netutils.IPFamilyOfCIDRString(cidrStr))
func GetIPFamilyFromIP(ip net.IP) v1.IPFamily {
return convertToV1IPFamily(netutils.IPFamilyOf(ip))
}
// Convert netutils.IPFamily to v1.IPFamily

29
pkg/proxy/util/utils_test.go
View File

@ -338,10 +338,18 @@ func TestMapIPsByIPFamily(t *testing.T) {
ipMap := MapIPsByIPFamily(testcase.ipString)
if !reflect.DeepEqual(testcase.expectCorrect, ipMap[ipFamily]) {
var ipStr []string
for _, ip := range ipMap[ipFamily] {
ipStr = append(ipStr, ip.String())
}
if !reflect.DeepEqual(testcase.expectCorrect, ipStr) {
t.Errorf("Test %v failed: expected %v, got %v", testcase.desc, testcase.expectCorrect, ipMap[ipFamily])
}
if !reflect.DeepEqual(testcase.expectIncorrect, ipMap[otherIPFamily]) {
ipStr = nil
for _, ip := range ipMap[otherIPFamily] {
ipStr = append(ipStr, ip.String())
}
if !reflect.DeepEqual(testcase.expectIncorrect, ipStr) {
t.Errorf("Test %v failed: expected %v, got %v", testcase.desc, testcase.expectIncorrect, ipMap[otherIPFamily])
}
})
@ -440,11 +448,20 @@ func TestMapCIDRsByIPFamily(t *testing.T) {
cidrMap := MapCIDRsByIPFamily(testcase.ipString)
if !reflect.DeepEqual(testcase.expectCorrect, cidrMap[ipFamily]) {
t.Errorf("Test %v failed: expected %v, got %v", testcase.desc, testcase.expectCorrect, cidrMap[ipFamily])
var cidrStr []string
for _, cidr := range cidrMap[ipFamily] {
cidrStr = append(cidrStr, cidr.String())
}
if !reflect.DeepEqual(testcase.expectIncorrect, cidrMap[otherIPFamily]) {
t.Errorf("Test %v failed: expected %v, got %v", testcase.desc, testcase.expectIncorrect, cidrMap[otherIPFamily])
var cidrStrOther []string
for _, cidr := range cidrMap[otherIPFamily] {
cidrStrOther = append(cidrStrOther, cidr.String())
}
if !reflect.DeepEqual(testcase.expectCorrect, cidrStr) {
t.Errorf("Test %v failed: expected %v, got %v", testcase.desc, testcase.expectCorrect, cidrStr)
}
if !reflect.DeepEqual(testcase.expectIncorrect, cidrStrOther) {
t.Errorf("Test %v failed: expected %v, got %v", testcase.desc, testcase.expectIncorrect, cidrStrOther)
}
})
}