Ensure port forwards do not select terminating pods and are ready before use

changelog/v1.18.0-beta13/port-forward-safety.yaml

@@ -0,0 +1,11 @@
+changelog:
+  - type: NON_USER_FACING
+    issueLink: https://github.com/solo-io/gloo/issues/9353
+    resolvesIssue: false
+    description: >-
+      Add safety around port forward util to ensure the port forward can be dialed
+      before returning nil.
+  - type: DEPENDENCY_BUMP
+    dependencyOwner: solo-io
+    dependencyRepo: go-utils
+    dependencyTag: v0.25.2

go.mod

@@ -48,7 +48,7 @@ require (
 	github.com/saiskee/gettercheck v0.0.0-20210820204958-38443d06ebe0
 	github.com/sergi/go-diff v1.1.0
 	github.com/solo-io/go-list-licenses v0.1.4
-	github.com/solo-io/go-utils v0.24.8
+	github.com/solo-io/go-utils v0.25.2
 	github.com/solo-io/k8s-utils v0.7.2
 	github.com/solo-io/protoc-gen-ext v0.0.18
 	github.com/solo-io/protoc-gen-openapi v0.2.4

go.sum

@@ -2000,8 +2000,8 @@ github.com/solo-io/go-control-plane-fork-v2 v1.29.0-patch1/go.mod h1:ZBTaoJ23lqI
 github.com/solo-io/go-list-licenses v0.1.4 h1:u4xh1OUORT4iSWuAp3Q4NsfHcDaeUV8QRDH8ACQqbxw=
 github.com/solo-io/go-list-licenses v0.1.4/go.mod h1:x6LSp/NrYgVXwNum7ZOiaAYTpg6B3F6TrWYfcdHVroA=
 github.com/solo-io/go-utils v0.20.2/go.mod h1:6e8K1spnMWwlnJRSNp/J84GEyJbrcK4Gm7i+ehzCi8c=
-github.com/solo-io/go-utils v0.24.8 h1:gukFEvQ0SSRzIwysulI6w0c/dG08TCohO9QxwCqW6Lg=
-github.com/solo-io/go-utils v0.24.8/go.mod h1:bFFKO4Ih+sPViwNdVxv3z5dRrzMcJjNMHlx4zA8vxSg=
+github.com/solo-io/go-utils v0.25.2 h1:FteDaEeDr67BcWtcY/yBm/2pcM7ZaMzek4CUJKEddG4=
+github.com/solo-io/go-utils v0.25.2/go.mod h1:bFFKO4Ih+sPViwNdVxv3z5dRrzMcJjNMHlx4zA8vxSg=
 github.com/solo-io/k8s-utils v0.7.2 h1:pIRiTOpwymdCHUOSjzKDi/Ay16FNtF7JV7NIRlC2ZNQ=
 github.com/solo-io/k8s-utils v0.7.2/go.mod h1:RrT6PVTSD1X0vteKCQmGzoAAfjI1U5oV/wA+T3T+NoM=
 github.com/solo-io/protoc-gen-ext v0.0.18 h1:zSAL8NzWpJUGYoA5IyjHiKASNyHjR0uxBQ7eQS94i3A=

pkg/utils/kubeutils/kubectl/cli.go

@@ -193,6 +193,7 @@ func (c *Cli) StartPortForward(ctx context.Context, options ...portforward.Optio
 		portforward.WithKubeContext(c.kubeContext),
 	}, options...)
 
+	// We use the Cli PortForwarder because this util is explicitly designed to be a convenience wrapper around kubectl
 	portForwarder := portforward.NewCliPortForwarder(options...)
 	err := portForwarder.Start(
 		ctx,

pkg/utils/kubeutils/pods.go

@@ -45,13 +45,21 @@ func GetReadyPodsForDeployment(
 	kubeClient *kubernetes.Clientset,
 	deploy metav1.ObjectMeta,
 ) ([]string, error) {
-	// This predicate will return true if and only if the pod is ready
+	// This predicate will return true if and only if the pod is ready.
 	readyPodPredicate := func(pod corev1.Pod) bool {
+
+		// Make sure our pod has not been marked for deletion.
+		if pod.DeletionTimestamp != nil {
+			return false
+		}
+
+		// Make sure the condition named "Ready" has the status "True".
 		for _, condition := range pod.Status.Conditions {
 			if condition.Type == corev1.PodReady {
-				return true
+				return condition.Status == corev1.ConditionTrue
 			}
 		}
+
 		return false
 	}
 

pkg/utils/kubeutils/portforward/api_forwarder.go

@@ -55,8 +55,24 @@ type apiPortForwarder struct {
 }
 
 func (f *apiPortForwarder) Start(ctx context.Context, options ...retry.Option) error {
-	return retry.Do(func() error {
+	// First we attempt to start the port-forward following the retry options provided.
+	if err := retry.Do(func() error {
 		return f.startOnce(ctx)
+	}, options...); err != nil {
+		return err
+	}
+
+	// Before returning nil, we make sure the port-forward can be dialed on the
+	// port-forward address configured. This prevents us returning nil while the
+	// port-forward is still initializing.
+	return retry.Do(func() error {
+		conn, err := net.Dial("tcp4", f.Address())
+		if err != nil {
+			return err
+		}
+		conn.Close()
+		return nil
+
 	}, options...)
 }
 

pkg/utils/kubeutils/portforward/cli_portforwarder.go

@@ -6,8 +6,11 @@ import (
 	"net"
 	"os/exec"
 	"strconv"
+	"strings"
+	"time"
 
 	errors "github.com/rotisserie/eris"
+	"github.com/solo-io/go-utils/testutils"
 
 	"github.com/avast/retry-go/v4"
 )
@@ -38,8 +41,24 @@ type cliPortForwarder struct {
 }
 
 func (c *cliPortForwarder) Start(ctx context.Context, options ...retry.Option) error {
-	return retry.Do(func() error {
+	// First we attempt to start the port-forward following the retry options provided.
+	if err := retry.Do(func() error {
 		return c.startOnce(ctx)
+	}, options...); err != nil {
+		return err
+	}
+
+	// Before returning nil, we make sure the port-forward can be dialed on the
+	// port-forward address configured. This prevents us returning nil while the
+	// port-forward is still initializing.
+	return retry.Do(func() error {
+		conn, err := net.Dial("tcp4", c.Address())
+		if err != nil {
+			return err
+		}
+		conn.Close()
+		return nil
+
 	}, options...)
 }
 
@@ -55,6 +74,13 @@ func (c *cliPortForwarder) startOnce(ctx context.Context) error {
 
 	cmdCtx, cmdCancel := context.WithCancel(ctx)
 
+	if strings.Contains(c.properties.resourceType, "deploy") || strings.Contains(c.properties.resourceType, "pod") {
+		if err := testutils.WaitPodsRunning(ctx, time.Millisecond*100, c.properties.resourceNamespace, fmt.Sprintf("gloo=%s", c.properties.resourceName)); err != nil {
+			cmdCancel()
+			return err
+		}
+	}
+
 	c.cmd = exec.CommandContext(
 		cmdCtx,
 		"kubectl",

pkg/utils/kubeutils/portforward/portforwarder.go

@@ -12,7 +12,8 @@ import (
 // Implementations are NOT thread-safe, as the goroutine that Starts the PortForward
 // should also be the one that Closes it
 type PortForwarder interface {
-	// Start runs this PortForwarder.
+	// Start will attempt to open a port forward using the retry options provided. A nil return
+	// value indicates the port-forward succeeded and is safe to dial.
 	Start(ctx context.Context, options ...retry.Option) error
 
 	// Address returns the local forwarded address. Only valid while the apiPortForwarder is running.

projects/gloo/cli/pkg/cmd/gateway/dump.go

@@ -6,14 +6,13 @@ import (
 	"io"
 	"log"
 	"net/http"
-	"os"
-	"os/exec"
-	"strconv"
 	"strings"
 	"time"
 
+	"github.com/avast/retry-go/v4"
 	"github.com/solo-io/go-utils/cliutils"
 
+	"github.com/solo-io/gloo/pkg/utils/kubeutils/portforward"
 	"github.com/solo-io/gloo/projects/gloo/cli/pkg/cmd/options"
 	"github.com/solo-io/gloo/projects/gloo/pkg/defaults"
 	"github.com/solo-io/solo-kit/pkg/errors"
@@ -43,19 +42,23 @@ func GetEnvoyAdminData(ctx context.Context, proxyName, namespace, path string, t
 	if !strings.HasPrefix(path, "/") {
 		path = "/" + path
 	}
-	adminPort := strconv.Itoa(int(defaults.EnvoyAdminPort))
-	portFwd := exec.Command("kubectl", "port-forward", "-n", namespace,
-		"deployment/"+proxyName, adminPort)
-	portFwd.Stdout = os.Stderr
-	portFwd.Stderr = os.Stderr
-	if err := portFwd.Start(); err != nil {
+
+	adminPort := int(defaults.EnvoyAdminPort)
+
+	pf := portforward.NewPortForwarder(portforward.WithDeployment(proxyName, namespace), portforward.WithPorts(adminPort, adminPort))
+	err := pf.Start(ctx,
+		retry.LastErrorOnly(true),
+		retry.Delay(100*time.Millisecond),
+		retry.DelayType(retry.BackOffDelay),
+		retry.Attempts(5))
+	if err != nil {
 		return "", errors.Wrapf(err, "failed to start port-forward")
 	}
 	defer func() {
-		if portFwd.Process != nil {
-			portFwd.Process.Kill()
-		}
+		pf.Close()
+		pf.WaitForStop()
 	}()
+
 	result := make(chan string)
 	errs := make(chan error)
 	go func() {
@@ -65,7 +68,7 @@ func GetEnvoyAdminData(ctx context.Context, proxyName, namespace, path string, t
 				return
 			default:
 			}
-			res, err := http.Get("http://localhost:" + adminPort + path)
+			res, err := http.Get("http://" + pf.Address() + path)
 			if err != nil {
 				errs <- err
 				time.Sleep(time.Millisecond * 250)

test/gomega/assertions/stats.go

@@ -77,18 +77,6 @@ func EventuallyWithOffsetStatisticsMatchAssertions(offset int, statsPortFwd Stat
 		portForwarder.WaitForStop()
 	}()
 
-	statsRequest, err := http.NewRequest(http.MethodGet, fmt.Sprintf("http://localhost:%d/", statsPortFwd.LocalPort), nil)
-	ExpectWithOffset(offset+1, err).NotTo(HaveOccurred())
-	EventuallyWithOffset(offset+1, func(g Gomega) {
-		resp, err := http.DefaultClient.Do(statsRequest)
-		g.Expect(err).NotTo(HaveOccurred())
-		defer resp.Body.Close()
-		g.Expect(resp).To(matchers.HaveHttpResponse(&matchers.HttpResponse{
-			StatusCode: http.StatusOK,
-			Body:       Not(BeEmpty()),
-		}))
-	}).Should(Succeed())
-
 	for _, assertion := range assertions {
 		assertion.WithOffset(offset + 1).ShouldNot(HaveOccurred())
 	}

test/kube2e/gloo/utils_test.go

@@ -0,0 +1,54 @@
+package gloo_test
+
+import (
+	"context"
+	"net"
+	"time"
+
+	"github.com/avast/retry-go/v4"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+
+	"github.com/solo-io/gloo/pkg/utils/kubeutils/portforward"
+)
+
+var _ = Describe("Utils", func() {
+
+	Context("portforward", func() {
+		var (
+			pf                  portforward.PortForwarder
+			ctx                 context.Context
+			cancel              context.CancelFunc
+			defaultRetryOptions = []retry.Option{
+				retry.LastErrorOnly(true),
+				retry.Delay(100 * time.Millisecond),
+				retry.DelayType(retry.BackOffDelay),
+				retry.Attempts(5),
+			}
+		)
+
+		AfterEach(func() {
+			cancel()
+			pf.Close()
+			pf.WaitForStop()
+		})
+
+		BeforeEach(func() {
+			ctx, cancel = context.WithCancel(context.Background())
+			// We are using kube-dns because we know we have a kubectl and a kind cluster on our CI
+			// env. If that ever changes we need to re-assess this test strategy.
+			// We are using port 30053 because 53 is privileged.
+			pf = portforward.NewPortForwarder(portforward.WithService("kube-dns", "kube-system"), portforward.WithPorts(30053, 53))
+		})
+
+		It("Creates a usable portforward with Start", func() {
+			err := pf.Start(ctx, defaultRetryOptions...)
+			Expect(err).ToNot(HaveOccurred())
+			_, err = net.Dial("tcp4", pf.Address())
+			Expect(err).ToNot(HaveOccurred())
+
+		})
+
+	})
+})

test/kubernetes/e2e/features/deployer/suite.go

@@ -15,7 +15,6 @@ import (
 	"github.com/solo-io/gloo/pkg/utils/kubeutils"
 	"github.com/solo-io/gloo/projects/gloo/pkg/syncer/setup"
 	"github.com/solo-io/gloo/test/kubernetes/e2e"
-	"github.com/solo-io/gloo/test/kubernetes/testutils/runtime"
 )
 
 var _ e2e.NewSuiteFunc = NewTestingSuite
@@ -76,16 +75,12 @@ func (s *testingSuite) TestConfigureProxiesFromGatewayParameters() {
 	s.testInstallation.Assertions.EventuallyRunningReplicas(s.ctx, proxyDeployment.ObjectMeta, gomega.Equal(1))
 
 	// We assert that we can port-forward requests to the proxy deployment, and then execute requests against the server
-	if s.testInstallation.RuntimeContext.RunSource == runtime.LocalDevelopment {
-		// There are failures when opening port-forwards to the Envoy Admin API in CI
-		// Those are currently being investigated
-		s.testInstallation.Assertions.AssertEnvoyAdminApi(
-			s.ctx,
-			proxyDeployment.ObjectMeta,
-			serverInfoLogLevelAssertion(s.testInstallation, "debug", "connection:trace,upstream:debug"),
-			xdsClusterAssertion(s.testInstallation),
-		)
-	}
+	s.testInstallation.Assertions.AssertEnvoyAdminApi(
+		s.ctx,
+		proxyDeployment.ObjectMeta,
+		serverInfoLogLevelAssertion(s.testInstallation, "debug", "connection:trace,upstream:debug"),
+		xdsClusterAssertion(s.testInstallation),
+	)
 }
 
 func (s *testingSuite) TestSelfManagedGateway() {