Deflake Services should be able to preserve UDP traffic when server pod cycles for a NodePort service

test/e2e/network/BUILD

@@ -8,6 +8,7 @@ load(
 go_library(
     name = "go_default_library",
     srcs = [
+        "conntrack.go",
         "dns.go",
         "dns_common.go",
         "dns_configmap.go",

test/e2e/network/conntrack.go

@@ -0,0 +1,246 @@
+/*
+Copyright 2020 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package network
+
+import (
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/onsi/ginkgo"
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/intstr"
+	clientset "k8s.io/client-go/kubernetes"
+	"k8s.io/kubernetes/test/e2e/framework"
+	e2enode "k8s.io/kubernetes/test/e2e/framework/node"
+	e2epod "k8s.io/kubernetes/test/e2e/framework/pod"
+	e2eservice "k8s.io/kubernetes/test/e2e/framework/service"
+	e2eskipper "k8s.io/kubernetes/test/e2e/framework/skipper"
+)
+
+const (
+	serviceName = "svc-udp"
+	podClient   = "pod-client"
+	podBackend1 = "pod-server-1"
+	podBackend2 = "pod-server-2"
+	srcPort     = 12345
+)
+
+// Linux NAT uses conntrack to perform NAT, everytime a new
+// flow is seen, a connection is created in the conntrack table, and it
+// is being used by the NAT module.
+// Each entry in the conntrack table has associated a timeout, that removes
+// the connection once it expires.
+// UDP is a connectionless protocol, so the conntrack module tracking functions
+// are not very advanced.
+// It uses a short timeout (30 sec by default) that is renewed if there are new flows
+// matching the connection. Otherwise it expires the entry.
+// This behaviour can cause issues in Kubernetes when one entry on the conntrack table
+// is never expired because the sender does not stop sending traffic, but the pods or
+// endpoints were deleted, blackholing the traffic
+// In order to mitigate this problem, Kubernetes delete the stale entries:
+// - when an endpoint is removed
+// - when a service goes from no endpoints to new endpoint
+
+// Ref: https://api.semanticscholar.org/CorpusID:198903401
+// Boye, Magnus. “Netfilter Connection Tracking and NAT Implementation.” (2012).
+
+var _ = SIGDescribe("Conntrack", func() {
+
+	fr := framework.NewDefaultFramework("conntrack")
+
+	type nodeInfo struct {
+		name   string
+		nodeIP string
+	}
+
+	var (
+		cs                             clientset.Interface
+		ns                             string
+		clientNodeInfo, serverNodeInfo nodeInfo
+	)
+
+	ginkgo.BeforeEach(func() {
+		cs = fr.ClientSet
+		ns = fr.Namespace.Name
+
+		nodes, err := e2enode.GetBoundedReadySchedulableNodes(cs, 2)
+		framework.ExpectNoError(err)
+		if len(nodes.Items) < 2 {
+			e2eskipper.Skipf(
+				"Test requires >= 2 Ready nodes, but there are only %v nodes",
+				len(nodes.Items))
+		}
+
+		ips := e2enode.CollectAddresses(nodes, v1.NodeInternalIP)
+
+		clientNodeInfo = nodeInfo{
+			name:   nodes.Items[0].Name,
+			nodeIP: ips[0],
+		}
+
+		serverNodeInfo = nodeInfo{
+			name:   nodes.Items[1].Name,
+			nodeIP: ips[1],
+		}
+	})
+
+	ginkgo.It("should be able to preserve UDP traffic when server pod cycles for a NodePort service", func() {
+
+		// Create a NodePort service
+		udpJig := e2eservice.NewTestJig(cs, ns, serviceName)
+		ginkgo.By("creating a UDP service " + serviceName + " with type=NodePort in " + ns)
+		udpService, err := udpJig.CreateUDPService(func(svc *v1.Service) {
+			svc.Spec.Type = v1.ServiceTypeNodePort
+			svc.Spec.Ports = []v1.ServicePort{
+				{Port: 80, Name: "udp", Protocol: v1.ProtocolUDP, TargetPort: intstr.FromInt(80)},
+			}
+		})
+		framework.ExpectNoError(err)
+
+		// Create a pod in one node to create the UDP traffic against the NodePort service every 5 seconds
+		ginkgo.By("creating a client pod for probing the service " + serviceName)
+		clientPod := newAgnhostPod(podClient, "")
+		clientPod.Spec.NodeName = clientNodeInfo.name
+		cmd := fmt.Sprintf(`date; for i in $(seq 1 300); do echo "$(date) Try: ${i}"; echo hostname | nc -u -w 5 -p %d %s %d; echo; done`, srcPort, serverNodeInfo.nodeIP, udpService.Spec.Ports[0].NodePort)
+		clientPod.Spec.Containers[0].Command = []string{"/bin/sh", "-c", cmd}
+		clientPod.Spec.Containers[0].Name = podClient
+		fr.PodClient().CreateSync(clientPod)
+
+		// Read the client pod logs
+		logs, err := e2epod.GetPodLogs(cs, ns, podClient, podClient)
+		framework.ExpectNoError(err)
+		framework.Logf("Pod client logs: %s", logs)
+
+		// Add a backend pod to the service in the other node
+		ginkgo.By("creating a backend pod " + podBackend1 + " for the service " + serviceName)
+		serverPod1 := newAgnhostPod(podBackend1, "netexec", fmt.Sprintf("--udp-port=%d", 80))
+		serverPod1.Labels = udpJig.Labels
+		serverPod1.Spec.NodeName = serverNodeInfo.name
+		fr.PodClient().CreateSync(serverPod1)
+
+		// Waiting for service to expose endpoint.
+		err = validateEndpointsPorts(cs, ns, serviceName, portsByPodName{podBackend1: {80}})
+		framework.ExpectNoError(err, "failed to validate endpoints for service %s in namespace: %s", serviceName, ns)
+
+		// Check that the pod receives the traffic
+		// UDP conntrack entries timeout is 30 sec by default
+		ginkgo.By("checking client pod connected to the backend 1 on Node IP " + serverNodeInfo.nodeIP)
+		time.Sleep(30 * time.Second)
+		logs, err = e2epod.GetPodLogs(cs, ns, podClient, podClient)
+		framework.ExpectNoError(err)
+		framework.Logf("Pod client logs: %s", logs)
+		if !strings.Contains(string(logs), podBackend1) {
+			framework.Failf("Failed to connecto to backend 1")
+		}
+
+		// Create a second pod
+		ginkgo.By("creating a second backend pod " + podBackend2 + " for the service " + serviceName)
+		serverPod2 := newAgnhostPod(podBackend2, "netexec", fmt.Sprintf("--udp-port=%d", 80))
+		serverPod2.Labels = udpJig.Labels
+		serverPod2.Spec.NodeName = serverNodeInfo.name
+		fr.PodClient().CreateSync(serverPod2)
+
+		// and delete the first pod
+		framework.Logf("Cleaning up %s pod", podBackend1)
+		fr.PodClient().DeleteSync(podBackend1, metav1.DeleteOptions{}, framework.DefaultPodDeletionTimeout)
+
+		// Check that the second pod keeps receiving traffic
+		// UDP conntrack entries timeout is 30 sec by default
+		ginkgo.By("checking client pod connected to the backend 2 on Node IP " + serverNodeInfo.nodeIP)
+		time.Sleep(30 * time.Second)
+		logs, err = e2epod.GetPodLogs(cs, ns, podClient, podClient)
+		framework.ExpectNoError(err)
+		framework.Logf("Pod client logs: %s", logs)
+		if !strings.Contains(string(logs), podBackend2) {
+			framework.Failf("Failed to connecto to backend 2")
+		}
+	})
+
+	ginkgo.It("should be able to preserve UDP traffic when server pod cycles for a ClusterIP service", func() {
+
+		// Create a NodePort service
+		udpJig := e2eservice.NewTestJig(cs, ns, serviceName)
+		ginkgo.By("creating a UDP service " + serviceName + " with type=ClusterIP in " + ns)
+		udpService, err := udpJig.CreateUDPService(func(svc *v1.Service) {
+			svc.Spec.Type = v1.ServiceTypeClusterIP
+			svc.Spec.Ports = []v1.ServicePort{
+				{Port: 80, Name: "udp", Protocol: v1.ProtocolUDP, TargetPort: intstr.FromInt(80)},
+			}
+		})
+		framework.ExpectNoError(err)
+
+		// Create a pod in one node to create the UDP traffic against the NodePort service every 5 seconds
+		ginkgo.By("creating a client pod for probing the service " + serviceName)
+		clientPod := newAgnhostPod(podClient, "")
+		clientPod.Spec.NodeName = clientNodeInfo.name
+		cmd := fmt.Sprintf(`date; for i in $(seq 1 300); do echo "$(date) Try: ${i}"; echo hostname | nc -u -w 5 -p %d %s %d; echo; done`, srcPort, udpService.Spec.ClusterIP, udpService.Spec.Ports[0].Port)
+		clientPod.Spec.Containers[0].Command = []string{"/bin/sh", "-c", cmd}
+		clientPod.Spec.Containers[0].Name = podClient
+		fr.PodClient().CreateSync(clientPod)
+
+		// Read the client pod logs
+		logs, err := e2epod.GetPodLogs(cs, ns, podClient, podClient)
+		framework.ExpectNoError(err)
+		framework.Logf("Pod client logs: %s", logs)
+
+		// Add a backend pod to the service in the other node
+		ginkgo.By("creating a backend pod " + podBackend1 + " for the service " + serviceName)
+		serverPod1 := newAgnhostPod(podBackend1, "netexec", fmt.Sprintf("--udp-port=%d", 80))
+		serverPod1.Labels = udpJig.Labels
+		serverPod1.Spec.NodeName = serverNodeInfo.name
+		fr.PodClient().CreateSync(serverPod1)
+
+		// Waiting for service to expose endpoint.
+		err = validateEndpointsPorts(cs, ns, serviceName, portsByPodName{podBackend1: {80}})
+		framework.ExpectNoError(err, "failed to validate endpoints for service %s in namespace: %s", serviceName, ns)
+
+		// Check that the pod receives the traffic
+		// UDP conntrack entries timeout is 30 sec by default
+		ginkgo.By("checking client pod connected to the backend 1 on Node IP " + serverNodeInfo.nodeIP)
+		time.Sleep(30 * time.Second)
+		logs, err = e2epod.GetPodLogs(cs, ns, podClient, podClient)
+		framework.ExpectNoError(err)
+		framework.Logf("Pod client logs: %s", logs)
+		if !strings.Contains(string(logs), podBackend1) {
+			framework.Failf("Failed to connecto to backend 1")
+		}
+
+		// Create a second pod
+		ginkgo.By("creating a second backend pod " + podBackend2 + " for the service " + serviceName)
+		serverPod2 := newAgnhostPod(podBackend2, "netexec", fmt.Sprintf("--udp-port=%d", 80))
+		serverPod2.Labels = udpJig.Labels
+		serverPod2.Spec.NodeName = serverNodeInfo.name
+		fr.PodClient().CreateSync(serverPod2)
+
+		// and delete the first pod
+		framework.Logf("Cleaning up %s pod", podBackend1)
+		fr.PodClient().DeleteSync(podBackend1, metav1.DeleteOptions{}, framework.DefaultPodDeletionTimeout)
+
+		// Check that the second pod keeps receiving traffic
+		// UDP conntrack entries timeout is 30 sec by default
+		ginkgo.By("checking client pod connected to the backend 2 on Node IP " + serverNodeInfo.nodeIP)
+		time.Sleep(30 * time.Second)
+		logs, err = e2epod.GetPodLogs(cs, ns, podClient, podClient)
+		framework.ExpectNoError(err)
+		framework.Logf("Pod client logs: %s", logs)
+		if !strings.Contains(string(logs), podBackend2) {
+			framework.Failf("Failed to connecto to backend 2")
+		}
+	})
+})