github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2025-07-27 05:27:21 +00:00
Code Issues Projects Releases Wiki Activity

Merge pull request #106158 from thockin/kp-cleanup

Browse Source 
kube-proxy iptables code clarity cleanups
This commit is contained in:
Kubernetes Prow Robot 2021-11-05 17:20:51 -07:00 committed by GitHub
commit bdb9c0824c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 179 additions and 198 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -223,10 +223,10 @@ type Proxier struct {
// that are significantly impacting performance. // that are significantly impacting performance.
iptablesData *bytes.Buffer iptablesData *bytes.Buffer
existingFilterChainsData *bytes.Buffer existingFilterChainsData *bytes.Buffer
filterChains *bytes.Buffer filterChains utilproxy.LineBuffer
filterRules *bytes.Buffer filterRules utilproxy.LineBuffer
natChains *bytes.Buffer natChains utilproxy.LineBuffer
natRules *bytes.Buffer natRules utilproxy.LineBuffer
// endpointChainsNumber is the total amount of endpointChains across all // endpointChainsNumber is the total amount of endpointChains across all
// services that we will generate (it is computed at the beginning of // services that we will generate (it is computed at the beginning of
@ -315,10 +315,10 @@ func NewProxier(ipt utiliptables.Interface,
precomputedProbabilities: make([]string, 0, 1001), precomputedProbabilities: make([]string, 0, 1001),
iptablesData: bytes.NewBuffer(nil), iptablesData: bytes.NewBuffer(nil),
existingFilterChainsData: bytes.NewBuffer(nil), existingFilterChainsData: bytes.NewBuffer(nil),
filterChains: bytes.NewBuffer(nil), filterChains: utilproxy.LineBuffer{},
filterRules: bytes.NewBuffer(nil), filterRules: utilproxy.LineBuffer{},
natChains: bytes.NewBuffer(nil), natChains: utilproxy.LineBuffer{},
natRules: bytes.NewBuffer(nil), natRules: utilproxy.LineBuffer{},
nodePortAddresses: nodePortAddresses, nodePortAddresses: nodePortAddresses,
networkInterfacer: utilproxy.RealNetwork{}, networkInterfacer: utilproxy.RealNetwork{},
} }
@ -432,26 +432,26 @@ func CleanupLeftovers(ipt utiliptables.Interface) (encounteredError bool) {
encounteredError = true encounteredError = true
} else { } else {
existingNATChains := utiliptables.GetChainLines(utiliptables.TableNAT, iptablesData.Bytes()) existingNATChains := utiliptables.GetChainLines(utiliptables.TableNAT, iptablesData.Bytes())
natChains := bytes.NewBuffer(nil) natChains := &utilproxy.LineBuffer{}
natRules := bytes.NewBuffer(nil) natRules := &utilproxy.LineBuffer{}
utilproxy.WriteLine(natChains, "*nat") natChains.Write("*nat")
// Start with chains we know we need to remove. // Start with chains we know we need to remove.
for _, chain := range []utiliptables.Chain{kubeServicesChain, kubeNodePortsChain, kubePostroutingChain} { for _, chain := range []utiliptables.Chain{kubeServicesChain, kubeNodePortsChain, kubePostroutingChain} {
if _, found := existingNATChains[chain]; found { if _, found := existingNATChains[chain]; found {
chainString := string(chain) chainString := string(chain)
utilproxy.WriteBytesLine(natChains, existingNATChains[chain]) // flush natChains.WriteBytes(existingNATChains[chain]) // flush
utilproxy.WriteLine(natRules, "-X", chainString) // delete natRules.Write("-X", chainString) // delete
} }
} }
// Hunt for service and endpoint chains. // Hunt for service and endpoint chains.
for chain := range existingNATChains { for chain := range existingNATChains {
chainString := string(chain) chainString := string(chain)
if strings.HasPrefix(chainString, "KUBE-SVC-") || strings.HasPrefix(chainString, "KUBE-SEP-") || strings.HasPrefix(chainString, "KUBE-FW-") || strings.HasPrefix(chainString, "KUBE-XLB-") { if strings.HasPrefix(chainString, "KUBE-SVC-") || strings.HasPrefix(chainString, "KUBE-SEP-") || strings.HasPrefix(chainString, "KUBE-FW-") || strings.HasPrefix(chainString, "KUBE-XLB-") {
utilproxy.WriteBytesLine(natChains, existingNATChains[chain]) // flush natChains.WriteBytes(existingNATChains[chain]) // flush
utilproxy.WriteLine(natRules, "-X", chainString) // delete natRules.Write("-X", chainString) // delete
} }
} }
utilproxy.WriteLine(natRules, "COMMIT") natRules.Write("COMMIT")
natLines := append(natChains.Bytes(), natRules.Bytes()...) natLines := append(natChains.Bytes(), natRules.Bytes()...)
// Write it. // Write it.
err = ipt.Restore(utiliptables.TableNAT, natLines, utiliptables.NoFlushTables, utiliptables.RestoreCounters) err = ipt.Restore(utiliptables.TableNAT, natLines, utiliptables.NoFlushTables, utiliptables.RestoreCounters)
@ -469,17 +469,17 @@ func CleanupLeftovers(ipt utiliptables.Interface) (encounteredError bool) {
encounteredError = true encounteredError = true
} else { } else {
existingFilterChains := utiliptables.GetChainLines(utiliptables.TableFilter, iptablesData.Bytes()) existingFilterChains := utiliptables.GetChainLines(utiliptables.TableFilter, iptablesData.Bytes())
filterChains := bytes.NewBuffer(nil) filterChains := &utilproxy.LineBuffer{}
filterRules := bytes.NewBuffer(nil) filterRules := &utilproxy.LineBuffer{}
utilproxy.WriteLine(filterChains, "*filter") filterChains.Write("*filter")
for _, chain := range []utiliptables.Chain{kubeServicesChain, kubeExternalServicesChain, kubeForwardChain, kubeNodePortsChain} { for _, chain := range []utiliptables.Chain{kubeServicesChain, kubeExternalServicesChain, kubeForwardChain, kubeNodePortsChain} {
if _, found := existingFilterChains[chain]; found { if _, found := existingFilterChains[chain]; found {
chainString := string(chain) chainString := string(chain)
utilproxy.WriteBytesLine(filterChains, existingFilterChains[chain]) filterChains.WriteBytes(existingFilterChains[chain])
utilproxy.WriteLine(filterRules, "-X", chainString) filterRules.Write("-X", chainString)
} }
} }
utilproxy.WriteLine(filterRules, "COMMIT") filterRules.Write("COMMIT")
filterLines := append(filterChains.Bytes(), filterRules.Bytes()...) filterLines := append(filterChains.Bytes(), filterRules.Bytes()...)
// Write it. // Write it.
if err := ipt.Restore(utiliptables.TableFilter, filterLines, utiliptables.NoFlushTables, utiliptables.RestoreCounters); err != nil { if err := ipt.Restore(utiliptables.TableFilter, filterLines, utiliptables.NoFlushTables, utiliptables.RestoreCounters); err != nil {
@ -760,6 +760,10 @@ func (proxier *Proxier) deleteEndpointConnections(connectionMap []proxy.ServiceE
} }
} }
func prepend(sl []string, args ...string) []string {
return append(args, sl...)
}
const endpointChainsNumberThreshold = 1000 const endpointChainsNumberThreshold = 1000
// Assumes proxier.mu is held. // Assumes proxier.mu is held.
@ -890,23 +894,23 @@ func (proxier *Proxier) syncProxyRules() {
proxier.natRules.Reset() proxier.natRules.Reset()
// Write table headers. // Write table headers.
utilproxy.WriteLine(proxier.filterChains, "*filter") proxier.filterChains.Write("*filter")
utilproxy.WriteLine(proxier.natChains, "*nat") proxier.natChains.Write("*nat")
// Make sure we keep stats for the top-level chains, if they existed // Make sure we keep stats for the top-level chains, if they existed
// (which most should have because we created them above). // (which most should have because we created them above).
for _, chainName := range []utiliptables.Chain{kubeServicesChain, kubeExternalServicesChain, kubeForwardChain, kubeNodePortsChain} { for _, chainName := range []utiliptables.Chain{kubeServicesChain, kubeExternalServicesChain, kubeForwardChain, kubeNodePortsChain} {
if chain, ok := existingFilterChains[chainName]; ok { if chain, ok := existingFilterChains[chainName]; ok {
utilproxy.WriteBytesLine(proxier.filterChains, chain) proxier.filterChains.WriteBytes(chain)
} else { } else {
utilproxy.WriteLine(proxier.filterChains, utiliptables.MakeChainLine(chainName)) proxier.filterChains.Write(utiliptables.MakeChainLine(chainName))
} }
} }
for _, chainName := range []utiliptables.Chain{kubeServicesChain, kubeNodePortsChain, kubePostroutingChain, KubeMarkMasqChain} { for _, chainName := range []utiliptables.Chain{kubeServicesChain, kubeNodePortsChain, kubePostroutingChain, KubeMarkMasqChain} {
if chain, ok := existingNATChains[chainName]; ok { if chain, ok := existingNATChains[chainName]; ok {
utilproxy.WriteBytesLine(proxier.natChains, chain) proxier.natChains.WriteBytes(chain)
} else { } else {
utilproxy.WriteLine(proxier.natChains, utiliptables.MakeChainLine(chainName)) proxier.natChains.Write(utiliptables.MakeChainLine(chainName))
} }
} }
@ -914,17 +918,16 @@ func (proxier *Proxier) syncProxyRules() {
// this so that it is easier to flush and change, for example if the mark // this so that it is easier to flush and change, for example if the mark
// value should ever change. // value should ever change.
// NB: THIS MUST MATCH the corresponding code in the kubelet // NB: THIS MUST MATCH the corresponding code in the kubelet
utilproxy.WriteLine(proxier.natRules, []string{ proxier.natRules.Write(
"-A", string(kubePostroutingChain), "-A", string(kubePostroutingChain),
"-m", "mark", "!", "--mark", fmt.Sprintf("%s/%s", proxier.masqueradeMark, proxier.masqueradeMark), "-m", "mark", "!", "--mark", fmt.Sprintf("%s/%s", proxier.masqueradeMark, proxier.masqueradeMark),
"-j", "RETURN", "-j", "RETURN",
}...) )
// Clear the mark to avoid re-masquerading if the packet re-traverses the network stack. // Clear the mark to avoid re-masquerading if the packet re-traverses the network stack.
utilproxy.WriteLine(proxier.natRules, []string{ proxier.natRules.Write(
"-A", string(kubePostroutingChain), "-A", string(kubePostroutingChain),
// XOR proxier.masqueradeMark to unset it
"-j", "MARK", "--xor-mark", proxier.masqueradeMark, "-j", "MARK", "--xor-mark", proxier.masqueradeMark,
}...) )
masqRule := []string{ masqRule := []string{
"-A", string(kubePostroutingChain), "-A", string(kubePostroutingChain),
"-m", "comment", "--comment", `"kubernetes service traffic requiring SNAT"`, "-m", "comment", "--comment", `"kubernetes service traffic requiring SNAT"`,
@ -933,15 +936,15 @@ func (proxier *Proxier) syncProxyRules() {
if proxier.iptables.HasRandomFully() { if proxier.iptables.HasRandomFully() {
masqRule = append(masqRule, "--random-fully") masqRule = append(masqRule, "--random-fully")
} }
utilproxy.WriteLine(proxier.natRules, masqRule...) proxier.natRules.Write(masqRule...)
// Install the kubernetes-specific masquerade mark rule. We use a whole chain for // Install the kubernetes-specific masquerade mark rule. We use a whole chain for
// this so that it is easier to flush and change, for example if the mark // this so that it is easier to flush and change, for example if the mark
// value should ever change. // value should ever change.
utilproxy.WriteLine(proxier.natRules, []string{ proxier.natRules.Write(
"-A", string(KubeMarkMasqChain), "-A", string(KubeMarkMasqChain),
"-j", "MARK", "--or-mark", proxier.masqueradeMark, "-j", "MARK", "--or-mark", proxier.masqueradeMark,
}...) )
// Accumulate NAT chains to keep. // Accumulate NAT chains to keep.
activeNATChains := map[utiliptables.Chain]bool{} // use a map as a set activeNATChains := map[utiliptables.Chain]bool{} // use a map as a set
@ -1054,16 +1057,16 @@ func (proxier *Proxier) syncProxyRules() {
// Create the endpoint chain, retaining counters if possible. // Create the endpoint chain, retaining counters if possible.
if chain, ok := existingNATChains[endpointChain]; ok { if chain, ok := existingNATChains[endpointChain]; ok {
utilproxy.WriteBytesLine(proxier.natChains, chain) proxier.natChains.WriteBytes(chain)
} else { } else {
utilproxy.WriteLine(proxier.natChains, utiliptables.MakeChainLine(endpointChain)) proxier.natChains.Write(utiliptables.MakeChainLine(endpointChain))
} }
activeNATChains[endpointChain] = true activeNATChains[endpointChain] = true
args = append(args[:0], "-A", string(endpointChain)) args = append(args[:0], "-A", string(endpointChain))
args = proxier.appendServiceCommentLocked(args, svcNameString) args = proxier.appendServiceCommentLocked(args, svcNameString)
// Handle traffic that loops back to the originator with SNAT. // Handle traffic that loops back to the originator with SNAT.
utilproxy.WriteLine(proxier.natRules, append(args, proxier.natRules.Write(append(args,
"-s", utilproxy.ToCIDR(netutils.ParseIPSloppy(epInfo.IP())), "-s", utilproxy.ToCIDR(netutils.ParseIPSloppy(epInfo.IP())),
"-j", string(KubeMarkMasqChain))...) "-j", string(KubeMarkMasqChain))...)
// Update client-affinity lists. // Update client-affinity lists.
@ -1072,16 +1075,16 @@ func (proxier *Proxier) syncProxyRules() {
} }
// DNAT to final destination. // DNAT to final destination.
args = append(args, "-m", protocol, "-p", protocol, "-j", "DNAT", "--to-destination", epInfo.Endpoint) args = append(args, "-m", protocol, "-p", protocol, "-j", "DNAT", "--to-destination", epInfo.Endpoint)
utilproxy.WriteLine(proxier.natRules, args...) proxier.natRules.Write(args...)
} }
svcChain := svcInfo.servicePortChainName svcChain := svcInfo.servicePortChainName
if hasEndpoints { if hasEndpoints {
// Create the per-service chain, retaining counters if possible. // Create the per-service chain, retaining counters if possible.
if chain, ok := existingNATChains[svcChain]; ok { if chain, ok := existingNATChains[svcChain]; ok {
utilproxy.WriteBytesLine(proxier.natChains, chain) proxier.natChains.WriteBytes(chain)
} else { } else {
utilproxy.WriteLine(proxier.natChains, utiliptables.MakeChainLine(svcChain)) proxier.natChains.Write(utiliptables.MakeChainLine(svcChain))
} }
activeNATChains[svcChain] = true activeNATChains[svcChain] = true
} }
@ -1091,9 +1094,9 @@ func (proxier *Proxier) syncProxyRules() {
// Only for services request OnlyLocal traffic // Only for services request OnlyLocal traffic
// create the per-service LB chain, retaining counters if possible. // create the per-service LB chain, retaining counters if possible.
if lbChain, ok := existingNATChains[svcXlbChain]; ok { if lbChain, ok := existingNATChains[svcXlbChain]; ok {
utilproxy.WriteBytesLine(proxier.natChains, lbChain) proxier.natChains.WriteBytes(lbChain)
} else { } else {
utilproxy.WriteLine(proxier.natChains, utiliptables.MakeChainLine(svcXlbChain)) proxier.natChains.Write(utiliptables.MakeChainLine(svcXlbChain))
} }
activeNATChains[svcXlbChain] = true activeNATChains[svcXlbChain] = true
} }
@ -1107,19 +1110,25 @@ func (proxier *Proxier) syncProxyRules() {
"--dport", strconv.Itoa(svcInfo.Port()), "--dport", strconv.Itoa(svcInfo.Port()),
) )
if proxier.masqueradeAll { if proxier.masqueradeAll {
utilproxy.WriteRuleLine(proxier.natRules, string(svcChain), append(args, "-j", string(KubeMarkMasqChain))...) args := prepend(args, "-A", string(svcChain))
args = append(args, "-j", string(KubeMarkMasqChain))
proxier.natRules.Write(args...)
} else if proxier.localDetector.IsImplemented() { } else if proxier.localDetector.IsImplemented() {
// This masquerades off-cluster traffic to a service VIP. The idea // This masquerades off-cluster traffic to a service VIP. The idea
// is that you can establish a static route for your Service range, // is that you can establish a static route for your Service range,
// routing to any node, and that node will bridge into the Service // routing to any node, and that node will bridge into the Service
// for you. Since that might bounce off-node, we masquerade here. // for you. Since that might bounce off-node, we masquerade here.
// If/when we support "Local" policy for VIPs, we should update this. // If/when we support "Local" policy for VIPs, we should update this.
utilproxy.WriteRuleLine(proxier.natRules, string(svcChain), proxier.localDetector.JumpIfNotLocal(args, string(KubeMarkMasqChain))...) args := prepend(args, "-A", string(svcChain))
args = proxier.localDetector.JumpIfNotLocal(args, string(KubeMarkMasqChain))
proxier.natRules.Write(args...)
} }
utilproxy.WriteRuleLine(proxier.natRules, string(kubeServicesChain), append(args, "-j", string(svcChain))...) args = prepend(args, "-A", string(kubeServicesChain))
args = append(args, "-j", string(svcChain))
proxier.natRules.Write(args...)
} else { } else {
// No endpoints. // No endpoints.
utilproxy.WriteLine(proxier.filterRules, proxier.filterRules.Write(
"-A", string(kubeServicesChain), "-A", string(kubeServicesChain),
"-m", "comment", "--comment", fmt.Sprintf(`"%s has no endpoints"`, svcNameString), "-m", "comment", "--comment", fmt.Sprintf(`"%s has no endpoints"`, svcNameString),
"-m", protocol, "-p", protocol, "-m", protocol, "-p", protocol,
@ -1180,19 +1189,24 @@ func (proxier *Proxier) syncProxyRules() {
// If we can't differentiate the local traffic we always SNAT. // If we can't differentiate the local traffic we always SNAT.
if !svcInfo.NodeLocalExternal() { if !svcInfo.NodeLocalExternal() {
destChain = svcChain destChain = svcChain
args := prepend(args, "-A", string(svcChain))
// This masquerades off-cluster traffic to a External IP. // This masquerades off-cluster traffic to a External IP.
if proxier.localDetector.IsImplemented() { if proxier.localDetector.IsImplemented() {
utilproxy.WriteRuleLine(proxier.natRules, string(svcChain), proxier.localDetector.JumpIfNotLocal(args, string(KubeMarkMasqChain))...) proxier.natRules.Write(
proxier.localDetector.JumpIfNotLocal(args, string(KubeMarkMasqChain))...)
} else { } else {
utilproxy.WriteRuleLine(proxier.natRules, string(svcChain), append(args, "-j", string(KubeMarkMasqChain))...) proxier.natRules.Write(
append(args, "-j", string(KubeMarkMasqChain))...)
} }
} }
// Send traffic bound for external IPs to the service chain. // Send traffic bound for external IPs to the service chain.
utilproxy.WriteRuleLine(proxier.natRules, string(kubeServicesChain), append(args, "-j", string(destChain))...) args = prepend(args, "-A", string(kubeServicesChain))
proxier.natRules.Write(
append(args, "-j", string(destChain))...)
} else { } else {
// No endpoints. // No endpoints.
utilproxy.WriteLine(proxier.filterRules, proxier.filterRules.Write(
"-A", string(kubeExternalServicesChain), "-A", string(kubeExternalServicesChain),
"-m", "comment", "--comment", fmt.Sprintf(`"%s has no endpoints"`, svcNameString), "-m", "comment", "--comment", fmt.Sprintf(`"%s has no endpoints"`, svcNameString),
"-m", protocol, "-p", protocol, "-m", protocol, "-p", protocol,
@ -1210,9 +1224,9 @@ func (proxier *Proxier) syncProxyRules() {
if hasEndpoints { if hasEndpoints {
// create service firewall chain // create service firewall chain
if chain, ok := existingNATChains[fwChain]; ok { if chain, ok := existingNATChains[fwChain]; ok {
utilproxy.WriteBytesLine(proxier.natChains, chain) proxier.natChains.WriteBytes(chain)
} else { } else {
utilproxy.WriteLine(proxier.natChains, utiliptables.MakeChainLine(fwChain)) proxier.natChains.Write(utiliptables.MakeChainLine(fwChain))
} }
activeNATChains[fwChain] = true activeNATChains[fwChain] = true
// The service firewall rules are created based on ServiceSpec.loadBalancerSourceRanges field. // The service firewall rules are created based on ServiceSpec.loadBalancerSourceRanges field.
@ -1227,7 +1241,7 @@ func (proxier *Proxier) syncProxyRules() {
"--dport", strconv.Itoa(svcInfo.Port()), "--dport", strconv.Itoa(svcInfo.Port()),
) )
// jump to service firewall chain // jump to service firewall chain
utilproxy.WriteLine(proxier.natRules, append(args, "-j", string(fwChain))...) proxier.natRules.Write(append(args, "-j", string(fwChain))...)
args = append(args[:0], args = append(args[:0],
"-A", string(fwChain), "-A", string(fwChain),
@ -1239,18 +1253,18 @@ func (proxier *Proxier) syncProxyRules() {
// If we are proxying globally, we need to masquerade in case we cross nodes. // If we are proxying globally, we need to masquerade in case we cross nodes.
// If we are proxying only locally, we can retain the source IP. // If we are proxying only locally, we can retain the source IP.
if !svcInfo.NodeLocalExternal() { if !svcInfo.NodeLocalExternal() {
utilproxy.WriteLine(proxier.natRules, append(args, "-j", string(KubeMarkMasqChain))...) proxier.natRules.Write(append(args, "-j", string(KubeMarkMasqChain))...)
chosenChain = svcChain chosenChain = svcChain
} }
if len(svcInfo.LoadBalancerSourceRanges()) == 0 { if len(svcInfo.LoadBalancerSourceRanges()) == 0 {
// allow all sources, so jump directly to the KUBE-SVC or KUBE-XLB chain // allow all sources, so jump directly to the KUBE-SVC or KUBE-XLB chain
utilproxy.WriteLine(proxier.natRules, append(args, "-j", string(chosenChain))...) proxier.natRules.Write(append(args, "-j", string(chosenChain))...)
} else { } else {
// firewall filter based on each source range // firewall filter based on each source range
allowFromNode := false allowFromNode := false
for _, src := range svcInfo.LoadBalancerSourceRanges() { for _, src := range svcInfo.LoadBalancerSourceRanges() {
utilproxy.WriteLine(proxier.natRules, append(args, "-s", src, "-j", string(chosenChain))...) proxier.natRules.Write(append(args, "-s", src, "-j", string(chosenChain))...)
_, cidr, err := netutils.ParseCIDRSloppy(src) _, cidr, err := netutils.ParseCIDRSloppy(src)
if err != nil { if err != nil {
klog.ErrorS(err, "Error parsing CIDR in LoadBalancerSourceRanges, dropping it", "cidr", cidr) klog.ErrorS(err, "Error parsing CIDR in LoadBalancerSourceRanges, dropping it", "cidr", cidr)
@ -1262,16 +1276,16 @@ func (proxier *Proxier) syncProxyRules() {
// loadbalancer's backend hosts. In this case, request will not hit the loadbalancer but loop back directly. // loadbalancer's backend hosts. In this case, request will not hit the loadbalancer but loop back directly.
// Need to add the following rule to allow request on host. // Need to add the following rule to allow request on host.
if allowFromNode { if allowFromNode {
utilproxy.WriteLine(proxier.natRules, append(args, "-s", utilproxy.ToCIDR(netutils.ParseIPSloppy(ingress)), "-j", string(chosenChain))...) proxier.natRules.Write(append(args, "-s", utilproxy.ToCIDR(netutils.ParseIPSloppy(ingress)), "-j", string(chosenChain))...)
} }
} }
// If the packet was able to reach the end of firewall chain, then it did not get DNATed. // If the packet was able to reach the end of firewall chain, then it did not get DNATed.
// It means the packet cannot go thru the firewall, then mark it for DROP // It means the packet cannot go thru the firewall, then mark it for DROP
utilproxy.WriteLine(proxier.natRules, append(args, "-j", string(KubeMarkDropChain))...) proxier.natRules.Write(append(args, "-j", string(KubeMarkDropChain))...)
} else { } else {
// No endpoints. // No endpoints.
utilproxy.WriteLine(proxier.filterRules, proxier.filterRules.Write(
"-A", string(kubeExternalServicesChain), "-A", string(kubeExternalServicesChain),
"-m", "comment", "--comment", fmt.Sprintf(`"%s has no endpoints"`, svcNameString), "-m", "comment", "--comment", fmt.Sprintf(`"%s has no endpoints"`, svcNameString),
"-m", protocol, "-p", protocol, "-m", protocol, "-p", protocol,
@ -1345,9 +1359,11 @@ func (proxier *Proxier) syncProxyRules() {
) )
if !svcInfo.NodeLocalExternal() { if !svcInfo.NodeLocalExternal() {
// Nodeports need SNAT, unless they're local. // Nodeports need SNAT, unless they're local.
utilproxy.WriteRuleLine(proxier.natRules, string(svcChain), append(args, "-j", string(KubeMarkMasqChain))...) proxier.natRules.Write(
append(prepend(args, "-A", string(svcChain)), "-j", string(KubeMarkMasqChain))...)
// Jump to the service chain. // Jump to the service chain.
utilproxy.WriteRuleLine(proxier.natRules, string(kubeNodePortsChain), append(args, "-j", string(svcChain))...) proxier.natRules.Write(
append(prepend(args, "-A", string(kubeNodePortsChain)), "-j", string(svcChain))...)
} else { } else {
// TODO: Make all nodePorts jump to the firewall chain. // TODO: Make all nodePorts jump to the firewall chain.
// Currently we only create it for loadbalancers (#33586). // Currently we only create it for loadbalancers (#33586).
@ -1357,12 +1373,15 @@ func (proxier *Proxier) syncProxyRules() {
if isIPv6 { if isIPv6 {
loopback = "::1/128" loopback = "::1/128"
} }
utilproxy.WriteRuleLine(proxier.natRules, string(kubeNodePortsChain), append(args, "-s", loopback, "-j", string(KubeMarkMasqChain))...) args = prepend(args, "-A", string(kubeNodePortsChain))
utilproxy.WriteRuleLine(proxier.natRules, string(kubeNodePortsChain), append(args, "-j", string(svcXlbChain))...) proxier.natRules.Write(
append(args, "-s", loopback, "-j", string(KubeMarkMasqChain))...)
proxier.natRules.Write(
append(args, "-j", string(svcXlbChain))...)
} }
} else { } else {
// No endpoints. // No endpoints.
utilproxy.WriteLine(proxier.filterRules, proxier.filterRules.Write(
"-A", string(kubeExternalServicesChain), "-A", string(kubeExternalServicesChain),
"-m", "comment", "--comment", fmt.Sprintf(`"%s has no endpoints"`, svcNameString), "-m", "comment", "--comment", fmt.Sprintf(`"%s has no endpoints"`, svcNameString),
"-m", "addrtype", "--dst-type", "LOCAL", "-m", "addrtype", "--dst-type", "LOCAL",
@ -1377,7 +1396,7 @@ func (proxier *Proxier) syncProxyRules() {
if svcInfo.HealthCheckNodePort() != 0 { if svcInfo.HealthCheckNodePort() != 0 {
// no matter if node has local endpoints, healthCheckNodePorts // no matter if node has local endpoints, healthCheckNodePorts
// need to add a rule to accept the incoming connection // need to add a rule to accept the incoming connection
utilproxy.WriteLine(proxier.filterRules, proxier.filterRules.Write(
"-A", string(kubeNodePortsChain), "-A", string(kubeNodePortsChain),
"-m", "comment", "--comment", fmt.Sprintf(`"%s health check node port"`, svcNameString), "-m", "comment", "--comment", fmt.Sprintf(`"%s health check node port"`, svcNameString),
"-m", "tcp", "-p", "tcp", "-m", "tcp", "-p", "tcp",
@ -1402,7 +1421,7 @@ func (proxier *Proxier) syncProxyRules() {
"--rcheck", "--seconds", strconv.Itoa(svcInfo.StickyMaxAgeSeconds()), "--reap", "--rcheck", "--seconds", strconv.Itoa(svcInfo.StickyMaxAgeSeconds()), "--reap",
"-j", string(endpointChain), "-j", string(endpointChain),
) )
utilproxy.WriteLine(proxier.natRules, args...) proxier.natRules.Write(args...)
} }
} }
@ -1421,7 +1440,7 @@ func (proxier *Proxier) syncProxyRules() {
} }
// The final (or only if n == 1) rule is a guaranteed match. // The final (or only if n == 1) rule is a guaranteed match.
args = append(args, "-j", string(endpointChain)) args = append(args, "-j", string(endpointChain))
utilproxy.WriteLine(proxier.natRules, args...) proxier.natRules.Write(args...)
} }
// The logic below this applies only if this service is marked as OnlyLocal // The logic below this applies only if this service is marked as OnlyLocal
@ -1438,17 +1457,17 @@ func (proxier *Proxier) syncProxyRules() {
"-m", "comment", "--comment", "-m", "comment", "--comment",
`"Redirect pods trying to reach external loadbalancer VIP to clusterIP"`, `"Redirect pods trying to reach external loadbalancer VIP to clusterIP"`,
) )
utilproxy.WriteLine(proxier.natRules, proxier.localDetector.JumpIfLocal(args, string(svcChain))...) proxier.natRules.Write(proxier.localDetector.JumpIfLocal(args, string(svcChain))...)
} }
// Next, redirect all src-type=LOCAL -> LB IP to the service chain for externalTrafficPolicy=Local // Next, redirect all src-type=LOCAL -> LB IP to the service chain for externalTrafficPolicy=Local
// This allows traffic originating from the host to be redirected to the service correctly, // This allows traffic originating from the host to be redirected to the service correctly,
// otherwise traffic to LB IPs are dropped if there are no local endpoints. // otherwise traffic to LB IPs are dropped if there are no local endpoints.
args = append(args[:0], "-A", string(svcXlbChain)) args = append(args[:0], "-A", string(svcXlbChain))
utilproxy.WriteLine(proxier.natRules, append(args, proxier.natRules.Write(append(args,
"-m", "comment", "--comment", fmt.Sprintf(`"masquerade LOCAL traffic for %s LB IP"`, svcNameString), "-m", "comment", "--comment", fmt.Sprintf(`"masquerade LOCAL traffic for %s LB IP"`, svcNameString),
"-m", "addrtype", "--src-type", "LOCAL", "-j", string(KubeMarkMasqChain))...) "-m", "addrtype", "--src-type", "LOCAL", "-j", string(KubeMarkMasqChain))...)
utilproxy.WriteLine(proxier.natRules, append(args, proxier.natRules.Write(append(args,
"-m", "comment", "--comment", fmt.Sprintf(`"route LOCAL traffic for %s LB IP to service chain"`, svcNameString), "-m", "comment", "--comment", fmt.Sprintf(`"route LOCAL traffic for %s LB IP to service chain"`, svcNameString),
"-m", "addrtype", "--src-type", "LOCAL", "-j", string(svcChain))...) "-m", "addrtype", "--src-type", "LOCAL", "-j", string(svcChain))...)
@ -1462,12 +1481,12 @@ func (proxier *Proxier) syncProxyRules() {
"-j", "-j",
string(KubeMarkDropChain), string(KubeMarkDropChain),
) )
utilproxy.WriteLine(proxier.natRules, args...) proxier.natRules.Write(args...)
} else { } else {
// First write session affinity rules only over local endpoints, if applicable. // First write session affinity rules only over local endpoints, if applicable.
if svcInfo.SessionAffinityType() == v1.ServiceAffinityClientIP { if svcInfo.SessionAffinityType() == v1.ServiceAffinityClientIP {
for _, endpointChain := range localEndpointChains { for _, endpointChain := range localEndpointChains {
utilproxy.WriteLine(proxier.natRules, proxier.natRules.Write(
"-A", string(svcXlbChain), "-A", string(svcXlbChain),
"-m", "comment", "--comment", svcNameString, "-m", "comment", "--comment", svcNameString,
"-m", "recent", "--name", string(endpointChain), "-m", "recent", "--name", string(endpointChain),
@ -1493,7 +1512,7 @@ func (proxier *Proxier) syncProxyRules() {
} }
// The final (or only if n == 1) rule is a guaranteed match. // The final (or only if n == 1) rule is a guaranteed match.
args = append(args, "-j", string(endpointChain)) args = append(args, "-j", string(endpointChain))
utilproxy.WriteLine(proxier.natRules, args...) proxier.natRules.Write(args...)
} }
} }
} }
@ -1509,8 +1528,8 @@ func (proxier *Proxier) syncProxyRules() {
// We must (as per iptables) write a chain-line for it, which has // We must (as per iptables) write a chain-line for it, which has
// the nice effect of flushing the chain. Then we can remove the // the nice effect of flushing the chain. Then we can remove the
// chain. // chain.
utilproxy.WriteBytesLine(proxier.natChains, existingNATChains[chain]) proxier.natChains.WriteBytes(existingNATChains[chain])
utilproxy.WriteLine(proxier.natRules, "-X", chainString) proxier.natRules.Write("-X", chainString)
} }
} }
@ -1525,7 +1544,7 @@ func (proxier *Proxier) syncProxyRules() {
"-m", "comment", "--comment", `"kubernetes service nodeports; NOTE: this must be the last rule in this chain"`, "-m", "comment", "--comment", `"kubernetes service nodeports; NOTE: this must be the last rule in this chain"`,
"-m", "addrtype", "--dst-type", "LOCAL", "-m", "addrtype", "--dst-type", "LOCAL",
"-j", string(kubeNodePortsChain)) "-j", string(kubeNodePortsChain))
utilproxy.WriteLine(proxier.natRules, args...) proxier.natRules.Write(args...)
// Nothing else matters after the zero CIDR. // Nothing else matters after the zero CIDR.
break break
} }
@ -1540,13 +1559,13 @@ func (proxier *Proxier) syncProxyRules() {
"-m", "comment", "--comment", `"kubernetes service nodeports; NOTE: this must be the last rule in this chain"`, "-m", "comment", "--comment", `"kubernetes service nodeports; NOTE: this must be the last rule in this chain"`,
"-d", address, "-d", address,
"-j", string(kubeNodePortsChain)) "-j", string(kubeNodePortsChain))
utilproxy.WriteLine(proxier.natRules, args...) proxier.natRules.Write(args...)
} }
// Drop the packets in INVALID state, which would potentially cause // Drop the packets in INVALID state, which would potentially cause
// unexpected connection reset. // unexpected connection reset.
// https://github.com/kubernetes/kubernetes/issues/74839 // https://github.com/kubernetes/kubernetes/issues/74839
utilproxy.WriteLine(proxier.filterRules, proxier.filterRules.Write(
"-A", string(kubeForwardChain), "-A", string(kubeForwardChain),
"-m", "conntrack", "-m", "conntrack",
"--ctstate", "INVALID", "--ctstate", "INVALID",
@ -1556,7 +1575,7 @@ func (proxier *Proxier) syncProxyRules() {
// If the masqueradeMark has been added then we want to forward that same // 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 // traffic, this allows NodePort traffic to be forwarded even if the default
// FORWARD policy is not accept. // FORWARD policy is not accept.
utilproxy.WriteLine(proxier.filterRules, proxier.filterRules.Write(
"-A", string(kubeForwardChain), "-A", string(kubeForwardChain),
"-m", "comment", "--comment", `"kubernetes forwarding rules"`, "-m", "comment", "--comment", `"kubernetes forwarding rules"`,
"-m", "mark", "--mark", fmt.Sprintf("%s/%s", proxier.masqueradeMark, proxier.masqueradeMark), "-m", "mark", "--mark", fmt.Sprintf("%s/%s", proxier.masqueradeMark, proxier.masqueradeMark),
@ -1566,14 +1585,14 @@ func (proxier *Proxier) syncProxyRules() {
// The following two rules ensure the traffic after the initial packet // The following two rules ensure the traffic after the initial packet
// accepted by the "kubernetes forwarding rules" rule above will be // accepted by the "kubernetes forwarding rules" rule above will be
// accepted. // accepted.
utilproxy.WriteLine(proxier.filterRules, proxier.filterRules.Write(
"-A", string(kubeForwardChain), "-A", string(kubeForwardChain),
"-m", "comment", "--comment", `"kubernetes forwarding conntrack pod source rule"`, "-m", "comment", "--comment", `"kubernetes forwarding conntrack pod source rule"`,
"-m", "conntrack", "-m", "conntrack",
"--ctstate", "RELATED,ESTABLISHED", "--ctstate", "RELATED,ESTABLISHED",
"-j", "ACCEPT", "-j", "ACCEPT",
) )
utilproxy.WriteLine(proxier.filterRules, proxier.filterRules.Write(
"-A", string(kubeForwardChain), "-A", string(kubeForwardChain),
"-m", "comment", "--comment", `"kubernetes forwarding conntrack pod destination rule"`, "-m", "comment", "--comment", `"kubernetes forwarding conntrack pod destination rule"`,
"-m", "conntrack", "-m", "conntrack",
@ -1587,8 +1606,8 @@ func (proxier *Proxier) syncProxyRules() {
metrics.IptablesRulesTotal.WithLabelValues(string(utiliptables.TableNAT)).Set(float64(numberNatIptablesRules)) metrics.IptablesRulesTotal.WithLabelValues(string(utiliptables.TableNAT)).Set(float64(numberNatIptablesRules))
// Write the end-of-table markers. // Write the end-of-table markers.
utilproxy.WriteLine(proxier.filterRules, "COMMIT") proxier.filterRules.Write("COMMIT")
utilproxy.WriteLine(proxier.natRules, "COMMIT") proxier.natRules.Write("COMMIT")
// Sync rules. // Sync rules.
// NOTE: NoFlushTables is used so we don't flush non-kubernetes chains in the table // NOTE: NoFlushTables is used so we don't flush non-kubernetes chains in the table

8
pkg/proxy/iptables/proxier_test.go
View File

@ -509,10 +509,10 @@ func NewFakeProxier(ipt utiliptables.Interface) *Proxier {
precomputedProbabilities: make([]string, 0, 1001), precomputedProbabilities: make([]string, 0, 1001),
iptablesData: bytes.NewBuffer(nil), iptablesData: bytes.NewBuffer(nil),
existingFilterChainsData: bytes.NewBuffer(nil), existingFilterChainsData: bytes.NewBuffer(nil),
filterChains: bytes.NewBuffer(nil), filterChains: utilproxy.LineBuffer{},
filterRules: bytes.NewBuffer(nil), filterRules: utilproxy.LineBuffer{},
natChains: bytes.NewBuffer(nil), natChains: utilproxy.LineBuffer{},
natRules: bytes.NewBuffer(nil), natRules: utilproxy.LineBuffer{},
nodePortAddresses: make([]string, 0), nodePortAddresses: make([]string, 0),
networkInterfacer: utilproxytest.NewFakeNetwork(), networkInterfacer: utilproxytest.NewFakeNetwork(),
} }

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

@ -261,10 +261,10 @@ type Proxier struct {
// that are significantly impacting performance. // that are significantly impacting performance.
iptablesData *bytes.Buffer iptablesData *bytes.Buffer
filterChainsData *bytes.Buffer filterChainsData *bytes.Buffer
natChains *bytes.Buffer natChains utilproxy.LineBuffer
filterChains *bytes.Buffer filterChains utilproxy.LineBuffer
natRules *bytes.Buffer natRules utilproxy.LineBuffer
filterRules *bytes.Buffer filterRules utilproxy.LineBuffer
// Added as a member to the struct to allow injection for testing. // Added as a member to the struct to allow injection for testing.
netlinkHandle NetLinkHandle netlinkHandle NetLinkHandle
// ipsetList is the list of ipsets that ipvs proxier used. // ipsetList is the list of ipsets that ipvs proxier used.
@ -479,10 +479,10 @@ func NewProxier(ipt utiliptables.Interface,
ipGetter: &realIPGetter{nl: NewNetLinkHandle(ipFamily == v1.IPv6Protocol)}, ipGetter: &realIPGetter{nl: NewNetLinkHandle(ipFamily == v1.IPv6Protocol)},
iptablesData: bytes.NewBuffer(nil), iptablesData: bytes.NewBuffer(nil),
filterChainsData: bytes.NewBuffer(nil), filterChainsData: bytes.NewBuffer(nil),
natChains: bytes.NewBuffer(nil), natChains: utilproxy.LineBuffer{},
natRules: bytes.NewBuffer(nil), natRules: utilproxy.LineBuffer{},
filterChains: bytes.NewBuffer(nil), filterChains: utilproxy.LineBuffer{},
filterRules: bytes.NewBuffer(nil), filterRules: utilproxy.LineBuffer{},
netlinkHandle: NewNetLinkHandle(ipFamily == v1.IPv6Protocol), netlinkHandle: NewNetLinkHandle(ipFamily == v1.IPv6Protocol),
ipset: ipset, ipset: ipset,
nodePortAddresses: nodePortAddresses, nodePortAddresses: nodePortAddresses,
@ -1043,8 +1043,8 @@ func (proxier *Proxier) syncProxyRules() {
proxier.filterRules.Reset() proxier.filterRules.Reset()
// Write table headers. // Write table headers.
utilproxy.WriteLine(proxier.filterChains, "*filter") proxier.filterChains.Write("*filter")
utilproxy.WriteLine(proxier.natChains, "*nat") proxier.natChains.Write("*nat")
proxier.createAndLinkKubeChain() proxier.createAndLinkKubeChain()
@ -1704,7 +1704,7 @@ func (proxier *Proxier) writeIptablesRules() {
"-m", "set", "--match-set", proxier.ipsetList[set.name].Name, "-m", "set", "--match-set", proxier.ipsetList[set.name].Name,
set.matchType, set.matchType,
) )
utilproxy.WriteLine(proxier.natRules, append(args, "-j", set.to)...) proxier.natRules.Write(append(args, "-j", set.to)...)
} }
} }
@ -1715,14 +1715,14 @@ func (proxier *Proxier) writeIptablesRules() {
"-m", "set", "--match-set", proxier.ipsetList[kubeClusterIPSet].Name, "-m", "set", "--match-set", proxier.ipsetList[kubeClusterIPSet].Name,
) )
if proxier.masqueradeAll { if proxier.masqueradeAll {
utilproxy.WriteLine(proxier.natRules, append(args, "dst,dst", "-j", string(KubeMarkMasqChain))...) proxier.natRules.Write(append(args, "dst,dst", "-j", string(KubeMarkMasqChain))...)
} else if proxier.localDetector.IsImplemented() { } else if proxier.localDetector.IsImplemented() {
// This masquerades off-cluster traffic to a service VIP. The idea // This masquerades off-cluster traffic to a service VIP. The idea
// is that you can establish a static route for your Service range, // is that you can establish a static route for your Service range,
// routing to any node, and that node will bridge into the Service // routing to any node, and that node will bridge into the Service
// for you. Since that might bounce off-node, we masquerade here. // for you. Since that might bounce off-node, we masquerade here.
// If/when we support "Local" policy for VIPs, we should update this. // If/when we support "Local" policy for VIPs, we should update this.
utilproxy.WriteLine(proxier.natRules, proxier.localDetector.JumpIfNotLocal(append(args, "dst,dst"), string(KubeMarkMasqChain))...) proxier.natRules.Write(proxier.localDetector.JumpIfNotLocal(append(args, "dst,dst"), string(KubeMarkMasqChain))...)
} else { } else {
// Masquerade all OUTPUT traffic coming from a service ip. // Masquerade all OUTPUT traffic coming from a service ip.
// The kube dummy interface has all service VIPs assigned which // The kube dummy interface has all service VIPs assigned which
@ -1731,7 +1731,7 @@ func (proxier *Proxier) writeIptablesRules() {
// VIP:<service port>. // VIP:<service port>.
// Always masquerading OUTPUT (node-originating) traffic with a VIP // Always masquerading OUTPUT (node-originating) traffic with a VIP
// source ip and service port destination fixes the outgoing connections. // source ip and service port destination fixes the outgoing connections.
utilproxy.WriteLine(proxier.natRules, append(args, "src,dst", "-j", string(KubeMarkMasqChain))...) proxier.natRules.Write(append(args, "src,dst", "-j", string(KubeMarkMasqChain))...)
} }
} }
@ -1744,11 +1744,11 @@ func (proxier *Proxier) writeIptablesRules() {
externalTrafficOnlyArgs := append(args, externalTrafficOnlyArgs := append(args,
"-m", "physdev", "!", "--physdev-is-in", "-m", "physdev", "!", "--physdev-is-in",
"-m", "addrtype", "!", "--src-type", "LOCAL") "-m", "addrtype", "!", "--src-type", "LOCAL")
utilproxy.WriteLine(proxier.natRules, append(externalTrafficOnlyArgs, "-j", "ACCEPT")...) proxier.natRules.Write(append(externalTrafficOnlyArgs, "-j", "ACCEPT")...)
dstLocalOnlyArgs := append(args, "-m", "addrtype", "--dst-type", "LOCAL") dstLocalOnlyArgs := append(args, "-m", "addrtype", "--dst-type", "LOCAL")
// Allow traffic bound for external IPs that happen to be recognized as local IPs to stay local. // Allow traffic bound for external IPs that happen to be recognized as local IPs to stay local.
// This covers cases like GCE load-balancers which get added to the local routing table. // This covers cases like GCE load-balancers which get added to the local routing table.
utilproxy.WriteLine(proxier.natRules, append(dstLocalOnlyArgs, "-j", "ACCEPT")...) proxier.natRules.Write(append(dstLocalOnlyArgs, "-j", "ACCEPT")...)
} }
if !proxier.ipsetList[kubeExternalIPSet].isEmpty() { if !proxier.ipsetList[kubeExternalIPSet].isEmpty() {
@ -1759,7 +1759,7 @@ func (proxier *Proxier) writeIptablesRules() {
"-m", "set", "--match-set", proxier.ipsetList[kubeExternalIPSet].Name, "-m", "set", "--match-set", proxier.ipsetList[kubeExternalIPSet].Name,
"dst,dst", "dst,dst",
) )
utilproxy.WriteLine(proxier.natRules, append(args, "-j", string(KubeMarkMasqChain))...) proxier.natRules.Write(append(args, "-j", string(KubeMarkMasqChain))...)
externalIPRules(args) externalIPRules(args)
} }
@ -1778,16 +1778,16 @@ func (proxier *Proxier) writeIptablesRules() {
"-A", string(kubeServicesChain), "-A", string(kubeServicesChain),
"-m", "addrtype", "--dst-type", "LOCAL", "-m", "addrtype", "--dst-type", "LOCAL",
) )
utilproxy.WriteLine(proxier.natRules, append(args, "-j", string(KubeNodePortChain))...) proxier.natRules.Write(append(args, "-j", string(KubeNodePortChain))...)
// mark drop for KUBE-LOAD-BALANCER // mark drop for KUBE-LOAD-BALANCER
utilproxy.WriteLine(proxier.natRules, []string{ proxier.natRules.Write([]string{
"-A", string(KubeLoadBalancerChain), "-A", string(KubeLoadBalancerChain),
"-j", string(KubeMarkMasqChain), "-j", string(KubeMarkMasqChain),
}...) }...)
// mark drop for KUBE-FIRE-WALL // mark drop for KUBE-FIRE-WALL
utilproxy.WriteLine(proxier.natRules, []string{ proxier.natRules.Write([]string{
"-A", string(KubeFireWallChain), "-A", string(KubeFireWallChain),
"-j", string(KubeMarkDropChain), "-j", string(KubeMarkDropChain),
}...) }...)
@ -1800,7 +1800,7 @@ func (proxier *Proxier) writeIptablesRules() {
// If the masqueradeMark has been added then we want to forward that same // 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 // traffic, this allows NodePort traffic to be forwarded even if the default
// FORWARD policy is not accept. // FORWARD policy is not accept.
utilproxy.WriteLine(proxier.filterRules, proxier.filterRules.Write(
"-A", string(KubeForwardChain), "-A", string(KubeForwardChain),
"-m", "comment", "--comment", `"kubernetes forwarding rules"`, "-m", "comment", "--comment", `"kubernetes forwarding rules"`,
"-m", "mark", "--mark", fmt.Sprintf("%s/%s", proxier.masqueradeMark, proxier.masqueradeMark), "-m", "mark", "--mark", fmt.Sprintf("%s/%s", proxier.masqueradeMark, proxier.masqueradeMark),
@ -1810,14 +1810,14 @@ func (proxier *Proxier) writeIptablesRules() {
// The following two rules ensure the traffic after the initial packet // The following two rules ensure the traffic after the initial packet
// accepted by the "kubernetes forwarding rules" rule above will be // accepted by the "kubernetes forwarding rules" rule above will be
// accepted. // accepted.
utilproxy.WriteLine(proxier.filterRules, proxier.filterRules.Write(
"-A", string(KubeForwardChain), "-A", string(KubeForwardChain),
"-m", "comment", "--comment", `"kubernetes forwarding conntrack pod source rule"`, "-m", "comment", "--comment", `"kubernetes forwarding conntrack pod source rule"`,
"-m", "conntrack", "-m", "conntrack",
"--ctstate", "RELATED,ESTABLISHED", "--ctstate", "RELATED,ESTABLISHED",
"-j", "ACCEPT", "-j", "ACCEPT",
) )
utilproxy.WriteLine(proxier.filterRules, proxier.filterRules.Write(
"-A", string(KubeForwardChain), "-A", string(KubeForwardChain),
"-m", "comment", "--comment", `"kubernetes forwarding conntrack pod destination rule"`, "-m", "comment", "--comment", `"kubernetes forwarding conntrack pod destination rule"`,
"-m", "conntrack", "-m", "conntrack",
@ -1826,7 +1826,7 @@ func (proxier *Proxier) writeIptablesRules() {
) )
// Add rule to accept traffic towards health check node port // Add rule to accept traffic towards health check node port
utilproxy.WriteLine(proxier.filterRules, proxier.filterRules.Write(
"-A", string(KubeNodePortChain), "-A", string(KubeNodePortChain),
"-m", "comment", "--comment", proxier.ipsetList[kubeHealthCheckNodePortSet].getComment(), "-m", "comment", "--comment", proxier.ipsetList[kubeHealthCheckNodePortSet].getComment(),
"-m", "set", "--match-set", proxier.ipsetList[kubeHealthCheckNodePortSet].Name, "dst", "-m", "set", "--match-set", proxier.ipsetList[kubeHealthCheckNodePortSet].Name, "dst",
@ -1837,13 +1837,13 @@ func (proxier *Proxier) writeIptablesRules() {
// this so that it is easier to flush and change, for example if the mark // this so that it is easier to flush and change, for example if the mark
// value should ever change. // value should ever change.
// NB: THIS MUST MATCH the corresponding code in the kubelet // NB: THIS MUST MATCH the corresponding code in the kubelet
utilproxy.WriteLine(proxier.natRules, []string{ proxier.natRules.Write([]string{
"-A", string(kubePostroutingChain), "-A", string(kubePostroutingChain),
"-m", "mark", "!", "--mark", fmt.Sprintf("%s/%s", proxier.masqueradeMark, proxier.masqueradeMark), "-m", "mark", "!", "--mark", fmt.Sprintf("%s/%s", proxier.masqueradeMark, proxier.masqueradeMark),
"-j", "RETURN", "-j", "RETURN",
}...) }...)
// Clear the mark to avoid re-masquerading if the packet re-traverses the network stack. // Clear the mark to avoid re-masquerading if the packet re-traverses the network stack.
utilproxy.WriteLine(proxier.natRules, []string{ proxier.natRules.Write([]string{
"-A", string(kubePostroutingChain), "-A", string(kubePostroutingChain),
// XOR proxier.masqueradeMark to unset it // XOR proxier.masqueradeMark to unset it
"-j", "MARK", "--xor-mark", proxier.masqueradeMark, "-j", "MARK", "--xor-mark", proxier.masqueradeMark,
@ -1856,19 +1856,19 @@ func (proxier *Proxier) writeIptablesRules() {
if proxier.iptables.HasRandomFully() { if proxier.iptables.HasRandomFully() {
masqRule = append(masqRule, "--random-fully") masqRule = append(masqRule, "--random-fully")
} }
utilproxy.WriteLine(proxier.natRules, masqRule...) proxier.natRules.Write(masqRule...)
// Install the kubernetes-specific masquerade mark rule. We use a whole chain for // Install the kubernetes-specific masquerade mark rule. We use a whole chain for
// this so that it is easier to flush and change, for example if the mark // this so that it is easier to flush and change, for example if the mark
// value should ever change. // value should ever change.
utilproxy.WriteLine(proxier.natRules, []string{ proxier.natRules.Write([]string{
"-A", string(KubeMarkMasqChain), "-A", string(KubeMarkMasqChain),
"-j", "MARK", "--or-mark", proxier.masqueradeMark, "-j", "MARK", "--or-mark", proxier.masqueradeMark,
}...) }...)
// Write the end-of-table markers. // Write the end-of-table markers.
utilproxy.WriteLine(proxier.filterRules, "COMMIT") proxier.filterRules.Write("COMMIT")
utilproxy.WriteLine(proxier.natRules, "COMMIT") proxier.natRules.Write("COMMIT")
} }
func (proxier *Proxier) acceptIPVSTraffic() { func (proxier *Proxier) acceptIPVSTraffic() {
@ -1882,7 +1882,7 @@ func (proxier *Proxier) acceptIPVSTraffic() {
default: default:
matchType = "dst,dst" matchType = "dst,dst"
} }
utilproxy.WriteLine(proxier.natRules, []string{ proxier.natRules.Write([]string{
"-A", string(kubeServicesChain), "-A", string(kubeServicesChain),
"-m", "set", "--match-set", proxier.ipsetList[set].Name, matchType, "-m", "set", "--match-set", proxier.ipsetList[set].Name, matchType,
"-j", "ACCEPT", "-j", "ACCEPT",
@ -1912,15 +1912,15 @@ func (proxier *Proxier) createAndLinkKubeChain() {
} }
if ch.table == utiliptables.TableNAT { if ch.table == utiliptables.TableNAT {
if chain, ok := existingNATChains[ch.chain]; ok { if chain, ok := existingNATChains[ch.chain]; ok {
utilproxy.WriteBytesLine(proxier.natChains, chain) proxier.natChains.WriteBytes(chain)
} else { } else {
utilproxy.WriteLine(proxier.natChains, utiliptables.MakeChainLine(ch.chain)) proxier.natChains.Write(utiliptables.MakeChainLine(ch.chain))
} }
} else { } else {
if chain, ok := existingFilterChains[ch.chain]; ok { if chain, ok := existingFilterChains[ch.chain]; ok {
utilproxy.WriteBytesLine(proxier.filterChains, chain) proxier.filterChains.WriteBytes(chain)
} else { } else {
utilproxy.WriteLine(proxier.filterChains, utiliptables.MakeChainLine(ch.chain)) proxier.filterChains.Write(utiliptables.MakeChainLine(ch.chain))
} }
} }
} }

13
pkg/proxy/ipvs/proxier_test.go
View File

@ -39,6 +39,7 @@ import (
"k8s.io/kubernetes/pkg/proxy" "k8s.io/kubernetes/pkg/proxy"
"k8s.io/kubernetes/pkg/proxy/healthcheck" "k8s.io/kubernetes/pkg/proxy/healthcheck"
netlinktest "k8s.io/kubernetes/pkg/proxy/ipvs/testing" netlinktest "k8s.io/kubernetes/pkg/proxy/ipvs/testing"
utilproxy "k8s.io/kubernetes/pkg/proxy/util"
proxyutiliptables "k8s.io/kubernetes/pkg/proxy/util/iptables" proxyutiliptables "k8s.io/kubernetes/pkg/proxy/util/iptables"
proxyutiltest "k8s.io/kubernetes/pkg/proxy/util/testing" proxyutiltest "k8s.io/kubernetes/pkg/proxy/util/testing"
"k8s.io/kubernetes/pkg/util/async" "k8s.io/kubernetes/pkg/util/async"
@ -159,10 +160,10 @@ func NewFakeProxier(ipt utiliptables.Interface, ipvs utilipvs.Interface, ipset u
ipGetter: &fakeIPGetter{nodeIPs: nodeIPs}, ipGetter: &fakeIPGetter{nodeIPs: nodeIPs},
iptablesData: bytes.NewBuffer(nil), iptablesData: bytes.NewBuffer(nil),
filterChainsData: bytes.NewBuffer(nil), filterChainsData: bytes.NewBuffer(nil),
natChains: bytes.NewBuffer(nil), natChains: utilproxy.LineBuffer{},
natRules: bytes.NewBuffer(nil), natRules: utilproxy.LineBuffer{},
filterChains: bytes.NewBuffer(nil), filterChains: utilproxy.LineBuffer{},
filterRules: bytes.NewBuffer(nil), filterRules: utilproxy.LineBuffer{},
netlinkHandle: netlinktest.NewFakeNetlinkHandle(), netlinkHandle: netlinktest.NewFakeNetlinkHandle(),
ipsetList: ipsetList, ipsetList: ipsetList,
nodePortAddresses: make([]string, 0), nodePortAddresses: make([]string, 0),
@ -4650,8 +4651,8 @@ func TestCreateAndLinkKubeChain(t *testing.T) {
expectedFilterChains := `:KUBE-FORWARD - [0:0] expectedFilterChains := `:KUBE-FORWARD - [0:0]
:KUBE-NODE-PORT - [0:0] :KUBE-NODE-PORT - [0:0]
` `
assert.Equal(t, expectedNATChains, fp.natChains.String()) assert.Equal(t, expectedNATChains, string(fp.natChains.Bytes()))
assert.Equal(t, expectedFilterChains, fp.filterChains.String()) assert.Equal(t, expectedFilterChains, string(fp.filterChains.Bytes()))
} }
// This test ensures that the iptables proxier supports translating Endpoints to // This test ensures that the iptables proxier supports translating Endpoints to

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

@ -468,35 +468,35 @@ func GetClusterIPByFamily(ipFamily v1.IPFamily, service *v1.Service) string {
return "" return ""
} }
// WriteLine join all words with spaces, terminate with newline and write to buff. type LineBuffer struct {
func WriteLine(buf *bytes.Buffer, words ...string) { b bytes.Buffer
}
// Write joins all words with spaces, terminates with newline and writes to buf.
func (buf *LineBuffer) Write(words ...string) {
// We avoid strings.Join for performance reasons. // We avoid strings.Join for performance reasons.
for i := range words { for i := range words {
buf.WriteString(words[i]) buf.b.WriteString(words[i])
if i < len(words)-1 { if i < len(words)-1 {
buf.WriteByte(' ') buf.b.WriteByte(' ')
} else { } else {
buf.WriteByte('\n') buf.b.WriteByte('\n')
} }
} }
} }
// WriteRuleLine prepends the strings "-A" and chainName to the buffer and calls // WriteBytes writes bytes to buffer, and terminates with newline.
// WriteLine to join all the words into the buffer and terminate with newline. func (buf *LineBuffer) WriteBytes(bytes []byte) {
func WriteRuleLine(buf *bytes.Buffer, chainName string, words ...string) { buf.b.Write(bytes)
if len(words) == 0 { buf.b.WriteByte('\n')
return
}
buf.WriteString("-A ")
buf.WriteString(chainName)
buf.WriteByte(' ')
WriteLine(buf, words...)
} }
// WriteBytesLine write bytes to buffer, terminate with newline func (buf *LineBuffer) Reset() {
func WriteBytesLine(buf *bytes.Buffer, bytes []byte) { buf.b.Reset()
buf.Write(bytes) }
buf.WriteByte('\n')
func (buf *LineBuffer) Bytes() []byte {
return buf.b.Bytes()
} }
// RevertPorts is closing ports in replacementPortsMap but not in originalPortsMap. In other words, it only // RevertPorts is closing ports in replacementPortsMap but not in originalPortsMap. In other words, it only

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

@ -17,7 +17,6 @@ limitations under the License.
package util package util
import ( import (
"bytes"
"context" "context"
"fmt" "fmt"
"math/rand" "math/rand"
@ -1171,51 +1170,13 @@ func TestWriteLine(t *testing.T) {
expected: "test1 test2 test3\n", expected: "test1 test2 test3\n",
}, },
} }
testBuffer := bytes.NewBuffer(nil) testBuffer := LineBuffer{}
for _, testCase := range testCases { for _, testCase := range testCases {
t.Run(testCase.name, func(t *testing.T) { t.Run(testCase.name, func(t *testing.T) {
testBuffer.Reset() testBuffer.Reset()
WriteLine(testBuffer, testCase.words...) testBuffer.Write(testCase.words...)
if !strings.EqualFold(testBuffer.String(), testCase.expected) { if want, got := testCase.expected, string(testBuffer.Bytes()); !strings.EqualFold(want, got) {
t.Fatalf("write word is %v\n expected: %s, got: %s", testCase.words, testCase.expected, testBuffer.String()) t.Fatalf("write word is %v\n expected: %s, got: %s", testCase.words, want, got)
}
})
}
}
func TestWriteRuleLine(t *testing.T) {
testCases := []struct {
name string
chainName string
words []string
expected string
}{
{
name: "write no line due to no words",
chainName: "KUBE-SVC-FOO",
words: []string{},
expected: "",
},
{
name: "write one line",
chainName: "KUBE-XLB-FOO",
words: []string{"test1"},
expected: "-A KUBE-XLB-FOO test1\n",
},
{
name: "write multi word line",
chainName: "lolChain",
words: []string{"test1", "test2", "test3"},
expected: "-A lolChain test1 test2 test3\n",
},
}
testBuffer := bytes.NewBuffer(nil)
for _, testCase := range testCases {
t.Run(testCase.name, func(t *testing.T) {
testBuffer.Reset()
WriteRuleLine(testBuffer, testCase.chainName, testCase.words...)
if !strings.EqualFold(testBuffer.String(), testCase.expected) {
t.Fatalf("write word is %v\n expected: %s, got: %s", testCase.words, testCase.expected, testBuffer.String())
} }
}) })
} }
@ -1239,13 +1200,13 @@ func TestWriteBytesLine(t *testing.T) {
}, },
} }
testBuffer := bytes.NewBuffer(nil) testBuffer := LineBuffer{}
for _, testCase := range testCases { for _, testCase := range testCases {
t.Run(testCase.name, func(t *testing.T) { t.Run(testCase.name, func(t *testing.T) {
testBuffer.Reset() testBuffer.Reset()
WriteBytesLine(testBuffer, testCase.bytes) testBuffer.WriteBytes(testCase.bytes)
if !strings.EqualFold(testBuffer.String(), testCase.expected) { if want, got := testCase.expected, string(testBuffer.Bytes()); !strings.EqualFold(want, got) {
t.Fatalf("write word is %v\n expected: %s, got: %s", testCase.bytes, testCase.expected, testBuffer.String()) t.Fatalf("write bytes is %v\n expected: %s, got: %s", testCase.bytes, want, got)
} }
}) })
} }
@ -1282,12 +1243,12 @@ func TestWriteCountLines(t *testing.T) {
expected: 100000, expected: 100000,
}, },
} }
testBuffer := bytes.NewBuffer(nil) testBuffer := LineBuffer{}
for _, testCase := range testCases { for _, testCase := range testCases {
t.Run(testCase.name, func(t *testing.T) { t.Run(testCase.name, func(t *testing.T) {
testBuffer.Reset() testBuffer.Reset()
for i := 0; i < testCase.expected; i++ { for i := 0; i < testCase.expected; i++ {
WriteLine(testBuffer, randSeq()) testBuffer.Write(randSeq())
} }
n := CountBytesLines(testBuffer.Bytes()) n := CountBytesLines(testBuffer.Bytes())
if n != testCase.expected { if n != testCase.expected {