Allow reads for cache events to go to RO DB

Signed-off-by: Faisal Memon <[email protected]>

Author: faisal-memon

pkg/server/datastore/datastore.go

@@ -170,6 +170,7 @@ type ListAttestedNodesResponse struct {
 }
 
 type ListAttestedNodeEventsRequest struct {
+	DataConsistency    DataConsistency
 	GreaterThanEventID uint
 	LessThanEventID    uint
 }
@@ -224,6 +225,7 @@ type ListRegistrationEntriesResponse struct {
 }
 
 type ListRegistrationEntryEventsRequest struct {
+	DataConsistency    DataConsistency
 	GreaterThanEventID uint
 	LessThanEventID    uint
 }

pkg/server/datastore/sqlstore/sqlstore.go

@@ -375,13 +375,10 @@ func (ds *Plugin) DeleteAttestedNode(ctx context.Context, spiffeID string) (atte
 
 // ListAttestedNodeEvents lists all attested node events
 func (ds *Plugin) ListAttestedNodeEvents(ctx context.Context, req *datastore.ListAttestedNodeEventsRequest) (resp *datastore.ListAttestedNodeEventsResponse, err error) {
-	if err = ds.withReadTx(ctx, func(tx *gorm.DB) (err error) {
-		resp, err = listAttestedNodeEvents(tx, req)
-		return err
-	}); err != nil {
-		return nil, err
+	if req.DataConsistency == datastore.TolerateStale && ds.roDb != nil {
+		return listAttestedNodeEvents(ds.roDb, req)
 	}
-	return resp, nil
+	return listAttestedNodeEvents(ds.db, req)
 }
 
 // PruneAttestedNodeEvents deletes all attested node events older than a specified duration (i.e. more than 24 hours old)
@@ -572,13 +569,10 @@ func (ds *Plugin) PruneRegistrationEntries(ctx context.Context, expiresBefore ti
 
 // ListRegistrationEntryEvents lists all registration entry events
 func (ds *Plugin) ListRegistrationEntryEvents(ctx context.Context, req *datastore.ListRegistrationEntryEventsRequest) (resp *datastore.ListRegistrationEntryEventsResponse, err error) {
-	if err = ds.withReadTx(ctx, func(tx *gorm.DB) (err error) {
-		resp, err = listRegistrationEntryEvents(tx, req)
-		return err
-	}); err != nil {
-		return nil, err
+	if req.DataConsistency == datastore.TolerateStale && ds.roDb != nil {
+		return listRegistrationEntryEvents(ds.roDb, req)
 	}
-	return resp, nil
+	return listRegistrationEntryEvents(ds.db, req)
 }
 
 // PruneRegistrationEntryEvents deletes all registration entry events older than a specified duration (i.e. more than 24 hours old)
@@ -1707,7 +1701,7 @@ func createAttestedNodeEvent(tx *gorm.DB, event *datastore.AttestedNodeEvent) er
 	return nil
 }
 
-func listAttestedNodeEvents(tx *gorm.DB, req *datastore.ListAttestedNodeEventsRequest) (*datastore.ListAttestedNodeEventsResponse, error) {
+func listAttestedNodeEvents(db *sqlDB, req *datastore.ListAttestedNodeEventsRequest) (*datastore.ListAttestedNodeEventsResponse, error) {
 	var events []AttestedNodeEvent
 
 	if req.GreaterThanEventID != 0 || req.LessThanEventID != 0 {
@@ -1716,11 +1710,11 @@ func listAttestedNodeEvents(tx *gorm.DB, req *datastore.ListAttestedNodeEventsRe
 			return nil, newWrappedSQLError(err)
 		}
 
-		if err := tx.Find(&events, query.String(), id).Order("id asc").Error; err != nil {
+		if err := db.Find(&events, query.String(), id).Order("id asc").Error; err != nil {
 			return nil, newWrappedSQLError(err)
 		}
 	} else {
-		if err := tx.Find(&events).Order("id asc").Error; err != nil {
+		if err := db.Find(&events).Order("id asc").Error; err != nil {
 			return nil, newWrappedSQLError(err)
 		}
 	}
@@ -4098,7 +4092,7 @@ func deleteRegistrationEntryEvent(tx *gorm.DB, eventID uint) error {
 	return nil
 }
 
-func listRegistrationEntryEvents(tx *gorm.DB, req *datastore.ListRegistrationEntryEventsRequest) (*datastore.ListRegistrationEntryEventsResponse, error) {
+func listRegistrationEntryEvents(db *sqlDB, req *datastore.ListRegistrationEntryEventsRequest) (*datastore.ListRegistrationEntryEventsResponse, error) {
 	var events []RegisteredEntryEvent
 
 	if req.GreaterThanEventID != 0 || req.LessThanEventID != 0 {
@@ -4107,11 +4101,11 @@ func listRegistrationEntryEvents(tx *gorm.DB, req *datastore.ListRegistrationEnt
 			return nil, newWrappedSQLError(err)
 		}
 
-		if err := tx.Find(&events, query.String(), id).Order("id asc").Error; err != nil {
+		if err := db.Find(&events, query.String(), id).Order("id asc").Error; err != nil {
 			return nil, newWrappedSQLError(err)
 		}
 	} else {
-		if err := tx.Find(&events).Order("id asc").Error; err != nil {
+		if err := db.Find(&events).Order("id asc").Error; err != nil {
 			return nil, newWrappedSQLError(err)
 		}
 	}

pkg/server/datastore/sqlstore/sqlstore_test.go

@@ -1594,7 +1594,7 @@ func (s *PluginSuite) TestListAttestedNodeEvents() {
 			name:               "Setting both greater and less than",
 			greaterThanEventID: 1,
 			lessThanEventID:    1,
-			expectedErr:        "rpc error: code = Unknown desc = datastore-sql: can't set both greater and less than event id",
+			expectedErr:        "datastore-sql: can't set both greater and less than event id",
 		},
 	}
 	for _, test := range tests {
@@ -4133,7 +4133,7 @@ func (s *PluginSuite) TestListRegistrationEntryEvents() {
 			name:               "Setting both greater and less than",
 			greaterThanEventID: 1,
 			lessThanEventID:    1,
-			expectedErr:        "rpc error: code = Unknown desc = datastore-sql: can't set both greater and less than event id",
+			expectedErr:        "datastore-sql: can't set both greater and less than event id",
 		},
 	}
 	for _, test := range tests {

pkg/server/endpoints/authorized_entryfetcher_attested_nodes.go

@@ -102,16 +102,10 @@ func (a *attestedNodes) selectPolledEvents(ctx context.Context) {
 }
 
 func (a *attestedNodes) scanForNewEvents(ctx context.Context) error {
-	// If we haven't seen an event, scan for all events; otherwise, scan from the last event.
-	var resp *datastore.ListAttestedNodeEventsResponse
-	var err error
-	if a.firstEventTime.IsZero() {
-		resp, err = a.ds.ListAttestedNodeEvents(ctx, &datastore.ListAttestedNodeEventsRequest{})
-	} else {
-		resp, err = a.ds.ListAttestedNodeEvents(ctx, &datastore.ListAttestedNodeEventsRequest{
-			GreaterThanEventID: a.lastEvent,
-		})
-	}
+	resp, err := a.ds.ListAttestedNodeEvents(ctx, &datastore.ListAttestedNodeEventsRequest{
+		DataConsistency:    datastore.TolerateStale,
+		GreaterThanEventID: a.lastEvent,
+	})
 	if err != nil {
 		return err
 	}

pkg/server/endpoints/authorized_entryfetcher_registration_entries.go

@@ -101,16 +101,10 @@ func (a *registrationEntries) selectPolledEvents(ctx context.Context) {
 }
 
 func (a *registrationEntries) scanForNewEvents(ctx context.Context) error {
-	// If we haven't seen an event, scan for all events; otherwise, scan from the last event.
-	var resp *datastore.ListRegistrationEntryEventsResponse
-	var err error
-	if a.firstEventTime.IsZero() {
-		resp, err = a.ds.ListRegistrationEntryEvents(ctx, &datastore.ListRegistrationEntryEventsRequest{})
-	} else {
-		resp, err = a.ds.ListRegistrationEntryEvents(ctx, &datastore.ListRegistrationEntryEventsRequest{
-			GreaterThanEventID: a.lastEvent,
-		})
-	}
+	resp, err := a.ds.ListRegistrationEntryEvents(ctx, &datastore.ListRegistrationEntryEventsRequest{
+		DataConsistency:    datastore.TolerateStale,
+		GreaterThanEventID: a.lastEvent,
+	})
 	if err != nil {
 		return err
 	}