diff --git a/.gitignore b/.gitignore index 4ca36a88b..a2c016e3b 100644 --- a/.gitignore +++ b/.gitignore @@ -29,4 +29,5 @@ venv /vendor pkg/services/.lock pkg/workceptor/status -pkg/workceptor/status.lock \ No newline at end of file +pkg/workceptor/status.lock +**/__debug_* \ No newline at end of file diff --git a/.golangci.yml b/.golangci.yml index afc41aa6f..b2b19d439 100644 --- a/.golangci.yml +++ b/.golangci.yml @@ -100,6 +100,7 @@ linters-settings: - "k8s.io/apimachinery/pkg/watch" - "k8s.io/client-go/kubernetes" - "k8s.io/client-go/rest" + - "k8s.io/client-go/testing" - "k8s.io/client-go/tools/remotecommand" - "github.com/quic-go/quic-go" - "github.com/quic-go/quic-go/logging" diff --git a/go.mod b/go.mod index a8a200f80..61057d478 100644 --- a/go.mod +++ b/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 diff --git a/pkg/workceptor/kubernetes.go b/pkg/workceptor/kubernetes.go index 096a113b5..8d8d3840e 100644 --- a/pkg/workceptor/kubernetes.go +++ b/pkg/workceptor/kubernetes.go @@ -41,6 +41,7 @@ import ( // KubeUnit implements the WorkUnit interface. type KubeUnit struct { BaseWorkUnitForWorkUnit + KubePodStateHelper KubeAPIWrapperInstance KubeAPIer authMethod string streamMethod string @@ -177,6 +178,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 containerName = "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 +252,7 @@ func (kw *KubeUnit) kubeLoggingConnectionHandler(timestamps bool, sinceTime time podNamespace := kw.Pod.Namespace podName := kw.Pod.Name podOptions := &corev1.PodLogOptions{ - Container: "worker", + Container: containerName, Follow: true, } if timestamps { @@ -400,6 +403,15 @@ func (kw *KubeUnit) KubeLoggingWithReconnect(streamWait *sync.WaitGroup, stdout podName, ) + // timeout := int64(10) + // _, err = kw.CapturePodStatus(kw.Pod, stdout.Size(), &timeout) + // if err != nil { + // kw.GetWorkceptor().nc.GetLogger().Info("Detected error while retrieving pod status for %s/%s.", + // podNamespace, + // podName, + // ) + // } + return } @@ -486,7 +498,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 == containerName { spec.Containers[i].Stdin = true spec.Containers[i].StdinOnce = true foundWorker = true @@ -517,7 +529,7 @@ func (kw *KubeUnit) CreatePod(env map[string]string) error { } spec = &corev1.PodSpec{ Containers: []corev1.Container{{ - Name: "worker", + Name: containerName, Image: ked.Image, Command: command, Args: params, @@ -601,7 +613,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 == containerName { 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 +644,7 @@ func (kw *KubeUnit) CreatePod(env map[string]string) error { } for _, cstat := range kw.Pod.Status.ContainerStatuses { - if cstat.Name == "worker" { + if cstat.Name == containerName { if cstat.State.Waiting != nil { return fmt.Errorf("%s, %s", err.Error(), cstat.State.Waiting.Reason) } @@ -732,7 +744,7 @@ func (kw *KubeUnit) runWorkUsingLogger() { req.VersionedParams( &corev1.PodExecOptions{ - Container: "worker", + Container: containerName, Stdin: true, Stdout: false, Stderr: false, @@ -912,7 +924,21 @@ func (kw *KubeUnit) runWorkUsingLogger() { return } - // only transition from WorkStateRunning to WorkStateSucceeded if WorkStateFailed is set we do not override + // pod, err := kw.KubeAPIWrapperInstance.Get(kw.GetContext(), kw.clientset, podNamespace, podName, metav1.GetOptions{}) + // if err != nil { + // kw.GetWorkceptor().nc.GetLogger().Warning("Failed to retrieve pod for diagnostics: %v", err) + // } + // timeout := int64(10) + // ok, _ := kw.CapturePodStatus(pod, stdout.Size(), &timeout) + // if !ok { + // // If the pod did not succeed, we already updated the status to WorkStateFailed + // // and we can return early. + // return + // } + + // Only transition to WorkStateSucceeded if the work unit is still running + // and has not already failed due to diagnostic or streaming errors. + // This ensures we don't override a previously set WorkStateFailed. if kw.GetContext().Err() != context.Canceled && kw.Status().State == WorkStateRunning { kw.UpdateBasicStatus(WorkStateSucceeded, "Finished", stdout.Size()) } diff --git a/pkg/workceptor/kubernetes_test.go b/pkg/workceptor/kubernetes_test.go index e4435d04a..95805de42 100644 --- a/pkg/workceptor/kubernetes_test.go +++ b/pkg/workceptor/kubernetes_test.go @@ -35,13 +35,18 @@ import ( "k8s.io/client-go/tools/remotecommand" ) -func startNetceptorNodeWithWorkceptor() (*workceptor.KubeUnit, error) { +func startNetceptorNodeWithWorkceptor(rlogger *logger.ReceptorLogger) (*workceptor.KubeUnit, error) { kw := &workceptor.KubeUnit{ BaseWorkUnitForWorkUnit: &workceptor.BaseWorkUnit{}, } + if rlogger == nil { + rlogger = logger.NewReceptorLogger("") + } + // Create Netceptor node using external backends n1 := netceptor.New(context.Background(), "node1") + n1.Logger = rlogger b1, err := netceptor.NewExternalBackend() if err != nil { return kw, err @@ -65,7 +70,7 @@ func startNetceptorNodeWithWorkceptor() (*workceptor.KubeUnit, error) { func TestShouldUseReconnect(t *testing.T) { const envVariable string = "RECEPTOR_KUBE_SUPPORT_RECONNECT" - kw, err := startNetceptorNodeWithWorkceptor() + kw, err := startNetceptorNodeWithWorkceptor(nil) if err != nil { t.Fatal(err) } @@ -115,7 +120,7 @@ func TestShouldUseReconnect(t *testing.T) { func TestGetTimeoutOpenLogstream(t *testing.T) { const envVariable string = "RECEPTOR_OPEN_LOGSTREAM_TIMEOUT" - kw, err := startNetceptorNodeWithWorkceptor() + kw, err := startNetceptorNodeWithWorkceptor(nil) if err != nil { t.Fatal(err) } @@ -338,7 +343,7 @@ func Test_IsCompatibleK8S(t *testing.T) { versionStr string } - kw, err := startNetceptorNodeWithWorkceptor() + kw, err := startNetceptorNodeWithWorkceptor(nil) if err != nil { t.Fatal(err) } @@ -1242,7 +1247,7 @@ func TestParseTimeExtended(t *testing.T) { // TestIsCompatibleK8SExtended tests the IsCompatibleK8S function with more cases. func TestIsCompatibleK8SExtended(t *testing.T) { - kw, err := startNetceptorNodeWithWorkceptor() + kw, err := startNetceptorNodeWithWorkceptor(nil) if err != nil { t.Fatal(err) } @@ -1302,7 +1307,7 @@ func TestIsCompatibleK8SExtended(t *testing.T) { func TestGetTimeoutOpenLogstreamExtended(t *testing.T) { const envVariable string = "RECEPTOR_OPEN_LOGSTREAM_TIMEOUT" - kw, err := startNetceptorNodeWithWorkceptor() + kw, err := startNetceptorNodeWithWorkceptor(nil) if err != nil { t.Fatal(err) } @@ -1636,7 +1641,7 @@ func TestKubeUnitRestart(t *testing.T) { } func TestProcessLogLine(t *testing.T) { - kw, err := startNetceptorNodeWithWorkceptor() + kw, err := startNetceptorNodeWithWorkceptor(nil) if err != nil { t.Fatal(err) } diff --git a/pkg/workceptor/pod.go b/pkg/workceptor/pod.go new file mode 100644 index 000000000..fa8bb8106 --- /dev/null +++ b/pkg/workceptor/pod.go @@ -0,0 +1,126 @@ +package workceptor + +import ( + "context" + "fmt" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/watch" + "k8s.io/client-go/kubernetes" +) + +type KubePodStateHelper interface { + GetPodStatus(pod *corev1.Pod) (bool, error) + PodContainerHealthy(pod *corev1.Pod, containerName string) (bool, error) + PodHealthy(pod *corev1.Pod, containerName string) (bool, error) + WaitForPodCompleted(pod *corev1.Pod, clientset kubernetes.Interface, timeoutSeconds *int64) (*corev1.Pod, error) +} + +// 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, containerName) + if !ok || err != nil { + return ok, err + } + + return ok, nil +} + +// 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: + ok, err := kw.PodContainerHealthy(pod, containerName) + if !ok || err != nil { + return false, err + } + + return true, nil + 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) +} + +func (kw KubeUnit) WaitForPodCompleted(ctx context.Context, pod *corev1.Pod, clientset kubernetes.Interface, timeoutSeconds *int64) (*corev1.Pod, error) { + if pod == nil { + return nil, fmt.Errorf("pod is nil") + } + + originalPhase := pod.Status.Phase + + watcher, err := clientset.CoreV1().Pods(pod.Namespace).Watch(ctx, metav1.ListOptions{ + TimeoutSeconds: timeoutSeconds, + FieldSelector: "involvedObject.kind=Pod,involvedObject.name=" + pod.Name, + }) + if err != nil { + return pod, err + } + + for event := range watcher.ResultChan() { + switch event.Type { + case watch.Error: + kw.GetWorkceptor().nc.GetLogger().Debug("Pod %s/%s event %s phase %s (error)", pod.Namespace, pod.Name, event.Type, pod.Status.Phase) + + return pod, event.Object.(error) + default: + pod = event.Object.(*corev1.Pod) + if pod.Status.Phase != originalPhase { + kw.GetWorkceptor().nc.GetLogger().Debug("Pod %s/%s phase changed from %s to %s", pod.Namespace, pod.Name, originalPhase, pod.Status.Phase) + + return pod, nil + } + kw.GetWorkceptor().nc.GetLogger().Debug("Pod %s/%s event %s phase %s (no change)", pod.Namespace, pod.Name, event.Type, pod.Status.Phase) + + return pod, nil + } + } + + kw.GetWorkceptor().nc.GetLogger().Debug("Pod %s/%s phase %s timeout (no change)", pod.Namespace, pod.Name, pod.Status.Phase) + + return pod, nil +} diff --git a/pkg/workceptor/pod_test.go b/pkg/workceptor/pod_test.go new file mode 100644 index 000000000..61e08319f --- /dev/null +++ b/pkg/workceptor/pod_test.go @@ -0,0 +1,573 @@ +package workceptor_test + +import ( + "bytes" + "context" + "fmt" + "strings" + "sync" + "testing" + "time" + + "github.com/ansible/receptor/pkg/logger" + "github.com/stretchr/testify/assert" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/watch" + "k8s.io/client-go/kubernetes/fake" + testcore "k8s.io/client-go/testing" +) + +var podSuccess = &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: "default", + Name: "test-pod", + }, + Status: corev1.PodStatus{ + Phase: corev1.PodSucceeded, + ContainerStatuses: []corev1.ContainerStatus{ + { + Name: "worker", + State: corev1.ContainerState{ + Waiting: nil, + Running: nil, + Terminated: &corev1.ContainerStateTerminated{ + ExitCode: 0, + Reason: "Success", + }, + }, + }, + }, + }, +} + +var podInfraError = &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: "default", + Name: "infra-error-pod", + }, + Status: corev1.PodStatus{ + Phase: corev1.PodFailed, + Reason: "OOMKilled", + ContainerStatuses: []corev1.ContainerStatus{ + { + Name: "worker", + State: corev1.ContainerState{ + Terminated: &corev1.ContainerStateTerminated{ + ExitCode: 137, + Reason: "OOMKill", + }, + }, + }, + }, + }, +} + +var podAppError = &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: "default", + Name: "app-error-pod", + }, + Status: corev1.PodStatus{ + Phase: corev1.PodFailed, + Reason: "Error", + ContainerStatuses: []corev1.ContainerStatus{ + { + Name: "worker", + State: corev1.ContainerState{ + Terminated: &corev1.ContainerStateTerminated{ + ExitCode: 1, + Reason: "Error", + }, + }, + }, + }, + }, +} + +var podPending = &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: "default", + Name: "pending-pod", + }, + Status: corev1.PodStatus{ + Phase: corev1.PodPending, + ContainerStatuses: []corev1.ContainerStatus{ + { + Name: "worker", + State: corev1.ContainerState{ + Waiting: &corev1.ContainerStateWaiting{ + Reason: "ContainerCreating", + Message: "Container is being created", + }, + }, + }, + }, + }, +} + +var podUnknownPhase = &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: "default", + Name: "unknown-phase-pod", + }, + Status: corev1.PodStatus{ + Phase: "NotARealPhase", + ContainerStatuses: []corev1.ContainerStatus{ + { + Name: "worker", + State: corev1.ContainerState{ + Waiting: &corev1.ContainerStateWaiting{ + Reason: "ContainerCreating", + Message: "Container is being created", + }, + }, + }, + }, + }, +} + +func TestPodHeathy(t *testing.T) { + kw, err := startNetceptorNodeWithWorkceptor(nil) + if err != nil { + t.Fatal(err) + } + + tests := []struct { + name string + pod *corev1.Pod + container string + wantOk bool + wantErr bool + wantError string + }{ + { + name: "nil pod", + pod: nil, + container: "worker", + wantOk: false, + wantErr: true, + wantError: "pod is nil", + }, + { + name: "pod not terminated", + pod: podPending, + container: "worker", + wantOk: true, + wantErr: false, + }, + { + name: "container missing", + pod: &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "no-container-pod", Namespace: "default"}}, + container: "worker", + wantOk: false, + wantErr: true, + wantError: "pod does not contain container worker", + }, + { + name: "container healthy", + pod: podSuccess, + container: "worker", + wantOk: true, + wantErr: false, + }, + { + name: "pod unknown phase", + pod: podUnknownPhase, + container: "worker", + wantOk: false, + wantErr: true, + wantError: "unknown phase: NotARealPhase", + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ok, err := kw.PodHealthy(tt.pod, tt.container) + if ok != tt.wantOk || (err != nil) != tt.wantErr { + t.Errorf("Failed %s case: ok=%v wantok=%v err=%v", tt.name, ok, tt.wantOk, err) + } + if err != nil && tt.wantErr == false { + t.Errorf("Expected error message got '%s'", err.Error()) + } + if tt.wantErr { + if err == nil { + t.Errorf("Expected error message '%s', got nil error", tt.wantError) + } else if !strings.Contains(err.Error(), tt.wantError) { + t.Errorf("Expected error message '%s', got '%s'", tt.wantError, err.Error()) + } + } + if tt.wantError == "" && err != nil { + t.Errorf("Unexpected error for %s case: %v", tt.name, err) + } + }) + } +} + +func TestPodContainerHealthy(t *testing.T) { + kw, err := startNetceptorNodeWithWorkceptor(nil) + if err != nil { + t.Fatal(err) + } + + tests := []struct { + name string + pod *corev1.Pod + container string + wantOk bool + wantErr bool + wantError string + }{ + { + name: "nil pod", + pod: nil, + container: "worker", + wantOk: false, + wantErr: true, + wantError: "pod is nil", + }, + { + name: "pod not terminated", + pod: podPending, + container: "worker", + wantOk: true, + wantErr: false, + }, + { + name: "container missing", + pod: &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "no-container-pod", Namespace: "default"}}, + container: "worker", + wantOk: false, + wantErr: true, + wantError: "pod does not contain container worker", + }, + { + name: "container healthy", + pod: podSuccess, + container: "worker", + wantOk: true, + wantErr: false, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ok, err := kw.PodContainerHealthy(tt.pod, tt.container) + if ok != tt.wantOk || (err != nil) != tt.wantErr { + t.Errorf("Failed %s case: ok=%v wantok=%v err=%v", tt.name, ok, tt.wantOk, err) + } + if err != nil && tt.wantErr == false { + t.Errorf("Expected error message got '%s'", err.Error()) + } + if tt.wantErr { + if err == nil { + t.Errorf("Expected error message '%s', got nil error", tt.wantError) + } else if !strings.Contains(err.Error(), tt.wantError) { + t.Errorf("Expected error message '%s', got '%s'", tt.wantError, err.Error()) + } + } + if tt.wantError == "" && err != nil { + t.Errorf("Unexpected error for %s case: %v", tt.name, err) + } + }) + } +} + +func TestGetPodStatus(t *testing.T) { + kw, err := startNetceptorNodeWithWorkceptor(nil) + if err != nil { + t.Fatal(err) + } + + tests := []struct { + name string + pod *corev1.Pod + wantOk bool + wantErr bool + wantError string + }{ + { + name: "nil pod", + pod: nil, + wantOk: false, + wantErr: true, + wantError: "pod is nil", + }, + { + name: "infrastructure failure", + pod: podInfraError, + wantOk: false, + wantErr: true, + wantError: "container worker exited with code 137: OOMKill", + }, + { + name: "application failure", + pod: podAppError, + wantOk: false, + wantErr: true, + wantError: "container worker exited with code 1: Error", + }, + { + name: "success case", + pod: podSuccess, + wantOk: true, + wantErr: false, + }, + { + name: "pending pod", + pod: podPending, + wantOk: true, + wantErr: false, + }, + { + name: "pod with no container", + pod: &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "no-container-pod", Namespace: "default"}}, + wantOk: false, + wantErr: true, + wantError: "pod does not contain container worker", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ok, err := kw.GetPodStatus(tt.pod) + if ok != tt.wantOk || (err != nil) != tt.wantErr { + t.Errorf("Failed %s case: ok=%v wantok=%v err=%v", tt.name, ok, tt.wantOk, err) + } + if err != nil && tt.wantErr == false { + t.Errorf("Expected error message got '%s'", err.Error()) + } + if tt.wantErr { + if err == nil { + t.Errorf("Expected error message '%s', got nil error", tt.wantError) + } else if !strings.Contains(err.Error(), tt.wantError) { + t.Errorf("Expected error message '%s', got '%s'", tt.wantError, err.Error()) + } + } + if tt.wantError == "" && err != nil { + t.Errorf("Unexpected error for %s case: %v", tt.name, err) + } + }) + } +} + +func TestWaitForPodCompleted(t *testing.T) { + var logBuffer bytes.Buffer + logger.SetGlobalLogLevel(logger.DebugLevel) + testLogger := logger.NewReceptorLogger("") + testLogger.SetOutput(&logBuffer) + + kw, err := startNetceptorNodeWithWorkceptor(testLogger) + if err != nil { + t.Fatal(err) + } + + tests := []struct { + name string + initialPod *corev1.Pod + updatePhase corev1.PodPhase + wantErr bool + wantErrorString string + debugLine string + }{ + { + name: "nil pod", + initialPod: nil, + wantErr: true, + wantErrorString: "pod is nil", + }, + { + name: "pending to success", + initialPod: podPending, + updatePhase: corev1.PodSucceeded, + wantErr: false, + debugLine: "Pod default/pending-pod phase changed from Pending to Succeeded", + }, + { + name: "pending to failed", + initialPod: podPending, + updatePhase: corev1.PodFailed, + wantErr: false, + debugLine: "Pod default/pending-pod phase changed from Pending to Failed", + }, + { + name: "pending to running", + initialPod: podPending, + updatePhase: corev1.PodRunning, + wantErr: false, + debugLine: "Pod default/pending-pod phase changed from Pending to Running", + }, + { + name: "pending to pending", // This simulates a pod that remains pending despite an update. + initialPod: podPending, + updatePhase: corev1.PodPending, + wantErr: false, + debugLine: "Pod default/pending-pod event MODIFIED phase Pending (no change)", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctx := context.Background() + timeoutSeconds := int64(2) + logBuffer.Reset() + + var clientset *fake.Clientset + if tt.initialPod != nil { + clientset = fake.NewSimpleClientset(tt.initialPod) + } else { + clientset = fake.NewSimpleClientset(&corev1.Pod{}) + } + + if tt.updatePhase != "" && tt.initialPod != nil { + go func() { + updatedPod := tt.initialPod.DeepCopy() + updatedPod.Status.Phase = tt.updatePhase + _, _ = clientset.CoreV1().Pods("default").Update(context.Background(), updatedPod, metav1.UpdateOptions{}) + }() + } + + _, err := kw.WaitForPodCompleted(ctx, tt.initialPod, clientset, &timeoutSeconds) + if (err != nil) != tt.wantErr { + t.Errorf("WaitForPodCompleted() error = %v, wantErr %v", err, tt.wantErr) + if tt.wantErrorString != "" && err != nil && !strings.Contains(err.Error(), tt.wantErrorString) { + t.Errorf("Expected error message '%s', got '%s'", tt.wantErrorString, err.Error()) + } + } + if tt.wantErr && err != nil { + if !strings.Contains(err.Error(), tt.wantErrorString) { + t.Errorf("Expected error message '%s', got '%s'", tt.wantErrorString, err.Error()) + } + } + if tt.debugLine != "" { + logOutput := logBuffer.String() + if !strings.Contains(logOutput, tt.debugLine) { + t.Errorf("Expected debug log '%s', got '%s'", tt.debugLine, logOutput) + } + } else { + assert.NotContains(t, logBuffer.String(), "Pod diagnostics failed") + } + }) + } +} + +type statusErrorForTesting struct { + *metav1.Status +} + +func (s *statusErrorForTesting) Error() string { + return s.Message +} + +func TestWaitForPodCompleted_HandlesTimeoutAndErrorEvents(t *testing.T) { + // Create error event with custom type + errStatus := &statusErrorForTesting{ + Status: &metav1.Status{ + Status: metav1.StatusFailure, + Message: "simulated error for unit test", + Reason: metav1.StatusReasonUnknown, + Code: 500, + }, + } + + errorEvent := watch.Event{ + Type: watch.Error, + Object: errStatus, + } + + tests := []struct { + name string + pod *corev1.Pod + watchEvents []watch.Event + expectedDebugLogs []string + expectedError string + cancelContextAfter time.Duration + timeoutSeconds int64 + }{ + { + name: "Timeout", + pod: &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: "test-pod", Namespace: "default"}, + Status: corev1.PodStatus{Phase: corev1.PodPending}, + }, + watchEvents: []watch.Event{}, + expectedDebugLogs: []string{ + "Pod default/test-pod phase Pending timeout (no change)\n", + }, + timeoutSeconds: 1, + }, + { + name: "Error Event", + pod: &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: "test-pod", Namespace: "default"}, + Status: corev1.PodStatus{Phase: corev1.PodPending}, + }, + watchEvents: []watch.Event{ + errorEvent, + }, + expectedDebugLogs: []string{ + "Pod default/test-pod event ERROR phase Pending (error)", + }, + expectedError: "simulated error for unit test", + timeoutSeconds: 2, + cancelContextAfter: 0, // No cancellation needed for this test + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + // Use a buffer to capture log output + var logBuffer bytes.Buffer + logger.SetGlobalLogLevel(logger.DebugLevel) + testLogger := logger.NewReceptorLogger("") + testLogger.SetOutput(&logBuffer) + + // Create a KubeUnit with a mock logger + kw, err := startNetceptorNodeWithWorkceptor(testLogger) + if err != nil { + t.Fatal(err) + } + + clientset := fake.NewSimpleClientset(tt.pod) + watcher := watch.NewFake() + clientset.PrependWatchReactor("pods", testcore.DefaultWatchReactor(watcher, nil)) + + // Use a WaitGroup to ensure the goroutine is done + var wg sync.WaitGroup + wg.Add(1) + + go func() { + defer wg.Done() + for _, event := range tt.watchEvents { + watcher.Action(event.Type, event.Object) + } + // Close the watcher to terminate the loop in WaitForPodCompleted + watcher.Stop() + }() + + ctx, cancel := context.WithCancel(context.Background()) + if tt.cancelContextAfter > 0 { + go func() { + time.Sleep(tt.cancelContextAfter) + cancel() + }() + } else { + defer cancel() + } + timeout := tt.timeoutSeconds + _, err = kw.WaitForPodCompleted(ctx, tt.pod, clientset, &timeout) + + wg.Wait() + + if tt.expectedError != "" { + assert.EqualError(t, err, tt.expectedError) + } else { + assert.NoError(t, err) + } + + logOutput := logBuffer.String() + for _, expectedLog := range tt.expectedDebugLogs { + assert.Contains(t, logOutput, expectedLog) + } + fmt.Println(logOutput) + }) + } +}