func parseLabels(labels map[string]string) string {
	query := []string{}
	for k, v := range labels {
		query = append(query, k+"="+v)
	}
	return strings.Join(query, ",")
}
parseLabels parses label sets, and build query string.
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
}
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
	}
}
func isPendingPod(pod corev1.Pod) bool {
	if pod.Status.Phase == corev1.PodPending {
		return true
	}
	return false
}