Clean-up a few things around global/job shutdown

This commit makes a few small tweaks around how shutdown occurs. I've added a
shared lock when setting a job as completed and also swapped the complete
channel to send a struct instead of a boolean (not much of a change).

Author: Justin Reagor

core/app.go

@@ -118,7 +118,7 @@ func (a *App) Run() {
 		// signal handler above, and reload endpoint, only need to fire a
 		// GlobalShutdown across the event bus. Context handles everything after
 		// that process.
-		completedCh := make(chan bool)
+		completedCh := make(chan struct{}, 1)
 		go func() {
 			for {
 				select {
@@ -159,7 +159,6 @@ func (a *App) Run() {
 			log.Error(err)
 			break
 		}
-		cancel()
 		close(completedCh)
 	}
 }
@@ -198,7 +197,7 @@ func (a *App) reload() error {
 
 // HandlePolling sets up polling functions and write their quit channels
 // back to our config
-func (a *App) runTasks(ctx context.Context, completedCh chan bool) {
+func (a *App) runTasks(ctx context.Context, completedCh chan struct{}) {
 	// we need to subscribe to events before we Run all the jobs
 	// to avoid races where a job finishes and fires events before
 	// other jobs are even subscribed to listen for them.

core/signals_test.go

@@ -62,7 +62,7 @@ func getSignalEventTestConfig(signals []string) *App {
 // by this same test, but that we don't have a separate unit test
 // because they'll interfere with each other's state.
 func TestTerminateSignal(t *testing.T) {
-	stopCh := make(chan bool)
+	stopCh := make(chan struct{}, 1)
 	app := getSignalTestConfig()
 	bus := app.Bus
 	ctx, cancel := context.WithCancel(context.Background())
@@ -93,7 +93,7 @@ func TestTerminateSignal(t *testing.T) {
 // Test handler for handling signal events SIGHUP (and SIGUSR2). Note that the
 // SIGUSR1 is currently setup to handle reloading ContainerPilot's log file.
 func TestSignalEvent(t *testing.T) {
-	stopCh := make(chan bool)
+	stopCh := make(chan struct{}, 1)
 	signals := []string{"SIGHUP", "SIGUSR2"}
 	app := getSignalEventTestConfig(signals)
 	bus := app.Bus

jobs/jobs.go

@@ -52,7 +52,8 @@ type Job struct {
 	frequency      time.Duration
 
 	// completed
-	IsComplete bool
+	IsComplete   bool
+	completeLock *sync.RWMutex
 
 	events.Subscriber
 	events.Publisher
@@ -76,6 +77,7 @@ func NewJob(cfg *Config) *Job {
 		frequency:         cfg.freqInterval,
 	}
 	job.statusLock = &sync.RWMutex{}
+	job.completeLock = &sync.RWMutex{}
 	job.Rx = make(chan events.Event, eventBufferSize)
 	if job.Name == "containerpilot" {
 		// right now this hardcodes the telemetry service to
@@ -119,6 +121,8 @@ func (job *Job) setStatus(status JobStatus) {
 }
 
 func (job *Job) setComplete() {
+	job.completeLock.Lock()
+	defer job.completeLock.Unlock()
 	job.IsComplete = true
 }
 
@@ -130,7 +134,7 @@ func (job *Job) Kill() {
 }
 
 // Run executes the event loop for the Job
-func (job *Job) Run(pctx context.Context, completedCh chan bool) {
+func (job *Job) Run(pctx context.Context, completedCh chan struct{}) {
 	ctx, cancel := context.WithCancel(pctx)
 
 	if job.frequency > 0 {
@@ -152,7 +156,7 @@ func (job *Job) Run(pctx context.Context, completedCh chan bool) {
 	go func() {
 		defer func() {
 			job.cleanup(ctx, cancel)
-			completedCh <- true
+			completedCh <- struct{}{}
 		}()
 		for {
 			select {

jobs/jobs_test.go

@@ -14,7 +14,7 @@ import (
 
 func TestJobRunSafeClose(t *testing.T) {
 	bus := events.NewEventBus()
-	stopCh := make(chan bool)
+	stopCh := make(chan struct{}, 1)
 	cfg := &Config{
 		Name: "myjob",
 		Exec: "sleep 10",
@@ -50,7 +50,7 @@ func TestJobRunSafeClose(t *testing.T) {
 // A Job should timeout if not started before the startupTimeout
 func TestJobRunStartupTimeout(t *testing.T) {
 	bus := events.NewEventBus()
-	stopCh := make(chan bool)
+	stopCh := make(chan struct{}, 1)
 	cfg := &Config{Name: "myjob", Exec: "true",
 		When: &WhenConfig{Source: "never", Once: "startup", Timeout: "100ms"}}
 	cfg.Validate(noop)
@@ -88,7 +88,7 @@ func TestJobRunStartupTimeout(t *testing.T) {
 // A Job should not timeout if started before the startupTimeout
 func TestJobRunStartupNoTimeout(t *testing.T) {
 	bus := events.NewEventBus()
-	stopCh := make(chan bool)
+	stopCh := make(chan struct{}, 1)
 	cfg := &Config{Name: "myjob", Exec: "sleep 5",
 		When: &WhenConfig{Timeout: "500ms"}}
 	cfg.Validate(noop)
@@ -127,7 +127,7 @@ func TestJobRunStartupNoTimeout(t *testing.T) {
 func TestJobRunRestarts(t *testing.T) {
 	runRestartsTest := func(restarts interface{}, expected int) {
 		bus := events.NewEventBus()
-		stopCh := make(chan bool)
+		stopCh := make(chan struct{}, 1)
 		cfg := &Config{
 			Name:            "myjob",
 			whenEvent:       events.GlobalStartup,
@@ -165,7 +165,7 @@ func TestJobRunRestarts(t *testing.T) {
 
 func TestJobRunPeriodic(t *testing.T) {
 	bus := events.NewEventBus()
-	stopCh := make(chan bool)
+	stopCh := make(chan struct{}, 1)
 	cfg := &Config{
 		Name: "myjob",
 		Exec: []string{"./testdata/test.sh", "doStuff", "runPeriodicTest"},
@@ -208,7 +208,7 @@ func TestJobRunPeriodic(t *testing.T) {
 func TestJobMaintenance(t *testing.T) {
 	testFunc := func(t *testing.T, startingState JobStatus, event events.Event) JobStatus {
 		bus := events.NewEventBus()
-		stopCh := make(chan bool)
+		stopCh := make(chan struct{}, 1)
 		cfg := &Config{
 			Name: "myjob",
 			Exec: "true",