Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
67 commits
Select commit Hold shift + click to select a range
55b3047
working WaitForPodCompleted and GetPodStatus
Jun 2, 2025
b27cda5
gitignore
Jun 6, 2025
a174a48
add test at eof time as well as when waitgroup completes
Jun 6, 2025
cb9a50b
remove bin
Jun 6, 2025
88414e8
working WaitForPodCompleted and GetPodStatus
Jun 2, 2025
64b7e79
gitignore
Jun 6, 2025
41a3d29
add test at eof time as well as when waitgroup completes
Jun 6, 2025
49afc98
remove bin
Jun 6, 2025
a6b1263
Merge branch 'capture-k8s-pod-diagnostics' of github.com:arrestle/rec…
Jun 10, 2025
2fb30e2
fix workunitbase.go by deleting and adding
Jun 10, 2025
362a739
add workunitbase.go
Jun 10, 2025
a5527cb
put mock back
Jun 10, 2025
f520d48
adding pod and container names
Jun 10, 2025
7c8926c
add pod reason and container name to failed pod containers
Jun 10, 2025
735f110
deactivate capturePodStatus until unit and integration tests are comp…
Jun 10, 2025
5dead7d
clean up error handling
Jun 11, 2025
d56c4c6
Merge branch 'ansible:devel' into capture-k8s-pod-diagnostics
Jun 11, 2025
3ecf827
lint
Jun 11, 2025
9b938c7
unit tests for CapturePodStatus
Jun 11, 2025
22939d4
unit tests
Jun 11, 2025
c528965
Update pkg/workceptor/kubernetes.go
Jun 12, 2025
c021f09
Fix test
Jun 12, 2025
bcbc220
cleanup linting
Jun 12, 2025
03d347a
pr comments
Jun 12, 2025
1f42903
lint
Jun 12, 2025
0f11125
feedback: only consider worker container, others tbd.
Jun 12, 2025
3a20984
Update pkg/workceptor/kubernetes.go
Jun 16, 2025
e0df53d
matoval review
Jun 16, 2025
c4db94b
merge to devel
Jun 17, 2025
30743b8
try to resolve merge conflict
Jun 17, 2025
9608c06
Merge branch 'devel' into capture-k8s-pod-diagnostics
Jun 17, 2025
bc6320c
pkg/
Jun 17, 2025
3f6de50
lint
Jun 17, 2025
0e37103
remove extraneous schema
Jun 18, 2025
f0c3698
Merge branch 'devel' into capture-k8s-pod-diagnostics
Jun 18, 2025
0db61e5
address review comments, add additional unit test
Jun 18, 2025
fd67a69
add unknown state
Jun 18, 2025
3a33387
Merge branch 'devel' into capture-k8s-pod-diagnostics
Jun 20, 2025
cc16392
Merge branch 'devel' into capture-k8s-pod-diagnostics
Jun 20, 2025
95e21e0
push todays changes
Jun 20, 2025
7424c77
Update watcher logic
lranjbar Jun 20, 2025
71c5e47
Fix GetPodStatus unit tests
lranjbar Jun 20, 2025
7002182
Fix WaitForPodCompleted tests
lranjbar Jun 20, 2025
677c2b3
Update test pods to be more realistic
lranjbar Jun 20, 2025
65604ab
Move new helper functions to new file pod.go
lranjbar Jun 23, 2025
c73562e
Merge branch 'devel' into AAP-46619-capture-k8s-pod-diagnostics
lranjbar Jun 23, 2025
568681c
Remove unused mocks in k8s.go
lranjbar Jun 23, 2025
78358bf
Merge branch 'devel' into capture-k8s-pod-diagnostics
lranjbar Jun 23, 2025
0c1f67d
Merge branch 'AAP-46619-capture-k8s-pod-diagnostics' into capture-k8s…
lranjbar Jun 23, 2025
1ad624c
Simplified pod status logic
lranjbar Jun 23, 2025
2fad163
revert accidental mock deletion
lranjbar Jun 23, 2025
6e02cf3
Fix some linting errors
lranjbar Jun 23, 2025
4ac5aca
Fix more linting errors
lranjbar Jun 24, 2025
6cb833e
Merge branch 'devel' into capture-k8s-pod-diagnostics
lranjbar Jun 24, 2025
b6c66a9
Fix more linting errors
lranjbar Jun 24, 2025
3f10768
Merge branch 'devel' into capture-k8s-pod-diagnostics
lranjbar Jun 27, 2025
5aaf190
Address review comments and fix broken unit test
lranjbar Jun 27, 2025
9c35093
add debug messages
Jun 27, 2025
3db8495
fix linting errors
lranjbar Jun 27, 2025
fd4a5c0
Merge branch 'AAP-46619-capture-k8s-pod-diagnostics' into capture-k8s…
lranjbar Jun 27, 2025
afca14c
fix linting errors
lranjbar Jun 27, 2025
0a7e38b
fix linting errors
lranjbar Jun 27, 2025
8bc7f36
fix linting errors
lranjbar Jun 27, 2025
e477eb8
add unit testing, add debug lines, add return for timeout
Jun 27, 2025
e5da583
Merge branch 'devel' into capture-k8s-pod-diagnostics
lranjbar Jun 30, 2025
bf86624
unit testing 95.7% remove CapturePodStatus for now
Jul 2, 2025
2bb31a5
lint
Jul 2, 2025
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -29,4 +29,5 @@ venv
/vendor
pkg/services/.lock
pkg/workceptor/status
pkg/workceptor/status.lock
pkg/workceptor/status.lock
**/__debug_*
Comment thread
arrestle marked this conversation as resolved.
1 change: 1 addition & 0 deletions .golangci.yml
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down
5 changes: 5 additions & 0 deletions go.mod
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
40 changes: 33 additions & 7 deletions pkg/workceptor/kubernetes.go
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@ import (
// KubeUnit implements the WorkUnit interface.
type KubeUnit struct {
BaseWorkUnitForWorkUnit
KubePodStateHelper
KubeAPIWrapperInstance KubeAPIer
authMethod string
streamMethod string
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -400,6 +403,15 @@ func (kw *KubeUnit) KubeLoggingWithReconnect(streamWait *sync.WaitGroup, stdout
podName,
)

// timeout := int64(10)

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Take out commented out code please,

The tests are enough for now

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@AaronH88 I have a conflict here because @lranjbar would like to keep the comment in the code so we know where it goes in the future.

// _, 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,
// )
// }

Comment thread
lranjbar marked this conversation as resolved.
return
}

Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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)
}
Expand Down Expand Up @@ -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)
}
Expand Down Expand Up @@ -732,7 +744,7 @@ func (kw *KubeUnit) runWorkUsingLogger() {

req.VersionedParams(
&corev1.PodExecOptions{
Container: "worker",
Container: containerName,
Stdin: true,
Stdout: false,
Stderr: false,
Expand Down Expand Up @@ -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{})

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

same as above
Take out commented out code please,

The tests are enough for now

// 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
// }
Comment thread
lranjbar marked this conversation as resolved.

// 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())
}
Expand Down
19 changes: 12 additions & 7 deletions pkg/workceptor/kubernetes_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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)
}
Expand Down Expand Up @@ -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)
}
Expand Down Expand Up @@ -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)
}
Expand Down Expand Up @@ -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)
}
Expand Down Expand Up @@ -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)
}
Expand Down Expand Up @@ -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)
}
Expand Down
126 changes: 126 additions & 0 deletions pkg/workceptor/pod.go
Original file line number Diff line number Diff line change
@@ -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
}
Loading
Loading