Update Kube Exec() to take Stdin & fix unit test (#5441)

pkg/function/backup_data.go

@@ -92,7 +92,7 @@ func (*backupDataFunc) Exec(ctx context.Context, tp param.TemplateParams, args m
 	// Create backup and dump it on the object store
 	backupTag := rand.String(10)
 	cmd := restic.BackupCommandByTag(tp.Profile, backupArtifactPrefix, backupTag, includePath, encryptionKey)
-	stdout, stderr, err := kube.Exec(cli, namespace, pod, container, cmd)
+	stdout, stderr, err := kube.Exec(cli, namespace, pod, container, cmd, nil)
 	format.Log(pod, container, stdout)
 	format.Log(pod, container, stderr)
 	if err != nil {

pkg/function/copy_volume_data.go

@@ -75,7 +75,7 @@ func copyVolumeDataPodFunc(cli kubernetes.Interface, tp param.TemplateParams, na
 		// Copy data to object store
 		backupTag := rand.String(10)
 		cmd := restic.BackupCommandByTag(tp.Profile, targetPath, backupTag, mountPoint, encryptionKey)
-		stdout, stderr, err := kube.Exec(cli, namespace, pod.Name, pod.Spec.Containers[0].Name, cmd)
+		stdout, stderr, err := kube.Exec(cli, namespace, pod.Name, pod.Spec.Containers[0].Name, cmd, nil)
 		format.Log(pod.Name, pod.Spec.Containers[0].Name, stdout)
 		format.Log(pod.Name, pod.Spec.Containers[0].Name, stderr)
 		if err != nil {

pkg/function/delete_data.go

@@ -65,7 +65,7 @@ func deleteDataPodFunc(cli kubernetes.Interface, tp param.TemplateParams, reclai
 		}
 		if deleteTag != "" {
 			cmd := restic.SnapshotsCommandByTag(tp.Profile, targetPath, deleteTag, encryptionKey)
-			stdout, stderr, err := kube.Exec(cli, namespace, pod.Name, pod.Spec.Containers[0].Name, cmd)
+			stdout, stderr, err := kube.Exec(cli, namespace, pod.Name, pod.Spec.Containers[0].Name, cmd, nil)
 			format.Log(pod.Name, pod.Spec.Containers[0].Name, stdout)
 			format.Log(pod.Name, pod.Spec.Containers[0].Name, stderr)
 			if err != nil {
@@ -78,7 +78,7 @@ func deleteDataPodFunc(cli kubernetes.Interface, tp param.TemplateParams, reclai
 		}
 		if deleteIdentifier != "" {
 			cmd := restic.ForgetCommandByID(tp.Profile, targetPath, deleteIdentifier, encryptionKey)
-			stdout, stderr, err := kube.Exec(cli, namespace, pod.Name, pod.Spec.Containers[0].Name, cmd)
+			stdout, stderr, err := kube.Exec(cli, namespace, pod.Name, pod.Spec.Containers[0].Name, cmd, nil)
 			format.Log(pod.Name, pod.Spec.Containers[0].Name, stdout)
 			format.Log(pod.Name, pod.Spec.Containers[0].Name, stderr)
 			if err != nil {
@@ -87,7 +87,7 @@ func deleteDataPodFunc(cli kubernetes.Interface, tp param.TemplateParams, reclai
 		}
 		if reclaimSpace {
 			cmd := restic.PruneCommand(tp.Profile, targetPath, encryptionKey)
-			stdout, stderr, err := kube.Exec(cli, namespace, pod.Name, pod.Spec.Containers[0].Name, cmd)
+			stdout, stderr, err := kube.Exec(cli, namespace, pod.Name, pod.Spec.Containers[0].Name, cmd, nil)
 			format.Log(pod.Name, pod.Spec.Containers[0].Name, stdout)
 			format.Log(pod.Name, pod.Spec.Containers[0].Name, stderr)
 			if err != nil {

pkg/function/kube_exec.go

@@ -79,7 +79,7 @@ func (kef *kubeExecFunc) Exec(ctx context.Context, tp param.TemplateParams, args
 		return nil, err
 	}
 
-	stdout, stderr, err := kube.Exec(cli, namespace, pod, container, cmd)
+	stdout, stderr, err := kube.Exec(cli, namespace, pod, container, cmd, nil)
 	format.Log(pod, container, stdout)
 	format.Log(pod, container, stderr)
 	if err != nil {

pkg/function/kube_exec_all.go

@@ -70,7 +70,7 @@ func execAll(cli kubernetes.Interface, namespace string, ps []string, cs []strin
 	for _, p := range ps {
 		for _, c := range cs {
 			go func(p string, c string) {
-				stdout, stderr, err := kube.Exec(cli, namespace, p, c, cmd)
+				stdout, stderr, err := kube.Exec(cli, namespace, p, c, cmd, nil)
 				format.Log(p, c, stdout)
 				format.Log(p, c, stderr)
 				errChan <- err

pkg/kube/exec.go

@@ -28,12 +28,13 @@ type ExecOptions struct {
 
 // Exec is our version of the call to `kubectl exec` that does not depend on
 // k8s.io/kubernetes.
-func Exec(cli kubernetes.Interface, namespace, pod, container string, command []string) (string, string, error) {
+func Exec(cli kubernetes.Interface, namespace, pod, container string, command []string, stdin io.Reader) (string, string, error) {
 	opts := ExecOptions{
 		Command:       command,
 		Namespace:     namespace,
 		PodName:       pod,
 		ContainerName: container,
+		Stdin:         stdin,
 		CaptureStdout: true,
 		CaptureStderr: true,
 	}

pkg/kube/exec_test.go

@@ -3,6 +3,10 @@
 package kube
 
 import (
+	"bytes"
+	"context"
+	"time"
+
 	. "gopkg.in/check.v1"
 	"k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -43,6 +47,11 @@ func (s *ExecSuite) SetUpSuite(c *C) {
 	}
 	s.pod, err = s.cli.Core().Pods(s.namespace).Create(pod)
 	c.Assert(err, IsNil)
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
+	defer cancel()
+	c.Assert(WaitForPodReady(ctx, s.cli, s.namespace, s.pod.Name), IsNil)
+	s.pod, err = s.cli.Core().Pods(s.namespace).Get(s.pod.Name, metav1.GetOptions{})
+	c.Assert(err, IsNil)
 }
 
 func (s *ExecSuite) TearDownSuite(c *C) {
@@ -52,9 +61,11 @@ func (s *ExecSuite) TearDownSuite(c *C) {
 	}
 }
 func (s *ExecSuite) TestExecEcho(c *C) {
-	cmd := []string{"sh", "-c", "echo badabing"}
+	cmd := []string{"sh", "-c", "cat -"}
+	c.Assert(s.pod.Status.Phase, Equals, v1.PodRunning)
+	c.Assert(len(s.pod.Status.ContainerStatuses) > 0, Equals, true)
 	for _, cs := range s.pod.Status.ContainerStatuses {
-		stdout, stderr, err := Exec(s.cli, s.pod.Namespace, s.pod.Name, cs.Name, cmd)
+		stdout, stderr, err := Exec(s.cli, s.pod.Namespace, s.pod.Name, cs.Name, cmd, bytes.NewBufferString("badabing"))
 		c.Assert(err, IsNil)
 		c.Assert(stdout, Equals, "badabing")
 		c.Assert(stderr, Equals, "")

pkg/kube/pod.go

@@ -83,7 +83,7 @@ func WaitForPodReady(ctx context.Context, cli kubernetes.Interface, namespace, n
 	err := poll.Wait(ctx, func(ctx context.Context) (bool, error) {
 		p, err := cli.Core().Pods(namespace).Get(name, metav1.GetOptions{})
 		if err != nil {
-			return true, err
+			return false, err
 		}
 		return (p.Status.Phase == v1.PodRunning), nil
 	})

pkg/restic/restic.go

@@ -124,15 +124,15 @@ func resticArgs(profile *param.Profile, repository, encryptionKey string) []stri
 func GetOrCreateRepository(cli kubernetes.Interface, namespace, pod, container, artifactPrefix, encryptionKey string, profile *param.Profile) error {
 	// Use the snapshots command to check if the repository exists
 	cmd := SnapshotsCommand(profile, artifactPrefix, encryptionKey)
-	stdout, stderr, err := kube.Exec(cli, namespace, pod, container, cmd)
+	stdout, stderr, err := kube.Exec(cli, namespace, pod, container, cmd, nil)
 	format.Log(pod, container, stdout)
 	format.Log(pod, container, stderr)
 	if err == nil {
 		return nil
 	}
 	// Create a repository
 	cmd = InitCommand(profile, artifactPrefix, encryptionKey)
-	stdout, stderr, err = kube.Exec(cli, namespace, pod, container, cmd)
+	stdout, stderr, err = kube.Exec(cli, namespace, pod, container, cmd, nil)
 	format.Log(pod, container, stdout)
 	format.Log(pod, container, stderr)
 	return errors.Wrapf(err, "Failed to create object store backup location")