func (f *Firehose) watch() {
	q := &nomad.QueryOptions{
		WaitIndex:  1,
		WaitTime:   5 * time.Minute,
		AllowStale: true,
	}
	newMax := f.lastChangeTime
	for {
		allocations, meta, err := f.nomadClient.Allocations().List(q)
		if err != nil {
			log.Errorf("Unable to fetch allocations: %s", err)
			time.Sleep(10 * time.Second)
			continue
		}
		remoteWaitIndex := meta.LastIndex
		localWaitIndex := q.WaitIndex
		if remoteWaitIndex == localWaitIndex {
			log.Debugf("Allocations index is unchanged (%d == %d)", remoteWaitIndex, localWaitIndex)
			continue
		}
		log.Debugf("Allocations index is changed (%d <> %d)", remoteWaitIndex, localWaitIndex)
		for _, allocation := range allocations {
			for taskName, taskInfo := range allocation.TaskStates {
				for _, taskEvent := range taskInfo.Events {
					if taskEvent.Time <= f.lastChangeTime {
						continue
					}
					if taskEvent.Time > newMax {
						newMax = taskEvent.Time
					}
					payload := &AllocationUpdate{
						Name:               allocation.Name,
						NodeID:             allocation.NodeID,
						AllocationID:       allocation.ID,
						EvalID:             allocation.EvalID,
						DesiredStatus:      allocation.DesiredStatus,
						DesiredDescription: allocation.DesiredDescription,
						ClientStatus:       allocation.ClientStatus,
						ClientDescription:  allocation.ClientDescription,
						JobID:              allocation.JobID,
						GroupName:          allocation.TaskGroup,
						TaskName:           taskName,
						TaskEvent:          taskEvent,
						TaskState:          taskInfo.State,
						TaskFailed:         taskInfo.Failed,
						TaskStartedAt:      &taskInfo.StartedAt,
						TaskFinishedAt:     &taskInfo.FinishedAt,
					}
					f.publish(payload)
				}
			}
		}
		q.WaitIndex = meta.LastIndex
		f.lastChangeTime = newMax
	}
}
Continously watch for changes to the allocation list and publish it as updates
package allocations
import (
	"encoding/json"
	"fmt"
	"time"
	nomad "github.com/hashicorp/nomad/api"
	"github.com/seatgeek/nomad-firehose/sink"
	log "github.com/sirupsen/logrus"
)
type Firehose struct {
	lastChangeTime   int64
	lastChangeTimeCh chan interface{}
	nomadClient      *nomad.Client
	sink             sink.Sink
	stopCh           chan struct{}
}
type AllocationUpdate struct {
	Name               string
	NodeID             string
	AllocationID       string
	DesiredStatus      string
	DesiredDescription string
	ClientStatus       string
	ClientDescription  string
	JobID              string
	GroupName          string
	TaskName           string
	EvalID             string
	TaskState          string
	TaskFailed         bool
	TaskStartedAt      *time.Time
	TaskFinishedAt     *time.Time
	TaskEvent          *nomad.TaskEvent
}
func NewFirehose() (*Firehose, error) {
	nomadClient, err := nomad.NewClient(nomad.DefaultConfig())
	if err != nil {
		return nil, err
	}
	sink, err := sink.GetSink("allocations")
	if err != nil {
		return nil, err
	}
	return &Firehose{
		nomadClient:      nomadClient,
		sink:             sink,
		stopCh:           make(chan struct{}, 1),
		lastChangeTimeCh: make(chan interface{}, 1),
	}, nil
}
func (f *Firehose) Name() string {
	return "allocations"
}
func (f *Firehose) UpdateCh() <-chan interface{} {
	return f.lastChangeTimeCh
}
func (f *Firehose) SetRestoreValue(restoreValue interface{}) error {
	switch restoreValue.(type) {
	case int:
		f.lastChangeTime = int64(restoreValue.(int))
	case int64:
		f.lastChangeTime = restoreValue.(int64)
	default:
		return fmt.Errorf("Unknown restore type '%T' with value '%+v'", restoreValue, restoreValue)
	}
	return nil
}
func (f *Firehose) Start() {
	go f.sink.Start()
	f.stopCh = make(chan struct{})
	go f.watch()
	go f.persistLastChangeTime(5 * time.Second)
	for {
		select {
		case <-f.stopCh:
			return
		}
	}
}
func (f *Firehose) Stop() {
	close(f.stopCh)
	f.sink.Stop()
}
func (f *Firehose) persistLastChangeTime(interval time.Duration) {
	ticker := time.NewTicker(interval)
	for {
		select {
		case <-f.stopCh:
			f.lastChangeTimeCh <- f.lastChangeTime
			break
		case <-ticker.C:
			f.lastChangeTimeCh <- f.lastChangeTime
		}
	}
}
func (f *Firehose) publish(update *AllocationUpdate) {
	b, err := json.Marshal(update)
	if err != nil {
		log.Error(err)
	}
	f.sink.Put(b)
}