fix: resolve issue where skaffold logger could hang indefinitely if k8s job pod wasn't created (#8717)

Author: aaron-prindle

pkg/skaffold/k8sjob/logger/log.go

@@ -18,6 +18,7 @@ package logger
 
 import (
 	"context"
+	"fmt"
 	"io"
 	"sync"
 	"sync/atomic"
@@ -154,7 +155,6 @@ func (l *Logger) streamLogsFromKubernetesJob(ctx context.Context, id, namespace
 					return true, nil
 				}
 			}
-
 			var podName string
 			w, err := clientset.CoreV1().Pods(namespace).Watch(context.TODO(),
 				metav1.ListOptions{
@@ -164,13 +164,25 @@ func (l *Logger) streamLogsFromKubernetesJob(ctx context.Context, id, namespace
 				return false, nil
 			}
 
-			for event := range w.ResultChan() {
-				pod, ok := event.Object.(*corev1.Pod)
-				if ok {
-					podName = pod.Name
-					// TODO(aaron-prindle) add support for jobs w/ multiple pods in the future
-					break
+			done := make(chan bool)
+			go func() {
+				for event := range w.ResultChan() {
+					pod, ok := event.Object.(*corev1.Pod)
+					if ok {
+						podName = pod.Name
+						done <- true
+						break
+					}
 				}
+			}()
+
+			select {
+			case <-ctx.Done():
+				return false, fmt.Errorf("context cancelled for k8s job logging of pod of kubernetes job: %s", "id")
+			case <-done:
+				// Continue
+			case <-time.After(30 * time.Second): // Timeout after 30 seconds
+				return false, fmt.Errorf("timeout waiting for event from pod of kubernetes job: %s", id)
 			}
 
 			podLogOptions := &corev1.PodLogOptions{

pkg/skaffold/verify/k8sjob/verify.go

@@ -222,7 +222,7 @@ func (v *Verifier) createAndRunJob(ctx context.Context, tc latest.VerifyTestCase
 	// This retrying is added as when attempting to kickoff multiple jobs simultaneously
 	// This is because the k8s API server can be unresponsive when hit with a large
 	// intitial set of Job CREATE requests
-	if waitErr := wait.Poll(100*time.Millisecond, 10*time.Second, func() (bool, error) {
+	if waitErr := wait.Poll(100*time.Millisecond, 30*time.Second, func() (bool, error) {
 		_, err = clientset.BatchV1().Jobs(job.Namespace).Create(ctx, job, metav1.CreateOptions{})
 		if err != nil {
 			return false, nil