Keepalives now work correctly when using postgres (#3650)

This commit fixes an issue where keepalives would not always fire
correctly when using the postgres store. It seems that this
regression was introduced several releases ago, but it was not
always apparent for all users.

The keepalive mechanism was always consulting etcd when retrieving
events, and as such, when postgres was enabled, keepalives could be
incorrectly passing.

Unfortunately, due to the current architecture of sensu-go and
sensu-enterprise-go, it is not possible to write a unit test that
proves this commit fixes the issue. A test will need to be added
to the sensu-enterprise-go repo instead.

An unrelated unit test has been added that shores up testing of
check history generation; this has been done to maintain parity
between sensu-go and sensu-enterprise-go for tests.

Signed-off-by: Eric Chlebek <[email protected]>

CHANGELOG.md

@@ -29,9 +29,10 @@ mistake.
 - Fixed a bug where the agent could connect to a backend using a namespace that
 doesn't exist.
 - Subscriptions can no longer be empty strings (#2932)
-### Fixed
 - The proper HTTP status codes are returned for unauthenticated & permission
 denied errors in the REST API.
+- Fixed a bug where keepalives would not always fire correctly when using
+the postgres event store.
 
 ## [5.18.1] - 2020-03-10
 

backend/backend.go

@@ -315,7 +315,7 @@ func Initialize(ctx context.Context, config *Config) (*Backend, error) {
 		DeregistrationHandler: config.DeregistrationHandler,
 		Bus:             bus,
 		Store:           stor,
-		EventStore:      stor,
+		EventStore:      eventStoreProxy,
 		LivenessFactory: liveness.EtcdFactory(b.runCtx, b.Client),
 		RingPool:        ringPool,
 		BufferSize:      viper.GetInt(FlagKeepalivedBufferSize),

backend/keepalived/keepalived.go

@@ -96,9 +96,9 @@ func New(c Config, opts ...Option) (*Keepalived, error) {
 	ctx, cancel := context.WithCancel(context.Background())
 
 	k := &Keepalived{
-		store:                 c.Store,
-		eventStore:            c.EventStore,
-		bus:                   c.Bus,
+		store:      c.Store,
+		eventStore: c.EventStore,
+		bus:        c.Bus,
 		deregistrationHandler: c.DeregistrationHandler,
 		livenessFactory:       c.LivenessFactory,
 		keepaliveChan:         make(chan interface{}, c.BufferSize),
@@ -179,7 +179,7 @@ func (k *Keepalived) initFromStore(ctx context.Context) error {
 		entityCtx := context.WithValue(ctx, corev2.NamespaceKey, keepalive.Namespace)
 		tctx, cancel := context.WithTimeout(entityCtx, k.storeTimeout)
 		defer cancel()
-		event, err := k.store.GetEventByEntityCheck(tctx, keepalive.Name, "keepalive")
+		event, err := k.eventStore.GetEventByEntityCheck(tctx, keepalive.Name, "keepalive")
 		if err != nil {
 			return err
 		}
@@ -480,6 +480,7 @@ func (k *Keepalived) dead(key string, prev liveness.State, leader bool) bool {
 	if entity == nil {
 		// The entity has been deleted, there is no longer a need to
 		// track keepalives for it.
+		lager.Debug("nil entity")
 		return true
 	}
 
@@ -493,16 +494,18 @@ func (k *Keepalived) dead(key string, prev liveness.State, leader bool) bool {
 		if err := deregisterer.Deregister(entity); err != nil {
 			lager.WithError(err).Error("error deregistering entity")
 		}
+		lager.Debug("deregistering entity")
 		return true
 	}
 
-	currentEvent, err := k.store.GetEventByEntityCheck(ctx, name, "keepalive")
+	currentEvent, err := k.eventStore.GetEventByEntityCheck(ctx, name, "keepalive")
 	if err != nil {
 		lager.WithError(err).Error("error while reading event")
 		return false
 	}
 	if currentEvent == nil {
 		// The keepalive was deleted, so bury the switch
+		lager.Debug("nil event")
 		return true
 	}
 

backend/keepalived/keepalived_test.go

@@ -61,6 +61,7 @@ func newKeepalivedTest(t *testing.T) *keepalivedTest {
 	require.NoError(t, err)
 	k, err := New(Config{
 		Store:           store,
+		EventStore:      store,
 		Bus:             bus,
 		LivenessFactory: fakeFactory,
 		BufferSize:      1,
@@ -250,6 +251,7 @@ func TestProcessRegistration(t *testing.T) {
 
 			keepalived, err := New(Config{
 				Store:           store,
+				EventStore:      store,
 				Bus:             messageBus,
 				LivenessFactory: fakeFactory,
 				WorkerCount:     1,
@@ -314,6 +316,7 @@ func TestDeadCallbackNoEntity(t *testing.T) {
 	store := &mockstore.MockStore{}
 	keepalived, err := New(Config{
 		Store:           store,
+		EventStore:      store,
 		Bus:             messageBus,
 		LivenessFactory: fakeFactory,
 		WorkerCount:     1,
@@ -347,6 +350,7 @@ func TestDeadCallbackNoEvent(t *testing.T) {
 	store := &mockstore.MockStore{}
 	keepalived, err := New(Config{
 		Store:           store,
+		EventStore:      store,
 		Bus:             messageBus,
 		LivenessFactory: fakeFactory,
 		WorkerCount:     1,

backend/store/etcd/event_store_test.go

@@ -704,3 +704,32 @@ func TestHandleExpireOnResolveEntries(t *testing.T) {
 		})
 	}
 }
+
+func TestEventStoreHistory(t *testing.T) {
+	testWithEtcd(t, func(s store.Store) {
+		ctx := store.NamespaceContext(context.Background(), "default")
+		event := corev2.FixtureEvent("foo", "bar")
+		want := []corev2.CheckHistory{}
+		for i := 0; i < 30; i++ {
+			event.Check.Executed = int64(i)
+			historyItem := corev2.CheckHistory{
+				Executed: int64(i),
+			}
+			want = append(want, historyItem)
+			_, _, err := s.UpdateEvent(ctx, event)
+			if err != nil {
+				t.Fatal(err)
+			}
+		}
+		want = want[len(want)-21:]
+		event, err := s.GetEventByEntityCheck(ctx, "foo", "bar")
+		if err != nil {
+			t.Fatal(err)
+		}
+		for i := 0; i < 21; i++ {
+		}
+		if got := event.Check.History; !reflect.DeepEqual(got, want) {
+			t.Fatalf("bad event history: got %v, want %v", got, want)
+		}
+	})
+}