Merge pull request #44 from pablochacin/check-pod-condition

Add wait option to Pod Create

Author: javaducky

examples/create-pod-wait.js

@@ -0,0 +1,25 @@
+
+import { sleep } from 'k6';
+import { Kubernetes } from 'k6/x/kubernetes';
+
+export default function () {
+  const kubernetes = new Kubernetes({
+  })
+  const namespace = "default"
+  const podName = "new-pod"
+  const image = "busybox"
+  const command = ["sh",  "-c", "/bin/false"]
+
+try {  
+  kubernetes.pods.create({
+    namespace: namespace,
+    name: podName,
+    image: image,
+    command: command,
+    wait: "5s"
+  })
+  console.log(podName + " has been created")
+} catch (err) {
+  console.log("error creating pod " + podName + ": " + err)
+}
+}

examples/wait-pod.js

@@ -0,0 +1,32 @@
+
+import { sleep } from 'k6';
+import { Kubernetes } from 'k6/x/kubernetes';
+
+export default function () {
+  const kubernetes = new Kubernetes({
+  })
+  const namespace = "default"
+  const podName = "new-pod"
+  const image = "busybox"
+  const command = ["sh",  "-c", "sleep 5"]
+
+  kubernetes.pods.create({
+    namespace: namespace,
+    name: podName,
+    image: image,
+    command: command
+  })
+  
+ 
+  const options = {
+    namespace: namespace,
+    name: podName,
+    status: "Succeeded",
+    timeout: "10s"
+  }
+  if (kubernetes.pods.wait(options)) {
+    console.log(podName + " pod completed successfully")
+  } else {
+    throw podName + " is not completed"
+  }
+}

internal/testutils/kube_resources.go

@@ -226,9 +226,19 @@ func NewPod(name string, namespace string) *coreV1.Pod {
 			},
 			EphemeralContainers: nil,
 		},
+		Status: coreV1.PodStatus{
+			Phase: coreV1.PodRunning,
+		},
 	}
 }
 
+// NewPodWithStatus is a helper for building Pods with a given Status
+func NewPodWithStatus(name string, namespace string, phase string) *coreV1.Pod {
+	pod := NewPod(name, namespace)
+	pod.Status.Phase = coreV1.PodPhase(phase)
+	return pod
+}
+
 // NewSecret is a helper to build a new Secret instance
 func NewSecret(name string, namespace string) *coreV1.Secret {
 	return &coreV1.Secret{

pkg/pods/pods.go

@@ -5,17 +5,27 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"fmt"
+	"time"
 
 	k8sTypes "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/strategicpatch"
+	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/remotecommand"
 )
 
+// ValidPod status
+const (
+	Running   string = "Running"
+	Succeeded string = "Succeeded"
+)
+
 func New(client kubernetes.Interface, config *rest.Config, metaOptions metav1.ListOptions, ctx context.Context) *Pods {
 	return &Pods{
 		client,
@@ -62,6 +72,7 @@ type PodOptions struct {
 	Image         string                 // image to be executed by the pod's container
 	Command       []string               // command to be executed by the pod's container and its arguments
 	RestartPolicy k8sTypes.RestartPolicy // policy for restarting containers in the pod. One of One of Always, OnFailure, Never
+	Wait          string                 // timeout for waiting until the pod is running
 }
 
 func (obj *Pods) List(namespace string) ([]k8sTypes.Pod, error) {
@@ -133,7 +144,81 @@ func (obj *Pods) Create(options PodOptions) (k8sTypes.Pod, error) {
 	if err != nil {
 		return k8sTypes.Pod{}, err
 	}
-	return *pod, nil
+
+	if options.Wait == "" {
+		return *pod, nil
+	}
+	waitOpts := WaitOptions{
+		Name:      options.Name,
+		Namespace: options.Namespace,
+		Status:    Running,
+		Timeout:   options.Wait,
+	}
+	status, err := obj.Wait(waitOpts)
+	if err != nil {
+		return k8sTypes.Pod{}, err
+	}
+	if !status {
+		return k8sTypes.Pod{}, errors.New("timeout exceeded waiting for pod to be running")
+	}
+
+	return obj.Get(options.Name, options.Namespace)
+}
+
+// Options for waiting for a Pod status
+type WaitOptions struct {
+	Name      string // Pod name
+	Namespace string // Namespace where the pod is running
+	Status    string // Wait until pod reaches the specified status. Must be one of "Running" or "Succeeded".
+	Timeout   string // Timeout for waiting condition to be true
+}
+
+// Wait for the Pod to be in a given status up to given timeout and returns a boolean indicating if the staus was reached. If the pod is Failed returns error.
+func (obj *Pods) Wait(options WaitOptions) (bool, error) {
+	if options.Status != Running && options.Status != Succeeded {
+		return false, errors.New("wait condition must be 'Running' or 'Succeeded'")
+	}
+	timeout, err := time.ParseDuration(options.Timeout)
+	if err != nil {
+		return false, err
+	}
+	selector := fields.Set{
+		"metadata.name": options.Name,
+	}.AsSelector()
+	watcher, err := obj.client.CoreV1().Pods(options.Namespace).Watch(
+		obj.ctx,
+		metav1.ListOptions{
+			FieldSelector: selector.String(),
+		},
+	)
+	if err != nil {
+		return false, err
+	}
+	defer watcher.Stop()
+
+	for {
+		select {
+		case <-time.After(timeout):
+			return false, nil
+		case event := <-watcher.ResultChan():
+			if event.Type == watch.Error {
+				return false, fmt.Errorf("error watching for pod: %v", event.Object)
+			}
+			if event.Type == watch.Modified {
+				pod, isPod := event.Object.(*k8sTypes.Pod)
+				if !isPod {
+					return false, errors.New("received unknown object while watching for pods")
+				}
+				if pod.Status.Phase == k8sTypes.PodFailed {
+					return false, errors.New("Pod has failed")
+				}
+				if string(pod.Status.Phase) == options.Status {
+					return true, nil
+				}
+			}
+		}
+	}
+	return false, nil
 }
 
 // Exec executes a non-interactive command described in options and returns the stdout and stderr outputs