Automated cherry pick of #37103 #38260

test/e2e/framework/util.go

@@ -78,6 +78,7 @@ import (
 	utilyaml "k8s.io/kubernetes/pkg/util/yaml"
 	"k8s.io/kubernetes/pkg/version"
 	"k8s.io/kubernetes/pkg/watch"
+	testutils "k8s.io/kubernetes/test/utils"
 
 	"github.com/blang/semver"
 	"golang.org/x/crypto/ssh"
@@ -1458,6 +1459,17 @@ func WaitForPodNotPending(c *client.Client, ns, podName, resourceVersion string)
 	return err
 }
 
+// WaitForTerminatedContainer returns an error if it took to long for the container
+// to terminate.
+func WaitForTerminatedContainer(c *client.Client, pod *api.Pod, namespace, containerName string) error {
+	return waitForPodCondition(c, namespace, pod.Name, "container terminated", PodStartTimeout, func(pod *api.Pod) (bool, error) {
+		if len(testutils.TerminatedContainers(pod)[containerName]) > 0 {
+			return true, nil
+		}
+		return false, nil
+	})
+}
+
 // waitForPodTerminatedInNamespace returns an error if it took too long for the pod
 // to terminate or if the pod terminated with an unexpected reason.
 func waitForPodTerminatedInNamespace(c *client.Client, podName, reason, namespace string) error {

test/e2e/portforward.go

@@ -54,9 +54,24 @@ func pfPod(expectedClientData, chunks, chunkSize, chunkIntervalMillis string) *a
 		},
 		Spec: api.PodSpec{
 			Containers: []api.Container{
+				{
+					Name:  "readiness",
+					Image: "gcr.io/google_containers/netexec:1.7",
+					ReadinessProbe: &api.Probe{
+						Handler: api.Handler{
+							Exec: &api.ExecAction{
+								Command: []string{
+									"sh", "-c", "netstat -na | grep LISTEN | grep -v 8080 | grep 80",
+								}},
+						},
+						InitialDelaySeconds: 5,
+						TimeoutSeconds:      60,
+						PeriodSeconds:       1,
+					},
+				},
 				{
 					Name:  "portforwardtester",
-					Image: "gcr.io/google_containers/portforwardtester:1.0",
+					Image: "gcr.io/google_containers/portforwardtester:1.2",
 					Env: []api.EnvVar{
 						{
 							Name:  "BIND_PORT",
@@ -180,7 +195,7 @@ var _ = framework.KubeDescribe("Port forwarding", func() {
 			if _, err := f.Client.Pods(f.Namespace.Name).Create(pod); err != nil {
 				framework.Failf("Couldn't create pod: %v", err)
 			}
-			if err := f.WaitForPodRunning(pod.Name); err != nil {
+			if err := f.WaitForPodReady(pod.Name); err != nil {
 				framework.Failf("Pod did not start running: %v", err)
 			}
 			defer func() {
@@ -206,8 +221,8 @@ var _ = framework.KubeDescribe("Port forwarding", func() {
 			conn.Close()
 
 			By("Waiting for the target pod to stop running")
-			if err := f.WaitForPodNoLongerRunning(pod.Name); err != nil {
-				framework.Failf("Pod did not stop running: %v", err)
+			if err := framework.WaitForTerminatedContainer(f.Client, pod, f.Namespace.Name, "portforwardtester"); err != nil {
+				framework.Failf("Container did not terminate: %v", err)
 			}
 
 			By("Verifying logs")
@@ -225,7 +240,7 @@ var _ = framework.KubeDescribe("Port forwarding", func() {
 			if _, err := f.Client.Pods(f.Namespace.Name).Create(pod); err != nil {
 				framework.Failf("Couldn't create pod: %v", err)
 			}
-			if err := f.WaitForPodRunning(pod.Name); err != nil {
+			if err := f.WaitForPodReady(pod.Name); err != nil {
 				framework.Failf("Pod did not start running: %v", err)
 			}
 			defer func() {
@@ -272,8 +287,8 @@ var _ = framework.KubeDescribe("Port forwarding", func() {
 			}
 
 			By("Waiting for the target pod to stop running")
-			if err := f.WaitForPodNoLongerRunning(pod.Name); err != nil {
-				framework.Failf("Pod did not stop running: %v", err)
+			if err := framework.WaitForTerminatedContainer(f.Client, pod, f.Namespace.Name, "portforwardtester"); err != nil {
+				framework.Failf("Container did not terminate: %v", err)
 			}
 
 			By("Verifying logs")
@@ -293,7 +308,7 @@ var _ = framework.KubeDescribe("Port forwarding", func() {
 			if _, err := f.Client.Pods(f.Namespace.Name).Create(pod); err != nil {
 				framework.Failf("Couldn't create pod: %v", err)
 			}
-			if err := f.WaitForPodRunning(pod.Name); err != nil {
+			if err := f.WaitForPodReady(pod.Name); err != nil {
 				framework.Failf("Pod did not start running: %v", err)
 			}
 			defer func() {
@@ -330,8 +345,8 @@ var _ = framework.KubeDescribe("Port forwarding", func() {
 			}
 
 			By("Waiting for the target pod to stop running")
-			if err := f.WaitForPodNoLongerRunning(pod.Name); err != nil {
-				framework.Failf("Pod did not stop running: %v", err)
+			if err := framework.WaitForTerminatedContainer(f.Client, pod, f.Namespace.Name, "portforwardtester"); err != nil {
+				framework.Failf("Container did not terminate: %v", err)
 			}
 
 			By("Verifying logs")

test/images/port-forward-tester/Makefile

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-TAG = 1.0
+TAG = 1.2
 PREFIX = gcr.io/google_containers
 
 all: push

test/images/port-forward-tester/portforwardtester.go

@@ -47,20 +47,40 @@ func getEnvInt(name string) int {
 	return value
 }
 
+// taken from net/http/server.go:
+//
+// rstAvoidanceDelay is the amount of time we sleep after closing the
+// write side of a TCP connection before closing the entire socket.
+// By sleeping, we increase the chances that the client sees our FIN
+// and processes its final data before they process the subsequent RST
+// from closing a connection with known unread data.
+// This RST seems to occur mostly on BSD systems. (And Windows?)
+// This timeout is somewhat arbitrary (~latency around the planet).
+const rstAvoidanceDelay = 500 * time.Millisecond
+
 func main() {
+	bindAddress := os.Getenv("BIND_ADDRESS")
+	if bindAddress == "" {
+		bindAddress = "localhost"
+	}
 	bindPort := os.Getenv("BIND_PORT")
-	listener, err := net.Listen("tcp", fmt.Sprintf("localhost:%s", bindPort))
+	addr, err := net.ResolveTCPAddr("tcp", net.JoinHostPort(bindAddress, bindPort))
+	if err != nil {
+		fmt.Printf("Error resolving: %v\n", err)
+		os.Exit(1)
+	}
+	listener, err := net.ListenTCP("tcp", addr)
 	if err != nil {
 		fmt.Printf("Error listening: %v\n", err)
 		os.Exit(1)
 	}
 
-	conn, err := listener.Accept()
+	conn, err := listener.AcceptTCP()
 	if err != nil {
 		fmt.Printf("Error accepting connection: %v\n", err)
 		os.Exit(1)
 	}
-	defer conn.Close()
+
 	fmt.Println("Accepted client connection")
 
 	expectedClientData := os.Getenv("EXPECTED_CLIENT_DATA")
@@ -102,5 +122,20 @@ func main() {
 		}
 	}
 
+	fmt.Println("Shutting down connection")
+
+	// set linger timeout to flush buffers. This is the official way according to the go api docs. But
+	// there are controversial discussions whether this value has any impact on most platforms
+	// (compare https://codereview.appspot.com/95320043).
+	conn.SetLinger(-1)
+
+	// Flush the connection cleanly, following https://blog.netherlabs.nl/articles/2009/01/18/the-ultimate-so_linger-page-or-why-is-my-tcp-not-reliable:
+	// 1. close write half of connection which sends a FIN packet
+	// 2. give client some time to receive the FIN
+	// 3. close the complete connection
+	conn.CloseWrite()
+	time.Sleep(rstAvoidanceDelay)
+	conn.Close()
+
 	fmt.Println("Done")
 }