Puberty commit

2017-10-29 23:06:41 +01:00
commit 34034b5223
63 changed files with 16011 additions and 0 deletions
@@ -0,0 +1,117 @@
+package runner
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"os/exec"
+	"time"
+
+	"github.com/juju/errors"
+
+	"github.com/localhots/cmdui/backend/commands"
+	"github.com/localhots/cmdui/backend/config"
+	"github.com/localhots/cmdui/backend/db"
+	"github.com/localhots/cmdui/backend/log"
+)
+
+func Start(cmd commands.Command, user db.User) (*Process, error) {
+	job := db.NewJob(cmd, user)
+	if err := job.Create(); err != nil {
+		return nil, errors.Annotate(err, "Failed to create a job")
+	}
+
+	p := &Process{
+		ID:  job.ID,
+		Job: &job,
+	}
+
+	basePath := config.Get().Commands.BasePath
+	p.exec = exec.Command(basePath, cmd.CombinedArgs()...)
+	fd, err := p.useLogfile(p.logfile())
+	if err != nil {
+		return nil, err
+	}
+	p.exec.Stdout = fd
+	p.exec.Stderr = fd
+
+	if err := pool.add(context.Background(), p); err != nil {
+		return p, errors.Annotate(err, "Failed to start a process")
+	}
+
+	return p, nil
+}
+
+func ReadLogUpdates(ctx context.Context, p *Process, w io.Writer) (done <-chan struct{}, err error) {
+	cmde := exec.CommandContext(ctx, "/usr/bin/tail", "-n", "100", "-f", p.logfile())
+	return readLog(ctx, cmde, p, w)
+}
+
+func ReadFullLog(ctx context.Context, p *Process, w io.Writer) (done <-chan struct{}, err error) {
+	cmde := exec.CommandContext(ctx, "/bin/cat", p.logfile())
+	return readLog(ctx, cmde, p, w)
+}
+
+func readLog(ctx context.Context, cmde *exec.Cmd, p *Process, w io.Writer) (done <-chan struct{}, err error) {
+	cmde.Stdout = w
+	cmde.Stderr = w
+
+	tp := &Process{
+		ID:   sysID("log-access"),
+		exec: cmde,
+	}
+
+	exited := make(chan struct{}, 2)
+	p.onExit(func(p *Process) {
+		exited <- struct{}{}
+	})
+	tp.onExit(func(p *Process) {
+		exited <- struct{}{}
+	})
+
+	if err := pool.add(ctx, tp); err != nil {
+		return nil, errors.Annotate(err, "Failed to start a tail process")
+	}
+
+	return exited, nil
+}
+
+func CommandsList() ([]commands.Command, error) {
+	var buf bytes.Buffer
+	cmde := exec.Command("/bin/bash", "-c", config.Get().Commands.ConfigCommand)
+	cmde.Stdout = &buf
+	cmde.Stderr = &buf
+
+	p := &Process{
+		ID:   sysID("commands-list"),
+		exec: cmde,
+	}
+
+	exited := make(chan struct{})
+	p.onExit(func(p *Process) {
+		close(exited)
+	})
+
+	if err := pool.add(context.Background(), p); err != nil {
+		return nil, errors.Annotate(err, "Failed to import commands")
+	}
+	<-exited
+
+	body := buf.Bytes()
+	var list []commands.Command
+	if err := json.Unmarshal(body, &list); err != nil {
+		log.WithFields(log.F{
+			"error": err,
+			"json":  string(body),
+		}).Error("Invalid commands JSON")
+		return nil, errors.Annotate(err, "Failed to decode commands JSON")
+	}
+
+	return list, nil
+}
+
+func sysID(name string) string {
+	return fmt.Sprintf("%s-%d", name, time.Now().UnixNano())
+}
@@ -0,0 +1,155 @@
+package runner
+
+import (
+	"context"
+	"os"
+	"strings"
+	"sync"
+
+	"github.com/juju/errors"
+
+	"github.com/localhots/cmdui/backend/config"
+	"github.com/localhots/cmdui/backend/db"
+	"github.com/localhots/cmdui/backend/log"
+)
+
+var pool = &processPool{}
+
+type processPool struct {
+	lock  sync.RWMutex
+	wg    sync.WaitGroup
+	procs map[string]*Process
+}
+
+func FindProcess(id string) *Process {
+	pool.lock.RLock()
+	defer pool.lock.RUnlock()
+	return pool.procs[id]
+}
+
+func Shutdown() {
+	pool.close()
+}
+
+func PrepareLogsDir() error {
+	return os.MkdirAll(config.Get().LogDir, 0755)
+}
+
+func (pp *processPool) add(ctx context.Context, p *Process) error {
+	// Validate process
+	if err := pp.validate(p); err != nil {
+		return errors.Annotate(err, "Process validation failed")
+	}
+
+	// Register process
+	pp.lock.Lock()
+	if pp.procs == nil {
+		pp.procs = make(map[string]*Process)
+	}
+	pp.procs[p.ID] = p
+	pp.lock.Unlock()
+
+	// Start process
+
+	if err := p.exec.Start(); err != nil {
+		log.WithFields(log.F{
+			"id":    p.ID,
+			"error": err,
+		}).Error("Failed to start a command")
+		return errors.Annotate(err, "Failed to start a process")
+	}
+	p.PID = p.exec.Process.Pid
+	log.WithFields(log.F{
+		"id":      p.ID,
+		"pid":     p.PID,
+		"command": strings.Join(p.exec.Args, " "),
+	}).Info("Command started")
+	tryUpdateState(p, db.JobStateStarted)
+
+	pp.wg.Add(1)
+	go pp.handleProcessExit(p)
+	return nil
+}
+
+func (pp *processPool) handleProcessExit(p *Process) {
+	defer pp.wg.Done()
+	err := p.exec.Wait()
+
+	pp.lock.Lock()
+	delete(pp.procs, p.ID)
+	pp.lock.Unlock()
+
+	if err != nil {
+		log.WithFields(log.F{
+			"id":    p.ID,
+			"pid":   p.PID,
+			"error": err,
+		}).Error("Command failed")
+		tryUpdateState(p, db.JobStateFailed)
+	} else {
+		log.WithFields(log.F{
+			"id":  p.ID,
+			"pid": p.PID,
+		}).Info("Command finished")
+		tryUpdateState(p, db.JobStateFinished)
+	}
+
+	if err := p.close(); err != nil {
+		log.WithFields(log.F{
+			"id":    p.ID,
+			"error": err,
+		}).Error("Failed to close a job")
+	}
+
+	for _, onExit := range p.exitCallbacks {
+		onExit(p)
+	}
+}
+
+func (pp *processPool) procsList() []*Process {
+	pp.lock.RLock()
+	defer pp.lock.RUnlock()
+
+	list := make([]*Process, len(pp.procs))
+	i := 0
+	for _, p := range pp.procs {
+		list[i] = p
+		i++
+	}
+
+	return list
+}
+
+func (pp *processPool) validate(p *Process) error {
+	switch {
+	case p == nil:
+		return errors.New("Can't add an empty process")
+	case p.ID == "":
+		return errors.New("Can't add a process without an ID")
+	case p.exec == nil:
+		return errors.New("Process executable can't be empty")
+	default:
+		return nil
+	}
+}
+
+func (pp *processPool) close() {
+	pp.wg.Wait()
+}
+
+func tryUpdateState(p *Process, s db.JobState) {
+	log.WithFields(log.F{
+		"id":    p.ID,
+		"state": s,
+	}).Debug("Job state changed")
+	if p.Job == nil {
+		return
+	}
+	if err := p.Job.UpdateState(s); err != nil {
+		log.WithFields(log.F{
+			"id":    p.ID,
+			"state": s,
+			"error": err,
+		}).Error("Job state changed")
+	}
+}
@@ -0,0 +1,60 @@
+package runner
+
+import (
+	"fmt"
+	"io"
+	"os"
+	"os/exec"
+	"syscall"
+
+	"github.com/juju/errors"
+
+	"github.com/localhots/cmdui/backend/commands"
+	"github.com/localhots/cmdui/backend/config"
+	"github.com/localhots/cmdui/backend/db"
+)
+
+type Process struct {
+	ID      string           `json:"id"`
+	PID     int              `json:"pid"`
+	Job     *db.Job          `json:"job"`
+	Command commands.Command `json:"command"`
+	Out     io.Writer        `json:"-"`
+
+	exec *exec.Cmd
+	log  *os.File
+
+	exitCallbacks []func(p *Process) `json:"-"`
+}
+
+func (p *Process) Signal(s syscall.Signal) error {
+	return p.exec.Process.Signal(s)
+}
+
+func (p *Process) logfile() string {
+	return fmt.Sprintf("%s/%s.log", config.Get().LogDir, p.ID)
+}
+
+func (p *Process) useLogfile(path string) (io.Writer, error) {
+	fd, err := os.OpenFile(p.logfile(), os.O_CREATE|os.O_WRONLY, 0744)
+	if err != nil {
+		return nil, errors.Annotate(err, "Failed to create log file")
+	}
+	p.log = fd
+
+	return fd, nil
+}
+
+func (p *Process) onExit(fn func(p *Process)) {
+	p.exitCallbacks = append(p.exitCallbacks, fn)
+}
+
+func (p *Process) close() error {
+	if p.log != nil {
+		err := p.log.Close()
+		if err != nil {
+			return errors.Annotate(err, "Failed to close log file")
+		}
+	}
+	return nil
+}