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,72 @@
+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")
+	}
+
+	var foundContainer *corev1.ContainerStatus = nil
+
+	for i, cs := range pod.Status.ContainerStatuses {
+		if cs.Name == containerName {
+			foundContainer = &pod.Status.ContainerStatuses[i]
+
+			break
+		}
+	}
+	if foundContainer == nil {
+		return false, fmt.Errorf("pod does not contain container %s", containerName)
+	}
+	if foundContainer.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 foundContainer.State.Terminated.ExitCode != 0 { // exit code of 0 means success
+		return false, fmt.Errorf("container %s exited with code %d: %s", containerName, foundContainer.State.Terminated.ExitCode, foundContainer.State.Terminated.Reason)
+	}
+
+	return true, nil // container terminated with exit code of 0
+}
+
+// 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")
+	}
+
+	containerDiag := fmt.Sprintf("container %s is healthy", containerName)
+	containerOk, containerError := kw.PodContainerHealthy(pod, containerName)
+	if containerError != nil {
+		containerDiag = fmt.Sprintf("%v", containerError)
+	}
+
+	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 message: %s", podError, pod.Status.Message)
+		}
+
+		return false, fmt.Errorf("%s %v", podError, containerDiag)
+
+	case corev1.PodSucceeded, corev1.PodRunning, corev1.PodPending:
+		return containerOk, containerError
+	default:
+		return false, fmt.Errorf("unknown phase: %s %s", pod.Status.Phase, containerDiag)
+	}
+}