Fix Consul SD loop exiting on EOF (#788)

* switched io.EOF for quit to errStopped

* adding unit test for consul sd EOF

sd/consul/instancer.go

@@ -1,8 +1,8 @@
 package consul
 
 import (
+	"errors"
 	"fmt"
-	"io"
 	"time"
 
 	consul "github.com/hashicorp/consul/api"
@@ -15,6 +15,9 @@ import (
 
 const defaultIndex = 0
 
+// errStopped notifies the loop to quit. aka stopped via quitc
+var errStopped = errors.New("quit and closed consul instancer")
+
 // Instancer yields instances for a service in Consul.
 type Instancer struct {
 	cache       *instance.Cache
@@ -66,7 +69,7 @@ func (s *Instancer) loop(lastIndex uint64) {
 	for {
 		instances, lastIndex, err = s.getInstances(lastIndex, s.quitc)
 		switch {
-		case err == io.EOF:
+		case err == errStopped:
 			return // stopped via quitc
 		case err != nil:
 			s.logger.Log("err", err)
@@ -125,7 +128,7 @@ func (s *Instancer) getInstances(lastIndex uint64, interruptc chan struct{}) ([]
 	case res := <-resc:
 		return res.instances, res.index, nil
 	case <-interruptc:
-		return nil, 0, io.EOF
+		return nil, 0, errStopped
 	}
 }
 

sd/consul/instancer_test.go

@@ -2,9 +2,10 @@ package consul
 
 import (
 	"context"
-	"testing"
-
 	consul "github.com/hashicorp/consul/api"
+	"io"
+	"testing"
+	"time"
 
 	"github.com/go-kit/kit/log"
 	"github.com/go-kit/kit/sd"
@@ -131,3 +132,73 @@ func TestInstancerAddressOverride(t *testing.T) {
 		t.Errorf("want %q, have %q", want, have)
 	}
 }
+
+type eofTestClient struct {
+	client *testClient
+	eofSig chan bool
+	called chan struct{}
+}
+
+func neweofTestClient(client *testClient, sig chan bool, called chan struct{}) Client {
+	return &eofTestClient{client: client, eofSig: sig, called: called}
+}
+
+func (c *eofTestClient) Register(r *consul.AgentServiceRegistration) error {
+	return c.client.Register(r)
+}
+
+func (c *eofTestClient) Deregister(r *consul.AgentServiceRegistration) error {
+	return c.client.Deregister(r)
+}
+
+func (c *eofTestClient) Service(service, tag string, passingOnly bool, queryOpts *consul.QueryOptions) ([]*consul.ServiceEntry, *consul.QueryMeta, error) {
+	c.called <- struct{}{}
+	shouldEOF := <-c.eofSig
+	if shouldEOF {
+		return nil, &consul.QueryMeta{}, io.EOF
+	}
+	return c.client.Service(service, tag, passingOnly, queryOpts)
+}
+
+func TestInstancerWithEOF(t *testing.T) {
+	var (
+		sig    = make(chan bool, 1)
+		called = make(chan struct{}, 1)
+		logger = log.NewNopLogger()
+		client = neweofTestClient(newTestClient(consulState), sig, called)
+	)
+
+	sig <- false
+	s := NewInstancer(client, logger, "search", []string{"api"}, true)
+	defer s.Stop()
+
+	select {
+	case <-called:
+	case <-time.Tick(time.Millisecond * 500):
+		t.Error("failed, to receive call")
+	}
+
+	state := s.cache.State()
+	if want, have := 2, len(state.Instances); want != have {
+		t.Errorf("want %d, have %d", want, have)
+	}
+
+	// some error occurred resulting in io.EOF
+	sig <- true
+
+	// Service Called Once
+	select {
+	case <-called:
+	case <-time.Tick(time.Millisecond * 500):
+		t.Error("failed, to receive call in time")
+	}
+
+	sig <- false
+
+	// loop should continue
+	select {
+	case <-called:
+	case <-time.Tick(time.Millisecond * 500):
+		t.Error("failed, to receive call in time")
+	}
+}