github/kubernetes
4
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

kube-proxy: store LoadBalancerVIPs as net.IP

Browse Source 
They were stored as strings which could be non-canonical
and cause problems
This commit is contained in:
Lars Ekman
2024-01-07 08:33:30 +01:00
parent 564b80b1e1
commit d2294007b0
8 changed files with 67 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -53,8 +53,8 @@ func deleteStaleServiceConntrackEntries(isIPv6 bool, exec utilexec.Interface, sv
for _, extIP := range svcInfo.ExternalIPStrings() {
conntrackCleanupServiceIPs.Insert(extIP)
}
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 {
@@ -103,8 +103,8 @@ func deleteStaleEndpointConntrackEntries(exec utilexec.Interface, svcPortMap pro
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)
}

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

@@ -1014,7 +1014,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
}
@@ -1105,13 +1105,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 +1121,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 +1130,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,
)
@@ -1309,10 +1309,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))
}
}

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

@@ -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,
@@ -1190,7 +1190,7 @@ func (proxier *Proxier) syncProxyRules() {
for _, src := range svcInfo.LoadBalancerSourceRanges() {
// ipset call
entry = &utilipset.Entry{
IP: ingress,
IP: ingress.String(),
Port: svcInfo.Port(),
Protocol: protocol,
Net: src,
@@ -1214,10 +1214,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 +1234,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,

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

@@ -1188,7 +1188,7 @@ 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)
@@ -1213,8 +1213,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 +1227,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 +1246,7 @@ func (proxier *Proxier) syncProxyRules() {
tx.Add(&knftables.Element{
Map: kubeFirewallIPsMap,
Key: []string{
lbip,
lbip.String(),
protocol,
strconv.Itoa(svcInfo.Port()),
},
@@ -1261,11 +1261,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,

36
pkg/proxy/servicechangetracker_test.go
View File

@@ -17,6 +17,7 @@ limitations under the License.
package proxy
import (
"net"
"reflect"
"testing"
"time"
@@ -86,6 +87,13 @@ 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 TestServiceToServiceMap(t *testing.T) {
testClusterIPv4 := "10.0.0.1"
@@ -187,10 +195,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 +216,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 +237,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 +259,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 +302,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")
}),
},
},
@@ -405,7 +413,7 @@ func TestServiceToServiceMap(t *testing.T) {
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.loadBalancerVIPs = makeIPs(testExternalIPv4)
}),
},
},
@@ -443,7 +451,7 @@ func TestServiceToServiceMap(t *testing.T) {
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.loadBalancerVIPs = makeIPs(testExternalIPv6)
}),
},
},
@@ -481,7 +489,7 @@ func TestServiceToServiceMap(t *testing.T) {
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.loadBalancerVIPs = makeIPs(testExternalIPv4)
}),
},
},
@@ -519,7 +527,7 @@ func TestServiceToServiceMap(t *testing.T) {
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.loadBalancerVIPs = makeIPs(testExternalIPv6)
}),
},
},

21
pkg/proxy/serviceport.go
View File

@@ -42,8 +42,8 @@ type ServicePort interface {
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
// 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
@@ -78,7 +78,7 @@ type BaseServicePortInfo struct {
port int
protocol v1.Protocol
nodePort int
loadBalancerVIPs []string
loadBalancerVIPs []net.IP
sessionAffinityType v1.ServiceAffinity
stickyMaxAgeSeconds int
externalIPs []string
@@ -141,8 +141,8 @@ func (bsvcPortInfo *BaseServicePortInfo) ExternalIPStrings() []string {
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
}
@@ -235,7 +235,7 @@ func newBaseServiceInfo(service *v1.Service, ipFamily v1.IPFamily, port *v1.Serv
}
// Obtain Load Balancer Ingress
var invalidIPs []string
var invalidIPs []net.IP
for _, ing := range service.Status.LoadBalancer.Ingress {
if ing.IP == "" {
continue
@@ -252,15 +252,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) {

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

@@ -182,19 +182,19 @@ 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 {
for _, ipStr := range ipStrings {
ip := netutils.ParseIPSloppy(ipStr)
// Handle only the valid IPs
if ipFamily := GetIPFamilyFromIP(ip); ipFamily != v1.IPFamilyUnknown {
ipFamilyMap[ipFamily] = append(ipFamilyMap[ipFamily], ip)
ipFamilyMap[ipFamily] = append(ipFamilyMap[ipFamily], ipStr)
} 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)
if len(strings.TrimSpace(ipStr)) != 0 {
klog.ErrorS(nil, "Skipping invalid IP", "ip", ipStr)
}
}
}
@@ -216,8 +216,8 @@ func MapCIDRsByIPFamily(cidrStrings []string) map[v1.IPFamily][]string {
}
// 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))
func GetIPFamilyFromIP(ip net.IP) v1.IPFamily {
return convertToV1IPFamily(netutils.IPFamilyOf(ip))
}
// Returns the IP family of cidrStr, or IPFamilyUnknown if cidrStr can't be parsed as a CIDR