Ignore records older than TOMBSTONE_LIFESPAN (#71)

Author: relistan

catalog/services_state.go

@@ -296,6 +296,17 @@ func (state *ServicesState) AddServiceEntry(newSvc service.Service) {
 	state.Lock()
 	defer state.Unlock()
 
+	// Some weird edge cases can cause very old stuff to get broadcast.  This
+	// can end up in a broadcast/tombstone/broadcast loop. We'll attempt to
+	// prevent that by dropping anything older than the tombstone window.
+	if newSvc.IsStale(TOMBSTONE_LIFESPAN) {
+		log.Warnf(
+			"Dropping stale service received on gossip: %s:%s (%s)",
+			newSvc.Hostname, newSvc.Name, newSvc.ID,
+		)
+		return
+	}
+
 	if !state.HasServer(newSvc.Hostname) {
 		state.Servers[newSvc.Hostname] = NewServer(newSvc.Hostname)
 	}

catalog/services_state_test.go

@@ -4,13 +4,15 @@ import (
 	"bytes"
 	"encoding/json"
 	"fmt"
+	"os"
 	"regexp"
 	"sync"
 	"testing"
 	"time"
 
 	"github.com/NinesStack/sidecar/service"
 	"github.com/relistan/go-director"
+	log "github.com/sirupsen/logrus"
 	. "github.com/smartystreets/goconvey/convey"
 )
 
@@ -130,7 +132,7 @@ func Test_ServicesStateWithData(t *testing.T) {
 				So(state.Servers[anotherHostname].Services[svc.ID], ShouldNotBeNil)
 			})
 
-			Convey("Doesn't merge a stale service", func() {
+			Convey("Doesn't merge an update that is older than what we have", func() {
 				state.AddServiceEntry(svc)
 
 				staleService := service.Service{
@@ -152,6 +154,26 @@ func Test_ServicesStateWithData(t *testing.T) {
 					ShouldEqual, "101deadbeef")
 			})
 
+			Convey("Doesn't merge an update that is past the tombstone lifespan", func() {
+				staleService := service.Service{
+					ID:       "deadbeef123",
+					Name:     "stale_service",
+					Image:    "stale",
+					Created:  baseTime,
+					Hostname: anotherHostname,
+					Updated:  baseTime.Add(0 - 1*time.Minute).Add(0 - TOMBSTONE_LIFESPAN),
+					Status:   service.ALIVE,
+				}
+
+				capture := LogCapture(func() {
+					state.AddServiceEntry(staleService)
+				})
+
+				_, ok := state.Servers[anotherHostname]
+				So(ok, ShouldBeFalse)
+				So(capture, ShouldContainSubstring, "Dropping stale service received on gossip")
+			})
+
 			Convey("Updates the LastUpdated time for the server", func() {
 				newDate := svc.Updated.AddDate(0, 0, 5)
 				svc.Updated = newDate
@@ -723,7 +745,7 @@ func Test_ClusterMembershipManagement(t *testing.T) {
 
 func Test_DecodeStream(t *testing.T) {
 	Convey("Test decoding stream", t, func() {
-		serv := service.Service{ID: "007", Name: "api", Hostname: "some-aws-host", Status: 1}
+		serv := service.Service{ID: "007", Name: "api", Hostname: "some-aws-host", Status: 1, Updated: time.Now().UTC()}
 		state := NewServicesState()
 		state.AddServiceEntry(serv)
 
@@ -841,3 +863,13 @@ func ShouldMatch(actual interface{}, expected ...interface{}) string {
 
 	return ""
 }
+
+// LogCapture logs for async testing where we can't get a nice handle on thigns
+func LogCapture(fn func()) string {
+	capture := &bytes.Buffer{}
+	log.SetOutput(capture)
+	fn()
+	log.SetOutput(os.Stdout)
+
+	return capture.String()
+}

catalog/url_listener_test.go

@@ -4,6 +4,7 @@ import (
 	"net/http"
 	"net/url"
 	"testing"
+	"time"
 
 	"github.com/NinesStack/sidecar/service"
 	"github.com/relistan/go-director"
@@ -65,7 +66,7 @@ func Test_Listen(t *testing.T) {
 		hostname := "grendel"
 
 		svcId1 := "deadbeef123"
-		service1 := service.Service{ID: svcId1, Hostname: hostname}
+		service1 := service.Service{ID: svcId1, Hostname: hostname, Updated: time.Now().UTC()}
 
 		state := NewServicesState()
 		state.Hostname = hostname

service/service.go

@@ -10,7 +10,7 @@ import (
 	"time"
 
 	"github.com/NinesStack/sidecar/output"
-	"github.com/fsouza/go-dockerclient"
+	docker "github.com/fsouza/go-dockerclient"
 	log "github.com/sirupsen/logrus"
 )
 
@@ -65,6 +65,12 @@ func (svc *Service) Invalidates(otherSvc *Service) bool {
 	return otherSvc != nil && svc.Updated.After(otherSvc.Updated)
 }
 
+func (svc *Service) IsStale(lifespan time.Duration) bool {
+	oldestAllowed := time.Now().UTC().Add(0 - lifespan)
+	// We add a fudge factor for clock drift
+	return svc.Updated.Before(oldestAllowed.Add(0 - 1*time.Minute))
+}
+
 func (svc *Service) Format() string {
 	var ports []string
 	for _, port := range svc.Ports {

service/service_test.go

@@ -3,8 +3,9 @@ package service
 import (
 	"os"
 	"testing"
+	"time"
 
-	"github.com/fsouza/go-dockerclient"
+	docker "github.com/fsouza/go-dockerclient"
 	. "github.com/smartystreets/goconvey/convey"
 )
 
@@ -137,3 +138,23 @@ func Test_ToService(t *testing.T) {
 		})
 	})
 }
+
+func Test_IsStale(t *testing.T) {
+	Convey("IsStale()", t, func() {
+		Convey("identifies records that are too old to process", func() {
+			lifespan := 1 * time.Hour
+			lastUpdated := time.Now().UTC().Add(0-lifespan).Add(0-2 * time.Minute)
+
+			svc := &Service{
+				Name:     "hrunting",
+				Updated:  lastUpdated,
+				Hostname: "beowulf",
+			}
+
+			So(svc.IsStale(lifespan), ShouldBeTrue)
+
+			svc.Updated = time.Now().UTC().Add(0-lifespan)
+			So(svc.IsStale(62*time.Minute), ShouldBeFalse)
+		})
+	})
+}