Resolve test issues and event driver bugs (gravitational#6990)

lib/events/dynamoevents/dynamoevents.go

@@ -58,10 +58,6 @@ var tableSchema = []*dynamodb.AttributeDefinition{
 		AttributeName: aws.String(keyEventIndex),
 		AttributeType: aws.String("N"),
 	},
-	{
-		AttributeName: aws.String(keyEventNamespace),
-		AttributeType: aws.String("S"),
-	},
 	{
 		AttributeName: aws.String(keyCreatedAt),
 		AttributeType: aws.String("N"),
@@ -181,9 +177,6 @@ const (
 	// keyEventIndex is EventIndex key
 	keyEventIndex = "EventIndex"
 
-	// keyEventNamespace
-	keyEventNamespace = "EventNamespace"
-
 	// keyCreatedAt identifies created at key
 	keyCreatedAt = "CreatedAt"
 
@@ -530,13 +523,17 @@ func (l *Log) PostSessionSlice(slice events.SessionSlice) error {
 		if err != nil {
 			return trace.Wrap(err)
 		}
+
+		timeAt := time.Unix(0, chunk.Time).In(time.UTC)
+
 		event := event{
 			SessionID:      slice.SessionID,
 			EventNamespace: defaults.Namespace,
 			EventType:      chunk.EventType,
 			EventIndex:     chunk.EventIndex,
-			CreatedAt:      time.Unix(0, chunk.Time).In(time.UTC).Unix(),
+			CreatedAt:      timeAt.Unix(),
 			Fields:         string(data),
+			CreatedAtDate:  timeAt.Format(iso8601DateFormat),
 		}
 		l.setExpiry(&event)
 		item, err := dynamodbattribute.MarshalMap(event)
@@ -667,7 +664,7 @@ func (l *Log) SearchEvents(fromUTC, toUTC time.Time, namespace string, eventType
 	eventArr := make([]events.AuditEvent, 0, len(rawEvents))
 	for _, rawEvent := range rawEvents {
 		var fields events.EventFields
-		if err := utils.FastUnmarshal([]byte(rawEvent.Fields), fields); err != nil {
+		if err := utils.FastUnmarshal([]byte(rawEvent.Fields), &fields); err != nil {
 			return nil, "", trace.Wrap(err)
 		}
 		event, err := events.FromEventFields(fields)

lib/events/dynamoevents/dynamoevents_test.go

@@ -44,6 +44,8 @@ import (
 	"github.com/gravitational/trace"
 )
 
+const dynamoDBLargeQueryRetries int = 10
+
 func TestMain(m *testing.M) {
 	utils.InitLoggerForTests()
 	os.Exit(m.Run())
@@ -75,8 +77,7 @@ func (s *DynamoeventsSuite) SetUpSuite(c *check.C) {
 	s.log = log
 	s.EventsSuite.Log = log
 	s.EventsSuite.Clock = fakeClock
-	s.EventsSuite.QueryDelay = time.Second
-
+	s.EventsSuite.QueryDelay = time.Second * 5
 }
 
 func (s *DynamoeventsSuite) SetUpTest(c *check.C) {
@@ -96,7 +97,8 @@ func (s *DynamoeventsSuite) TestSessionEventsCRUD(c *check.C) {
 	err := s.log.deleteAllItems()
 	c.Assert(err, check.IsNil)
 
-	for i := 0; i < 4000; i++ {
+	const eventCount int = 4000
+	for i := 0; i < eventCount; i++ {
 		err := s.Log.EmitAuditEventLegacy(events.UserLocalLoginE, events.EventFields{
 			events.LoginMethod:        events.LoginMethodSAML,
 			events.AuthAttemptSuccess: true,
@@ -106,13 +108,21 @@ func (s *DynamoeventsSuite) TestSessionEventsCRUD(c *check.C) {
 		c.Assert(err, check.IsNil)
 	}
 
-	time.Sleep(s.EventsSuite.QueryDelay)
+	var history []events.AuditEvent
 
-	history, _, err := s.Log.SearchEvents(s.Clock.Now().Add(-1*time.Hour), s.Clock.Now().Add(time.Hour), defaults.Namespace, nil, 0, "")
-	c.Assert(err, check.IsNil)
+	for i := 0; i < dynamoDBLargeQueryRetries; i++ {
+		time.Sleep(s.EventsSuite.QueryDelay)
+
+		history, _, err = s.Log.SearchEvents(s.Clock.Now().Add(-1*time.Hour), s.Clock.Now().Add(time.Hour), defaults.Namespace, nil, 0, "")
+		c.Assert(err, check.IsNil)
+
+		if len(history) == eventCount {
+			break
+		}
+	}
 
 	// `check.HasLen` prints the entire array on failure, which pollutes the output
-	c.Assert(len(history), check.Equals, 4000)
+	c.Assert(len(history), check.Equals, eventCount)
 }
 
 // TestIndexExists tests functionality of the `Log.indexExists` function.

lib/events/firestoreevents/firestoreevents.go

@@ -465,7 +465,7 @@ func (l *Log) SearchEvents(fromUTC, toUTC time.Time, namespace string, eventType
 
 	var lastKey int64
 	var values []events.EventFields
-	var parsedStartKey int64 = -1
+	var parsedStartKey int64
 	var err error
 
 	if startKey != "" {
@@ -477,7 +477,7 @@ func (l *Log) SearchEvents(fromUTC, toUTC time.Time, namespace string, eventType
 
 	modifyquery := func(query firestore.Query) firestore.Query {
 		if startKey != "" {
-			return query.StartAt(parsedStartKey)
+			return query.StartAfter(parsedStartKey)
 		}
 
 		return query

lib/events/firestoreevents/firestoreevents_test.go

@@ -68,7 +68,7 @@ func (s *FirestoreeventsSuite) SetUpSuite(c *check.C) {
 	s.log = log
 	s.EventsSuite.Log = log
 	s.EventsSuite.Clock = fakeClock
-	s.EventsSuite.QueryDelay = time.Second
+	s.EventsSuite.QueryDelay = time.Second * 5
 }
 
 func emulatorRunning() bool {

lib/events/test/suite.go

@@ -111,7 +111,7 @@ func (s *EventsSuite) EventPagination(c *check.C) {
 	}
 
 	checkpoint = ""
-	for i := range []int{0, 2} {
+	for _, i := range []int{0, 2} {
 		nameA := names[i]
 		nameB := names[i+1]
 		arr, checkpoint, err = s.Log.SearchEvents(baseTime, toTime, defaults.Namespace, nil, 2, checkpoint)
@@ -174,7 +174,7 @@ func (s *EventsSuite) SessionEventsCRUD(c *check.C) {
 	// read the session event
 	historyEvents, err := s.Log.GetSessionEvents(defaults.Namespace, sessionID, 0, false)
 	c.Assert(err, check.IsNil)
-	c.Assert(history, check.HasLen, 2)
+	c.Assert(historyEvents, check.HasLen, 2)
 	c.Assert(historyEvents[0].GetString(events.EventType), check.Equals, events.SessionStartEvent)
 	c.Assert(historyEvents[1].GetString(events.EventType), check.Equals, events.SessionEndEvent)