fix: pass errors in case of pod failure

.github/workflows/test.yml

@@ -2,34 +2,36 @@ name: Code build and checks
 
 on:
   push:
-    branches: [ main ]
+    branches: [main]
   pull_request:
-    branches: [ main ]
+    branches: [main]
 
 jobs:
-
   build:
     runs-on: ubuntu-latest
     services:
       mongo:
         image: bitnami/mongodb
         ports:
           - 27017:27017
-          
+
     steps:
-    - uses: actions/checkout@v2
+      - uses: actions/checkout@v2
+
+      - name: Set up Go
+        uses: actions/setup-go@v2
+        with:
+          go-version: 1.17
 
-    - name: Set up Go
-      uses: actions/setup-go@v2
-      with:
-        go-version: 1.17
+      - name: Unit test
+        run: go test -v ./...
 
-    - name: Unit test
-      run: go test -v ./...
+      - name: Integration tests
+        run: go test --tags=integration -v ./...
 
-    # Don't work yet as expected https://github.com/nwestfall/openapi-action/issues/3
-    - name: OpenAPI Lint Checks
-      uses: nwestfall/[email protected]
-      with:
-        github_token: ${{ secrets.GITHUB_TOKEN }}
-        file: api/v1/testkube.yaml
+      # Don't work yet as expected https://github.com/nwestfall/openapi-action/issues/3
+      - name: OpenAPI Lint Checks
+        uses: nwestfall/[email protected]
+        with:
+          github_token: ${{ secrets.GITHUB_TOKEN }}
+          file: api/v1/testkube.yaml

Makefile

@@ -56,11 +56,15 @@ openapi-generate-model-testkube:
 
 
 test: 
-	go test ./... -cover
+	go test ./... -cover -v 
 
 test-e2e:
 	go test --tags=e2e -v ./test/e2e
 
+test-integration:
+	go test --tags=integration -v ./...
+
+
 test-e2e-namespace:
 	NAMESPACE=$(NAMESPACE) go test --tags=e2e -v  ./test/e2e 
 

cmd/kubectl-testkube/commands/scripts/common.go

@@ -49,7 +49,7 @@ func watchLogs(id string, client client.Client) {
 	for l := range logs {
 		switch l.Type_ {
 		case output.TypeError:
-			ui.Warn(l.Content)
+			ui.Errf(l.Content)
 		case output.TypeResult:
 			ui.Info("Execution completed", l.Result.Output)
 		default:

internal/pkg/api/repository/result/mongo_test.go

@@ -1,3 +1,5 @@
+//go:build integration
+
 package result
 
 import (

pkg/executor/client/job.go

@@ -66,24 +66,26 @@ func (c JobExecutor) Get(id string) (execution testkube.ExecutionResult, err err
 	return *exec.ExecutionResult, nil
 }
 
-// Logs returns job logs
-// TODO too many goroutines - need to be simplified
+// Logs returns job logs using kubernetes api
 func (c JobExecutor) Logs(id string) (out chan output.Output, err error) {
 	out = make(chan output.Output)
 	logs := make(chan []byte)
 
-	if err := c.Client.TailJobLogs(id, logs); err != nil {
-		return out, err
-	}
-
 	go func() {
 		defer func() {
 			c.Log.Debug("closing JobExecutor.Logs out log")
 			close(out)
 		}()
+
+		if err := c.Client.TailJobLogs(id, logs); err != nil {
+			out <- output.NewOutputError(err)
+			return
+		}
+
 		for l := range logs {
 			entry, err := output.GetLogEntry(l)
 			if err != nil {
+				out <- output.NewOutputError(err)
 				return
 			}
 			out <- entry

pkg/jobs/jobclient.go

@@ -18,6 +18,7 @@ import (
 	"github.com/kubeshop/testkube/pkg/secret"
 	"go.uber.org/zap"
 	batchv1 "k8s.io/api/batch/v1"
+	corev1 "k8s.io/api/core/v1"
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -214,28 +215,44 @@ func (c *JobClient) GetJobPods(podsClient pods.PodInterface, jobName string, ret
 
 // TailJobLogs - locates logs for job pod(s)
 func (c *JobClient) TailJobLogs(id string, logs chan []byte) (err error) {
+
+	const pollTimeout = 24 * time.Hour
+	const pollInterval = 200 * time.Millisecond
 	podsClient := c.ClientSet.CoreV1().Pods(c.Namespace)
 	ctx := context.Background()
-	pods, err := c.GetJobPods(podsClient, id, 1, 10)
 
+	pods, err := c.GetJobPods(podsClient, id, 1, 10)
 	if err != nil {
 		close(logs)
 		return err
 	}
 
 	for _, pod := range pods.Items {
 		if pod.Labels["job-name"] == id {
-			if pod.Status.Phase != v1.PodRunning {
-				c.Log.Debugw("Waiting for pod to be ready", "pod", pod.Name)
-				if err = wait.PollImmediate(100*time.Millisecond, time.Duration(0)*time.Second, k8sclient.IsPodReady(c.ClientSet, pod.Name, c.Namespace)); err != nil {
-					c.Log.Errorw("poll immediate error when tailing logs", "error", err)
+
+			l := c.Log.With("namespace", pod.Namespace, "pod", pod.Name)
+
+			switch pod.Status.Phase {
+
+			case v1.PodRunning:
+				l.Debug("Tailing pod logs immediately")
+				return c.TailPodLogs(ctx, pod.Name, logs)
+
+			case v1.PodFailed:
+				err := fmt.Errorf("can't get pod logs, pod failed: %s/%s", pod.Namespace, pod.Name)
+				l.Errorw(err.Error())
+				return err
+
+			default:
+				l.Debugw("Waiting for pod to be ready")
+				if err = wait.PollImmediate(pollInterval, pollTimeout, IsPodReady(c.ClientSet, pod.Name, c.Namespace)); err != nil {
+					l.Errorw("poll immediate error when tailing logs", "error", err)
 					close(logs)
 					return err
 				}
-				c.Log.Debug("Tailing pod logs")
-				return c.TailPodLogs(ctx, pod.Name, logs)
-			} else if pod.Status.Phase == v1.PodRunning {
+				l.Debug("Tailing pod logs")
 				return c.TailPodLogs(ctx, pod.Name, logs)
+
 			}
 		}
 	}
@@ -472,3 +489,21 @@ var envVars = []v1.EnvVar{
 		Value: os.Getenv("SCRAPPERENABLED"),
 	},
 }
+
+// IsPodReady defines if pod is ready or failed for logs scrapping
+func IsPodReady(c *kubernetes.Clientset, podName, namespace string) wait.ConditionFunc {
+	return func() (bool, error) {
+		pod, err := c.CoreV1().Pods(namespace).Get(context.Background(), podName, metav1.GetOptions{})
+		if err != nil {
+			return false, err
+		}
+
+		switch pod.Status.Phase {
+		case corev1.PodSucceeded:
+			return true, nil
+		case corev1.PodFailed:
+			return true, fmt.Errorf("pod %s/%s failed", pod.Namespace, pod.Name)
+		}
+		return false, nil
+	}
+}

test/e2e/e2e_test.go

@@ -1,5 +1,4 @@
 //go:build e2e
-// +build e2e
 
 package main