Add support for running a pod with specified PVCs attached (#4067)

* Add helpers to run a pod

* Address review comments

pkg/kube/pod.go

@@ -0,0 +1,69 @@
+package kube
+
+import (
+	"context"
+
+	"github.com/pkg/errors"
+	log "github.com/sirupsen/logrus"
+	"k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+
+	"github.com/kanisterio/kanister/pkg/poll"
+)
+
+// PodOptions specifies options for `CreatePod`
+type PodOptions struct {
+	Namespace    string
+	GenerateName string
+	Image        string
+	Command      []string
+	Volumes      map[string]string
+}
+
+// CreatePod creates a pod with a single container based on the specified image
+func CreatePod(ctx context.Context, cli kubernetes.Interface, opts *PodOptions) (*v1.Pod, error) {
+	volumeMounts, podVolumes := createVolumeSpecs(opts.Volumes)
+	pod := &v1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			GenerateName: opts.GenerateName,
+			Namespace:    opts.Namespace,
+		},
+		Spec: v1.PodSpec{
+			Containers: []v1.Container{
+				v1.Container{
+					Name:            "container",
+					Image:           opts.Image,
+					Command:         opts.Command,
+					ImagePullPolicy: v1.PullPolicy(v1.PullIfNotPresent),
+					VolumeMounts:    volumeMounts,
+				},
+			},
+			Volumes: podVolumes,
+		},
+	}
+	pod, err := cli.Core().Pods(opts.Namespace).Create(pod)
+	if err != nil {
+		return nil, errors.Wrapf(err, "Failed to create pod. Namespace: %s, NameFmt: %s", opts.Namespace, opts.GenerateName)
+	}
+	err = poll.Wait(ctx, func(ctx context.Context) (bool, error) {
+		p, err := cli.Core().Pods(pod.Namespace).Get(pod.Name, metav1.GetOptions{})
+		if err != nil {
+			return true, err
+		}
+		return (p.Status.Phase == v1.PodRunning), nil
+	})
+	if err != nil {
+		defer DeletePod(context.Background(), cli, pod)
+		return nil, errors.Wrapf(err, "Pod did not transition to running state. Namespace:%s, Name:%s", pod.Namespace, pod.Name)
+	}
+	return pod, nil
+}
+
+// DeletePod deletes the specified pod
+func DeletePod(ctx context.Context, cli kubernetes.Interface, pod *v1.Pod) error {
+	if err := cli.Core().Pods(pod.Namespace).Delete(pod.Name, nil); err != nil {
+		log.Errorf("DeletePod failed: %v", err)
+	}
+	return nil
+}

pkg/kube/pod_test.go

@@ -0,0 +1,87 @@
+// +build !unit
+
+package kube
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	. "gopkg.in/check.v1"
+	"k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/kubernetes/fake"
+	"k8s.io/client-go/testing"
+)
+
+type PodSuite struct {
+	cli       kubernetes.Interface
+	namespace string
+}
+
+var _ = Suite(&PodSuite{})
+
+func (s *PodSuite) SetUpSuite(c *C) {
+	var err error
+	s.cli, err = NewClient()
+	c.Assert(err, IsNil)
+	ns := &v1.Namespace{
+		ObjectMeta: metav1.ObjectMeta{
+			GenerateName: "podtest-",
+		},
+	}
+	ns, err = s.cli.Core().Namespaces().Create(ns)
+	c.Assert(err, IsNil)
+	s.namespace = ns.Name
+}
+
+func (s *PodSuite) TearDownSuite(c *C) {
+	if s.namespace != "" {
+		err := s.cli.Core().Namespaces().Delete(s.namespace, nil)
+		c.Assert(err, IsNil)
+	}
+}
+
+func (s *PodSuite) TestPod(c *C) {
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
+	defer cancel()
+	pod, err := CreatePod(ctx, s.cli, &PodOptions{
+		Namespace:    s.namespace,
+		GenerateName: "test-",
+		Image:        "kanisterio/kanister-tools:0.12.0",
+		Command:      []string{"sh", "-c", "tail -f /dev/null"},
+	})
+	c.Assert(err, IsNil)
+	c.Assert(DeletePod(context.Background(), s.cli, pod), IsNil)
+}
+
+func (s *PodSuite) TestPodWithVolumes(c *C) {
+	cli := fake.NewSimpleClientset()
+	vols := map[string]string{"pvc-test": "/mnt/data1"}
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
+	defer cancel()
+	var p *v1.Pod
+	cli.PrependReactor("create", "pods", func(action testing.Action) (handled bool, ret runtime.Object, err error) {
+		fmt.Println("found pod")
+		ca := action.(testing.CreateAction)
+		p = ca.GetObject().(*v1.Pod)
+		return false, nil, nil
+	})
+	cli.PrependReactor("get", "pods", func(action testing.Action) (handled bool, ret runtime.Object, err error) {
+		p.Status.Phase = v1.PodRunning
+		return true, p, nil
+	})
+	pod, err := CreatePod(ctx, cli, &PodOptions{
+		Namespace:    s.namespace,
+		GenerateName: "test-",
+		Image:        "kanisterio/kanister-tools:0.12.0",
+		Command:      []string{"sh", "-c", "tail -f /dev/null"},
+		Volumes:      vols,
+	})
+	c.Assert(err, IsNil)
+	c.Assert(pod.Spec.Volumes, HasLen, 1)
+	c.Assert(pod.Spec.Volumes[0].VolumeSource.PersistentVolumeClaim.ClaimName, Equals, "pvc-test")
+	c.Assert(pod.Spec.Containers[0].VolumeMounts[0].MountPath, Equals, "/mnt/data1")
+}