Bug fixes, testing, and CI

.github/workflows/test.yml

@@ -0,0 +1,28 @@
+name: Tests
+
+on:
+  push:
+    branches: [main]
+  pull_request:
+    branches: [main]
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: actions/setup-go@v5
+        with:
+          go-version-file: go.mod
+
+      - name: Build binaries
+        run: |
+          go build -o /tmp/taskmasterd cmd/server/main.go
+          go build -o /tmp/taskmasterctl cmd/client/main.go
+
+      - name: Prepare test directory
+        run: mkdir -p /tmp/test_taskmaster
+
+      - name: Run tests
+        run: bash tests/full_test.sh

cmd/client/main.go

@@ -69,7 +69,7 @@ func main() {
 
 		res := client.Read(server.DEL)
 		if res.Err != nil {
-			utils.Errorf(err.Error())
+			utils.Errorf(res.Err.Error())
 		} else if res.HasContent() {
 			utils.Logf(res.Data)
 		}

compose.yaml

@@ -7,3 +7,5 @@ services:
       dockerfile: Dockerfile
     ports:
       - "8080:8080"
+    volumes:
+      - /dev/log:/dev/log

internal/job/job.go

@@ -36,14 +36,15 @@ type Job struct {
 	StderrWriter  *utils.DynamicWriter
 	NumProcs      int
 	pgid          int
-	mustart       sync.Mutex
+	startReady    chan struct{}
+	startOnce     sync.Once
 	mustop        sync.Mutex
 }
 
 func NewJob(name string, prog *config.Program) *Job {
 	has_zero := false
 	exit_codes := prog.ExitCodes
-	for exit := range exit_codes {
+	for _, exit := range exit_codes {
 		if exit == 0 {
 			has_zero = true
 			break
@@ -64,6 +65,9 @@ func NewJob(name string, prog *config.Program) *Job {
 		running[i] = false
 	}
 
+	ch := make(chan struct{})
+	close(ch)
+
 	return &Job{
 		Name:          name,
 		Command:       prog.Command,
@@ -86,69 +90,72 @@ func NewJob(name string, prog *config.Program) *Job {
 		NumProcs:      prog.NumProcs,
 		cmds:          make([]*exec.Cmd, prog.NumProcs),
 		pgid:          0,
+		startReady:    ch,
 	}
 }
 
+func (j *Job) closeStartReady() {
+	j.startOnce.Do(func() {
+		close(j.startReady)
+	})
+}
+
 type WorkerFn = func(j *Job, wg *sync.WaitGroup, _done chan bool) error
 
 func (j *Job) Start(wg *sync.WaitGroup, _done chan bool) error {
 	defer func() { _done <- true }()
 
-	j.mustart.Lock()
-	done := make(chan bool, j.NumProcs)
-	defer close(done)
+	j.mustop.Lock()
+	defer j.mustop.Unlock()
+
+	startReady := make(chan struct{})
+	j.startReady = startReady
+	j.startOnce = sync.Once{}
 
 	for i := range j.NumProcs {
 		if j.Is(STOPPING, i) || j._running[i] {
-			done <- true
 			continue
 		}
 
 		j._running[i] = true
 
 		if j.HasPgid() {
-			go j.startJobWorker(wg, i, done, j.pgid)
+			go j.startJobWorker(wg, i, j.pgid)
 			continue
 		}
 
-		go j.startJobWorker(wg, i, done, 0)
-		<-done
-		done <- true
+		go j.startJobWorker(wg, i, 0)
+		<-startReady
 		if !j.Is(RUNNING, 0) {
 			return fmt.Errorf("process could not be running")
 		}
 		j.pgid = j.cmds[i].Process.Pid
 	}
 
-	for range j.NumProcs {
-		<-done
-	}
-	j.mustart.Unlock()
 	return nil
 }
 
-func (j *Job) startJobWorker(wg *sync.WaitGroup, id int, done chan bool, pgid int) {
+func (j *Job) startJobWorker(wg *sync.WaitGroup, id int, pgid int) {
 	wg.Add(1)
 	defer wg.Done()
 
-	cmd_list := []string{
-		"sh",
-		"-c",
-		fmt.Sprintf("umask %v && %v", j.Umask, j.Command),
-	}
-
-	cmd := exec.Command(cmd_list[0], cmd_list[1:]...)
-	cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true, Pgid: pgid}
-
-	j.cmds[id] = cmd
-
 	retries := 0
 	for {
+		usePgid := pgid
+		if j.cmds[id] != nil && j.cmds[id].Process != nil {
+			usePgid = 0
+		}
+
+		cmd := exec.Command("sh", "-c", fmt.Sprintf("umask %v && %v", j.Umask, j.Command))
+		cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true, Pgid: usePgid}
+		j.cmds[id] = cmd
+
 		j.SetState(STARTING, id)
 		err := j.tryStart(id)
 		if err != nil {
 			logger.Error(err)
 			j.SetState(BACKOFF, id)
+			j.closeStartReady()
 			retries++
 			if j.StartRetries == retries {
 				break
@@ -159,7 +166,7 @@ func (j *Job) startJobWorker(wg *sync.WaitGroup, id int, done chan bool, pgid in
 
 		cur_ts := int(time.Now().Unix())
 		j.SetState(RUNNING, id)
-		done <- true
+		j.closeStartReady()
 		state, _ := j.cmds[id].Process.Wait()
 		j.cmds[id].ProcessState = state
 
@@ -181,11 +188,16 @@ func (j *Job) startJobWorker(wg *sync.WaitGroup, id int, done chan bool, pgid in
 			break
 		}
 		if j.Autorestart == AUTORESTART_UNEXPECTED {
-			for exit := range j.ExitCodes {
+			expected := false
+			for _, exit := range j.ExitCodes {
 				if exit == j.cmds[id].ProcessState.ExitCode() {
+					expected = true
 					break
 				}
 			}
+			if expected {
+				break
+			}
 		}
 	}
 	if j.Is(BACKOFF, id) {
@@ -245,25 +257,30 @@ func (j *Job) Restart(wg *sync.WaitGroup, _done chan bool) error {
 
 func (j *Job) Stop(wg *sync.WaitGroup, _done chan bool) error {
 	defer func() { _done <- true }()
+	<-j.startReady
 	j.mustop.Lock()
+	defer j.mustop.Unlock()
 
 	if j.HasPgid() {
 		for i := range j.NumProcs {
 			j.SetState(STOPPING, i)
 		}
-		cur := time.Now().Unix()
-		err := syscall.Kill(-j.pgid, syscall.SIGKILL)
+
+		err := syscall.Kill(-j.pgid, j.StopSignal)
 		if err != nil {
 			return err
 		}
 
+		cur := time.Now().Unix()
 		for time.Now().Unix()-cur < int64(j.StopWaitSecs) && j.IsRunning() {
 			time.Sleep(100 * time.Millisecond)
 		}
 
-		if j.HasPgid() {
-			err = syscall.Kill(-j.pgid, j.StopSignal)
-			return err
+		if j.HasPgid() && j.IsRunning() {
+			err = syscall.Kill(-j.pgid, syscall.SIGKILL)
+			if err != nil {
+				return err
+			}
 		}
 
 		for j.IsRunning() {
@@ -272,7 +289,6 @@ func (j *Job) Stop(wg *sync.WaitGroup, _done chan bool) error {
 	}
 
 	j.SetPgid(0)
-	j.mustop.Unlock()
 	return nil
 }
 

internal/logger/logger.go

@@ -6,8 +6,6 @@ import (
 	"os"
 	"sync"
 	"time"
-
-	"github.com/Archer-01/taskmaster/internal/utils"
 )
 
 type LogLevel uint8
@@ -46,7 +44,11 @@ func Init() {
 	syslogger, err := syslog.New(syslog.LOG_INFO, "taskmaster")
 
 	if err != nil {
-		utils.Errorf(err.Error())
+		fmt.Fprintln(os.Stderr, "ERROR: syslog unavailable. To fix this, configure a syslog log driver in Docker:")
+		fmt.Fprintln(os.Stderr, "  compose.yaml:  logging:\n                  driver: syslog\n                  options:\n                    syslog-address: \"unix:///dev/log\"\n                    tag: \"taskmaster\"")
+		fmt.Fprintln(os.Stderr, "  or run with:    docker run --log-driver=syslog --log-opt syslog-address=unix:///dev/log ...")
+		fmt.Fprintln(os.Stderr, "  or set as default in /etc/docker/daemon.json: { \"log-driver\": \"syslog\" }")
+		fmt.Fprintln(os.Stderr, "Original error:", err)
 		os.Exit(1)
 	}
 

internal/manager/manager.go

@@ -161,10 +161,14 @@ func (m *JobManager) Run() {
 			action.Done <- true
 			return
 
-		case RELOAD:
-			logger.Warn("Reloading...")
-			m.reload()
+	case RELOAD:
+		logger.Warn("Reloading...")
+		if err := m.reload(); err != nil {
+			action.Data <- err.Error()
+			action.Done <- false
+		} else {
 			action.Done <- true
+		}
 
 		case START:
 			m.setJobs("STARTING", (*job.Job).Start, action)

tests/ar_config.toml

@@ -0,0 +1,7 @@
+[program.shortlived]
+command = "sleep 0.5"
+autostart = true
+autorestart = "true"
+numprocs = 1
+startsecs = 0
+startretries = 5