AAP-46619 Get pod container status

.gitignore

@@ -29,4 +29,5 @@ venv
 /vendor
 pkg/services/.lock
 pkg/workceptor/status
-pkg/workceptor/status.lock
+pkg/workceptor/status.lock
+**/__debug_*

go.mod

@@ -32,6 +32,11 @@ require (
 	k8s.io/client-go v0.31.3
 )
 
+require (
+	github.com/pkg/errors v0.9.1 // indirect
+	gopkg.in/evanphx/json-patch.v4 v4.12.0 // indirect
+)
+
 require (
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/stretchr/testify v1.10.0

pkg/workceptor/kubernetes.go

@@ -177,6 +177,8 @@ var ErrPodFailed = fmt.Errorf("pod failed to start")
 // ErrImagePullBackOff is returned when the image for the container in the Pod cannot be pulled.
 var ErrImagePullBackOff = fmt.Errorf("container failed to start")
 
+const WorkerContainerName = "worker"
+
 // podRunningAndReady is a completion criterion for pod ready to be attached to.
 func podRunningAndReady(kw KubeUnit) func(event watch.Event) (bool, error) {
 	imagePullBackOffRetries := 3
@@ -249,7 +251,7 @@ func (kw *KubeUnit) kubeLoggingConnectionHandler(timestamps bool, sinceTime time
 	podNamespace := kw.Pod.Namespace
 	podName := kw.Pod.Name
 	podOptions := &corev1.PodLogOptions{
-		Container: "worker",
+		Container: WorkerContainerName,
 		Follow:    true,
 	}
 	if timestamps {
@@ -486,7 +488,7 @@ func (kw *KubeUnit) CreatePod(env map[string]string) error {
 		foundWorker := false
 		spec = &pod.Spec
 		for i := range spec.Containers {
-			if spec.Containers[i].Name == "worker" {
+			if spec.Containers[i].Name == WorkerContainerName {
 				spec.Containers[i].Stdin = true
 				spec.Containers[i].StdinOnce = true
 				foundWorker = true
@@ -517,7 +519,7 @@ func (kw *KubeUnit) CreatePod(env map[string]string) error {
 		}
 		spec = &corev1.PodSpec{
 			Containers: []corev1.Container{{
-				Name:      "worker",
+				Name:      WorkerContainerName,
 				Image:     ked.Image,
 				Command:   command,
 				Args:      params,
@@ -601,7 +603,7 @@ func (kw *KubeUnit) CreatePod(env map[string]string) error {
 	if err == ErrPodCompleted {
 		// Hao: shouldn't we also call kw.Cancel() in these cases?
 		for _, cstat := range kw.Pod.Status.ContainerStatuses {
-			if cstat.Name == "worker" {
+			if cstat.Name == WorkerContainerName {
 				if cstat.State.Terminated != nil && cstat.State.Terminated.ExitCode != 0 {
 					return fmt.Errorf("container failed with exit code %d: %s", cstat.State.Terminated.ExitCode, cstat.State.Terminated.Message)
 				}
@@ -632,7 +634,7 @@ func (kw *KubeUnit) CreatePod(env map[string]string) error {
 			}
 
 			for _, cstat := range kw.Pod.Status.ContainerStatuses {
-				if cstat.Name == "worker" {
+				if cstat.Name == WorkerContainerName {
 					if cstat.State.Waiting != nil {
 						return fmt.Errorf("%s, %s", err.Error(), cstat.State.Waiting.Reason)
 					}
@@ -732,7 +734,7 @@ func (kw *KubeUnit) runWorkUsingLogger() {
 
 		req.VersionedParams(
 			&corev1.PodExecOptions{
-				Container: "worker",
+				Container: WorkerContainerName,
 				Stdin:     true,
 				Stdout:    false,
 				Stderr:    false,

pkg/workceptor/pod.go

@@ -0,0 +1,87 @@
+package workceptor
+
+import (
+	"fmt"
+
+	corev1 "k8s.io/api/core/v1"
+)
+
+type KubePodStateHelper interface {
+	GetPodStatus(pod *corev1.Pod) (bool, error)
+	PodHealthy(pod *corev1.Pod, containerName string) (bool, error)
+	PodContainerHealthy(pod *corev1.Pod, containerName string) (bool, error)
+}
+
+// PodContainerHealthy checks if the pod has successfully completed its application logic.
+// this is called after podInfrastructureSuccess has confirmed the pod is in a terminal state.
+func (kw KubeUnit) PodContainerHealthy(pod *corev1.Pod, containerName string) (bool, error) {
+	if pod == nil {
+		return false, fmt.Errorf("pod is nil")
+	}
+
+	for _, cs := range pod.Status.ContainerStatuses {
+		if cs.Name == containerName {
+			if cs.State.Terminated == nil { // means it is waiting or running, so application logic has not completed yet. Normal behavior when job completes successfully.
+				return true, nil
+			}
+
+			if cs.State.Terminated.ExitCode != 0 { // exit code of 0 means success
+				return false, fmt.Errorf("container %s exited with code %d: %s", cs.Name, cs.State.Terminated.ExitCode, cs.State.Terminated.Reason)
+			}
+
+			return true, nil // container terminated with exit code of 0
+		}
+	}
+
+	return false, fmt.Errorf("pod does not contain container %s", containerName)
+}
+
+// PodInfrastructureSuccess checks if the pod has either successfully started, is pending or is running, or has is successfully terminated.
+// Any other state is considered an infrastructure failure.
+func (kw KubeUnit) PodHealthy(pod *corev1.Pod, containerName string) (bool, error) {
+	if pod == nil {
+		return false, fmt.Errorf("pod is nil")
+	}
+
+	for _, cs := range pod.Status.ContainerStatuses {
+		// Check if this is the container we care about first
+		if cs.Name == containerName {
+			switch pod.Status.Phase {
+			case corev1.PodFailed:
+				podError := fmt.Errorf("pod failed with reason: %s", pod.Status.Reason)
+				if pod.Status.Message != "" {
+					podError = fmt.Errorf("%s: %s", podError.Error(), pod.Status.Message)
+				}
+				ok, containerError := kw.PodContainerHealthy(pod, containerName)
+				if !ok || containerError != nil {
+					return false, fmt.Errorf("%s %v", podError.Error(), containerError)
+				}
+
+				return false, podError
+
+			case corev1.PodSucceeded:
+				return true, nil
+			case corev1.PodRunning, corev1.PodPending:
+				return true, nil
+			default:
+				return false, fmt.Errorf("unknown phase: %s", pod.Status.Phase)
+			}
+		}
+	}
+
+	return false, fmt.Errorf("pod does not contain container %s", containerName)
+}
+
+// GetPodStatus checks if the pod has successfully completed its application logic and infrastructure is healthy.
+func (kw KubeUnit) GetPodStatus(pod *corev1.Pod) (bool, error) {
+	if pod == nil {
+		return false, fmt.Errorf("pod is nil")
+	}
+
+	ok, err := kw.PodHealthy(pod, WorkerContainerName)
+	if !ok || err != nil {
+		return ok, err
+	}
+
+	return ok, nil
+}