func (js *Executor) Logs(name string) io.Reader {
	return listenToLogs(js.Client, name, js.Config.Namespace, js.labels)
}
Logs provides the log output of a running job. If the job is unknown, nil is returned.
package executor
import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"os"
	"strings"
	"sync"
	"time"
	werftv1 "github.com/csweichel/werft/pkg/api/v1"
	"github.com/gogo/protobuf/jsonpb"
	"github.com/golang/protobuf/ptypes"
	log "github.com/sirupsen/logrus"
	"github.com/technosophos/moniker"
	"golang.org/x/xerrors"
	corev1 "k8s.io/api/core/v1"
	k8serr "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/watch"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/util/retry"
)
type Config struct {
	Namespace       string    `yaml:"namespace"`
	EventTraceLog   string    `yaml:"eventTraceLog,omitempty"`
	JobPrepTimeout  *Duration `yaml:"preperationTimeout"`
	JobTotalTimeout *Duration `yaml:"totalTimeout"`
	LabelPrefix     string    `json:"labelPrefix"`
}
type Duration struct {
	time.Duration
}
func (d *Duration) UnmarshalYAML(unmarshal func(interface{}) error) error {
	var v string
	err := unmarshal(&v)
	if err != nil {
		return err
	}
	d.Duration, err = time.ParseDuration(v)
	if err != nil {
		return err
	}
	return nil
}
func NewExecutor(config Config, kubeConfig *rest.Config) (*Executor, error) {
	kubeClient, err := kubernetes.NewForConfig(kubeConfig)
	if err != nil {
		return nil, err
	}
	if config.JobPrepTimeout == nil {
		return nil, xerrors.Errorf("job preperation timeout is required")
	}
	if config.JobTotalTimeout == nil {
		return nil, xerrors.Errorf("total job timeout is required")
	}
	if config.JobTotalTimeout.Duration < config.JobPrepTimeout.Duration {
		return nil, xerrors.Errorf("total job timeout must be greater than the preparation timeout")
	}
	return &Executor{
		OnUpdate: func(pod *corev1.Pod, status *werftv1.JobStatus) {},
		Config:     config,
		Client:     kubeClient,
		KubeConfig: kubeConfig,
		labels:      newLabelSetet(config.LabelPrefix),
		waitingJobs: make(map[string]*waitingJob),
	}, nil
}
type Executor struct {
	OnUpdate func(pod *corev1.Pod, status *werftv1.JobStatus)
	Client     kubernetes.Interface
	Config     Config
	KubeConfig *rest.Config
	labels      labelSet
	waitingJobs map[string]*waitingJob
	mu          sync.RWMutex
}
type waitingJob struct {
	Cancel func(reason string)
	Start  func()
	Mutex  string
	Status *werftv1.JobStatus
}
func (js *Executor) Run() {
	go js.monitorJobs()
	go js.doHousekeeping()
}
type startOptions struct {
	JobName      string
	Modifier     []func(*corev1.Pod)
	Annotations  map[string]string
	BackoffLimit int
	Mutex        string
	CanReplay    bool
	WaitUntil    time.Time
	Sidecars     []string
}
type StartOpt func(*startOptions)
func WithBackoff(limit int) StartOpt {
	return func(opts *startOptions) {
		opts.Modifier = append(opts.Modifier, func(j *corev1.Pod) {
			opts.BackoffLimit = limit
		})
	}
}
func WithAnnotation(key, value string) StartOpt {
	return func(opts *startOptions) {
		if opts.Annotations == nil {
			opts.Annotations = make(map[string]string)
		}
		opts.Annotations[key] = value
	}
}
func WithAnnotations(annotations map[string]string) StartOpt {
	return func(opts *startOptions) {
		opts.Annotations = annotations
	}
}
func WithName(name string) StartOpt {
	return func(opts *startOptions) {
		opts.JobName = name
	}
}
func WithMutex(name string) StartOpt {
	return func(opts *startOptions) {
		opts.Mutex = name
	}
}
func WithCanReplay(canReplay bool) StartOpt {
	return func(opts *startOptions) {
		opts.CanReplay = canReplay
	}
}
func WithWaitUntil(t time.Time) StartOpt {
	return func(opts *startOptions) {
		opts.WaitUntil = t
	}
}
func WithSidecars(names []string) StartOpt {
	return func(opts *startOptions) {
		opts.Sidecars = names
	}
}
func (js *Executor) Start(podspec corev1.PodSpec, metadata werftv1.JobMetadata, options ...StartOpt) (status *werftv1.JobStatus, err error) {
	opts := startOptions{
		JobName: fmt.Sprintf("werft-%s", strings.ReplaceAll(moniker.New().Name(), " ", "-")),
	}
	for _, opt := range options {
		opt(&opts)
	}
	annotations := make(map[string]string)
	for key, val := range opts.Annotations {
		annotations[fmt.Sprintf("%s/%s", js.labels.UserDataAnnotationPrefix, key)] = val
	}
	if opts.CanReplay {
		annotations[js.labels.AnnotationCanReplay] = "true"
	}
	if !opts.WaitUntil.IsZero() {
		annotations[js.labels.AnnotationWaitUntil] = opts.WaitUntil.Format(time.RFC3339)
	}
	if opts.BackoffLimit > 0 {
		annotations[js.labels.AnnotationFailureLimit] = fmt.Sprintf("%d", opts.BackoffLimit)
	}
	if len(opts.Sidecars) > 0 {
		annotations[js.labels.AnnotationSidecars] = strings.Join(opts.Sidecars, " ")
	}
	metadata.Created = ptypes.TimestampNow()
	mdjson, err := (&jsonpb.Marshaler{
		EnumsAsInts: true,
	}).MarshalToString(&metadata)
	if err != nil {
		return nil, xerrors.Errorf("cannot marshal metadata: %w", err)
	}
	annotations[js.labels.AnnotationMetadata] = mdjson
	if podspec.RestartPolicy != corev1.RestartPolicyNever && podspec.RestartPolicy != corev1.RestartPolicyOnFailure {
		podspec.RestartPolicy = corev1.RestartPolicyOnFailure
	}
	meta := metav1.ObjectMeta{
		Name: opts.JobName,
		Labels: map[string]string{
			js.labels.LabelWerftMarker: "true",
			js.labels.LabelJobName:     opts.JobName,
		},
		Annotations: annotations,
	}
	poddesc := corev1.Pod{
		ObjectMeta: meta,
		Spec:       podspec,
	}
	for _, opt := range opts.Modifier {
		opt(&poddesc)
	}
	mutexCancelationMsg := fmt.Sprintf("a newer job (%s) with the same mutex (%s) started", opts.JobName, opts.Mutex)
	if opts.Mutex != "" {
		labelMutex := js.labels.LabelMutex
		poddesc.ObjectMeta.Labels[labelMutex] = opts.Mutex
		pods, err := js.Client.CoreV1().Pods(js.Config.Namespace).List(context.Background(), metav1.ListOptions{LabelSelector: fmt.Sprintf("%s=%s", labelMutex, opts.Mutex)})
		if err != nil {
			return nil, xerrors.Errorf("cannot enforce mutex: %w", err)
		}
		for _, pod := range pods.Items {
			err := js.addAnnotation(pod.Name, map[string]string{
				js.labels.AnnotationFailed: mutexCancelationMsg,
			})
			if err, ok := err.(*k8serr.StatusError); ok && err.ErrStatus.Code == http.StatusNotFound {
				continue
			}
			if err != nil {
				return nil, xerrors.Errorf("cannot enforce mutex: %w", err)
			}
		}
		js.mu.Lock()
		for k, wj := range js.waitingJobs {
			if wj.Mutex == opts.Mutex {
				wj.Cancel(mutexCancelationMsg)
				delete(js.waitingJobs, k)
			}
		}
		js.mu.Unlock()
	}
	startJob := func() (*werftv1.JobStatus, error) {
		if log.GetLevel() == log.DebugLevel {
			dbg, _ := json.MarshalIndent(poddesc, "", "  ")
			log.Debugf("scheduling job\n%s", dbg)
		}
		job, err := js.Client.CoreV1().Pods(js.Config.Namespace).Create(context.Background(), &poddesc, metav1.CreateOptions{})
		if err != nil {
			return nil, err
		}
		return getStatus(job, js.labels)
	}
	log.WithField("wait-until", opts.WaitUntil).Debug("waiting until")
	if !opts.WaitUntil.IsZero() && opts.WaitUntil.After(time.Now()) {
		status, err := getStatus(&poddesc, js.labels)
		if err != nil {
			return nil, err
		}
		startChan, cancelChan := make(chan struct{}), make(chan string)
		js.mu.Lock()
		js.waitingJobs[opts.JobName] = &waitingJob{
			Cancel: func(reason string) { cancelChan <- reason },
			Start:  func() { close(startChan) },
			Mutex:  opts.Mutex,
			Status: status,
		}
		js.mu.Unlock()
		run := func() {
			js.mu.Lock()
			delete(js.waitingJobs, opts.JobName)
			js.mu.Unlock()
			startJob()
		}
		go func() {
			d := opts.WaitUntil.Sub(time.Now())
			select {
			case <-time.After(d):
				run()
			case <-startChan:
				run()
			case reason := <-cancelChan:
				log.WithField("name", opts.JobName).Debug("canceled this waiting job")
				status.Phase = werftv1.JobPhase_PHASE_DONE
				status.Conditions.Success = false
				status.Details = reason
				js.OnUpdate(&poddesc, status)
			}
		}()
		status.Phase = werftv1.JobPhase_PHASE_WAITING
		js.OnUpdate(&poddesc, status)
		return status, nil
	}
	return startJob()
}
func (js *Executor) monitorJobs() {
	reconnectionTimeout := 500 * time.Millisecond
	for {
		incoming, err := js.Client.CoreV1().Pods(js.Config.Namespace).Watch(context.Background(), metav1.ListOptions{
			LabelSelector: fmt.Sprintf("%s=true", js.labels.LabelWerftMarker),
		})
		if err != nil {
			log.WithError(err).Error("cannot watch jobs - retrying")
			time.Sleep(reconnectionTimeout)
			continue
		}
		log.Info("connected to Kubernetes master")
		for evt := range incoming.ResultChan() {
			if evt.Object == nil {
				break
			}
			obj, ok := evt.Object.(*corev1.Pod)
			if !ok {
				continue
			}
			js.handleJobEvent(evt.Type, obj)
		}
		log.Warn("lost connection to Kubernetes master")
		time.Sleep(reconnectionTimeout)
	}
}
func (js *Executor) handleJobEvent(evttpe watch.EventType, obj *corev1.Pod) {
	status, err := getStatus(obj, js.labels)
	js.writeEventTraceLog(status, obj)
	if err != nil {
		log.WithError(err).WithField("name", obj.Name).Error("cannot compute status")
		return
	}
	js.OnUpdate(obj, status)
	err = js.actOnUpdate(status, obj)
	if err != nil {
		log.WithError(err).WithField("name", obj.Name).Error("cannot act on status update")
		return
	}
}
func (js *Executor) actOnUpdate(status *werftv1.JobStatus, obj *corev1.Pod) error {
	if status.Phase == werftv1.JobPhase_PHASE_DONE {
		gracePeriod := int64(5)
		policy := metav1.DeletePropagationForeground
		err := js.Client.CoreV1().Pods(js.Config.Namespace).Delete(context.Background(), obj.Name, metav1.DeleteOptions{
			GracePeriodSeconds: &gracePeriod,
			PropagationPolicy:  &policy,
		})
		if err != nil {
			log.WithError(err).WithField("name", obj.Name).Error("cannot delete job pod")
		}
		return nil
	}
	return nil
}
func (js *Executor) writeEventTraceLog(status *werftv1.JobStatus, obj *corev1.Pod) {
	defer recover()
	if js.Config.EventTraceLog == "" {
		return
	}
	var out io.Writer
	if js.Config.EventTraceLog == "-" {
		out = os.Stdout
	} else {
		f, err := os.OpenFile(js.Config.EventTraceLog, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
		if err != nil {
			return
		}
		defer f.Close()
		out = f
	}
	type eventTraceEntry struct {
		Time   string             `yaml:"time"`
		Status *werftv1.JobStatus `yaml:"status"`
		Job    *corev1.Pod        `yaml:"job"`
	}
	json.NewEncoder(out).Encode(eventTraceEntry{Time: time.Now().Format(time.RFC3339), Status: status, Job: obj})
}