Merge pull request #986 from hashicorp/scripts

Implement script checks

api/tasks.go

@@ -19,7 +19,8 @@ type ServiceCheck struct {
 	Id       string
 	Name     string
 	Type     string
-	Script   string
+	Cmd      string
+	Args     []string
 	Path     string
 	Protocol string
 	Interval time.Duration

client/client.go

@@ -1180,11 +1180,11 @@ func (c *Client) setupConsulClient() error {
 
 // syncConsul removes services of tasks which are no longer in running state
 func (c *Client) syncConsul() {
-	sync := time.After(consulSyncInterval)
+	sync := time.NewTicker(consulSyncInterval)
 	for {
 		select {
-		case <-sync:
-			var runningTasks []*structs.Task
+		case <-sync.C:
+			services := make(map[string]struct{})
 			// Get the existing allocs
 			c.allocLock.RLock()
 			allocs := make([]*AllocRunner, 0, len(c.allocs))
@@ -1199,15 +1199,18 @@ func (c *Client) syncConsul() {
 				for taskName, taskState := range taskStates {
 					if taskState.State == structs.TaskStateRunning {
 						if tr, ok := ar.tasks[taskName]; ok {
-							runningTasks = append(runningTasks, tr.task)
+							for _, service := range tr.task.Services {
+								services[service.ID(ar.alloc.ID, tr.task.Name)] = struct{}{}
+							}
 						}
 					}
 				}
 			}
-			if err := c.consulService.KeepServices(runningTasks); err != nil {
+			if err := c.consulService.KeepServices(services); err != nil {
 				c.logger.Printf("[DEBUG] client: error removing services from non-running tasks: %v", err)
 			}
 		case <-c.shutdownCh:
+			sync.Stop()
 			c.logger.Printf("[INFO] client: shutting down consul sync")
 			return
 		}

client/consul/check.go

@@ -0,0 +1,88 @@
+package consul
+
+import (
+	"log"
+	"math/rand"
+	"sync"
+	"time"
+
+	cstructs "github.com/hashicorp/nomad/client/driver/structs"
+)
+
+// CheckRunner runs a given check in a specific interval and update a
+// corresponding Consul TTL check
+type CheckRunner struct {
+	check    Check
+	runCheck func(Check)
+	logger   *log.Logger
+	stop     bool
+	stopCh   chan struct{}
+	stopLock sync.Mutex
+
+	started     bool
+	startedLock sync.Mutex
+}
+
+// NewCheckRunner configures and returns a CheckRunner
+func NewCheckRunner(check Check, runCheck func(Check), logger *log.Logger) *CheckRunner {
+	cr := CheckRunner{
+		check:    check,
+		runCheck: runCheck,
+		logger:   logger,
+		stopCh:   make(chan struct{}),
+	}
+	return &cr
+}
+
+// Start is used to start the check. The check runs until stop is called
+func (r *CheckRunner) Start() {
+	r.startedLock.Lock()
+	defer r.startedLock.Unlock()
+	if r.started {
+		return
+	}
+	r.stopLock.Lock()
+	defer r.stopLock.Unlock()
+	go r.run()
+	r.started = true
+}
+
+// Stop is used to stop the check.
+func (r *CheckRunner) Stop() {
+	r.stopLock.Lock()
+	defer r.stopLock.Unlock()
+	if !r.stop {
+		r.stop = true
+		close(r.stopCh)
+	}
+}
+
+// run is invoked by a goroutine to run until Stop() is called
+func (r *CheckRunner) run() {
+	// Get the randomized initial pause time
+	initialPauseTime := randomStagger(r.check.Interval())
+	r.logger.Printf("[DEBUG] agent: pausing %v before first invocation of %s", initialPauseTime, r.check.ID())
+	next := time.NewTimer(initialPauseTime)
+	for {
+		select {
+		case <-next.C:
+			r.runCheck(r.check)
+			next.Reset(r.check.Interval())
+		case <-r.stopCh:
+			next.Stop()
+			return
+		}
+	}
+}
+
+// Check is an interface which check providers can implement for Nomad to run
+type Check interface {
+	Run() *cstructs.CheckResult
+	ID() string
+	Interval() time.Duration
+}
+
+// Returns a random stagger interval between 0 and the duration
+func randomStagger(intv time.Duration) time.Duration {
+	return time.Duration(uint64(rand.Int63()) % uint64(intv))
+}