func (srv *Service) GetJob(ctx context.Context, req *v1.GetJobRequest) (resp *v1.GetJobResponse, err error) {
	job, err := srv.Jobs.Get(ctx, req.Name)
	if err != nil {
		return nil, status.Error(codes.Internal, err.Error())
	}
	if job == nil {
		return nil, status.Error(codes.NotFound, "not found")
	}
	return &v1.GetJobResponse{
		Result: job,
	}, nil
}
GetJob returns the information about a particular job
package werft
import (
	"bytes"
	"context"
	"fmt"
	"io"
	"io/ioutil"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"time"
	termtohtml "github.com/buildkite/terminal-to-html"
	"github.com/csweichel/werft/pkg/api/repoconfig"
	v1 "github.com/csweichel/werft/pkg/api/v1"
	"github.com/csweichel/werft/pkg/filterexpr"
	"github.com/csweichel/werft/pkg/logcutter"
	"github.com/csweichel/werft/pkg/store"
	"github.com/gogo/protobuf/proto"
	"github.com/golang/protobuf/ptypes"
	log "github.com/sirupsen/logrus"
	"github.com/technosophos/moniker"
	"golang.org/x/xerrors"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"gopkg.in/yaml.v3"
)
func (srv *Service) StartLocalJob(inc v1.WerftService_StartLocalJobServer) error {
	req, err := inc.Recv()
	if err != nil {
		return err
	}
	if req.GetMetadata() == nil {
		return status.Error(codes.InvalidArgument, "first request must contain metadata")
	}
	md := *req.GetMetadata()
	log.WithField("name", md).Debug("StartLocalJob - received metadata")
	dfs, err := ioutil.TempFile(os.TempDir(), "werft-lcp")
	if err != nil {
		return err
	}
	defer dfs.Close()
	defer os.Remove(dfs.Name())
	var (
		configYAML []byte
		jobYAML    []byte
		phase      int
	)
	const (
		phaseConfigYaml   = 0
		phaseJobYaml      = 1
		phaseWorkspaceTar = 2
	)
	for {
		req, err = inc.Recv()
		if err != nil {
			return err
		}
		if req.GetConfigYaml() != nil {
			if phase != phaseConfigYaml {
				return status.Error(codes.InvalidArgument, "expected config yaml")
			}
			configYAML = append(configYAML, req.GetConfigYaml()...)
			continue
		}
		if req.GetJobYaml() != nil {
			if phase == phaseConfigYaml {
				phase = phaseJobYaml
			}
			if phase != phaseJobYaml {
				return status.Error(codes.InvalidArgument, "expected job yaml")
			}
			jobYAML = append(jobYAML, req.GetJobYaml()...)
			continue
		}
		if req.GetWorkspaceTar() != nil {
			if phase == phaseJobYaml {
				phase = phaseWorkspaceTar
			}
			if phase != phaseWorkspaceTar {
				return status.Error(codes.InvalidArgument, "expected workspace tar")
			}
			data := req.GetWorkspaceTar()
			n, err := dfs.Write(data)
			if err != nil {
				return status.Error(codes.Internal, err.Error())
			}
			if n != len(data) {
				return status.Error(codes.Internal, io.ErrShortWrite.Error())
			}
		}
		if req.GetWorkspaceTarDone() {
			if phase != phaseWorkspaceTar {
				return status.Error(codes.InvalidArgument, "expected prior workspace tar")
			}
			break
		}
	}
	_, err = dfs.Seek(0, 0)
	if len(configYAML) == 0 && len(jobYAML) == 0 {
		return status.Error(codes.InvalidArgument, "either config or job YAML must not be empty")
	}
	cp := &LocalContentProvider{
		TarStream:  dfs,
		Namespace:  srv.Executor.Config.Namespace,
		Kubeconfig: srv.Executor.KubeConfig,
		Clientset:  srv.Executor.Client,
	}
	flatOwner := strings.ReplaceAll(strings.ToLower(md.Owner), " ", "")
	name := cleanupPodName(fmt.Sprintf("local-%s-%s", flatOwner, moniker.New().NameSep("-")))
	jobStatus, err := srv.RunJob(inc.Context(), name, md, cp, jobYAML, false, time.Time{})
	if err != nil {
		return status.Error(codes.Internal, err.Error())
	}
	log.WithField("status", jobStatus).Info(("started new local job"))
	return inc.SendAndClose(&v1.StartJobResponse{
		Status: jobStatus,
	})
}
func (srv *Service) StartGitHubJob(ctx context.Context, req *v1.StartGitHubJobRequest) (resp *v1.StartJobResponse, err error) {
	if req.GithubToken != "" {
		return nil, status.Errorf(codes.InvalidArgument, "Per-job GitHub tokens are no longer supported")
	}
	if req.Metadata.Repository.Host == "" {
		req.Metadata.Repository.Host = "github.com"
	}
	return srv.StartJob(ctx, &v1.StartJobRequest{
		JobPath:    req.JobPath,
		JobYaml:    req.JobYaml,
		Metadata:   req.Metadata,
		Sideload:   req.Sideload,
		WaitUntil:  req.WaitUntil,
		NameSuffix: req.NameSuffix,
	})
}
func (srv *Service) StartJob(ctx context.Context, req *v1.StartJobRequest) (resp *v1.StartJobResponse, err error) {
	log.WithField("req", proto.MarshalTextString(req)).Info("StartJob request")
	md := req.Metadata
	err = srv.RepositoryProvider.Resolve(ctx, md.Repository)
	if err != nil {
		return nil, status.Errorf(codes.Internal, "cannot resolve request: %q", err)
	}
	atns, err := srv.RepositoryProvider.RemoteAnnotations(ctx, md.Repository)
	if err != nil {
		return nil, err
	}
	for k, v := range atns {
		md.Annotations = append(md.Annotations, &v1.Annotation{
			Key:   k,
			Value: v,
		})
	}
	var cp ContentProvider
	cp, err = srv.RepositoryProvider.ContentProvider(ctx, md.Repository)
	if err != nil {
		return nil, status.Errorf(codes.Internal, "cannot produce content provider: %q", err)
	}
	if len(req.Sideload) > 0 {
		cp = &SideloadingContentProvider{
			Delegate: cp,
			TarStream:  bytes.NewReader(req.Sideload),
			Namespace:  srv.Executor.Config.Namespace,
			Kubeconfig: srv.Executor.KubeConfig,
			Clientset:  srv.Executor.Client,
		}
	}
	var fp FileProvider
	fp, err = srv.RepositoryProvider.FileProvider(ctx, md.Repository)
	if err != nil {
		return nil, status.Errorf(codes.Internal, "cannot produce file provider: %q", err)
	}
	var (
		jobYAML     = req.JobYaml
		tplpath     = req.JobPath
		jobSpecName = "custom"
	)
	if jobYAML == nil {
		if tplpath == "" {
			repoCfg, err := getRepoCfg(ctx, fp)
			if err != nil {
				return nil, status.Error(codes.Internal, err.Error())
			}
			tplpath = repoCfg.TemplatePath(req.Metadata)
		}
		if tplpath == "" {
			return nil, status.Errorf(codes.NotFound, "no jobspec found in repo config")
		}
		in, err := fp.Download(ctx, tplpath)
		if err != nil {
			return nil, status.Errorf(codes.Internal, "cannot download jobspec from %s: %s", tplpath, err.Error())
		}
		jobYAML, err = ioutil.ReadAll(in)
		in.Close()
		if err != nil {
			return nil, status.Errorf(codes.Internal, "cannot download jobspec from %s: %s", tplpath, err.Error())
		}
	}
	if tplpath != "" {
		jobSpecName = strings.TrimSpace(strings.TrimSuffix(filepath.Base(tplpath), filepath.Ext(tplpath)))
	}
	md.JobSpecName = jobSpecName
	refname := md.Repository.Ref
	refname = strings.TrimPrefix(refname, "refs/heads/")
	refname = strings.TrimPrefix(refname, "refs/tags/")
	refname = strings.ReplaceAll(refname, "/", "-")
	refname = strings.ReplaceAll(refname, "_", "-")
	refname = strings.ReplaceAll(refname, "@", "-")
	refname = strings.ToLower(refname)
	if refname == "" {
		refname = moniker.New().NameSep("-")
	}
	name := cleanupPodName(fmt.Sprintf("%s-%s-%s", md.Repository.Repo, jobSpecName, refname))
	if req.NameSuffix != "" {
		if len(req.NameSuffix) > 20 {
			return nil, status.Error(codes.InvalidArgument, "name suffix must be less than 20 characters")
		}
		name += "-" + req.NameSuffix
	}
	if refname != "" {
		t, err := srv.Groups.Next(name)
		if err != nil {
			return nil, status.Error(codes.Internal, err.Error())
		}
		name = fmt.Sprintf("%s.%d", name, t)
	}
	canReplay := len(req.Sideload) == 0
	var waitUntil time.Time
	if req.WaitUntil != nil {
		waitUntil, err = ptypes.Timestamp(req.WaitUntil)
		if err != nil {
			return nil, status.Errorf(codes.InvalidArgument, "waitUntil is invalid: %v", err)
		}
		if !canReplay {
			return nil, status.Error(codes.InvalidArgument, "cannot delay the execution of non-replayable jobs (i.e. jobs with custom GitHub token or sideload)")
		}
	}
	jobStatus, err := srv.RunJob(ctx, name, *md, cp, jobYAML, canReplay, waitUntil)
	if err != nil {
		return nil, status.Error(codes.Internal, err.Error())
	}
	log.WithField("status", jobStatus).Info(("started new GitHub job"))
	return &v1.StartJobResponse{
		Status: jobStatus,
	}, nil
}
func getRepoCfg(ctx context.Context, fp FileProvider) (*repoconfig.C, error) {
	werftYAML, err := fp.Download(ctx, PathWerftConfig)
	if err != nil {
		return nil, xerrors.Errorf("cannot get repo config: %w", err)
	}
	var repoCfg repoconfig.C
	err = yaml.NewDecoder(werftYAML).Decode(&repoCfg)
	if err != nil {
		return nil, xerrors.Errorf("cannot unmarshal repo config: %w", err)
	}
	return &repoCfg, nil
}
func cleanupPodName(name string) string {
	name = strings.TrimSpace(name)
	if len(name) == 0 {
		name = "unknown"
	}
	if len(name) > 58 {
		name = name[:58]
	}
	segs := strings.Split(name, ".")
	for i, n := range segs {
		s := strings.ToLower(n)[0]
		if !(('a' <= s && s <= 'z') || ('0' <= s && s <= '9')) {
			n = "a" + n[1:]
		}
		e := strings.ToLower(n)[len(n)-1]
		if !(('a' <= e && e <= 'z') || ('0' <= e && e <= '9')) {
			n = n[:len(n)-1] + "a"
		}
		segs[i] = n
	}
	name = strings.Join(segs, ".")
	return name
}
func (srv *Service) StartFromPreviousJob(ctx context.Context, req *v1.StartFromPreviousJobRequest) (*v1.StartJobResponse, error) {
	oldJobStatus, err := srv.Jobs.Get(ctx, req.PreviousJob)
	if err == store.ErrNotFound {
		return nil, status.Error(codes.NotFound, "job spec not found")
	}
	if err != nil {
		return nil, status.Error(codes.Internal, err.Error())
	}
	jobYAML, err := srv.Jobs.GetJobSpec(req.PreviousJob)
	if err == store.ErrNotFound {
		return nil, status.Error(codes.NotFound, "job spec not found")
	}
	if err != nil {
		return nil, status.Error(codes.Internal, err.Error())
	}
	name := req.PreviousJob
	if strings.Contains(name, ".") {
		segs := strings.Split(name, ".")
		name = strings.Join(segs[0:len(segs)-1], ".")
	}
	nr, err := srv.Groups.Next(name)
	if err != nil {
		return nil, status.Error(codes.Internal, err.Error())
	}
	name = fmt.Sprintf("%s.%d", name, nr)
	md := oldJobStatus.Metadata
	md.Finished = nil
	cp, err := srv.RepositoryProvider.ContentProvider(ctx, md.Repository)
	if err != nil {
		return nil, status.Error(codes.Internal, err.Error())
	}
	canReplay := true
	var waitUntil time.Time
	if req.WaitUntil != nil {
		waitUntil, err = ptypes.Timestamp(req.WaitUntil)
		if err != nil {
			return nil, status.Errorf(codes.InvalidArgument, "waitUntil is invalid: %v", err)
		}
	}
	jobStatus, err := srv.RunJob(ctx, name, *oldJobStatus.Metadata, cp, jobYAML, canReplay, waitUntil)
	if err != nil {
		return nil, status.Error(codes.Internal, err.Error())
	}
	log.WithField("name", req.PreviousJob).WithField("old-name", name).Info(("started new job from an old one"))
	return &v1.StartJobResponse{
		Status: jobStatus,
	}, nil
}
func newTarStreamAdapter(inc v1.WerftService_StartLocalJobServer, initial []byte) io.Reader {
	return &tarStreamAdapter{
		inc:       inc,
		remainder: initial,
	}
}
type tarStreamAdapter struct {
	inc       v1.WerftService_StartLocalJobServer
	remainder []byte
}
func (tsa *tarStreamAdapter) Read(p []byte) (n int, err error) {
	if len(tsa.remainder) == 0 {
		var msg *v1.StartLocalJobRequest
		msg, err = tsa.inc.Recv()
		if err != nil {
			return 0, err
		}
		data := msg.GetWorkspaceTar()
		if data == nil {
			log.Debug("tar upload done")
			return 0, io.EOF
		}
		n = copy(p, data)
		tsa.remainder = data[n:]
		return
	}
	n = copy(p, tsa.remainder)
	tsa.remainder = tsa.remainder[n:]
	return n, nil
}
func (srv *Service) ListJobs(ctx context.Context, req *v1.ListJobsRequest) (resp *v1.ListJobsResponse, err error) {
	result, total, err := srv.Jobs.Find(ctx, req.Filter, req.Order, int(req.Start), int(req.Limit))
	if err != nil {
		return nil, status.Error(codes.Internal, err.Error())
	}
	res := make([]*v1.JobStatus, len(result))
	for i := range result {
		res[i] = &result[i]
	}
	return &v1.ListJobsResponse{
		Total:  int32(total),
		Result: res,
	}, nil
}
func (srv *Service) Subscribe(req *v1.SubscribeRequest, resp v1.WerftService_SubscribeServer) (err error) {
	evts := srv.events.On("job")
	for evt := range evts {
		job := evt.Args[0].(*v1.JobStatus)
		if !filterexpr.MatchesFilter(job, req.Filter) {
			continue
		}
		resp.Send(&v1.SubscribeResponse{
			Result: job,
		})
	}
	return nil
}