Fix port forwarding on Windows (#4373)

Wrap kubectl portforward call in job object for Windows

pkg/skaffold/kubectl/cli.go

@@ -80,6 +80,12 @@ func (c *CLI) RunOut(ctx context.Context, command string, arg ...string) ([]byte
 	return util.RunCmdOut(cmd)
 }
 
+// CommandWithStrictCancellation ensures for windows OS that all child process get terminated on cancellation
+func (c *CLI) CommandWithStrictCancellation(ctx context.Context, command string, arg ...string) *Cmd {
+	args := c.args(command, "", arg...)
+	return CommandContext(ctx, "kubectl", args...)
+}
+
 // args builds an argument list for calling kubectl and consistently
 // adds the `--context` and `--namespace` flags.
 func (c *CLI) args(command string, namespace string, arg ...string) []string {

pkg/skaffold/kubectl/cli_test.go

@@ -101,4 +101,26 @@ func TestCLI(t *testing.T) {
 			t.CheckDeepEqual(string(out), output)
 		})
 	}
+
+	// test cli.CommandWithStrictCancellation()
+	for _, test := range tests {
+		testutil.Run(t, test.name, func(t *testutil.T) {
+			t.Override(&util.DefaultExecCommand, testutil.CmdRunOut(
+				test.expectedCommand,
+				output,
+			))
+
+			cli := NewFromRunContext(&runcontext.RunContext{
+				Opts: config.SkaffoldOptions{
+					Namespace:  test.namespace,
+					KubeConfig: test.kubeconfig,
+				},
+				KubeContext: kubeContext,
+			})
+			cmd := cli.CommandWithStrictCancellation(context.Background(), "exec", "arg1", "arg2")
+			out, err := util.RunCmdOut(cmd.Cmd)
+			t.CheckNoError(err)
+			t.CheckDeepEqual(string(out), output)
+		})
+	}
 }

pkg/skaffold/kubectl/exec.go

@@ -0,0 +1,39 @@
+// +build !windows
+
+/*
+Copyright 2020 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 kubectl
+
+import (
+	"context"
+	"os/exec"
+)
+
+// Cmd is a wrapper on exec.Cmd
+type Cmd struct {
+	*exec.Cmd
+}
+
+// CommandContext creates a new Cmd
+func CommandContext(ctx context.Context, name string, arg ...string) *Cmd {
+	return &Cmd{Cmd: exec.CommandContext(ctx, name, arg...)}
+}
+
+// Terminate kills the underlying process
+func (c *Cmd) Terminate() error {
+	return c.Process.Kill()
+}

pkg/skaffold/kubectl/exec_windows.go

@@ -0,0 +1,95 @@
+/*
+Copyright 2020 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 kubectl
+
+import (
+	"context"
+	"os/exec"
+	"unsafe"
+
+	"golang.org/x/sys/windows"
+)
+
+// Cmd represents an external command being prepared to run within a job object
+type Cmd struct {
+	*exec.Cmd
+	handle windows.Handle
+	ctx    context.Context
+}
+
+type process struct {
+	Pid    int
+	Handle uintptr
+}
+
+// CommandContext creates a new Cmd
+func CommandContext(ctx context.Context, name string, arg ...string) *Cmd {
+	return &Cmd{Cmd: exec.CommandContext(ctx, name, arg...), ctx: ctx}
+}
+
+// Start starts the specified command in a job object but does not wait for it to complete
+func (c *Cmd) Start() (err error) {
+	var handle windows.Handle
+	handle, err = windows.CreateJobObject(nil, nil)
+	if err != nil {
+		return
+	}
+
+	go func() {
+		<-c.ctx.Done()
+		c.Terminate()
+	}()
+
+	// https://gist.github.com/hallazzang/76f3970bfc949831808bbebc8ca15209
+	info := windows.JOBOBJECT_EXTENDED_LIMIT_INFORMATION{
+		BasicLimitInformation: windows.JOBOBJECT_BASIC_LIMIT_INFORMATION{
+			LimitFlags: windows.JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE,
+		},
+	}
+	if _, err = windows.SetInformationJobObject(
+		handle,
+		windows.JobObjectExtendedLimitInformation,
+		uintptr(unsafe.Pointer(&info)),
+		uint32(unsafe.Sizeof(info))); err != nil {
+		return
+	}
+
+	if err = c.Cmd.Start(); err != nil {
+		return
+	}
+
+	if err = windows.AssignProcessToJobObject(
+		handle,
+		windows.Handle((*process)(unsafe.Pointer(c.Process)).Handle)); err != nil {
+		return
+	}
+	c.handle = handle
+	return
+}
+
+// Run starts the specified command in a job object and waits for it to complete
+func (c *Cmd) Run() error {
+	if err := c.Start(); err != nil {
+		return err
+	}
+	return c.Wait()
+}
+
+// Terminate closes the job object handle which kills all connected processes
+func (c *Cmd) Terminate() error {
+	return windows.CloseHandle(c.handle)
+}

pkg/skaffold/kubernetes/portforward/kubectl_forwarder.go

@@ -21,7 +21,6 @@ import (
 	"context"
 	"fmt"
 	"io"
-	"os/exec"
 	"strings"
 	"time"
 
@@ -127,7 +126,7 @@ func (k *KubectlForwarder) forward(parentCtx context.Context, pfe *portForwardEn
 	}
 }
 
-func portForwardCommand(ctx context.Context, k *kubectl.CLI, pfe *portForwardEntry, buf io.Writer) *exec.Cmd {
+func portForwardCommand(ctx context.Context, k *kubectl.CLI, pfe *portForwardEntry, buf io.Writer) *kubectl.Cmd {
 	args := []string{
 		"--pod-running-timeout", "1s",
 		fmt.Sprintf("%s/%s", pfe.resource.Type, pfe.resource.Name),
@@ -137,7 +136,7 @@ func portForwardCommand(ctx context.Context, k *kubectl.CLI, pfe *portForwardEnt
 	if pfe.resource.Address != "" && pfe.resource.Address != util.Loopback {
 		args = append(args, []string{"--address", pfe.resource.Address}...)
 	}
-	cmd := k.Command(ctx, "port-forward", args...)
+	cmd := k.CommandWithStrictCancellation(ctx, "port-forward", args...)
 	cmd.Stdout = buf
 	cmd.Stderr = buf
 	return cmd
@@ -159,7 +158,7 @@ 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) monitorErrorLogs(ctx context.Context, buf *bytes.Buffer, cmd *exec.Cmd, p *portForwardEntry) {
+func (*KubectlForwarder) monitorErrorLogs(ctx context.Context, buf *bytes.Buffer, cmd *kubectl.Cmd, p *portForwardEntry) {
 	for {
 		select {
 		case <-ctx.Done():
@@ -179,7 +178,7 @@ func (*KubectlForwarder) monitorErrorLogs(ctx context.Context, buf *bytes.Buffer
 				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 {
+				if err := cmd.Terminate(); err != nil {
 					logrus.Tracef("failed to kill port forwarding %v, err: %s", p, err)
 				}
 				return

pkg/skaffold/kubernetes/portforward/kubectl_forwarder_test.go

@@ -19,7 +19,7 @@ package portforward
 import (
 	"bytes"
 	"context"
-	"os/exec"
+	"runtime"
 	"sync"
 	"testing"
 	"time"
@@ -117,8 +117,11 @@ func TestMonitorErrorLogs(t *testing.T) {
 			t.Override(&waitErrorLogs, 50*time.Millisecond)
 
 			ctx, cancel := context.WithCancel(context.Background())
-
-			cmd := exec.Command("sleep", "5")
+			cmdStr := "sleep"
+			if runtime.GOOS == "windows" {
+				cmdStr = "timeout"
+			}
+			cmd := kubectl.CommandContext(ctx, cmdStr, "5")
 			if err := cmd.Start(); err != nil {
 				t.Fatal("error starting command")
 			}
@@ -145,21 +148,21 @@ func TestMonitorErrorLogs(t *testing.T) {
 			// make sure the command is running or killed based on what's expected
 			if test.cmdRunning {
 				assertCmdIsRunning(t, cmd)
-				cmd.Process.Kill()
+				cmd.Terminate()
 			} else {
 				assertCmdWasKilled(t, cmd)
 			}
 		})
 	}
 }
 
-func assertCmdIsRunning(t *testutil.T, cmd *exec.Cmd) {
+func assertCmdIsRunning(t *testutil.T, cmd *kubectl.Cmd) {
 	if cmd.ProcessState != nil {
 		t.Fatal("cmd was killed but expected to continue running")
 	}
 }
 
-func assertCmdWasKilled(t *testutil.T, cmd *exec.Cmd) {
+func assertCmdWasKilled(t *testutil.T, cmd *kubectl.Cmd) {
 	if err := cmd.Wait(); err == nil {
 		t.Fatal("cmd was not killed but expected to be killed")
 	}
@@ -189,7 +192,7 @@ func TestDefaultAddressArg(t *testing.T) {
 	assertCmdContainsArgs(t, cmd, false, "--address")
 }
 
-func assertCmdContainsArgs(t *testing.T, cmd *exec.Cmd, expected bool, args ...string) {
+func assertCmdContainsArgs(t *testing.T, cmd *kubectl.Cmd, expected bool, args ...string) {
 	if len(args) == 0 {
 		return
 	}