Merge pull request #43415 from thockin/fix-nodeport-close-wait

Automatic merge from submit-queue

Install a REJECT rule for nodeport with no backend

Rather than actually accepting the connection, REJECT.  This will avoid
CLOSE_WAIT.

Fixes #43212


@justinsb @felipejfc @spiddy

pkg/proxy/iptables/proxier.go

@@ -1108,6 +1108,21 @@ func (proxier *Proxier) syncProxyRules() {
 				// Currently we only create it for loadbalancers (#33586).
 				writeLine(natRules, append(args, "-j", string(svcXlbChain))...)
 			}
+
+			// If the service has no endpoints then reject packets.  The filter
+			// table doesn't currently have the same per-service structure that
+			// the nat table does, so we just stick this into the kube-services
+			// chain.
+			if len(proxier.endpointsMap[svcName]) == 0 {
+				writeLine(filterRules,
+					"-A", string(kubeServicesChain),
+					"-m", "comment", "--comment", fmt.Sprintf(`"%s has no endpoints"`, svcName.String()),
+					"-m", "addrtype", "--dst-type", "LOCAL",
+					"-m", protocol, "-p", protocol,
+					"--dport", fmt.Sprintf("%d", svcInfo.nodePort),
+					"-j", "REJECT",
+				)
+			}
 		}
 
 		// If the service has no endpoints then reject packets.
@@ -1123,6 +1138,8 @@ func (proxier *Proxier) syncProxyRules() {
 			continue
 		}
 
+		// From here on, we assume there are active endpoints.
+
 		// Generate the per-endpoint chains.  We do this in multiple passes so we
 		// can group rules together.
 		// These two slices parallel each other - keep in sync

pkg/proxy/iptables/proxier_test.go

@@ -694,6 +694,25 @@ func TestNodePort(t *testing.T) {
 	}
 }
 
+func TestNodePortReject(t *testing.T) {
+	ipt := iptablestest.NewFake()
+	fp := NewFakeProxier(ipt)
+	svcName := "svc1"
+	svcIP := net.IPv4(10, 20, 30, 41)
+
+	svc := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "ns1", Name: svcName}, Port: "p80"}
+	svcInfo := newFakeServiceInfo(svc, svcIP, 80, api.ProtocolTCP, false)
+	svcInfo.nodePort = 3001
+	fp.serviceMap[svc] = svcInfo
+
+	fp.syncProxyRules()
+
+	kubeSvcRules := ipt.GetRules(string(kubeServicesChain))
+	if !hasJump(kubeSvcRules, iptablestest.Reject, svcIP.String(), 3001) {
+		errorf(fmt.Sprintf("Failed to find a %v rule for service %v with no endpoints", iptablestest.Reject, svcName), kubeSvcRules, t)
+	}
+}
+
 func strPtr(s string) *string {
 	return &s
 }