Make deploy.RunPod wait for pod readiness rather than running phase

e2e/disruptors/pod_e2e_test.go

@@ -70,11 +70,12 @@ func Test_PodDisruptor(t *testing.T) {
 			check    checks.Check
 		}{
 			{
-				title:    "Inject Http error 500",
-				pod:      fixtures.BuildHttpbinPod(),
+
+				title:   "Inject Http error 500",
+				pod:     fixtures.BuildHttpbinPodWithoutProbes(),
 				replicas: 1,
-				service:  fixtures.BuildHttpbinService(),
-				port:     80,
+				service: fixtures.BuildHttpbinService(),
+				port:    80,
 				injector: func(d disruptors.PodDisruptor) error {
 					fault := disruptors.HTTPFault{
 						Port:      intstr.FromInt32(80),
@@ -97,11 +98,11 @@ func Test_PodDisruptor(t *testing.T) {
 				},
 			},
 			{
-				title:    "Inject Grpc error",
-				pod:      fixtures.BuildGrpcpbinPod(),
+				title:   "Inject Grpc error",
+				pod:     fixtures.BuildGrpcpbinPod(),
 				replicas: 1,
-				service:  fixtures.BuildGrpcbinService(),
-				port:     9000,
+				service: fixtures.BuildGrpcbinService(),
+				port:    9000,
 				injector: func(d disruptors.PodDisruptor) error {
 					fault := disruptors.GrpcFault{
 						Port:       intstr.FromInt32(9000),
@@ -231,7 +232,7 @@ func Test_PodDisruptor(t *testing.T) {
 		err = deploy.ExposeApp(
 			k8s,
 			namespace,
-			fixtures.BuildHttpbinPod(),
+			fixtures.BuildHttpbinPodWithoutProbes(), // Probes generate requests, invalidating the test.
 			1,
 			service,
 			k8sintstr.FromInt(80),

e2e/disruptors/service_e2e_test.go

@@ -63,11 +63,11 @@ func Test_ServiceDisruptor(t *testing.T) {
 			check    checks.Check
 		}{
 			{
-				title:    "Inject Http error 500",
-				pod:      fixtures.BuildHttpbinPod(),
-				replicas: 1,
-				service:  fixtures.BuildHttpbinService(),
-				port:     80,
+
+				title:   "Inject Http error 500",
+				pod:     fixtures.BuildHttpbinPodWithoutProbes(),
+				service: fixtures.BuildHttpbinService(),
+				port:    80,
 				injector: func(d disruptors.ServiceDisruptor) error {
 					fault := disruptors.HTTPFault{
 						Port:      intstr.FromInt32(80),

pkg/kubernetes/helpers/pods.go

@@ -22,9 +22,9 @@ import (
 
 // PodHelper defines helper methods for handling Pods
 type PodHelper interface {
-	// WaitPodRunning waits for the Pod to be running for up to given timeout and returns a boolean indicating
+	// WaitPodReady waits for the Pod to be running for up to given timeout and returns a boolean indicating
 	// if the status was reached. If the pod is Failed returns error.
-	WaitPodRunning(ctx context.Context, name string, timeout time.Duration) (bool, error)
+	WaitPodReady(ctx context.Context, name string, timeout time.Duration) (bool, error)
 	// WaitPodDeleted waits for the Pod to be deleted for up to given timeout
 	WaitPodDeleted(ctx context.Context, name string, timeout time.Duration) error
 	// Exec executes a non-interactive command described in options and returns the stdout and stderr outputs
@@ -140,7 +140,7 @@ func (h *podHelper) waitForCondition(
 	}
 }
 
-func (h *podHelper) WaitPodRunning(ctx context.Context, name string, timeout time.Duration) (bool, error) {
+func (h *podHelper) WaitPodReady(ctx context.Context, name string, timeout time.Duration) (bool, error) {
 	return h.waitForCondition(
 		ctx,
 		h.namespace,
@@ -150,9 +150,13 @@ func (h *podHelper) WaitPodRunning(ctx context.Context, name string, timeout tim
 			if pod.Status.Phase == corev1.PodFailed {
 				return false, errors.New("pod has failed")
 			}
-			if pod.Status.Phase == corev1.PodRunning {
-				return true, nil
+
+			for _, condition := range pod.Status.Conditions {
+				if condition.Type == corev1.PodReady && condition.Status == corev1.ConditionTrue {
+					return true, nil
+				}
 			}
+
 			return false, nil
 		},
 	)

pkg/kubernetes/helpers/pods_test.go

@@ -26,6 +26,7 @@ func TestPods_Wait(t *testing.T) {
 		test           string
 		name           string
 		phase          corev1.PodPhase
+		conditions     []corev1.PodCondition
 		delay          time.Duration
 		expectError    bool
 		expectedResult bool
@@ -34,18 +35,47 @@ func TestPods_Wait(t *testing.T) {
 
 	testCases := []TestCase{
 		{
-			test:           "wait pod running",
-			name:           "pod-running",
-			delay:          1 * time.Second,
-			phase:          corev1.PodRunning,
+			test:  "pod running and ready",
+			name:  "pod-running-ready",
+			delay: 1 * time.Second,
+			phase: corev1.PodRunning,
+			conditions: []corev1.PodCondition{
+				{
+					Type:   corev1.PodReady,
+					Status: corev1.ConditionTrue,
+				},
+			},
 			expectError:    false,
 			expectedResult: true,
 			timeout:        5 * time.Second,
 		},
 		{
-			test:           "timeout waiting pod running",
-			name:           "pod-running",
+			test:  "pod running but not ready",
+			name:  "pod-running-not-ready",
+			delay: 1 * time.Second,
+			phase: corev1.PodRunning,
+			conditions: []corev1.PodCondition{
+				{
+					Type:   corev1.PodReady,
+					Status: corev1.ConditionFalse,
+				},
+			},
+			expectError:    false,
+			expectedResult: false,
+			timeout:        5 * time.Second,
+		},
+		{
+			test:           "pod running no conditions",
+			name:           "pod-running-no-conditions",
+			delay:          1 * time.Second,
 			phase:          corev1.PodRunning,
+			expectError:    false,
+			expectedResult: false,
+			timeout:        5 * time.Second,
+		},
+		{
+			test:           "pod not running",
+			name:           "pod-not-running",
 			delay:          10 * time.Second,
 			expectError:    false,
 			expectedResult: false,
@@ -71,6 +101,7 @@ func TestPods_Wait(t *testing.T) {
 			observer := func(event builders.ObjectEvent, pod *corev1.Pod) (*corev1.Pod, bool, error) {
 				time.Sleep(tc.delay)
 				pod.Status.Phase = tc.phase
+				pod.Status.Conditions = tc.conditions
 				// update pod and stop watching updates
 				return pod, false, nil
 			}
@@ -95,7 +126,7 @@ func TestPods_Wait(t *testing.T) {
 			}
 
 			h := NewPodHelper(client, nil, testNamespace)
-			result, err := h.WaitPodRunning(
+			result, err := h.WaitPodReady(
 				context.TODO(),
 				tc.name,
 				tc.timeout,

pkg/kubernetes/integration_test.go

@@ -92,7 +92,7 @@ func Test_Kubernetes(t *testing.T) {
 		}
 
 		// wait for the pod to be ready for accepting requests
-		running, err := k8s.PodHelper(namespace).WaitPodRunning(context.TODO(), "nginx", time.Second*20)
+		running, err := k8s.PodHelper(namespace).WaitPodReady(context.TODO(), "nginx", time.Second*20)
 		if err != nil {
 			t.Fatalf("error waiting for pod %v", err)
 		}
@@ -114,7 +114,7 @@ func Test_Kubernetes(t *testing.T) {
 		})
 
 		// Deploy nginx and expose it as a service. Intentionally not using e2e fixures
-		// because these functions rely on WaitPodRunning and WaitServiceReady which we
+		// because these functions rely on WaitPodReady and WaitServiceReady which we
 		// are testing here.
 		nginxPod := fixtures.BuildNginxPod()
 		_, err = k8s.Client().CoreV1().Pods(namespace).Create(
@@ -165,7 +165,7 @@ func Test_Kubernetes(t *testing.T) {
 		}
 
 		// wait for the pod to be ready
-		running, err := k8s.PodHelper(namespace).WaitPodRunning(context.TODO(), "busybox", time.Second*20)
+		running, err := k8s.PodHelper(namespace).WaitPodReady(context.TODO(), "busybox", time.Second*20)
 		if err != nil {
 			t.Fatalf("error waiting for pod %v", err)
 		}

pkg/testutils/e2e/deploy/deploy.go

@@ -26,7 +26,7 @@ func RunPod(k8s kubernetes.Kubernetes, ns string, pod corev1.Pod, timeout time.D
 		return fmt.Errorf("error creating pod %s: %w", pod.Name, err)
 	}
 
-	running, err := k8s.PodHelper(ns).WaitPodRunning(
+	running, err := k8s.PodHelper(ns).WaitPodReady(
 		context.TODO(),
 		pod.Name,
 		timeout,

pkg/testutils/e2e/fixtures/fixtures.go

@@ -14,6 +14,7 @@ func BuildHttpbinPod() corev1.Pod {
 		WithImage("kennethreitz/httpbin").
 		WithPullPolicy(corev1.PullIfNotPresent).
 		WithPort("http", 80).
+		WithHTTPReadinessProbe().
 		Build()
 
 	return builders.NewPodBuilder("httpbin").
@@ -22,6 +23,14 @@ func BuildHttpbinPod() corev1.Pod {
 		Build()
 }
 
+// BuildHttpbinPodWithoutProbes returns the same pod as BuildHttpbinPod would, but without any probes.
+func BuildHttpbinPodWithoutProbes() corev1.Pod {
+	pod := BuildHttpbinPod()
+	pod.Spec.Containers[0].ReadinessProbe = nil
+
+	return pod
+}
+
 // BuildGrpcpbinPod returns the definition for deploying grpcbin as a Pod
 func BuildGrpcpbinPod() corev1.Pod {
 	c := builders.NewContainerBuilder("grpcbin").
@@ -83,10 +92,11 @@ func BuildPausedPod() corev1.Pod {
 
 // BuildNginxPod returns the definition of a Pod that runs Nginx
 func BuildNginxPod() corev1.Pod {
-	c := builders.NewContainerBuilder("busybox").
+	c := builders.NewContainerBuilder("nginx").
 		WithImage("nginx").
 		WithPullPolicy(corev1.PullIfNotPresent).
 		WithPort("http", 80).
+		WithHTTPReadinessProbe().
 		Build()
 
 	return builders.NewPodBuilder("nginx").
@@ -95,6 +105,14 @@ func BuildNginxPod() corev1.Pod {
 		Build()
 }
 
+// BuildNginxPodWithoutProbes returns the same pod as BuildNginxPod would, but without any probes.
+func BuildNginxPodWithoutProbes() corev1.Pod {
+	pod := BuildNginxPod()
+	pod.Spec.Containers[0].ReadinessProbe = nil
+
+	return pod
+}
+
 // BuildNginxService returns the definition of a Service that exposes the nginx pod(s)
 func BuildNginxService() corev1.Service {
 	return builders.NewServiceBuilder("nginx").

pkg/testutils/e2e/kubectl/integration_test.go

@@ -4,7 +4,6 @@
 package kubectl
 
 import (
-	"bytes"
 	"context"
 	"fmt"
 	"net/http"
@@ -13,9 +12,8 @@ import (
 
 	"github.com/grafana/xk6-disruptor/pkg/kubernetes"
 	"github.com/grafana/xk6-disruptor/pkg/testutils/e2e/deploy"
+	"github.com/grafana/xk6-disruptor/pkg/testutils/e2e/fixtures"
 	"github.com/grafana/xk6-disruptor/pkg/testutils/e2e/kubernetes/namespace"
-	"github.com/grafana/xk6-disruptor/pkg/testutils/kubernetes/builders"
-
 	"github.com/testcontainers/testcontainers-go/modules/k3s"
 
 	"k8s.io/client-go/tools/clientcmd"
@@ -61,14 +59,7 @@ func Test_Kubectl(t *testing.T) {
 		}
 
 		// Deploy nginx
-		nginx := builders.NewPodBuilder("nginx").
-			WithContainer(
-				builders.NewContainerBuilder("nginx").
-					WithImage("nginx").
-					WithPort("http", 80).
-					Build(),
-			).
-			Build()
+		nginx := fixtures.BuildNginxPod()
 
 		err = deploy.RunPod(k8s, namespace, nginx, 20*time.Second)
 		if err != nil {
@@ -93,7 +84,7 @@ func Test_Kubectl(t *testing.T) {
 		}
 
 		url := fmt.Sprintf("http://localhost:%d", port)
-		request, err := http.NewRequest("GET", url, bytes.NewReader([]byte{}))
+		request, err := http.NewRequest("GET", url, nil)
 		if err != nil {
 			t.Errorf("failed to create request: %v", err)
 			return

pkg/testutils/kubernetes/builders/container.go

@@ -1,6 +1,9 @@
 package builders
 
-import corev1 "k8s.io/api/core/v1"
+import (
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/util/intstr"
+)
 
 // ContainerBuilder defines the methods for building a Container
 type ContainerBuilder interface {
@@ -21,17 +24,20 @@ type ContainerBuilder interface {
 	// WithEnvVarFromField adds an environment variable to the container referencing a field
 	// Example: "PodName", "metadata.name"
 	WithEnvVarFromField(name string, path string) ContainerBuilder
+	// WithHTTPReadinessProbe adds an HTTP GET readiness probe to the first pod of the container.
+	WithHTTPReadinessProbe() ContainerBuilder
 }
 
 // containerBuilder maintains the configuration for building a container
 type containerBuilder struct {
-	name         string
-	image        string
-	imagePolicy  corev1.PullPolicy
-	command      []string
-	ports        []corev1.ContainerPort
-	capabilities []corev1.Capability
-	vars         []corev1.EnvVar
+	name           string
+	image          string
+	imagePolicy    corev1.PullPolicy
+	command        []string
+	ports          []corev1.ContainerPort
+	capabilities   []corev1.Capability
+	vars           []corev1.EnvVar
+	readinessProbe *corev1.Probe
 }
 
 // NewContainerBuilder returns a new ContainerBuilder
@@ -87,6 +93,23 @@ func (b *containerBuilder) WithEnvVarFromField(name string, path string) Contain
 	return b
 }
 
+func (b *containerBuilder) WithHTTPReadinessProbe() ContainerBuilder {
+	b.readinessProbe = &corev1.Probe{
+		ProbeHandler: corev1.ProbeHandler{
+			HTTPGet: &corev1.HTTPGetAction{
+				Path: "/",
+				Port: intstr.FromString(b.ports[0].Name),
+			},
+		},
+		TimeoutSeconds:   1,
+		PeriodSeconds:    1,
+		SuccessThreshold: 1,
+		FailureThreshold: 1,
+	}
+
+	return b
+}
+
 func (b *containerBuilder) Build() corev1.Container {
 	return corev1.Container{
 		Name:            b.name,
@@ -99,6 +122,7 @@ func (b *containerBuilder) Build() corev1.Container {
 				Add: b.capabilities,
 			},
 		},
-		Env: b.vars,
+		ReadinessProbe: b.readinessProbe,
+		Env:            b.vars,
 	}
 }