Create a channel to surface error from server

Created a channel to surface server errors

Create a channel to surface error from server

Dummy commit

Edited linters

manager/runner.go

@@ -23,7 +23,7 @@ import (
 	"github.com/hashicorp/consul-template/template"
 	"github.com/hashicorp/consul-template/watch"
 
-	multierror "github.com/hashicorp/go-multierror"
+	"github.com/hashicorp/go-multierror"
 )
 
 const (
@@ -38,6 +38,10 @@ type Runner struct {
 	ErrCh  chan error
 	DoneCh chan struct{}
 
+	// ServerCh is a channel to surface error responses from the server up the calling stack
+	// and will only hold a maximum of one error at a time
+	ServerCh chan error
+
 	// config is the Config that created this Runner. It is used internally to
 	// construct other objects and pass data.
 	config *config.Config
@@ -198,7 +202,8 @@ func NewRunner(config *config.Config, dry bool) (*Runner, error) {
 
 	runner := &Runner{
 		ErrCh:         make(chan error),
-		DoneCh:        make(chan struct{}),
+		ServerCh:      make(chan error),
+		DoneCh:        make(chan struct{}, 1),
 		config:        config,
 		dry:           dry,
 		inStream:      os.Stdin,
@@ -434,11 +439,20 @@ func (r *Runner) Start() {
 			log.Printf("[ERR] (runner) watcher reported error: %s", err)
 			r.ErrCh <- err
 			return
-
+		case err := <-r.watcher.ServerCh():
+			// If we got a server error we push the error up the stack
+			log.Printf("[ERR] (runner) sending server error back to caller")
+			// Drain the error channel if anything already exists
+			select {
+			case <-r.ServerCh:
+				continue
+			default:
+			}
+			r.ServerCh <- err
+			goto OUTER
 		case err := <-r.vaultTokenWatcher.ErrCh():
 			// Push the error back up the stack
 			log.Printf("[ERR] (runner): %s", err)
-			r.ErrCh <- err
 			return
 
 		case tmpl := <-r.quiescenceCh:

watch/view.go

@@ -121,7 +121,7 @@ func (v *View) DataAndLastIndex() (interface{}, uint64) {
 // accounts for interrupts on the interrupt channel. This allows the poll
 // function to be fired in a goroutine, but then halted even if the fetch
 // function is in the middle of a blocking query.
-func (v *View) poll(viewCh chan<- *View, errCh chan<- error) {
+func (v *View) poll(viewCh chan<- *View, errCh chan<- error, serverErrCh chan<- error) {
 	var retries int
 
 	for {
@@ -162,6 +162,7 @@ func (v *View) poll(viewCh chan<- *View, errCh chan<- error) {
 		case err := <-fetchErrCh:
 			if !errors.Is(err, errLookup) && v.retryFunc != nil {
 				retry, sleep := v.retryFunc(retries)
+				serverErrCh <- err
 				if retry {
 					log.Printf("[WARN] (view) %s (retry attempt %d after %q)",
 						err, retries+1, sleep)

watch/view_test.go

@@ -20,8 +20,9 @@ func TestPoll_returnsViewCh(t *testing.T) {
 
 	viewCh := make(chan *View)
 	errCh := make(chan error)
+	serverErrCh := make(chan error)
 
-	go view.poll(viewCh, errCh)
+	go view.poll(viewCh, errCh, serverErrCh)
 	defer view.stop()
 
 	select {
@@ -44,8 +45,9 @@ func TestPoll_returnsErrCh(t *testing.T) {
 
 	viewCh := make(chan *View)
 	errCh := make(chan error)
+	serverErrCh := make(chan error)
 
-	go view.poll(viewCh, errCh)
+	go view.poll(viewCh, errCh, serverErrCh)
 	defer view.stop()
 
 	select {
@@ -71,8 +73,9 @@ func TestPoll_stopsViewStopCh(t *testing.T) {
 
 	viewCh := make(chan *View)
 	errCh := make(chan error)
+	serverErrCh := make(chan error)
 
-	go view.poll(viewCh, errCh)
+	go view.poll(viewCh, errCh, serverErrCh)
 	view.stop()
 
 	select {
@@ -95,8 +98,9 @@ func TestPoll_once(t *testing.T) {
 
 	viewCh := make(chan *View)
 	errCh := make(chan error)
+	serverErrCh := make(chan error)
 
-	go view.poll(viewCh, errCh)
+	go view.poll(viewCh, errCh, serverErrCh)
 	defer view.stop()
 
 	select {
@@ -133,8 +137,9 @@ func TestPoll_retries(t *testing.T) {
 
 	viewCh := make(chan *View)
 	errCh := make(chan error)
+	serverErrCh := make(chan error)
 
-	go view.poll(viewCh, errCh)
+	go view.poll(viewCh, errCh, serverErrCh)
 	defer view.stop()
 
 	select {
@@ -143,6 +148,9 @@ func TestPoll_retries(t *testing.T) {
 	case <-time.After(100 * time.Millisecond):
 	}
 
+	// need to receive error to avoid timeout
+	<-serverErrCh
+
 	select {
 	case <-viewCh:
 		// Got this far, so the test passes
@@ -291,8 +299,9 @@ func TestStop_stopsPolling(t *testing.T) {
 
 	viewCh := make(chan *View)
 	errCh := make(chan error)
+	serverErrCh := make(chan error)
 
-	go view.poll(viewCh, errCh)
+	go view.poll(viewCh, errCh, serverErrCh)
 	view.stop()
 
 	select {

watch/watcher.go

@@ -30,6 +30,9 @@ type Watcher struct {
 	// errCh is the chan where any errors will be published.
 	errCh chan error
 
+	// serverCh is the chan where response errors from the server will be published
+	serverCh chan error
+
 	// blockQueryWaitTime is amount of time in seconds to do a blocking query for
 	blockQueryWaitTime time.Duration
 
@@ -95,6 +98,7 @@ func NewWatcher(i *NewWatcherInput) *Watcher {
 		depViewMap:         make(map[string]*View),
 		dataCh:             make(chan *View, dataBufferSize),
 		errCh:              make(chan error),
+		serverCh:           make(chan error),
 		maxStale:           i.MaxStale,
 		once:               i.Once,
 		blockQueryWaitTime: i.BlockQueryWaitTime,
@@ -121,6 +125,15 @@ func (w *Watcher) ErrCh() <-chan error {
 	return w.errCh
 }
 
+// ServerCh returns a read-only channel of errors returned by the server
+// as a response to each Consul instance
+func (w *Watcher) ServerCh() <-chan error {
+	if w == nil {
+		return nil
+	}
+	return w.serverCh
+}
+
 // Add adds the given dependency to the list of monitored dependencies
 // and start the associated view. If the dependency already exists, no action is
 // taken.
@@ -169,7 +182,7 @@ func (w *Watcher) Add(d dep.Dependency) (bool, error) {
 	log.Printf("[TRACE] (watcher) %s starting", d)
 
 	w.depViewMap[d.String()] = v
-	go v.poll(w.dataCh, w.errCh)
+	go v.poll(w.dataCh, w.errCh, w.serverCh)
 
 	return true, nil
 }