add generic exec in ephemeral tool

go.mod

@@ -16,7 +16,7 @@ require (
 	github.com/onsi/gomega v1.30.0
 	github.com/pkg/errors v0.9.1
 	github.com/rotisserie/eris v0.1.1
-	github.com/solo-io/go-utils v0.24.8
+	github.com/solo-io/go-utils v0.25.2-0.20240614172406-03937cdb7953
 	github.com/spf13/afero v1.6.0
 	go.uber.org/zap v1.26.0
 	golang.org/x/sync v0.5.0

go.sum

@@ -460,8 +460,10 @@ github.com/shopspring/decimal v1.3.1/go.mod h1:DKyhrW/HYNuLGql+MJL6WCR6knT2jwCFR
 github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
 github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ=
 github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
-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-0.20240614141144-3122157d8c19 h1:a2chiSsCe6w0rXITfhr/WTNYWU5ljYJTuQc1hJz/7dg=
+github.com/solo-io/go-utils v0.25.2-0.20240614141144-3122157d8c19/go.mod h1:bFFKO4Ih+sPViwNdVxv3z5dRrzMcJjNMHlx4zA8vxSg=
+github.com/solo-io/go-utils v0.25.2-0.20240614172406-03937cdb7953 h1:NoZjLbsrYGyt9G2/eszWtTFFNCOVWIRp/ABQB2ENJCw=
+github.com/solo-io/go-utils v0.25.2-0.20240614172406-03937cdb7953/go.mod h1:bFFKO4Ih+sPViwNdVxv3z5dRrzMcJjNMHlx4zA8vxSg=
 github.com/spf13/afero v1.6.0 h1:xoax2sJ2DT8S8xA2paPFjDCScCNeWsg75VG0DLRreiY=
 github.com/spf13/afero v1.6.0/go.mod h1:Ai8FlHk4v/PARR026UzYexafAt9roJ7LcLMAmO6Z93I=
 github.com/spf13/cast v1.3.1/go.mod h1:Qx5cxh0v+4UWYiBimWS+eyWzqEqokIECu5etghLkUJE=

testutils/kube/curl.go

@@ -1,106 +1,103 @@
 package kube
 
 import (
-	"bytes"
 	"context"
 	"fmt"
 	"io"
-	"io/ioutil"
 
-	"github.com/solo-io/go-utils/testutils"
-	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 
 	. "github.com/onsi/gomega"
+	"github.com/solo-io/go-utils/testutils/kubectl"
 )
 
-func CurlWithEphemeralPod(ctx context.Context, logger io.Writer, kubecontext, fromns, frompod string, args ...string) string {
-	createargs := []string{"alpha", "debug", "--quiet",
-		"--image=curlimages/curl@sha256:aa45e9d93122a3cfdf8d7de272e2798ea63733eeee6d06bd2ee4f2f8c4027d7c",
-		"--container=curl", frompod, "-n", fromns, "--", "sleep", "10h"}
-	// Execute curl commands from the same pod each time to avoid creating a burdensome number of ephemeral pods.
-	// create the curl pod; we do this every time and it will only work the first time, so ignore failures
-	executeNoFail(ctx, logger, kubecontext, createargs...)
-	args = append([]string{"exec",
-		"--container=curl", frompod, "-n", fromns, "--", "curl", "--connect-timeout", "1", "--max-time", "5"}, args...)
-	return execute(ctx, logger, kubecontext, args...)
-}
-
-func CurlWithEphemeralPodStable(ctx context.Context, logger io.Writer, kubeContext, fromNs, fromPod string, args ...string) string {
-	createArgs := []string{
-		"debug",
-		"--quiet",
-		"--image=curlimages/curl@sha256:aa45e9d93122a3cfdf8d7de272e2798ea63733eeee6d06bd2ee4f2f8c4027d7c",
-		"--container=curl",
-		"--image-pull-policy=IfNotPresent",
-		fromPod,
-		"-n",
-		fromNs,
-		"--",
-		"sleep",
-		"10h",
+var (
+	DefaultCurlImage = Image{
+		Registry:   "curlimages",
+		Repository: "curl",
+		Tag:        "",
+		Digest:     "aa45e9d93122a3cfdf8d7de272e2798ea63733eeee6d06bd2ee4f2f8c4027d7c",
+		PullPolicy: "IfNotPresent",
 	}
-	// Execute curl commands from the same pod each time to avoid creating a burdensome number of ephemeral pods.
-	// create the curl pod; we do this every time and it will only work the first time, so ignore failures
-	_, _ = executeNoFail(ctx, logger, kubeContext, createArgs...)
+)
 
-	args = append([]string{
-		"exec",
-		"--container=curl",
-		fromPod,
-		"-n",
-		fromNs,
-		"--",
-		"curl",
-		"--connect-timeout",
-		"1",
-		"--max-time",
-		"5",
-	}, args...)
-	return execute(ctx, logger, kubeContext, args...)
+func CurlWithEphemeralPod(ctx context.Context, logger io.Writer, kubeContext, fromns, frompod string, args ...string) string {
+	execParams := EphemeralPodParams{
+		Logger:        logger,
+		KubeContext:   kubeContext,
+		Image:         DefaultCurlImage,
+		FromContainer: "curl",
+		FromNamespace: fromns,
+		FromPod:       frompod,
+		ExecCmdPath:   "curl",
+		Args:          args,
+	}
+	out, _ := ExecFromEphemeralPod(ctx, execParams)
+	return out
 }
 
 // labelSelector is a string map e.g. gloo=gateway-proxy
 func FindPodNameByLabel(cfg *rest.Config, ctx context.Context, ns, labelSelector string) string {
 	clientset, err := kubernetes.NewForConfig(cfg)
 	Expect(err).NotTo(HaveOccurred())
-	pl, err := clientset.CoreV1().Pods(ns).List(ctx, v1.ListOptions{LabelSelector: labelSelector})
+	pl, err := clientset.CoreV1().Pods(ns).List(ctx, metav1.ListOptions{LabelSelector: labelSelector})
 	Expect(err).NotTo(HaveOccurred())
 	Expect(pl.Items).NotTo(BeEmpty())
 	return pl.Items[0].GetName()
 }
 
-func WaitForRollout(ctx context.Context, logger io.Writer, kubecontext, ns, deployment string) {
+func WaitForRollout(ctx context.Context, logger io.Writer, kubeContext, ns, deployment string) {
 	args := []string{"-n", ns, "rollout", "status", "deployment", deployment}
-	execute(ctx, logger, kubecontext, args...)
+	mustExecute(ctx, KubectlParams{
+		KubectlCmdParams: kubectl.NewParams(args...),
+		KubeContext:      kubeContext,
+		Logger:           logger,
+	})
 }
 
-func Curl(ctx context.Context, logger io.Writer, kubecontext, ns, fromDeployment, fromContainer, url string) string {
+func Curl(ctx context.Context, logger io.Writer, kubeContext, ns, fromDeployment, fromContainer, url string) string {
 	args := []string{
 		"-n", ns,
 		"exec", fmt.Sprintf("deployment/%s", fromDeployment),
 		"-c", fromContainer,
 		"--", "curl", url,
 	}
-	return execute(ctx, logger, kubecontext, args...)
+	return mustExecute(ctx, KubectlParams{
+		KubectlCmdParams: kubectl.NewParams(args...),
+		KubeContext:      kubeContext,
+		Logger:           logger,
+	})
 }
 
 func CreateNamespace(ctx context.Context, logger io.Writer, kubeContext, ns string) {
 	args := []string{"create", "namespace", ns}
-	out := execute(ctx, logger, kubeContext, args...)
+	out := mustExecute(ctx, KubectlParams{
+		KubectlCmdParams: kubectl.NewParams(args...),
+		KubeContext:      kubeContext,
+		Logger:           logger,
+	})
 	fmt.Fprintln(logger, out)
 }
 
 func DeleteNamespace(ctx context.Context, logger io.Writer, kubeContext, ns string) {
 	args := []string{"delete", "namespace", ns}
-	out := execute(ctx, logger, kubeContext, args...)
+	out := mustExecute(ctx, KubectlParams{
+		KubectlCmdParams: kubectl.NewParams(args...),
+		KubeContext:      kubeContext,
+		Logger:           logger,
+	})
 	fmt.Fprintln(logger, out)
 }
 
 func LabelNamespace(ctx context.Context, logger io.Writer, kubeContext, ns, label string) {
 	args := []string{"label", "namespace", ns, label}
-	out := execute(ctx, logger, kubeContext, args...)
+	out := mustExecute(ctx, KubectlParams{
+		KubectlCmdParams: kubectl.NewParams(args...),
+		KubeContext:      kubeContext,
+		Logger:           logger,
+	})
 	fmt.Fprintln(logger, out)
 }
 
@@ -118,7 +115,11 @@ func SetDeploymentEnvVars(
 		envVarStrings = append(envVarStrings, fmt.Sprintf("%s=%s", name, value))
 	}
 	args := append([]string{"set", "env", "-n", ns, fmt.Sprintf("deployment/%s", deploymentName), "-c", containerName}, envVarStrings...)
-	out := execute(ctx, logger, kubeContext, args...)
+	out := mustExecute(ctx, KubectlParams{
+		KubectlCmdParams: kubectl.NewParams(args...),
+		KubeContext:      kubeContext,
+		Logger:           logger,
+	})
 	fmt.Fprintln(logger, out)
 }
 
@@ -130,14 +131,18 @@ func DisableContainer(
 	deploymentName string,
 	containerName string,
 ) {
-	args := append([]string{
+	args := []string{
 		"-n", ns,
 		"patch", "deployment", deploymentName,
 		"--patch",
-		fmt.Sprintf("{\"spec\": {\"template\": {\"spec\": {\"containers\": [{\"name\": \"%s\",\"command\": [\"sleep\", \"20h\"]}]}}}}",
+		fmt.Sprintf(`{"spec": {"template": {"spec": {"containers": [{"name": "%s","command": ["sleep", "20h"]}]}}}}`,
 			containerName),
+	}
+	out := mustExecute(ctx, KubectlParams{
+		KubectlCmdParams: kubectl.NewParams(args...),
+		KubeContext:      kubeContext,
+		Logger:           logger,
 	})
-	out := execute(ctx, logger, kubeContext, args...)
 	fmt.Fprintln(logger, out)
 }
 
@@ -148,39 +153,16 @@ func EnableContainer(
 	ns string,
 	deploymentName string,
 ) {
-	args := append([]string{
+	args := []string{
 		"-n", ns,
 		"patch", "deployment", deploymentName,
 		"--type", "json",
-		"-p", "[{\"op\": \"remove\", \"path\": \"/spec/template/spec/containers/0/command\"}]",
+		"-p", `[{"op": "remove", "path": "/spec/template/spec/containers/0/command"}]`,
+	}
+	out := mustExecute(ctx, KubectlParams{
+		KubectlCmdParams: kubectl.NewParams(args...),
+		KubeContext:      kubeContext,
+		Logger:           logger,
 	})
-	out := execute(ctx, logger, kubeContext, args...)
 	fmt.Fprintln(logger, out)
 }
-
-func execute(ctx context.Context, logger io.Writer, kubeContext string, args ...string) string {
-	data, err := executeNoFail(ctx, logger, kubeContext, args...)
-	Expect(err).NotTo(HaveOccurred())
-	return data
-}
-
-func executeNoFail(ctx context.Context, logger io.Writer, kubeContext string, args ...string) (string, error) {
-	args = append([]string{"--context", kubeContext}, args...)
-	fmt.Fprintf(logger, "Executing: kubectl %v \n", args)
-	readerChan, done, err := testutils.KubectlOutChan(&bytes.Buffer{}, args...)
-	if err != nil {
-		return "", err
-	}
-	defer close(done)
-	select {
-	case <-ctx.Done():
-		return "", nil
-	case reader := <-readerChan:
-		data, err := ioutil.ReadAll(reader)
-		if err != nil {
-			return "", err
-		}
-		fmt.Fprintf(logger, "<kubectl %v> output: %v\n", args, string(data))
-		return string(data), nil
-	}
-}

testutils/kube/exec.go

@@ -0,0 +1,110 @@
+package kube
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"strings"
+
+	"github.com/onsi/gomega"
+	"github.com/solo-io/go-utils/testutils/kubectl"
+)
+
+type Image struct {
+	Registry, Repository, Tag, Digest string
+	PullPolicy                        string
+}
+
+func (i Image) String() string {
+	b := strings.Builder{}
+	b.WriteString(i.Registry)
+	b.WriteRune('/')
+	b.WriteString(i.Repository)
+	if i.Tag != "" {
+		b.WriteRune(':')
+		b.WriteString(i.Tag)
+	}
+	if i.Digest != "" {
+		b.WriteString("@sha256:")
+		b.WriteString(i.Digest)
+	}
+
+	return b.String()
+}
+
+type EphemeralPodParams struct {
+	Logger        io.Writer
+	KubeContext   string
+	Image         Image
+	FromContainer string
+	FromNamespace string
+	FromPod       string
+	ExecCmdPath   string
+	Args          []string
+	Env           []string
+	Stdin         io.Reader
+	Stdout        io.Writer
+	Stderr        io.Writer
+}
+
+func ExecFromEphemeralPod(ctx context.Context, params EphemeralPodParams) (string, error) {
+	createargs := []string{
+		"debug",
+		"--quiet",
+		fmt.Sprintf("--image=%s", params.Image),
+		fmt.Sprintf("--container=%s", params.FromContainer),
+		fmt.Sprintf("--image-pull-policy=%s", params.Image.PullPolicy),
+		params.FromPod,
+		"-n", params.FromNamespace,
+		"--", "sleep", "10h",
+	}
+
+	// create the params to the kubectl commands we will invoke.
+	// we will use the same params but just switch out the args for the
+	// different commands we execute.
+	kParams := kubectl.NewParams(createargs...)
+	if params.Stdin != nil {
+		kParams.Stdin = params.Stdin
+	}
+	if params.Stdout != nil {
+		kParams.Stdout = params.Stdout
+	}
+	if params.Stderr != nil {
+		kParams.Stderr = params.Stderr
+	}
+	if params.Env != nil {
+		kParams.Env = params.Env
+	}
+
+	// Execute curl commands from the same pod each time to avoid creating a burdensome number of ephemeral pods.
+	// create the curl pod; we do this every time and it will only work the first time, so ignore failures
+	_, _ = execute(ctx, KubectlParams{
+		KubectlCmdParams: kParams,
+		KubeContext:      params.KubeContext,
+		Logger:           params.Logger,
+	})
+
+	// Assert that eventually the ephemeral container is created before attempting to exec against it
+	gomega.Eventually(func(g gomega.Gomega) {
+		out, err := kubectl.KubectlOut(ctx, kubectl.Params{
+			Args: []string{"get", "pod", "-n", params.FromNamespace, params.FromPod, "-o=jsonpath='{.status.ephemeralContainerStatuses[*].name}'"},
+		})
+		g.Expect(err).ToNot(gomega.HaveOccurred())
+		g.Expect(out).To(gomega.ContainSubstring(params.FromContainer))
+	}).Should(gomega.Succeed())
+
+	execArgs := []string{
+		"exec",
+		fmt.Sprintf("--container=%s", params.FromContainer),
+		params.FromPod,
+		"-n", params.FromNamespace,
+		"--", params.ExecCmdPath,
+	}
+
+	kParams.Args = append(execArgs, params.Args...)
+	return execute(ctx, KubectlParams{
+		KubectlCmdParams: kParams,
+		KubeContext:      params.KubeContext,
+		Logger:           params.Logger,
+	})
+}