gitdeploy/internal/jobs/job.go

package jobs

import (
	"encoding/base64"
	"encoding/json"
	"errors"
	"io"
	"io/ioutil"
	"os"
	"os/exec"
	"path/filepath"
	"strings"
	"sync"
	"time"

	"git.rootprojects.org/root/gitdeploy/internal/log"
	"git.rootprojects.org/root/gitdeploy/internal/options"
	"git.rootprojects.org/root/gitdeploy/internal/webhooks"
)

// Pending is the map of backlog jobs
// map[webhooks.RefID]*webhooks.GitRef
var Pending sync.Map

// Actives is the map of jobs
// map[webhooks.RefID]*Job
var Actives sync.Map

// Recents are jobs that are dead, but recent
// map[webhooks.RevID]*Job
var Recents sync.Map

// Job represents a job started by the git webhook
// and also the JSON we send back through the API about jobs
type Job struct {
	// normal json
	StartedAt *time.Time    `json:"started_at,omitempty"` // empty when pending
	ID        string        `json:"id"`                   // could be URLSafeRefID or URLSafeRevID
	GitRef    *webhooks.Ref `json:"ref"`                  // always present
	PromoteTo string        `json:"promote_to,omitempty"` // empty when deploy and test
	Promote   bool          `json:"promote,omitempty"`    // empty when deploy and test
	EndedAt   *time.Time    `json:"ended_at,omitempty"`   // empty when running
	ExitCode  *int          `json:"exit_code,omitempty"`  // empty when running
	// full json
	Logs   []Log   `json:"logs,omitempty"`   // exist when requested
	Report *Result `json:"report,omitempty"` // empty unless given
	// internal only
	cmd *exec.Cmd  `json:"-"`
	mux sync.Mutex `json:"-"`
}

// TODO move cmd and mux here
// type LockingJob struct { }

// Result may have many items and sub-items
type Result struct {
	Name    string      `json:"name"`
	Status  string      `json:"status,omitempty"`
	Message string      `json:"message,omitempty"`
	Results []Result    `json:"results,omitempty"`
	Detail  interface{} `json:"_detail,omitempty"`
}

var initialized = false
var done = make(chan struct{})
var deathRow = make(chan webhooks.RefID)
var debounced = make(chan *webhooks.Ref)
var debacklog = make(chan *webhooks.Ref)

// Start starts the job loop, channels, and cleanup routines
func Start(runOpts *options.ServerConfig) {
	go Run(runOpts)
}

// Run starts the job loop and waits for it to be stopped
func Run(runOpts *options.ServerConfig) {
	log.Printf("[gitdeploy] Starting")
	if initialized {
		panic(errors.New("should not double initialize 'jobs'"))
	}
	initialized = true

	// TODO load the backlog from disk too

	oldJobs, err := WalkLogs(runOpts)
	if nil != err {
		panic(err)
	}
	for i := range oldJobs {
		job := oldJobs[i]
		job.ID = string(job.GitRef.GetRevID())
		val, ok := Recents.Load(job.GitRef.GetRevID())
		if !ok || val.(*Job).GitRef.Timestamp.Sub(job.GitRef.Timestamp) < 0 {
			Recents.Store(job.GitRef.GetRevID(), job)
		}
	}

	ticker := time.NewTicker(runOpts.StaleJobAge / 2)
	for {
		select {
		case h := <-webhooks.Hooks:
			hook := webhooks.New(h)
			//log.Printf("[%s] debouncing...", hook.GetRefID())
			saveBacklog(hook, runOpts)
			debounce(hook, runOpts)
		case hook := <-debacklog:
			//log.Printf("[%s] checking for backlog...", hook.GetRefID())
			debounce(hook, runOpts)
		case hook := <-debounced:
			//log.Printf("[%s] debounced!", hook.GetRefID())
			run(hook, runOpts)
		case activeID := <-deathRow:
			//log.Printf("[%s] done", activeID)
			remove(runOpts, activeID /*, false*/)
		case promotion := <-Promotions:
			log.Printf("[%s] promoting to %s", promotion.GitRef.GetRefID(), promotion.PromoteTo)
			promote(webhooks.New(*promotion.GitRef), promotion.PromoteTo, runOpts)
		case <-ticker.C:
			log.Printf("[gitdeploy] cleaning old jobs")
			expire(runOpts)
		case <-done:
			log.Printf("[gitdeploy] stopping")
			// TODO kill jobs
			ticker.Stop()
		}
	}
}

// Stop will cancel the job loop and its timers
func Stop() {
	done <- struct{}{}
	initialized = false
}

// All returns all jobs, including active, recent, and (TODO) historical
func All(then time.Time) []*Job {
	jobsTimersMux.Lock()
	defer jobsTimersMux.Unlock()

	jobsCopy := []*Job{}

	Pending.Range(func(key, value interface{}) bool {
		hook := value.(*webhooks.Ref)
		if hook.Timestamp.Sub(then) <= 0 {
			return true
		}

		jobCopy := &Job{
			//StartedAt: job.StartedAt,
			ID:     string(hook.GetURLSafeRefID()),
			GitRef: hook,
			//Promote:   job.Promote,
			//EndedAt:   job.EndedAt,
		}
		jobsCopy = append(jobsCopy, jobCopy)
		return true
	})

	Actives.Range(func(key, value interface{}) bool {
		job := value.(*Job)
		if job.GitRef.Timestamp.Sub(then) <= 0 {
			return true
		}

		jobCopy := &Job{
			StartedAt: job.StartedAt,
			ID:        string(job.GitRef.GetURLSafeRefID()),
			GitRef:    job.GitRef,
			//Promote:   job.Promote,
		}
		if nil != job.ExitCode {
			jobCopy.ExitCode = &(*job.ExitCode)
		}
		jobsCopy = append(jobsCopy, jobCopy)
		return true
	})

	Recents.Range(func(key, value interface{}) bool {
		job := value.(*Job)
		if job.GitRef.Timestamp.Sub(then) <= 0 {
			return true
		}

		jobCopy := &Job{
			StartedAt: job.StartedAt,
			ID:        string(job.GitRef.GetURLSafeRevID()),
			GitRef:    job.GitRef,
			EndedAt:   job.EndedAt,
			//Promote:   job.Promote,
		}
		if nil != job.ExitCode {
			jobCopy.ExitCode = &(*job.ExitCode)
		}
		jobsCopy = append(jobsCopy, jobCopy)
		return true
	})

	return jobsCopy
}

// SetReport will update jobs' logs
func SetReport(urlRefID webhooks.URLSafeRefID, result *Result) error {
	b, err := base64.RawURLEncoding.DecodeString(string(urlRefID))
	if nil != err {
		return err
	}
	refID := webhooks.RefID(b)

	value, ok := Actives.Load(refID)
	if !ok {
		return errors.New("active job not found by " + string(refID))
	}
	job := value.(*Job)

	// needs mutex (prevent double update)
	job.Report = result
	Actives.Store(refID, job)

	return nil
}

// Remove will put a job on death row
func Remove(gitID webhooks.URLSafeRefID /*, nokill bool*/) {
	activeID, err :=
		base64.RawURLEncoding.DecodeString(string(gitID))
	if nil != err {
		log.Printf("bad id: %s", activeID)
		return
	}
	deathRow <- webhooks.RefID(activeID)
}

func getJobFilePath(baseDir string, hook *webhooks.Ref, suffix string) (string, string, error) {
	baseDir, _ = filepath.Abs(baseDir)
	fileTime := hook.Timestamp.UTC().Format(options.TimeFile)
	fileName := fileTime + "." + hook.RefName + "." + hook.Rev[:7] + suffix // ".log" or ".json"
	fileDir := filepath.Join(baseDir, hook.RepoID)

	err := os.MkdirAll(fileDir, 0755)

	return fileDir, fileName, err
}

func getJobFile(baseDir string, hook *webhooks.Ref, suffix string) (*os.File, error) {
	repoDir, repoFile, err := getJobFilePath(baseDir, hook, suffix)
	if nil != err {
		//log.Printf("[warn] could not create log directory '%s': %v", repoDir, err)
		return nil, err
	}

	path := filepath.Join(repoDir, repoFile)
	return os.OpenFile(path, os.O_CREATE|os.O_WRONLY, 0644)
	//return fmt.Sprintf("%s#%s", strings.ReplaceAll(hook.RepoID, "/", "-"), hook.RefName)
}

func openJobFile(baseDir string, hook *webhooks.Ref, suffix string) (*os.File, error) {
	repoDir, repoFile, _ := getJobFilePath(baseDir, hook, suffix)
	return os.Open(filepath.Join(repoDir, repoFile))
}

func setOutput(logDir string, job *Job) *os.File {
	var f *os.File = nil

	defer func() {
		// TODO write to append-only log rather than keep in-memory
		// (noting that we want to keep Stdout vs Stderr and timing)
		cmd := job.cmd
		wout := &outWriter{job: job}
		werr := &errWriter{job: job}
		if nil != f {
			cmd.Stdout = io.MultiWriter(f, wout)
			cmd.Stderr = io.MultiWriter(f, werr)
		} else {
			cmd.Stdout = io.MultiWriter(os.Stdout, wout)
			cmd.Stderr = io.MultiWriter(os.Stderr, werr)
		}
	}()

	if "" == logDir {
		return nil
	}

	hook := job.GitRef
	f, err := getJobFile(logDir, hook, ".log")
	if nil != err {
		// f.Name() should be the full path
		log.Printf("[warn] could not create log file '%s': %v", logDir, err)
		return nil
	}

	cwd, _ := os.Getwd()
	log.Printf("[%s] log to ./%s", hook.GetRefID(), f.Name()[len(cwd)+1:])
	return f
}

// Debounce puts a job in the queue, in time
func Debounce(hook webhooks.Ref) {
	webhooks.Hooks <- hook
}

var jobsTimersMux sync.Mutex
var debounceTimers = make(map[webhooks.RefID]*time.Timer)

func debounce(hook *webhooks.Ref, runOpts *options.ServerConfig) {
	jobsTimersMux.Lock()
	defer jobsTimersMux.Unlock()

	activeID := hook.GetRefID()
	if _, ok := Actives.Load(activeID); ok {
		//log.Printf("[%s] will run again after current job", hook.GetRefID())
		return
	}

	refID := hook.GetRefID()
	timer, ok := debounceTimers[refID]
	if ok {
		//log.Printf("[%s] replaced debounce timer", hook.GetRefID())
		timer.Stop()
	}
	// this will not cause a mutual lock because it is async
	debounceTimers[refID] = time.AfterFunc(runOpts.DebounceDelay, func() {
		jobsTimersMux.Lock()
		delete(debounceTimers, refID)
		jobsTimersMux.Unlock()

		debounced <- hook
	})
}

func saveBacklog(hook *webhooks.Ref, runOpts *options.ServerConfig) {
	pendingID := hook.GetRefID()
	Pending.Store(pendingID, hook)

	repoDir, repoFile, err := getBacklogFilePath(runOpts.TmpDir, hook)
	if nil != err {
		log.Printf("[warn] could not create backlog dir %s:\n%v", repoDir, err)
		return
	}
	f, err := ioutil.TempFile(repoDir, "tmp-*")
	if nil != err {
		log.Printf("[warn] could not create backlog file %s:\n%v", f.Name(), err)
		return
	}

	b, _ := json.MarshalIndent(hook, "", "  ")
	if _, err := f.Write(b); nil != err {
		log.Printf("[warn] could not write backlog file %s:\n%v", f.Name(), err)
		return
	}

	replace := false
	backlogPath := filepath.Join(repoDir, repoFile)
	if _, err := os.Stat(backlogPath); nil == err {
		replace = true
		_ = os.Remove(backlogPath)
	}
	if err := os.Rename(f.Name(), backlogPath); nil != err {
		log.Printf("[warn] rename backlog json failed:\n%v", err)
		return
	}

	if replace {
		log.Printf("[%s] updated in queue", hook.GetRefID())
	} else {
		log.Printf("[%s] added to queue", hook.GetRefID())
	}
}

func getBacklogFilePath(baseDir string, hook *webhooks.Ref) (string, string, error) {
	baseDir, _ = filepath.Abs(baseDir)
	fileName := hook.RefName + ".json"
	fileDir := filepath.Join(baseDir, hook.RepoID)

	err := os.MkdirAll(fileDir, 0755)

	return fileDir, fileName, err
}

func run(curHook *webhooks.Ref, runOpts *options.ServerConfig) {
	// because we want to lock the whole transaction all of the state
	jobsTimersMux.Lock()
	defer jobsTimersMux.Unlock()

	pendingID := curHook.GetRefID()
	if _, ok := Actives.Load(pendingID); ok {
		log.Printf("[%s] already in progress", pendingID)
		return
	}

	var hook *webhooks.Ref
	// Legacy, but would be nice to repurpose for resuming on reload
	repoDir, repoFile, _ := getBacklogFilePath(runOpts.TmpDir, curHook)
	backlogFile := filepath.Join(repoDir, repoFile)
	if value, ok := Pending.Load(pendingID); ok {
		hook = value.(*webhooks.Ref)
	} else {
		// TODO add mutex (should not affect temp files)
		_ = os.Remove(backlogFile + ".cur")
		_ = os.Rename(backlogFile, backlogFile+".cur")
		b, err := ioutil.ReadFile(backlogFile + ".cur")
		if nil != err {
			if !os.IsNotExist(err) {
				log.Printf("[warn] could not read backlog file %s:\n%v", backlogFile, err)
				return
			}
			// doesn't exist => no backlog
			log.Printf("[%s] no backlog", pendingID)
			return
		}

		hook = &webhooks.Ref{}
		if err := json.Unmarshal(b, hook); nil != err {
			log.Printf("[warn] could not parse backlog %s:\n%v", backlogFile, err)
			return
		}
		hook = webhooks.New(*hook)
	}

	Pending.Delete(pendingID)
	_ = os.Remove(backlogFile)
	_ = os.Remove(backlogFile + ".cur")

	env := os.Environ()
	envs := getEnvs(runOpts.Addr, string(pendingID), runOpts.RepoList, hook)
	envs = append(envs, "GIT_DEPLOY_JOB_ID="+string(pendingID))

	scriptPath, _ := filepath.Abs(runOpts.ScriptsPath + "/deploy.sh")
	args := []string{"-i", "--", scriptPath}

	log.Printf("[%s] bash %s %s %s", hook.GetRefID(), args[0], args[1], args[2])
	cmd := exec.Command("bash", append(args, []string{
		string(pendingID),
		hook.RefName,
		hook.RefType,
		hook.Owner,
		hook.Repo,
		hook.HTTPSURL,
	}...)...)
	cmd.Env = append(env, envs...)

	now := time.Now()
	j := &Job{
		StartedAt: &now,
		cmd:       cmd,
		GitRef:    hook,
		Logs:      []Log{},
		Promote:   false,
	}
	// TODO jobs.New()
	// Sets cmd.Stdout and cmd.Stderr
	txtFile := setOutput(runOpts.LogDir, j)

	if err := cmd.Start(); nil != err {
		log.Printf("[ERROR] failed to exec: %s\n", err)
		return
	}

	Actives.Store(pendingID, j)

	go func() {
		//log.Printf("[%s] job started", pendingID)
		if err := cmd.Wait(); nil != err {
			log.Printf("[%s] exited with error: %v", pendingID, err)
		} else {
			log.Printf("[%s] exited successfully", pendingID)
		}
		if nil != txtFile {
			_ = txtFile.Close()
		}

		// this will completely clear the finished job
		deathRow <- pendingID

		// debounces without saving in the backlog
		// TODO move this into deathRow?
		debacklog <- hook
	}()
}

func getEnvs(addr, activeID string, repoList string, hook *webhooks.Ref) []string {

	port := strings.Split(addr, ":")[1]

	envs := []string{
		"GIT_DEPLOY_JOB_ID=" + activeID,
		"GIT_DEPLOY_TIMESTAMP=" + hook.Timestamp.Format(time.RFC3339),
		"GIT_DEPLOY_CALLBACK_URL=" + "http://localhost:" + port + "/api/local/jobs/" + string(hook.GetURLSafeRefID()),
		"GIT_REF_NAME=" + hook.RefName,
		"GIT_REF_TYPE=" + hook.RefType,
		"GIT_REPO_ID=" + hook.RepoID,
		"GIT_REPO_OWNER=" + hook.Owner,
		"GIT_REPO_NAME=" + hook.Repo,
		"GIT_CLONE_URL=" + hook.HTTPSURL, // deprecated
		"GIT_HTTPS_URL=" + hook.HTTPSURL,
		"GIT_SSH_URL=" + hook.SSHURL,
	}

	// GIT_REPO_TRUSTED
	// Set GIT_REPO_TRUSTED=TRUE if the repo matches exactly, or by pattern
	repoID := strings.ToLower(hook.RepoID)
	for _, repo := range strings.Fields(repoList) {
		last := len(repo) - 1
		if len(repo) < 0 {
			continue
		}
		repo = strings.ToLower(repo)
		if '*' == repo[last] {
			// Wildcard match a prefix, for example:
			// github.com/whatever/*					MATCHES github.com/whatever/foo
			// github.com/whatever/ProjectX-* MATCHES github.com/whatever/ProjectX-Foo
			if strings.HasPrefix(repoID, repo[:last]) {
				envs = append(envs, "GIT_REPO_TRUSTED=true")
				break
			}
		} else if repo == repoID {
			envs = append(envs, "GIT_REPO_TRUSTED=true")
			break
		}
	}

	return envs
}

// Remove kills the job and moves it to recents
func remove(runOpts *options.ServerConfig, activeID webhooks.RefID /*, nokill bool*/) {
	// Encapsulate the whole transaction
	jobsTimersMux.Lock()
	defer jobsTimersMux.Unlock()

	value, ok := Actives.Load(activeID)
	if !ok {
		log.Printf("[warn] could not find job to kill by RefID %s", activeID)
		return
	}
	job := value.(*Job)
	Actives.Delete(activeID)

	if nil == job.cmd.ProcessState {
		// is not yet finished
		if nil != job.cmd.Process {
			// but definitely was started
			if err := job.cmd.Process.Kill(); nil != err {
				log.Printf("error killing job:\n%v", err)
			}
		}
	}
	if nil != job.cmd.ProcessState {
		//*job.ExitCode = job.cmd.ProcessState.ExitCode()
		exitCode := job.cmd.ProcessState.ExitCode()
		job.ExitCode = &exitCode
	}
	now := time.Now()
	job.EndedAt = &now

	// Switch ID to the more specific RevID
	job.ID = string(job.GitRef.GetRevID())
	// replace the text log with a json log
	if jsonFile, err := getJobFile(runOpts.LogDir, job.GitRef, ".json"); nil != err {
		// jsonFile.Name() should be the full path
		log.Printf("[warn] could not create log file '%s': %v", runOpts.LogDir, err)
	} else {
		enc := json.NewEncoder(jsonFile)
		enc.SetIndent("", "  ")
		if err := enc.Encode(job); nil != err {
			log.Printf("[warn] could not encode json log '%s': %v", jsonFile.Name(), err)
		} else {
			logdir, logname, _ := getJobFilePath(runOpts.LogDir, job.GitRef, ".log")
			_ = os.Remove(filepath.Join(logdir, logname))
		}
		_ = jsonFile.Close()
	}
	job.Logs = []Log{}

	// transition to RevID for non-active, non-pending jobs
	job.ID = string(job.GitRef.GetRevID())
	Recents.Store(job.GitRef.GetRevID(), job)
}

func expire(runOpts *options.ServerConfig) {
	staleJobIDs := []webhooks.URLSafeRevID{}

	Recents.Range(func(key, value interface{}) bool {
		revID := key.(webhooks.URLSafeRevID)
		age := time.Now().Sub(value.(*Job).GitRef.Timestamp)
		if age > runOpts.StaleJobAge {
			staleJobIDs = append(staleJobIDs, revID)
		}
		return true
	})

	for _, revID := range staleJobIDs {
		Recents.Delete(revID)
	}
}