Add port forwarding integration test

integration/dev_test.go

@@ -27,8 +27,10 @@ import (
 	"time"
 
 	"github.com/GoogleContainerTools/skaffold/integration/skaffold"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/util"
 	"github.com/GoogleContainerTools/skaffold/proto"
 	"github.com/GoogleContainerTools/skaffold/testutil"
+	"github.com/sirupsen/logrus"
 	"k8s.io/apimachinery/pkg/util/wait"
 )
 
@@ -212,7 +214,7 @@ func TestDevPortForward(t *testing.T) {
 	originalResponse := "leeroooooy app!!"
 	replacementResponse := "test string"
 
-	waitForPortForwardEvent(t, entries, "leeroy-app", "service", originalResponse+"\n")
+	waitForPortForwardEvent(t, entries, "leeroy-app", "service", ns.Name, originalResponse+"\n")
 
 	original, perms, fErr := replaceInFile(originalResponse, replacementResponse, "examples/microservices/leeroy-app/app.go")
 	if fErr != nil {
@@ -224,7 +226,7 @@ func TestDevPortForward(t *testing.T) {
 		}
 	}()
 
-	waitForPortForwardEvent(t, entries, "leeroy-app", "service", replacementResponse+"\n")
+	waitForPortForwardEvent(t, entries, "leeroy-app", "service", ns.Name, replacementResponse+"\n")
 }
 
 func TestDevPortForwardGKELoadBalancer(t *testing.T) {
@@ -267,46 +269,66 @@ func TestDevPortForwardGKELoadBalancer(t *testing.T) {
 		}
 	}()
 
-	waitForPortForwardEvent(t, entries, "gke-loadbalancer", "service", "hello!!\n")
+	waitForPortForwardEvent(t, entries, "gke-loadbalancer", "service", ns.Name, "hello!!\n")
 }
 
-func waitForPortForwardEvent(t *testing.T, entries chan *proto.LogEntry, resourceName, resourceType, expected string) {
+func getLocalPortFromPortForwardEvent(t *testing.T, entries chan *proto.LogEntry, resourceName, resourceType, namespace string) int {
 	timeout := time.After(1 * time.Minute)
-	var port int32
-portForwardEvent:
 	for {
 		select {
 		case <-timeout:
-			t.Errorf("timed out waiting for port forwarding event")
-			break portForwardEvent
+			t.Fatalf("timed out waiting for port forwarding event")
 		case e := <-entries:
 			switch e.Event.GetEventType().(type) {
 			case *proto.Event_PortEvent:
+				t.Logf("event received %v", e)
 				if e.Event.GetPortEvent().ResourceName == resourceName &&
-					e.Event.GetPortEvent().ResourceType == resourceType {
-					port = e.Event.GetPortEvent().LocalPort
+					e.Event.GetPortEvent().ResourceType == resourceType &&
+					e.Event.GetPortEvent().Namespace == namespace {
+					port := e.Event.GetPortEvent().LocalPort
 					t.Logf("Detected %s/%s is forwarded to port %d", resourceType, resourceName, port)
-					break portForwardEvent
+					return int(port)
 				}
 			default:
 				t.Logf("event received %v", e)
 			}
 		}
 	}
-	var body []byte
-	err := wait.PollImmediate(time.Millisecond*2000, 1*time.Minute, func() (bool, error) {
-		resp, err := http.Get(fmt.Sprintf("http://localhost:%d", port))
-		if err != nil {
-			t.Logf("could not get %s/%s due to %s", resourceType, resourceName, err)
-			return false, nil
-		}
-		defer resp.Body.Close()
-		body, err = ioutil.ReadAll(resp.Body)
-		t.Logf("got %s from port %d but wanted %s", string(body), port, expected)
-		return string(body) == expected, err
-	})
+}
+
+func waitForPortForwardEvent(t *testing.T, entries chan *proto.LogEntry, resourceName, resourceType, namespace, expected string) {
+	port := getLocalPortFromPortForwardEvent(t, entries, resourceName, resourceType, namespace)
+	assertResponseFromPort(t, port, expected)
+}
+
+// assertResponseFromPort waits for two minutes for the expected response at port.
+func assertResponseFromPort(t *testing.T, port int, expected string) {
+	logrus.Infof("Waiting for response %s from port %d", expected, port)
+	ctx, cancelTimeout := context.WithTimeout(context.Background(), 2*time.Minute)
+	defer cancelTimeout()
 
-	testutil.CheckErrorAndDeepEqual(t, false, err, string(body), expected)
+	for {
+		select {
+		case <-ctx.Done():
+			t.Fatalf("Timed out waiting for response from port %d", port)
+		case <-time.After(1 * time.Second):
+			resp, err := http.Get(fmt.Sprintf("http://%s:%d", util.Loopback, port))
+			if err != nil {
+				logrus.Infof("error getting response from port %d: %v", port, err)
+				continue
+			}
+			defer resp.Body.Close()
+			body, err := ioutil.ReadAll(resp.Body)
+			if err != nil {
+				logrus.Infof("error reading response: %v", err)
+				continue
+			}
+			if string(body) == expected {
+				return
+			}
+			logrus.Infof("didn't get expected response from port. got: %s, expected: %s", string(body), expected)
+		}
+	}
 }
 
 func replaceInFile(target, replacement, filepath string) ([]byte, os.FileMode, error) {

integration/port_forward_test.go

@@ -0,0 +1,137 @@
+/*
+Copyright 2019 The Skaffold 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 integration
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/constants"
+	"github.com/GoogleContainerTools/skaffold/proto"
+	"github.com/sirupsen/logrus"
+
+	"github.com/GoogleContainerTools/skaffold/integration/skaffold"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/config"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubectl"
+	kubectx "github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubernetes/context"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubernetes/portforward"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/runner/runcontext"
+)
+
+func TestPortForward(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test")
+	}
+	if ShouldRunGCPOnlyTests() {
+		t.Skip("skipping test that is not gcp only")
+	}
+
+	ns, _, deleteNs := SetupNamespace(t)
+	defer deleteNs()
+
+	dir := "examples/microservices"
+	skaffold.Run().InDir(dir).InNs(ns.Name).RunOrFailOutput(t)
+
+	cfg, err := kubectx.CurrentConfig()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	kubectlCLI := kubectl.NewFromRunContext(&runcontext.RunContext{
+		KubeContext: cfg.CurrentContext,
+		Opts: config.SkaffoldOptions{
+			Namespace: ns.Name,
+		},
+	})
+
+	logrus.SetLevel(logrus.TraceLevel)
+	portforward.WhiteBoxPortForwardCycle(t, kubectlCLI, ns.Name)
+}
+
+// TestPortForwardDeletePod tests that port forwarding works
+// as expected. Then, the test force deletes a pod,
+// and tests that the pod eventually comes up at the same port again.
+func TestPortForwardDeletePod(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test")
+	}
+	if ShouldRunGCPOnlyTests() {
+		t.Skip("skipping test that is not gcp only")
+	}
+
+	ns, _, deleteNs := SetupNamespace(t)
+	defer deleteNs()
+
+	rpcAddr := randomPort()
+	env := []string{fmt.Sprintf("TEST_NS=%s", ns.Name)}
+	cmd := skaffold.Dev("--cache-artifacts=true", "--port-forward", "--rpc-port", rpcAddr, "-v=info").InDir("examples/microservices").InNs(ns.Name).WithEnv(env)
+	stop := cmd.RunBackground(t)
+	defer stop()
+
+	client, shutdown := setupRPCClient(t, rpcAddr)
+	defer shutdown()
+
+	// create a grpc connection. Increase number of reties for helm.
+	stream, err := readEventAPIStream(client, t, 20)
+	if stream == nil {
+		t.Fatalf("error retrieving event log: %v\n", err)
+	}
+
+	// read entries from the log
+	entries := make(chan *proto.LogEntry)
+	go func() {
+		for {
+			entry, _ := stream.Recv()
+			if entry != nil {
+				entries <- entry
+			}
+		}
+	}()
+
+	localPort := getLocalPortFromPortForwardEvent(t, entries, "leeroy-app", "service", ns.Name)
+	assertResponseFromPort(t, localPort, constants.LeeroyAppResponse)
+
+	// now, delete all pods in this namespace.
+	logrus.Infof("Deleting all pods in namespace %s", ns.Name)
+	kubectlCLI := getKubectlCLI(t, ns.Name)
+	killPodsCmd := kubectlCLI.Command(context.Background(),
+		"delete",
+		"pods", "--all",
+		"-n", ns.Name,
+	)
+
+	if output, err := killPodsCmd.CombinedOutput(); err != nil {
+		t.Fatalf("error deleting all pods: %v \n %s", err, string(output))
+	}
+	// port forwarding should come up again on the same port
+	assertResponseFromPort(t, localPort, constants.LeeroyAppResponse)
+}
+
+func getKubectlCLI(t *testing.T, ns string) *kubectl.CLI {
+	cfg, err := kubectx.CurrentConfig()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	return kubectl.NewFromRunContext(&runcontext.RunContext{
+		KubeContext: cfg.CurrentContext,
+		Opts: config.SkaffoldOptions{
+			Namespace: ns,
+		},
+	})
+}

pkg/skaffold/constants/constants.go

@@ -67,6 +67,8 @@ const (
 	DefaultRPCHTTPPort = 50052
 
 	DefaultPortForwardNamespace = "default"
+
+	LeeroyAppResponse = "leeroooooy app!!\n"
 )
 
 var (

pkg/skaffold/kubernetes/portforward/port_forward_integration.go

@@ -23,11 +23,10 @@ import (
 	"testing"
 	"time"
 
-	"github.com/sirupsen/logrus"
-	"k8s.io/apimachinery/pkg/util/wait"
-
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubectl"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/schema/latest"
+	"github.com/sirupsen/logrus"
+	"k8s.io/apimachinery/pkg/util/wait"
 )
 
 // For WhiteBox testing only