fix databroker restart versioning, handle missing sessions (#1145)

* fix databroker restart versioning, handle missing sessions

* send empty server version to detect change

* only rebuild if there are updated records

authorize/authorize.go

@@ -58,9 +58,8 @@ type Authorize struct {
 
 	dataBrokerClient databroker.DataBrokerServiceClient
 
-	dataBrokerDataLock             sync.RWMutex
+	dataBrokerDataLock sync.RWMutex
-	dataBrokerData                 evaluator.DataBrokerData
+	dataBrokerData     evaluator.DataBrokerData
-	dataBrokerSessionServerVersion string
 }
 
 // New validates and creates a new Authorize service from a set of config options.

authorize/evaluator/evaluator.go

@@ -402,6 +402,11 @@ func getDenyVar(vars rego.Vars) []Result {
 // DataBrokerData stores the data broker data by type => id => record
 type DataBrokerData map[string]map[string]interface{}
 
+// Clear removes all the data for the given type URL from the databroekr data.
+func (dbd DataBrokerData) Clear(typeURL string) {
+	delete(dbd, typeURL)
+}
+
 // Get gets a record from the DataBrokerData.
 func (dbd DataBrokerData) Get(typeURL, id string) interface{} {
 	m, ok := dbd[typeURL]

authorize/evaluator/store.go

@@ -24,6 +24,12 @@ func NewStore() *Store {
 	}
 }
 
+// ClearRecords removes all the records from the store.
+func (s *Store) ClearRecords(typeURL string) {
+	rawPath := fmt.Sprintf("/databroker_data/%s", typeURL)
+	s.delete(rawPath)
+}
+
 // UpdateAdmins updates the admins in the store.
 func (s *Store) UpdateAdmins(admins []string) {
 	s.write("/admins", admins)

authorize/grpc.go

@@ -49,19 +49,6 @@ func (a *Authorize) Check(ctx context.Context, in *envoy_service_auth_v2.CheckRe
 	rawJWT, _ := loadRawSession(hreq, a.currentOptions.Load(), a.currentEncoder.Load())
 	sessionState, _ := loadSession(a.currentEncoder.Load(), rawJWT)
 
-	// only accept sessions whose databroker server versions match
-	if sessionState != nil {
-		a.dataBrokerDataLock.RLock()
-		if a.dataBrokerSessionServerVersion != sessionState.Version.String() {
-			log.Warn().
-				Str("server_version", a.dataBrokerSessionServerVersion).
-				Str("session_version", sessionState.Version.String()).
-				Msg("clearing session due to invalid version")
-			sessionState = nil
-		}
-		a.dataBrokerDataLock.RUnlock()
-	}
-
 	if err := a.forceSync(ctx, sessionState); err != nil {
 		log.Warn().Err(err).Msg("clearing session due to force sync failed")
 		sessionState = nil
@@ -98,7 +85,7 @@ func (a *Authorize) forceSync(ctx context.Context, ss *sessions.State) error {
 	}
 	s := a.forceSyncSession(ctx, ss.ID)
 	if s == nil {
-		return nil
+		return errors.New("session not found")
 	}
 	if s.DeletedAt != nil {
 		return errors.New("session was deleted")

authorize/run.go

@@ -24,13 +24,8 @@ func (a *Authorize) Run(ctx context.Context) error {
 		return a.runTypesSyncer(ctx, updateTypes)
 	})
 
-	updateRecord := make(chan *databroker.Record)
 	eg.Go(func() error {
-		return a.runDataSyncer(ctx, updateTypes, updateRecord)
+		return a.runDataSyncer(ctx, updateTypes)
-	})
-
-	eg.Go(func() error {
-		return a.runDataUpdater(ctx, updateRecord)
 	})
 
 	return eg.Wait()
@@ -65,7 +60,7 @@ func (a *Authorize) runTypesSyncer(ctx context.Context, updateTypes chan<- []str
 	})
 }
 
-func (a *Authorize) runDataSyncer(ctx context.Context, updateTypes <-chan []string, updateRecord chan<- *databroker.Record) error {
+func (a *Authorize) runDataSyncer(ctx context.Context, updateTypes <-chan []string) error {
 	eg, ctx := errgroup.WithContext(ctx)
 	eg.Go(func() error {
 		seen := map[string]struct{}{}
@@ -78,7 +73,7 @@ func (a *Authorize) runDataSyncer(ctx context.Context, updateTypes <-chan []stri
 					dataType := dataType
 					if _, ok := seen[dataType]; !ok {
 						eg.Go(func() error {
-							return a.runDataTypeSyncer(ctx, dataType, updateRecord)
+							return a.runDataTypeSyncer(ctx, dataType)
 						})
 						seen[dataType] = struct{}{}
 					}
@@ -89,7 +84,7 @@ func (a *Authorize) runDataSyncer(ctx context.Context, updateTypes <-chan []stri
 	return eg.Wait()
 }
 
-func (a *Authorize) runDataTypeSyncer(ctx context.Context, typeURL string, updateRecord chan<- *databroker.Record) error {
+func (a *Authorize) runDataTypeSyncer(ctx context.Context, typeURL string) error {
 	var serverVersion, recordVersion string
 
 	log.Info().Str("type_url", typeURL).Msg("starting data initial load")
@@ -110,19 +105,10 @@ func (a *Authorize) runDataTypeSyncer(ctx context.Context, typeURL string, updat
 		}
 
 		serverVersion = res.GetServerVersion()
-		if typeURL == sessionTypeURL {
-			a.dataBrokerDataLock.Lock()
-			a.dataBrokerSessionServerVersion = serverVersion
-			a.dataBrokerDataLock.Unlock()
-		}
 		recordVersion = res.GetRecordVersion()
 
 		for _, record := range res.GetRecords() {
-			select {
+			a.updateRecord(record)
-			case <-ctx.Done():
-				return ctx.Err()
-			case updateRecord <- record:
-			}
 		}
 
 		break
@@ -151,7 +137,16 @@ func (a *Authorize) runDataTypeSyncer(ctx context.Context, typeURL string, updat
 			}
 
 			backoff.Reset()
-			serverVersion = res.GetServerVersion()
+			if res.GetServerVersion() != serverVersion {
+				log.Info().
+					Str("old_version", serverVersion).
+					Str("new_version", res.GetServerVersion()).
+					Str("type_url", typeURL).
+					Msg("detected new server version, clearing data")
+				serverVersion = res.GetServerVersion()
+				recordVersion = ""
+				a.clearRecords(typeURL)
+			}
 			for _, record := range res.GetRecords() {
 				if record.GetVersion() > recordVersion {
 					recordVersion = record.GetVersion()
@@ -159,33 +154,24 @@ func (a *Authorize) runDataTypeSyncer(ctx context.Context, typeURL string, updat
 			}
 
 			for _, record := range res.GetRecords() {
-				select {
+				a.updateRecord(record)
-				case <-stream.Context().Done():
-					return stream.Context().Err()
-				case updateRecord <- record:
-				}
 			}
 		}
 	})
 }
 
-func (a *Authorize) runDataUpdater(ctx context.Context, updateRecord <-chan *databroker.Record) error {
+func (a *Authorize) clearRecords(typeURL string) {
-	log.Info().Msg("starting data updater")
+	a.store.ClearRecords(typeURL)
-	for {
+	a.dataBrokerDataLock.Lock()
-		var record *databroker.Record
+	a.dataBrokerData.Clear(typeURL)
+	a.dataBrokerDataLock.Unlock()
+}
 
-		select {
+func (a *Authorize) updateRecord(record *databroker.Record) {
-		case <-ctx.Done():
+	a.store.UpdateRecord(record)
-			return ctx.Err()
+	a.dataBrokerDataLock.Lock()
-		case record = <-updateRecord:
+	a.dataBrokerData.Update(record)
-		}
+	a.dataBrokerDataLock.Unlock()
-
-		a.store.UpdateRecord(record)
-
-		a.dataBrokerDataLock.Lock()
-		a.dataBrokerData.Update(record)
-		a.dataBrokerDataLock.Unlock()
-	}
 }
 
 func tryForever(ctx context.Context, callback func(onSuccess interface{ Reset() }) error) error {

internal/databroker/config_source.go

@@ -178,10 +178,11 @@ func (src *ConfigSource) runUpdater(cfg *config.Config) {
 			}
 			onSuccess()
 
-			src.onSync(res.GetRecords())
+			if len(res.GetRecords()) > 0 {
-
+				src.onSync(res.GetRecords())
-			for _, record := range res.GetRecords() {
+				for _, record := range res.GetRecords() {
-				recordVersion = record.GetVersion()
+					recordVersion = record.GetVersion()
+				}
 			}
 
 			src.mu.Lock()

internal/databroker/server.go

@@ -230,6 +230,13 @@ func (srv *Server) Sync(req *databroker.SyncRequest, stream databroker.DataBroke
 	// reset record version if the server versions don't match
 	if req.GetServerVersion() != srv.version {
 		recordVersion = ""
+		// send the new server version to the client
+		err := stream.Send(&databroker.SyncResponse{
+			ServerVersion: srv.version,
+		})
+		if err != nil {
+			return err
+		}
 	}
 
 	db, err := srv.getDB(req.GetType())