Merge pull request #21602 from ncdc/port-forward-kubectl-timeout-flake

test/e2e/framework.go

@@ -248,6 +248,12 @@ func (f *Framework) WaitForPodRunningSlow(podName string) error {
 	return waitForPodRunningInNamespaceSlow(f.Client, podName, f.Namespace.Name)
 }
 
+// WaitForPodNoLongerRunning waits for the pod to no longer be running in the namespace, for either
+// success or failure.
+func (f *Framework) WaitForPodNoLongerRunning(podName string) error {
+	return waitForPodNoLongerRunningInNamespace(f.Client, podName, f.Namespace.Name)
+}
+
 // Runs the given pod and verifies that the output of exact container matches the desired output.
 func (f *Framework) TestContainerOutput(scenarioName string, pod *api.Pod, containerIndex int, expectedOutput []string) {
 	testContainerOutput(scenarioName, f.Client, pod, containerIndex, expectedOutput, f.Namespace.Name)

test/e2e/portforward.go

@@ -24,10 +24,8 @@ import (
 	"regexp"
 	"strconv"
 	"strings"
-	"time"
 
 	"k8s.io/kubernetes/pkg/api"
-	"k8s.io/kubernetes/pkg/util/wait"
 
 	. "github.com/onsi/ginkgo"
 )
@@ -112,21 +110,6 @@ func runPortForward(ns, podName string, port int) (*exec.Cmd, int) {
 	return cmd, listenPort
 }
 
-func runKubectlWithTimeout(timeout time.Duration, args ...string) string {
-	logOutput := make(chan string)
-	go func() {
-		defer GinkgoRecover()
-		logOutput <- runKubectlOrDie(args...)
-	}()
-	select {
-	case <-time.After(timeout):
-		Failf("kubectl timed out")
-		return ""
-	case o := <-logOutput:
-		return o
-	}
-}
-
 var _ = Describe("Port forwarding", func() {
 	framework := NewFramework("port-forwarding")
 
@@ -134,8 +117,12 @@ var _ = Describe("Port forwarding", func() {
 		It("should support a client that connects, sends no data, and disconnects [Conformance]", func() {
 			By("creating the target pod")
 			pod := pfPod("abc", "1", "1", "1")
-			framework.Client.Pods(framework.Namespace.Name).Create(pod)
-			framework.WaitForPodRunning(pod.Name)
+			if _, err := framework.Client.Pods(framework.Namespace.Name).Create(pod); err != nil {
+				Failf("Couldn't create pod: %v", err)
+			}
+			if err := framework.WaitForPodRunning(pod.Name); err != nil {
+				Failf("Pod did not start running: %v", err)
+			}
 
 			By("Running 'kubectl port-forward'")
 			cmd, listenPort := runPortForward(framework.Namespace.Name, pod.Name, 80)
@@ -150,7 +137,18 @@ var _ = Describe("Port forwarding", func() {
 			By("Closing the connection to the local port")
 			conn.Close()
 
-			logOutput := runKubectlWithTimeout(wait.ForeverTestTimeout, "logs", fmt.Sprintf("--namespace=%v", framework.Namespace.Name), "-f", podName)
+			By("Waiting for the target pod to stop running")
+			if err := framework.WaitForPodNoLongerRunning(pod.Name); err != nil {
+				Failf("Pod did not stop running: %v", err)
+			}
+
+			By("Retrieving logs from the target pod")
+			logOutput, err := getPodLogs(framework.Client, framework.Namespace.Name, pod.Name, "portforwardtester")
+			if err != nil {
+				Failf("Error retrieving logs: %v", err)
+			}
+
+			By("Verifying logs")
 			verifyLogMessage(logOutput, "Accepted client connection")
 			verifyLogMessage(logOutput, "Expected to read 3 bytes from client, but got 0 instead")
 		})
@@ -158,8 +156,12 @@ var _ = Describe("Port forwarding", func() {
 		It("should support a client that connects, sends data, and disconnects [Conformance]", func() {
 			By("creating the target pod")
 			pod := pfPod("abc", "10", "10", "100")
-			framework.Client.Pods(framework.Namespace.Name).Create(pod)
-			framework.WaitForPodRunning(pod.Name)
+			if _, err := framework.Client.Pods(framework.Namespace.Name).Create(pod); err != nil {
+				Failf("Couldn't create pod: %v", err)
+			}
+			if err := framework.WaitForPodRunning(pod.Name); err != nil {
+				Failf("Pod did not start running: %v", err)
+			}
 
 			By("Running 'kubectl port-forward'")
 			cmd, listenPort := runPortForward(framework.Namespace.Name, pod.Name, 80)
@@ -195,7 +197,18 @@ var _ = Describe("Port forwarding", func() {
 				Failf("Expected %q from server, got %q", e, a)
 			}
 
-			logOutput := runKubectlWithTimeout(wait.ForeverTestTimeout, "logs", fmt.Sprintf("--namespace=%v", framework.Namespace.Name), "-f", podName)
+			By("Waiting for the target pod to stop running")
+			if err := framework.WaitForPodNoLongerRunning(pod.Name); err != nil {
+				Failf("Pod did not stop running: %v", err)
+			}
+
+			By("Retrieving logs from the target pod")
+			logOutput, err := getPodLogs(framework.Client, framework.Namespace.Name, pod.Name, "portforwardtester")
+			if err != nil {
+				Failf("Error retrieving logs: %v", err)
+			}
+
+			By("Verifying logs")
 			verifyLogMessage(logOutput, "^Accepted client connection$")
 			verifyLogMessage(logOutput, "^Received expected client data$")
 			verifyLogMessage(logOutput, "^Done$")
@@ -205,8 +218,12 @@ var _ = Describe("Port forwarding", func() {
 		It("should support a client that connects, sends no data, and disconnects [Conformance]", func() {
 			By("creating the target pod")
 			pod := pfPod("", "10", "10", "100")
-			framework.Client.Pods(framework.Namespace.Name).Create(pod)
-			framework.WaitForPodRunning(pod.Name)
+			if _, err := framework.Client.Pods(framework.Namespace.Name).Create(pod); err != nil {
+				Failf("Couldn't create pod: %v", err)
+			}
+			if err := framework.WaitForPodRunning(pod.Name); err != nil {
+				Failf("Pod did not start running: %v", err)
+			}
 
 			By("Running 'kubectl port-forward'")
 			cmd, listenPort := runPortForward(framework.Namespace.Name, pod.Name, 80)
@@ -232,7 +249,18 @@ var _ = Describe("Port forwarding", func() {
 				Failf("Expected %q from server, got %q", e, a)
 			}
 
-			logOutput := runKubectlWithTimeout(wait.ForeverTestTimeout, "logs", fmt.Sprintf("--namespace=%v", framework.Namespace.Name), "-f", podName)
+			By("Waiting for the target pod to stop running")
+			if err := framework.WaitForPodNoLongerRunning(pod.Name); err != nil {
+				Failf("Pod did not stop running: %v", err)
+			}
+
+			By("Retrieving logs from the target pod")
+			logOutput, err := getPodLogs(framework.Client, framework.Namespace.Name, pod.Name, "portforwardtester")
+			if err != nil {
+				Failf("Error retrieving logs: %v", err)
+			}
+
+			By("Verifying logs")
 			verifyLogMessage(logOutput, "Accepted client connection")
 			verifyLogMessage(logOutput, "Done")
 		})

test/e2e/util.go

@@ -76,6 +76,9 @@ const (
 	// TODO: Make this 30 seconds once #4566 is resolved.
 	podStartTimeout = 5 * time.Minute
 
+	// How long to wait for the pod to no longer be running
+	podNoLongerRunningTimeout = 30 * time.Second
+
 	// If there are any orphaned namespaces to clean up, this test is running
 	// on a long lived cluster. A long wait here is preferably to spurious test
 	// failures caused by leaked resources from a previous test run.
@@ -799,6 +802,22 @@ func waitTimeoutForPodRunningInNamespace(c *client.Client, podName string, names
 	})
 }
 
+// Waits default amount of time (podNoLongerRunningTimeout) for the specified pod to stop running.
+// Returns an error if timeout occurs first.
+func waitForPodNoLongerRunningInNamespace(c *client.Client, podName string, namespace string) error {
+	return waitTimeoutForPodNoLongerRunningInNamespace(c, podName, namespace, podNoLongerRunningTimeout)
+}
+
+func waitTimeoutForPodNoLongerRunningInNamespace(c *client.Client, podName string, namespace string, timeout time.Duration) error {
+	return waitForPodCondition(c, namespace, podName, "no longer running", timeout, func(pod *api.Pod) (bool, error) {
+		if pod.Status.Phase == api.PodSucceeded || pod.Status.Phase == api.PodFailed {
+			Logf("Found pod '%s' with status '%s' on node '%s'", podName, pod.Status.Phase, pod.Spec.NodeName)
+			return true, nil
+		}
+		return false, nil
+	})
+}
+
 // waitForPodNotPending returns an error if it took too long for the pod to go out of pending state.
 func waitForPodNotPending(c *client.Client, ns, podName string) error {
 	return waitForPodCondition(c, ns, podName, "!pending", podStartTimeout, func(pod *api.Pod) (bool, error) {