func (w *Watcher) WaitToStartPod(ctx context.Context, pod corev1.Pod) (corev1.Pod, error) {
retry:
	for {
		targetPod, err := w.client.CoreV1().Pods(pod.Namespace).Get(ctx, pod.Name, metav1.GetOptions{})
		if err != nil {
			return pod, err
		}
		if !isPendingPod(*targetPod) {
			return *targetPod, nil
		}
		time.Sleep(1 * time.Second)
		continue retry
	}
}
WaitToStartPod wait until starting the pod.
Because the job does not start immediately after call kubernetes API.
So we have to wait to start the pod, before watch logs.
package job
import (
	"context"
	"io"
	"os"
	"strconv"
	"strings"
	"sync"
	"time"
	log "github.com/sirupsen/logrus"
	v1 "k8s.io/api/batch/v1"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/kubernetes"
	restclient "k8s.io/client-go/rest"
)
type Watcher struct {
	client kubernetes.Interface
	Container string
}
func NewWatcher(client kubernetes.Interface, container string) *Watcher {
	return &Watcher{
		client,
		container,
	}
}
func (w *Watcher) Watch(job *v1.Job, ctx context.Context) error {
	currentPodList := []corev1.Pod{}
retry:
	for {
		newPodList, err := w.FindPods(ctx, job)
		if err != nil {
			return err
		}
		incrementalPodList := diffPods(currentPodList, newPodList)
		go w.WatchPods(ctx, incrementalPodList)
		time.Sleep(1 * time.Second)
		currentPodList = newPodList
		continue retry
	}
}
func (w *Watcher) WatchPods(ctx context.Context, pods []corev1.Pod) error {
	var wg sync.WaitGroup
	errCh := make(chan error, len(pods))
	for _, pod := range pods {
		wg.Add(1)
		go func(p corev1.Pod) {
			defer wg.Done()
			startedPod, err := w.WaitToStartPod(ctx, p)
			if err != nil {
				errCh <- err
				return
			}
			logOptions := corev1.PodLogOptions{
				Container: w.Container,
				Follow:    true,
			}
			request := w.client.CoreV1().Pods(startedPod.Namespace).GetLogs(startedPod.Name, &logOptions).
				Param("follow", strconv.FormatBool(true)).
				Param("container", w.Container).
				Param("timestamps", strconv.FormatBool(false))
			err = readStreamLog(ctx, request, startedPod)
			errCh <- err
		}(pod)
	}
	select {
	case err := <-errCh:
		if err != nil {
			log.Error(err)
			return err
		}
	}
	wg.Wait()
	return nil
}
func (w *Watcher) FindPods(ctx context.Context, job *v1.Job) ([]corev1.Pod, error) {
	labels := parseLabels(job.Spec.Template.Labels)
	listOptions := metav1.ListOptions{
		LabelSelector: labels,
	}
	podList, err := w.client.CoreV1().Pods(job.Namespace).List(ctx, listOptions)
	if err != nil {
		return []corev1.Pod{}, err
	}
	return podList.Items, err
}