Ensure output stream is set before port forwarding (#6061)

* Ensure output stream is set before port forwarding

* fix test

Author: nkubala

integration/dev_test.go

@@ -306,7 +306,7 @@ func getLocalPortFromPortForwardEvent(t *testing.T, entries chan *proto.LogEntry
 		case e := <-entries:
 			switch e.Event.GetEventType().(type) {
 			case *proto.Event_PortEvent:
-				t.Logf("event received %v", e)
+				t.Logf("port event received: %v", e)
 				if e.Event.GetPortEvent().ResourceName == resourceName &&
 					e.Event.GetPortEvent().ResourceType == resourceType &&
 					e.Event.GetPortEvent().Namespace == namespace {
@@ -316,7 +316,7 @@ func getLocalPortFromPortForwardEvent(t *testing.T, entries chan *proto.LogEntry
 					return address, int(port)
 				}
 			default:
-				t.Logf("event received %v", e)
+				t.Logf("event received: %v", e)
 			}
 		}
 	}

pkg/skaffold/kubernetes/portforward/entry_manager.go

@@ -137,6 +137,11 @@ func (b *EntryManager) forwardPortForwardEntry(ctx context.Context, out io.Write
 	portForwardEventV2(entry)
 }
 
+// Start ensures the underlying entryForwarder is ready to forward.
+func (b *EntryManager) Start(out io.Writer) {
+	b.entryForwarder.Start(out)
+}
+
 // Stop terminates all kubectl port-forward commands.
 func (b *EntryManager) Stop() {
 	for _, pfe := range b.forwardedResources.resources {

pkg/skaffold/kubernetes/portforward/forwarder_manager.go

@@ -48,7 +48,8 @@ type Forwarder interface {
 
 // ForwarderManager manages all forwarders
 type ForwarderManager struct {
-	forwarders []Forwarder
+	forwarders   []Forwarder
+	entryManager *EntryManager
 }
 
 // NewForwarderManager returns a new port manager which handles starting and stopping port forwarding
@@ -73,7 +74,8 @@ func NewForwarderManager(cli *kubectl.CLI, podSelector kubernetes.PodSelector, l
 	}
 
 	return &ForwarderManager{
-		forwarders: forwarders,
+		forwarders:   forwarders,
+		entryManager: entryManager,
 	}
 }
 
@@ -120,6 +122,7 @@ func (p *ForwarderManager) Start(ctx context.Context, out io.Writer, namespaces
 	ctx, endTrace := instrumentation.StartTrace(ctx, "Start")
 	defer endTrace()
 
+	p.entryManager.Start(out)
 	for _, f := range p.forwarders {
 		if err := f.Start(ctx, out, namespaces); err != nil {
 			eventV2.TaskFailed(constants.PortForward, err)

pkg/skaffold/kubernetes/portforward/kubectl_forwarder.go

@@ -25,6 +25,7 @@ import (
 	"os"
 	"sort"
 	"strings"
+	"sync/atomic"
 	"time"
 
 	"github.com/sirupsen/logrus"
@@ -41,11 +42,13 @@ import (
 )
 
 type EntryForwarder interface {
+	Start(io.Writer)
 	Forward(parentCtx context.Context, pfe *portForwardEntry) error
 	Terminate(p *portForwardEntry)
 }
 
 type KubectlForwarder struct {
+	started int32
 	out     io.Writer
 	kubectl *kubectl.CLI
 }
@@ -66,6 +69,11 @@ var (
 	waitErrorLogs       = 1 * time.Second
 )
 
+func (k *KubectlForwarder) Start(out io.Writer) {
+	atomic.StoreInt32(&k.started, 1)
+	k.out = out
+}
+
 // 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
@@ -77,6 +85,10 @@ func (k *KubectlForwarder) Forward(parentCtx context.Context, pfe *portForwardEn
 }
 
 func (k *KubectlForwarder) forward(parentCtx context.Context, pfe *portForwardEntry, errChan chan error) {
+	if atomic.LoadInt32(&k.started) == 0 {
+		errChan <- fmt.Errorf("Forward() called before kubectl forwarder was started")
+		return
+	}
 	var notifiedUser bool
 	defer deferFunc()
 

pkg/skaffold/kubernetes/portforward/kubectl_forwarder_test.go

@@ -20,6 +20,7 @@ import (
 	"bytes"
 	"context"
 	"errors"
+	"io/ioutil"
 	"runtime"
 	"sort"
 	"strings"
@@ -69,6 +70,7 @@ func TestUnavailablePort(t *testing.T) {
 		}
 		pfe := newPortForwardEntry(0, latestV1.PortForwardResource{}, "", "", "", "", 8080, false)
 
+		k.Start(&buf)
 		go k.Forward(context.Background(), pfe)
 
 		// wait for isPortFree to be called
@@ -451,3 +453,31 @@ func mockPod(name string, ports []corev1.ContainerPort, creationTime time.Time)
 		},
 	}
 }
+
+func TestStartAndForward(t *testing.T) {
+	tests := []struct {
+		description string
+		startFirst  bool
+	}{
+		{
+			description: "Forward() before Start() errors",
+			startFirst:  false,
+		}, {
+			description: "Start() before Forward()",
+			startFirst:  true,
+		},
+	}
+
+	for _, test := range tests {
+		testutil.Run(t, test.description, func(_ *testutil.T) {
+			k := &KubectlForwarder{}
+			if test.startFirst {
+				k.Start(ioutil.Discard)
+				testutil.CheckDeepEqual(t, k.started, int32(1))
+			} else {
+				err := k.Forward(context.Background(), nil)
+				testutil.CheckError(t, true, err)
+			}
+		})
+	}
+}

pkg/skaffold/kubernetes/portforward/resource_forwarder_test.go

@@ -19,6 +19,7 @@ package portforward
 import (
 	"context"
 	"fmt"
+	"io"
 	"io/ioutil"
 	"sync"
 	"testing"
@@ -60,6 +61,8 @@ func (f *testForwarder) Terminate(pfe *portForwardEntry) {
 	f.forwardedPorts.Delete(pfe.localPort)
 }
 
+func (f *testForwarder) Start(io.Writer) {}
+
 func newTestForwarder() *testForwarder {
 	return &testForwarder{}
 }

pkg/skaffold/runner/v1/dev.go

@@ -336,11 +336,8 @@ func (r *SkaffoldRunner) Dev(ctx context.Context, out io.Writer, artifacts []*la
 
 	defer r.deployer.GetAccessor().Stop()
 
-	// forwarderManager := r.createForwarder(out)
-	// defer forwarderManager.Stop()
-
 	if err := r.deployer.GetAccessor().Start(ctx, out, r.runCtx.GetNamespaces()); err != nil {
-		logrus.Warnln("Error starting port forwarding:", err)
+		logrus.Warnln("Error starting resource accessor:", err)
 	}
 	if err := r.deployer.GetDebugger().Start(ctx, r.runCtx.GetNamespaces()); err != nil {
 		logrus.Warnln("Error starting debug container notification:", err)