AAP-47455 Wait for pod complete and GetPodStatus

.gitignore

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

.golangci.yml

@@ -99,6 +99,7 @@ linters-settings:
           - "k8s.io/apimachinery/pkg/selection"
           - "k8s.io/apimachinery/pkg/watch"
           - "k8s.io/client-go/kubernetes"
+          - "k8s.io/client-go/testing"
           - "k8s.io/client-go/rest"
           - "k8s.io/client-go/tools/remotecommand"
           - "github.com/quic-go/quic-go"

generate.go

@@ -22,4 +22,5 @@ package main
 //go:generate mockgen -source=pkg/workceptor/kubernetes.go -destination=pkg/workceptor/mock_workceptor/kubernetes.go
 //go:generate mockgen -source=pkg/workceptor/stdio_utils.go -destination=pkg/workceptor/mock_workceptor/stdio_utils.go
 //go:generate mockgen -source=pkg/workceptor/workceptor.go -destination=pkg/workceptor/mock_workceptor/workceptor.go
+//go:generate mockgen -source=pkg/workceptor/pod.go -destination=pkg/workceptor/mock_workceptor/pod.go
 //go:generate mockgen -package=mock_utils -destination=pkg/utils/mock_utils/io.go io ReadWriteCloser

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/kubernetes_test.go

@@ -64,6 +64,39 @@ func startNetceptorNodeWithWorkceptor() (*workceptor.KubeUnit, error) {
 	return kw, nil
 }
 
+func startNetceptorNodeWithWorkceptorWithLogger(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
+	}
+
+	err = n1.AddBackend(b1)
+	if err != nil {
+		return kw, err
+	}
+
+	w, err := workceptor.New(context.Background(), n1, "")
+	if err != nil {
+		return kw, err
+	}
+
+	kw.SetWorkceptor(w)
+
+	return kw, nil
+}
+
 func TestShouldUseReconnect(t *testing.T) {
 	const envVariable string = "RECEPTOR_KUBE_SUPPORT_RECONNECT"