Zanzana: periodic sync of team members (#94752)

* Rewrite zanzana collector to fetch all available pages

* Register access control as a background service

* If zanzana is enabled we run Syncs and start Reconciliation job

* Update pkg/services/authz/zanzana/client/client.go

Co-authored-by: Alexander Zobnin <alexanderzobnin@gmail.com>

* Use server lock when doing performing reconciliation

pkg/services/accesscontrol/acimpl/service.go

@@ -18,12 +18,14 @@ import (
 	"github.com/grafana/grafana/pkg/infra/localcache"
 	"github.com/grafana/grafana/pkg/infra/log"
 	"github.com/grafana/grafana/pkg/infra/metrics"
+	"github.com/grafana/grafana/pkg/infra/serverlock"
 	"github.com/grafana/grafana/pkg/infra/slugify"
 	"github.com/grafana/grafana/pkg/infra/tracing"
 	"github.com/grafana/grafana/pkg/plugins"
 	"github.com/grafana/grafana/pkg/services/accesscontrol"
 	"github.com/grafana/grafana/pkg/services/accesscontrol/api"
 	"github.com/grafana/grafana/pkg/services/accesscontrol/database"
+	"github.com/grafana/grafana/pkg/services/accesscontrol/dualwrite"
 	"github.com/grafana/grafana/pkg/services/accesscontrol/migrator"
 	"github.com/grafana/grafana/pkg/services/accesscontrol/permreg"
 	"github.com/grafana/grafana/pkg/services/accesscontrol/pluginutils"
@@ -53,8 +55,20 @@ func ProvideService(
 	cfg *setting.Cfg, db db.DB, routeRegister routing.RouteRegister, cache *localcache.CacheService,
 	accessControl accesscontrol.AccessControl, actionResolver accesscontrol.ActionResolver,
 	features featuremgmt.FeatureToggles, tracer tracing.Tracer, zclient zanzana.Client, permRegistry permreg.PermissionRegistry,
+	lock *serverlock.ServerLockService,
 ) (*Service, error) {
-	service := ProvideOSSService(cfg, database.ProvideService(db), actionResolver, cache, features, tracer, zclient, db, permRegistry)
+	service := ProvideOSSService(
+		cfg,
+		database.ProvideService(db),
+		actionResolver,
+		cache,
+		features,
+		tracer,
+		zclient,
+		db,
+		permRegistry,
+		lock,
+	)
 
 	api.NewAccessControlAPI(routeRegister, accessControl, service, features).RegisterAPIEndpoints()
 	if err := accesscontrol.DeclareFixedRoles(service, cfg); err != nil {
@@ -75,7 +89,7 @@ func ProvideService(
 func ProvideOSSService(
 	cfg *setting.Cfg, store accesscontrol.Store, actionResolver accesscontrol.ActionResolver,
 	cache *localcache.CacheService, features featuremgmt.FeatureToggles, tracer tracing.Tracer,
-	zclient zanzana.Client, db db.DB, permRegistry permreg.PermissionRegistry,
+	zclient zanzana.Client, db db.DB, permRegistry permreg.PermissionRegistry, lock *serverlock.ServerLockService,
 ) *Service {
 	s := &Service{
 		actionResolver: actionResolver,
@@ -85,7 +99,7 @@ func ProvideOSSService(
 		log:            log.New("accesscontrol.service"),
 		roles:          accesscontrol.BuildBasicRoleDefinitions(),
 		store:          store,
-		sync:           migrator.NewZanzanaSynchroniser(zclient, db),
+		reconciler:     dualwrite.NewZanzanaReconciler(zclient, db, lock),
 		permRegistry:   permRegistry,
 	}
 
@@ -102,10 +116,22 @@ type Service struct {
 	registrations  accesscontrol.RegistrationList
 	roles          map[string]*accesscontrol.RoleDTO
 	store          accesscontrol.Store
-	sync           *migrator.ZanzanaSynchroniser
+	reconciler     *dualwrite.ZanzanaReconciler
 	permRegistry   permreg.PermissionRegistry
 }
 
+// Run implements accesscontrol.Service.
+func (s *Service) Run(ctx context.Context) error {
+	if s.features.IsEnabledGlobally(featuremgmt.FlagZanzana) {
+		if err := s.reconciler.Sync(context.Background()); err != nil {
+			s.log.Error("Failed to synchronise permissions to zanzana ", "err", err)
+		}
+
+		return s.reconciler.Reconcile(ctx)
+	}
+	return nil
+}
+
 func (s *Service) GetUsageStats(_ context.Context) map[string]any {
 	return map[string]any{
 		"stats.oss.accesscontrol.enabled.count": 1,
@@ -448,12 +474,6 @@ func (s *Service) RegisterFixedRoles(ctx context.Context) error {
 		return true
 	})
 
-	if s.features.IsEnabledGlobally(featuremgmt.FlagZanzana) {
-		if err := s.sync.Sync(context.Background()); err != nil {
-			s.log.Error("Failed to synchronise permissions to zanzana ", "err", err)
-		}
-	}
-
 	return nil
 }
 

pkg/services/accesscontrol/acimpl/service_test.go

@@ -74,6 +74,7 @@ func TestUsageMetrics(t *testing.T) {
 				nil,
 				nil,
 				permreg.ProvidePermissionRegistry(),
+				nil,
 			)
 			assert.Equal(t, tt.expectedValue, s.GetUsageStats(context.Background())["stats.oss.accesscontrol.enabled.count"])
 		})