Cleaner kubectl port-forward retry logic

examples/microservices/leeroy-app/app.go

@@ -7,7 +7,7 @@ import (
 )
 
 func handler(w http.ResponseWriter, r *http.Request) {
-	fmt.Fprintf(w, "leeroooooy app!!\n")
+	fmt.Fprintf(w, "6\n")
 }
 
 func main() {

integration/dev_test.go

@@ -267,26 +267,38 @@ func TestDevPortForwardGKELoadBalancer(t *testing.T) {
 	}()
 
 	body := []byte{}
-	err = wait.PollImmediate(time.Millisecond*500, 5*time.Minute, func() (bool, error) {
-		e := <-entries
-		switch e.Event.GetEventType().(type) {
-		case *proto.Event_PortEvent:
-			if e.Event.GetPortEvent().ResourceName == "gke-loadbalancer" &&
-				e.Event.GetPortEvent().ResourceType == "service" {
-				port := e.Event.GetPortEvent().LocalPort
-				t.Logf("Detected service/gke-loadbalancer is forwarded to port %d", port)
-				resp, err := http.Get(fmt.Sprintf("http://localhost:%d", port))
-				if err != nil {
-					t.Errorf("could not get service/gke-loadbalancer due to %s", err)
+	var port int32
+	timeout := time.After(1 * time.Minute)
+
+portForwardEvent:
+	for {
+		select {
+		case <-timeout:
+			t.Errorf("timed out waiting for port forwarding event")
+			break portForwardEvent
+		case e := <-entries:
+			switch e.Event.GetEventType().(type) {
+			case *proto.Event_PortEvent:
+				if e.Event.GetPortEvent().ResourceName == "gke-loadbalancer" &&
+					e.Event.GetPortEvent().ResourceType == "service" {
+					port = e.Event.GetPortEvent().LocalPort
+					t.Logf("Detected service/gke-loadbalancer is forwarded to port %d", port)
+					break portForwardEvent
 				}
-				defer resp.Body.Close()
-				body, err = ioutil.ReadAll(resp.Body)
-				return true, err
+			default:
+				t.Logf("event received %v", e)
 			}
-			return false, nil
-		default:
+		}
+	}
+	err = wait.PollImmediate(time.Millisecond*500, 3*time.Minute, func() (bool, error) {
+		resp, err := http.Get(fmt.Sprintf("http://localhost:%d", port))
+		if err != nil {
+			t.Logf("could not get service/gke-loadbalancer due to %s", err)
 			return false, nil
 		}
+		defer resp.Body.Close()
+		body, err = ioutil.ReadAll(resp.Body)
+		return true, err
 	})
 
 	testutil.CheckErrorAndDeepEqual(t, false, err, string(body), "hello!!\n")

pkg/skaffold/kubernetes/portforward/entry_manager.go

@@ -21,19 +21,14 @@ import (
 	"fmt"
 	"io"
 	"sync"
-	"time"
-
-	"k8s.io/apimachinery/pkg/util/wait"
 
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/color"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/event"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubectl"
 )
 
 var (
-	// For testing
-	forwardingTimeoutTime = time.Minute
-	portForwardEvent      = func(entry *portForwardEntry) {
+	portForwardEvent = func(entry *portForwardEntry) {
 		// TODO priyawadhwa@, change event API to accept ports of type int
 		event.PortForwarded(
 			int32(entry.localPort),
@@ -157,35 +152,27 @@ func NewEntryManager(out io.Writer, cli *kubectl.CLI) EntryManager {
 		output:             out,
 		forwardedPorts:     newForwardedPorts(),
 		forwardedResources: newForwardedResources(),
-		EntryForwarder:     &KubectlForwarder{kubectl: cli},
+		EntryForwarder:     &KubectlForwarder{kubectl: cli, out: out},
 	}
 }
 
-func (b *EntryManager) forwardPortForwardEntry(ctx context.Context, entry *portForwardEntry) error {
+func (b *EntryManager) forwardPortForwardEntry(ctx context.Context, entry *portForwardEntry) {
 	// Check if this resource has already been forwarded
 	if _, ok := b.forwardedResources.Load(entry.key()); ok {
-		return nil
+		return
 	}
 	b.forwardedResources.Store(entry.key(), entry)
-	err := wait.PollImmediate(time.Second, forwardingTimeoutTime, func() (bool, error) {
-		if err := b.Forward(ctx, entry); err != nil {
-			return false, nil
-		}
-		return true, nil
-	})
-
-	go b.Monitor(entry, func() {
-		b.Retry(ctx, entry)
-	})
-
-	if err != nil {
-		return err
-	}
 
-	color.Default.Fprintln(b.output, fmt.Sprintf("Port forwarded %s/%s from remote port %d to local port %d", entry.resource.Type, entry.resource.Name, entry.resource.Port, entry.localPort))
+	b.Forward(ctx, entry)
 
+	color.Default.Fprintln(
+		b.output,
+		fmt.Sprintf("Port forwarding %s/%s from remote port %d to local port %d",
+			entry.resource.Type,
+			entry.resource.Name,
+			entry.resource.Port,
+			entry.localPort))
 	portForwardEvent(entry)
-	return nil
 }
 
 // Stop terminates all kubectl port-forward commands.
@@ -201,8 +188,3 @@ func (b *EntryManager) Terminate(p *portForwardEntry) {
 	b.forwardedPorts.Delete(p.localPort)
 	b.EntryForwarder.Terminate(p)
 }
-
-func (b *EntryManager) Retry(ctx context.Context, p *portForwardEntry) error {
-	b.Terminate(p)
-	return b.forwardPortForwardEntry(ctx, p)
-}

pkg/skaffold/kubernetes/portforward/entry_manager_test.go

@@ -69,7 +69,7 @@ func TestStop(t *testing.T) {
 	em.forwardedPorts.Store(9000, struct{}{})
 	em.forwardedPorts.Store(9001, struct{}{})
 
-	fakeForwarder := newTestForwarder(nil)
+	fakeForwarder := newTestForwarder()
 	fakeForwarder.forwardedResources = em.forwardedResources
 	em.EntryForwarder = fakeForwarder
 

pkg/skaffold/kubernetes/portforward/kubectl_forwarder.go

@@ -20,79 +20,99 @@ import (
 	"bytes"
 	"context"
 	"fmt"
-	"net"
+	"io"
+	"os/exec"
 	"strings"
 	"time"
 
-	"github.com/pkg/errors"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/color"
+
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/util"
+
 	"github.com/sirupsen/logrus"
-	"k8s.io/apimachinery/pkg/util/wait"
 
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubectl"
-	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/util"
 )
 
 type EntryForwarder interface {
-	Forward(parentCtx context.Context, pfe *portForwardEntry) error
+	Forward(parentCtx context.Context, pfe *portForwardEntry)
 	Terminate(p *portForwardEntry)
-	Monitor(*portForwardEntry, func())
 }
 
 type KubectlForwarder struct {
 	kubectl *kubectl.CLI
+	out     io.Writer
 }
 
-// Forward port-forwards a pod using kubectl port-forward
-// It returns an error only if the process fails or was terminated by a signal other than SIGTERM
-func (k *KubectlForwarder) Forward(parentCtx context.Context, pfe *portForwardEntry) error {
-	ctx, cancel := context.WithCancel(parentCtx)
-	// when retrying a portforwarding entry, it might already have a context running
-	if pfe.cancel != nil {
-		pfe.cancel()
-	}
-	pfe.cancel = cancel
-
-	cmd := k.kubectl.Command(ctx,
-		"port-forward",
-		"--pod-running-timeout", "5s",
-		fmt.Sprintf("%s/%s", pfe.resource.Type, pfe.resource.Name),
-		fmt.Sprintf("%d:%d", pfe.localPort, pfe.resource.Port),
-		"--namespace", pfe.resource.Namespace,
-	)
-	pfe.logBuffer = &bytes.Buffer{}
-	cmd.Stdout = pfe.logBuffer
-	cmd.Stderr = pfe.logBuffer
-
-	if err := cmd.Start(); err != nil {
-		if errors.Cause(err) == context.Canceled {
-			return nil
+// Forward port-forwards a pod using kubectl port-forward in the background
+// It kills the command on errors in the kubectl port-forward log
+// It restarts the command if it was not cancelled by skaffold
+// It retries in case the port is taken
+func (k *KubectlForwarder) Forward(parentCtx context.Context, pfe *portForwardEntry) {
+	go k.forward(parentCtx, pfe)
+}
+
+func (k *KubectlForwarder) forward(parentCtx context.Context, pfe *portForwardEntry) {
+	var notifiedUser bool
+	for {
+		if parentCtx.Err() == context.Canceled {
+			logrus.Debugf("port forwarding %v cancelled...", pfe)
+			return
+		}
+		if !util.IsPortFree(pfe.localPort) {
+			//assuming that Skaffold brokered ports don't overlap, this has to be an external process that started
+			//since the dev loop kicked off. We are notifying the user in the hope that they can fix it
+			color.Red.Fprintf(k.out, "failed to port forward %v, port %d is taken, retrying...\n", pfe, pfe.localPort)
+			notifiedUser = true
+			time.Sleep(5 * time.Second)
+			continue
 		}
-		return errors.Wrapf(err, "port forwarding %s/%s, port: %d to local port: %d, err: %s", pfe.resource.Type, pfe.resource.Name, pfe.resource.Port, pfe.localPort, pfe.logBuffer.String())
-	}
 
-	resultChan := make(chan error, 1)
-	go func() {
-		err := cmd.Wait()
-		if err != nil {
-			logrus.Debugf("port forwarding %v terminated: %s, output: %s", pfe, err, pfe.logBuffer.String())
-			resultChan <- err
+		if notifiedUser {
+			color.Green.Fprintf(k.out, "port forwarding %v recovered on port %d\n", pfe, pfe.localPort)
+			notifiedUser = false
 		}
-	}()
-
-	go func() {
-		err := wait.PollImmediate(200*time.Millisecond, 5*time.Second, func() (bool, error) {
-			// creating a listening port should not succeed
-			if ln, err := net.Listen("tcp", fmt.Sprintf("%s:%d", util.Loopback, pfe.localPort)); err == nil {
-				ln.Close()
-				return false, nil
+
+		ctx, cancel := context.WithCancel(parentCtx)
+		// when retrying a portforwarding entry, it might already have a context running
+		if pfe.cancel != nil {
+			pfe.cancel()
+		}
+		pfe.cancel = cancel
+		cmd := k.kubectl.Command(ctx,
+			"port-forward",
+			"--pod-running-timeout", "1s",
+			fmt.Sprintf("%s/%s", pfe.resource.Type, pfe.resource.Name),
+			fmt.Sprintf("%d:%d", pfe.localPort, pfe.resource.Port),
+			"--namespace", pfe.resource.Namespace,
+		)
+		buf := &bytes.Buffer{}
+		cmd.Stdout = buf
+		cmd.Stderr = buf
+
+		if err := cmd.Start(); err != nil {
+			if ctx.Err() == context.Canceled {
+				logrus.Debugf("couldn't start %v due to context cancellation", pfe)
+				return
 			}
-			return true, nil
-		})
-		resultChan <- err
-	}()
+			//retry on exit at Start()
+			logrus.Debugf("error starting port forwarding %v: %s, output: %s", pfe, err, buf.String())
+			time.Sleep(500 * time.Millisecond)
+			continue
+		}
 
-	err := <-resultChan
-	return err
+		//kill kubectl on port forwarding error logs
+		go k.monitorErrorLogs(ctx, buf, cmd, pfe)
+
+		if err := cmd.Wait(); err != nil {
+			if ctx.Err() == context.Canceled {
+				logrus.Debugf("terminated %v due to context cancellation", pfe)
+				return
+			}
+			logrus.Debugf("port forwarding %v got terminated: %s, output: %s", pfe, err, buf.String())
+			time.Sleep(500 * time.Millisecond)
+		}
+	}
 }
 
 // Terminate terminates an existing kubectl port-forward command using SIGTERM
@@ -107,18 +127,29 @@ func (*KubectlForwarder) Terminate(p *portForwardEntry) {
 // Monitor monitors the logs for a kubectl port forward command
 // If it sees an error, it calls back to the EntryManager to
 // retry the entire port forward operation.
-func (*KubectlForwarder) Monitor(p *portForwardEntry, retryFunc func()) {
+func (*KubectlForwarder) monitorErrorLogs(ctx context.Context, buf *bytes.Buffer, cmd *exec.Cmd, p *portForwardEntry) {
 	for {
-		time.Sleep(1 * time.Second)
-		s, _ := p.logBuffer.ReadString(byte('\n'))
-		if s != "" {
-			logrus.Tracef("[port-forward] %s", s)
-			if strings.Contains(s, "error forwarding port") || strings.Contains(s, "unable to forward") {
-				// kubectl is having an error. retry the command
-				logrus.Infof("retrying kubectl port-forward due to error: %s", s)
-				go retryFunc()
-				return
+		select {
+		case <-ctx.Done():
+			return
+		default:
+			time.Sleep(1 * time.Second)
+			s, _ := buf.ReadString(byte('\n'))
+			if s != "" {
+				logrus.Tracef("[port-forward] %s", s)
+
+				if strings.Contains(s, "error forwarding port") ||
+					strings.Contains(s, "unable to forward") ||
+					strings.Contains(s, "error upgrading connection") {
+					// kubectl is having an error. retry the command
+					logrus.Tracef("killing port forwarding %v", p)
+					if err := cmd.Process.Kill(); err != nil {
+						logrus.Tracef("failed to kill port forwarding %v, err: %s", p, err)
+					}
+					return
+				}
 			}
 		}
+
 	}
 }

pkg/skaffold/kubernetes/portforward/pod_forwarder.go

@@ -125,9 +125,8 @@ func (p *WatchingPodForwarder) portForwardPod(ctx context.Context, pod *v1.Pod)
 					p.Terminate(prevEntry)
 				}
 			}
-			if err := p.forwardPortForwardEntry(ctx, entry); err != nil {
-				return err
-			}
+			p.forwardPortForwardEntry(ctx, entry)
+
 		}
 	}
 	return nil