Merge pull request #23 from amfern/job-overview

jobliststubs command

Author: jippi

README.md

@@ -152,7 +152,13 @@ The output will be equal to the [Nomad Evaluation API structure](https://www.nom
 
 `nomad-firehose jobs` will monitor all job changes in the Nomad cluster and emit a firehose event per change to the configured sink.
 
-The output will be equal to the *full* [Nomad Job API structure](https://www.nomadproject.io/api/jobs.html)
+The output will be equal to the *full* [Nomad Job API structure](https://www.nomadproject.io/api/jobs.html#read-job)
+
+### `jobliststubs`
+
+`nomad-firehose jobliststubs` will monitor all job changes in the Nomad cluster and emit a firehose event per change to the configured sink.
+
+The output will be equal to the job list [Nomad Job API structure](https://www.nomadproject.io/api/jobs.html#list-jobs)
 
 ### `deployments`
 

command/jobs/app.go renamed to command/jobs/base.go

@@ -1,7 +1,6 @@
 package jobs
 
 import (
-	"encoding/json"
 	"fmt"
 	"time"
 
@@ -10,8 +9,12 @@ import (
 	log "github.com/sirupsen/logrus"
 )
 
+
+type WatchJobListFunc func(job *nomad.JobListStub)
+
+
 // Firehose ...
-type Firehose struct {
+type FirehoseBase struct {
 	lastChangeIndex  uint64
 	lastChangeTimeCh chan interface{}
 	nomadClient      *nomad.Client
@@ -20,7 +23,7 @@ type Firehose struct {
 }
 
 // NewFirehose ...
-func NewFirehose() (*Firehose, error) {
+func NewFirehoseBase() (*FirehoseBase, error) {
 	nomadClient, err := nomad.NewClient(nomad.DefaultConfig())
 	if err != nil {
 		return nil, err
@@ -31,23 +34,19 @@ func NewFirehose() (*Firehose, error) {
 		return nil, err
 	}
 
-	return &Firehose{
+	return &FirehoseBase{
 		nomadClient:      nomadClient,
 		sink:             sink,
 		stopCh:           make(chan struct{}, 1),
 		lastChangeTimeCh: make(chan interface{}, 1),
 	}, nil
 }
 
-func (f *Firehose) Name() string {
-	return "jobs"
-}
-
-func (f *Firehose) UpdateCh() <-chan interface{} {
+func (f *FirehoseBase) UpdateCh() <-chan interface{} {
 	return f.lastChangeTimeCh
 }
 
-func (f *Firehose) SetRestoreValue(restoreValue interface{}) error {
+func (f *FirehoseBase) SetRestoreValue(restoreValue interface{}) error {
 	switch restoreValue.(type) {
 	case int:
 		f.lastChangeIndex = uint64(restoreValue.(int))
@@ -60,11 +59,11 @@ func (f *Firehose) SetRestoreValue(restoreValue interface{}) error {
 }
 
 // Start the firehose
-func (f *Firehose) Start() {
+func (f *FirehoseBase) Start(w WatchJobListFunc) {
 	go f.sink.Start()
 
 	// watch for allocation changes
-	go f.watch()
+	go f.watch(w)
 
 	// Save the last event time every 5s
 	go f.persistLastChangeTime(5 * time.Second)
@@ -77,15 +76,15 @@ func (f *Firehose) Start() {
 }
 
 // Stop the firehose
-func (f *Firehose) Stop() {
+func (f *FirehoseBase) Stop() {
 	close(f.stopCh)
 	f.sink.Stop()
 }
 
 // Write the Last Change Time to Consul so if the process restarts,
 // it will try to resume from where it left off, not emitting tons of double events for
 // old events
-func (f *Firehose) persistLastChangeTime(interval time.Duration) {
+func (f *FirehoseBase) persistLastChangeTime(interval time.Duration) {
 	ticker := time.NewTicker(interval)
 
 	for {
@@ -99,18 +98,8 @@ func (f *Firehose) persistLastChangeTime(interval time.Duration) {
 	}
 }
 
-// Publish an update from the firehose
-func (f *Firehose) Publish(update *nomad.Job) {
-	b, err := json.Marshal(update)
-	if err != nil {
-		log.Error(err)
-	}
-
-	f.sink.Put(b)
-}
-
 // Continously watch for changes to the allocation list and publish it as updates
-func (f *Firehose) watch() {
+func (f *FirehoseBase) watch(w WatchJobListFunc) {
 	q := &nomad.QueryOptions{
 		WaitIndex:  f.lastChangeIndex,
 		WaitTime:   5 * time.Minute,
@@ -148,15 +137,7 @@ func (f *Firehose) watch() {
 				newMax = job.ModifyIndex
 			}
 
-			go func(jobID string) {
-				fullJob, _, err := f.nomadClient.Jobs().Info(jobID, &nomad.QueryOptions{})
-				if err != nil {
-					log.Errorf("Could not read job %s: %s", jobID, err)
-					return
-				}
-
-				f.Publish(fullJob)
-			}(job.ID)
+			w(job)
 		}
 
 		// Update WaitIndex and Last Change Time for next iteration

command/jobs/job.go

@@ -0,0 +1,55 @@
+package jobs
+
+import (
+	"encoding/json"
+
+	nomad "github.com/hashicorp/nomad/api"
+	log "github.com/sirupsen/logrus"
+)
+
+// Firehose ...
+type JobFirehose struct {
+	FirehoseBase
+}
+
+// NewFirehose ...
+func NewJobFirehose() (*JobFirehose, error) {
+	base, err := NewFirehoseBase()
+	if err != nil {
+		return nil, err
+	}
+
+	return &JobFirehose{FirehoseBase: *base}, nil
+}
+
+
+func (f *JobFirehose) Name() string {
+	return "jobs"
+}
+
+// Publish an update from the firehose
+func (f *JobFirehose) Publish(update *nomad.Job) {
+	b, err := json.Marshal(update)
+	if err != nil {
+		log.Error(err)
+	}
+
+	f.sink.Put(b)
+}
+
+
+func (f *JobFirehose) Start() {
+	f.FirehoseBase.Start(f.watchJobList)
+}
+
+func (f *JobFirehose) watchJobList(job *nomad.JobListStub) {
+	go func(jobID string) {
+		fullJob, _, err := f.nomadClient.Jobs().Info(jobID, &nomad.QueryOptions{})
+		if err != nil {
+			log.Errorf("Could not read job %s: %s", jobID, err)
+			return
+		}
+
+		f.Publish(fullJob)
+	}(job.ID)
+}

command/jobs/jobsliststub.go

@@ -0,0 +1,47 @@
+package jobs
+
+import (
+	"encoding/json"
+
+	nomad "github.com/hashicorp/nomad/api"
+	log "github.com/sirupsen/logrus"
+)
+
+// Firehose ...
+type JobListStubFirehose struct {
+	FirehoseBase
+}
+
+// NewFirehose ...
+func NewJobListStubFirehose() (*JobListStubFirehose, error) {
+	base, err := NewFirehoseBase()
+	if err != nil {
+		return nil, err
+	}
+
+	return &JobListStubFirehose{FirehoseBase: *base}, nil
+}
+
+func (f *JobListStubFirehose) Name() string {
+	return "jobliststub"
+}
+
+func (f *JobListStubFirehose) Start() {
+	f.FirehoseBase.Start(f.watchJobList)
+}
+
+// Publish an update from the firehose
+func (f *JobListStubFirehose) Publish(update *nomad.JobListStub) {
+	b, err := json.Marshal(update)
+	if err != nil {
+		log.Error(err)
+	}
+
+	f.sink.Put(b)
+}
+
+func (f *JobListStubFirehose) watchJobList(job *nomad.JobListStub) {
+	f.Publish(job)
+}
+
+

main.go

@@ -94,7 +94,25 @@ func main() {
 			Name:  "jobs",
 			Usage: "Firehose nomad job changes",
 			Action: func(c *cli.Context) error {
-				firehose, err := jobs.NewFirehose()
+				firehose, err := jobs.NewJobFirehose()
+				if err != nil {
+					return err
+				}
+
+				manager := helper.NewManager(firehose)
+				if err := manager.Start(); err != nil {
+					log.Fatal(err)
+					return err
+				}
+
+				return nil
+			},
+		},
+		{
+			Name:  "jobliststubs",
+			Usage: "Firehose nomad job info changes",
+			Action: func(c *cli.Context) error {
+				firehose, err := jobs.NewJobListStubFirehose()
 				if err != nil {
 					return err
 				}