Repository: go-kit/kit
Pull request #1215 (2022-02-13 01:41:59)

Fixed instancer loop logic that was causing Consul to get spammed
Address issue #1214 - Instancer spamming Consul with requests

The instancer loop was never updating the lastIndex but passing it to getInstances which caused Instancer to spam Consul repeatedly and nearly killed our Consul agent pods and time instances were added or removed from Consul for that service.

peterbourgon: Cool! Can you add a test which fails on the previous version and passes on this branch?
jkratz55: Hey @peterbourgon, I added a test case to the PR which fails on the old version and passes with the fix. Since the logic was in a loop that ran forever, the unit tests seem odd but seem to work. I'm just pumping out an error on a channel when the mock client detects the same index value was passed twice in a row and assumes every lookup Consul returns a new index since it isn't doing blocking queries in the mock. 

Patch:
diff --git a/README.md b/README.md
index ce92374..bea6ff4 100644
--- a/README.md
+++ b/README.md
@@ -117,7 +117,7 @@ Projects with a ★ have had particular influence on Go kit's design (or vice-ve
 - [Negroni](https://github.com/codegangsta/negroni)
 - [Goji](https://github.com/zenazn/goji)
 - [Martini](https://github.com/go-martini/martini)
-- [Beego](http://beego.me/)
+- [Beego](https://beego.vip/)
 - [Revel](https://revel.github.io/) (considered [harmful](https://github.com/go-kit/kit/issues/350))
 - [GoBuffalo](https://gobuffalo.io/)
 
diff --git a/sd/consul/instancer.go b/sd/consul/instancer.go
index 1110665..3215695 100644
--- a/sd/consul/instancer.go
+++ b/sd/consul/instancer.go
@@ -65,9 +65,9 @@ func (s *Instancer) loop(lastIndex uint64) {
 		instances []string
 		err       error
 		d         time.Duration = 10 * time.Millisecond
+		index     uint64
 	)
 	for {
-		index := lastIndex
 		instances, index, err = s.getInstances(lastIndex, s.quitc)
 		switch {
 		case errors.Is(err, errStopped):
@@ -82,11 +82,12 @@ func (s *Instancer) loop(lastIndex uint64) {
 			time.Sleep(d)
 			d = conn.Exponential(d)
 		case index < lastIndex:
-			s.logger.Log("err", "index is less than previous; reseting to default")
+			s.logger.Log("err", "index is less than previous; resetting to default")
 			lastIndex = defaultIndex
 			time.Sleep(d)
 			d = conn.Exponential(d)
 		default:
+			lastIndex = index
 			s.cache.Update(sd.Event{Instances: instances})
 			d = 10 * time.Millisecond
 		}


Test patch:
diff --git a/sd/consul/instancer_test.go b/sd/consul/instancer_test.go
index 03aa181..84737d1 100644
--- a/sd/consul/instancer_test.go
+++ b/sd/consul/instancer_test.go
@@ -2,6 +2,7 @@ package consul
 
 import (
 	"context"
+	"fmt"
 	"io"
 	"testing"
 	"time"
@@ -261,3 +262,66 @@ func TestInstancerWithInvalidIndex(t *testing.T) {
 		t.Error("failed, to receive call in time")
 	}
 }
+
+type indexTestClient struct {
+	client *testClient
+	index  uint64
+	errs   chan error
+}
+
+func newIndexTestClient(c *testClient, errs chan error) *indexTestClient {
+	return &indexTestClient{
+		client: c,
+		index:  0,
+		errs:   errs,
+	}
+}
+
+func (i *indexTestClient) Register(r *consul.AgentServiceRegistration) error {
+	return i.client.Register(r)
+}
+
+func (i *indexTestClient) Deregister(r *consul.AgentServiceRegistration) error {
+	return i.client.Deregister(r)
+}
+
+func (i *indexTestClient) Service(service, tag string, passingOnly bool, queryOpts *consul.QueryOptions) ([]*consul.ServiceEntry, *consul.QueryMeta, error) {
+
+	// Assumes this is the first call Service, loop hasn't begun running yet
+	if i.index == 0 && queryOpts.WaitIndex == 0 {
+		i.index = 100
+		entries, meta, err := i.client.Service(service, tag, passingOnly, queryOpts)
+		meta.LastIndex = i.index
+		return entries, meta, err
+	}
+
+	if queryOpts.WaitIndex < i.index {
+		i.errs <- fmt.Errorf("wait index %d is less than or equal to previous value", queryOpts.WaitIndex)
+	}
+
+	entries, meta, err := i.client.Service(service, tag, passingOnly, queryOpts)
+	i.index++
+	meta.LastIndex = i.index
+	return entries, meta, err
+}
+
+func TestInstancerLoopIndex(t *testing.T) {
+
+	var (
+		errs   = make(chan error, 1)
+		logger = log.NewNopLogger()
+		client = newIndexTestClient(newTestClient(consulState), errs)
+	)
+
+	go func() {
+		for err := range errs {
+			t.Error(err)
+			t.FailNow()
+		}
+	}()
+
+	instancer := NewInstancer(client, logger, "search", []string{"api"}, true)
+	defer instancer.Stop()
+
+	time.Sleep(2 * time.Second)
+}