update

authorize/authorize.go

@@ -48,7 +48,7 @@ func New(ctx context.Context, cfg *config.Config) (*Authorize, error) {
 	}
 	a.accessTracker = NewAccessTracker(a, accessTrackerMaxSize, accessTrackerDebouncePeriod)
 
-	state, err := newAuthorizeStateFromConfig(ctx, tracerProvider, cfg, a.store, nil)
+	state, err := newAuthorizeStateFromConfig(ctx, nil, tracerProvider, cfg, a.store)
 	if err != nil {
 		return nil, err
 	}
@@ -154,7 +154,7 @@ func newPolicyEvaluator(
 func (a *Authorize) OnConfigChange(ctx context.Context, cfg *config.Config) {
 	currentState := a.state.Load()
 	a.currentConfig.Store(cfg)
-	if newState, err := newAuthorizeStateFromConfig(ctx, a.tracerProvider, cfg, a.store, currentState.evaluator); err != nil {
+	if newState, err := newAuthorizeStateFromConfig(ctx, currentState, a.tracerProvider, cfg, a.store); err != nil {
 		log.Ctx(ctx).Error().Err(err).Msg("authorize: error updating state")
 	} else {
 		a.state.Store(newState)

authorize/grpc.go

@@ -169,11 +169,19 @@ func (a *Authorize) getMatchingPolicy(routeID uint64) *config.Policy {
 }
 
 func (a *Authorize) withQuerierForCheckRequest(ctx context.Context) context.Context {
-	querier := storage.NewCachingQuerier(
-		storage.NewQuerier(a.state.Load().dataBrokerClient),
+	state := a.state.Load()
+	q := storage.NewCachingQuerier(
+		storage.NewQuerier(state.dataBrokerClient),
 		storage.GlobalCache,
 	)
-	return storage.WithQuerier(ctx, querier)
+	if len(state.syncQueriers) > 0 {
+		m := map[string]storage.Querier{}
+		for recordType, sq := range state.syncQueriers {
+			m[recordType] = storage.NewFallbackQuerier(sq, q)
+		}
+		q = storage.NewTypedQuerier(q, m)
+	}
+	return storage.WithQuerier(ctx, q)
 }
 
 func getHTTPRequestFromCheckRequest(req *envoy_service_auth_v3.CheckRequest) *http.Request {

authorize/state.go

@@ -9,12 +9,14 @@ import (
 	oteltrace "go.opentelemetry.io/otel/trace"
 	googlegrpc "google.golang.org/grpc"
 
+	"github.com/pomerium/datasource/pkg/directory"
 	"github.com/pomerium/pomerium/authorize/evaluator"
 	"github.com/pomerium/pomerium/authorize/internal/store"
 	"github.com/pomerium/pomerium/config"
 	"github.com/pomerium/pomerium/internal/authenticateflow"
 	"github.com/pomerium/pomerium/pkg/grpc"
 	"github.com/pomerium/pomerium/pkg/grpc/databroker"
+	"github.com/pomerium/pomerium/pkg/storage"
 )
 
 var outboundGRPCConnection = new(grpc.CachedOutboundGRPClientConn)
@@ -30,14 +32,15 @@ type authorizeState struct {
 	dataBrokerClient           databroker.DataBrokerServiceClient
 	sessionStore               *config.SessionStore
 	authenticateFlow           authenticateFlow
+	syncQueriers               map[string]storage.Querier
 }
 
 func newAuthorizeStateFromConfig(
 	ctx context.Context,
+	previousState *authorizeState,
 	tracerProvider oteltrace.TracerProvider,
 	cfg *config.Config,
 	store *store.Store,
-	previousPolicyEvaluator *evaluator.Evaluator,
 ) (*authorizeState, error) {
 	if err := validateOptions(cfg.Options); err != nil {
 		return nil, fmt.Errorf("authorize: bad options: %w", err)
@@ -47,7 +50,12 @@ func newAuthorizeStateFromConfig(
 
 	var err error
 
-	state.evaluator, err = newPolicyEvaluator(ctx, cfg.Options, store, previousPolicyEvaluator)
+	var previousEvaluator *evaluator.Evaluator
+	if previousState != nil {
+		previousEvaluator = previousState.evaluator
+	}
+
+	state.evaluator, err = newPolicyEvaluator(ctx, cfg.Options, store, previousEvaluator)
 	if err != nil {
 		return nil, fmt.Errorf("authorize: failed to update policy with options: %w", err)
 	}
@@ -88,5 +96,26 @@ func newAuthorizeStateFromConfig(
 		return nil, err
 	}
 
+	state.syncQueriers = make(map[string]storage.Querier)
+	if previousState != nil {
+		if previousState.dataBrokerClientConnection == state.dataBrokerClientConnection {
+			state.syncQueriers = previousState.syncQueriers
+		} else {
+			for _, v := range previousState.syncQueriers {
+				v.Stop()
+			}
+		}
+	}
+	if cfg.Options.IsRuntimeFlagSet(config.RuntimeFlagAuthorizeUseSyncedData) {
+		for _, recordType := range []string{
+			directory.GroupRecordType,
+			directory.UserRecordType,
+		} {
+			if _, ok := state.syncQueriers[recordType]; !ok {
+				state.syncQueriers[recordType] = storage.NewSyncQuerier(state.dataBrokerClient, recordType)
+			}
+		}
+	}
+
 	return state, nil
 }

config/runtime_flags.go

@@ -28,7 +28,7 @@ var (
 
 	// RuntimeFlagAuthorizeUseSyncedData enables synced data for querying the databroker for
 	// certain types of data.
-	RuntimeFlagAuthorizeUseSyncedData = runtimeFlag("authorize_use_synced_data", false)
+	RuntimeFlagAuthorizeUseSyncedData = runtimeFlag("authorize_use_synced_data", true)
 )
 
 // RuntimeFlag is a runtime flag that can flip on/off certain features