Zanzana: Run reconciliation in its own service (#100361)

* Zanzana: Start reconciliation in its own service

* cleanup

* update go workspaces

* refactor

* remove unused code

* move func definition

pkg/services/accesscontrol/accesscontrol.go

@@ -32,7 +32,6 @@ type AccessControl interface {
 }
 
 type Service interface {
-	registry.BackgroundService
 	registry.ProvidesUsageStats
 	// GetRoleByName returns a role by name
 	GetRoleByName(ctx context.Context, orgID int64, roleName string) (*RoleDTO, error)

pkg/services/accesscontrol/acimpl/service.go

@@ -12,6 +12,7 @@ import (
 	"go.opentelemetry.io/otel/attribute"
 
 	claims "github.com/grafana/authlib/types"
+
 	"github.com/grafana/grafana/pkg/api/routing"
 	"github.com/grafana/grafana/pkg/apimachinery/identity"
 	"github.com/grafana/grafana/pkg/infra/db"
@@ -25,11 +26,9 @@ import (
 	"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"
-	"github.com/grafana/grafana/pkg/services/authz/zanzana"
 	"github.com/grafana/grafana/pkg/services/dashboards"
 	"github.com/grafana/grafana/pkg/services/featuremgmt"
 	"github.com/grafana/grafana/pkg/services/folder"
@@ -54,7 +53,7 @@ var OSSRolesPrefixes = []string{accesscontrol.ManagedRolePrefix, accesscontrol.E
 func ProvideService(
 	cfg *setting.Cfg, db db.DB, routeRegister routing.RouteRegister, cache *localcache.CacheService,
 	accessControl accesscontrol.AccessControl, userService user.Service, actionResolver accesscontrol.ActionResolver,
-	features featuremgmt.FeatureToggles, tracer tracing.Tracer, zclient zanzana.Client, permRegistry permreg.PermissionRegistry,
+	features featuremgmt.FeatureToggles, tracer tracing.Tracer, permRegistry permreg.PermissionRegistry,
 	lock *serverlock.ServerLockService, folderService folder.Service,
 ) (*Service, error) {
 	service := ProvideOSSService(
@@ -64,7 +63,6 @@ func ProvideService(
 		cache,
 		features,
 		tracer,
-		zclient,
 		db,
 		permRegistry,
 		lock,
@@ -90,8 +88,8 @@ 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,
-	lock *serverlock.ServerLockService, folderService folder.Service,
+	db db.DB, permRegistry permreg.PermissionRegistry, lock *serverlock.ServerLockService,
+	folderService folder.Service,
 ) *Service {
 	s := &Service{
 		actionResolver: actionResolver,
@@ -101,7 +99,6 @@ func ProvideOSSService(
 		log:            log.New("accesscontrol.service"),
 		roles:          accesscontrol.BuildBasicRoleDefinitions(),
 		store:          store,
-		reconciler:     dualwrite.NewZanzanaReconciler(cfg, zclient, db, lock, folderService),
 		permRegistry:   permRegistry,
 	}
 
@@ -118,18 +115,9 @@ type Service struct {
 	registrations  accesscontrol.RegistrationList
 	roles          map[string]*accesscontrol.RoleDTO
 	store          accesscontrol.Store
-	reconciler     *dualwrite.ZanzanaReconciler
 	permRegistry   permreg.PermissionRegistry
 }
 
-// Run implements accesscontrol.Service.
-func (s *Service) Run(ctx context.Context) error {
-	if s.features.IsEnabledGlobally(featuremgmt.FlagZanzana) {
-		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,

pkg/services/accesscontrol/acimpl/service_test.go

@@ -71,7 +71,6 @@ func TestUsageMetrics(t *testing.T) {
 				featuremgmt.WithFeatures(),
 				tracing.InitializeTracerForTest(),
 				nil,
-				nil,
 				permreg.ProvidePermissionRegistry(),
 				nil,
 				nil,

pkg/services/accesscontrol/dualwrite/reconciler.go

@@ -14,6 +14,7 @@ import (
 	"github.com/grafana/grafana/pkg/infra/log"
 	"github.com/grafana/grafana/pkg/infra/serverlock"
 	"github.com/grafana/grafana/pkg/services/authz/zanzana"
+	"github.com/grafana/grafana/pkg/services/featuremgmt"
 	"github.com/grafana/grafana/pkg/services/folder"
 	"github.com/grafana/grafana/pkg/services/org"
 	"github.com/grafana/grafana/pkg/setting"
@@ -25,12 +26,12 @@ var tracer = otel.Tracer("github.com/grafana/grafana/pkg/accesscontrol/migrator"
 // We should rewrite the migration after we have "migrated" all possible actions
 // into our schema.
 type ZanzanaReconciler struct {
-	cfg *setting.Cfg
-	log log.Logger
-
-	store  db.DB
-	client zanzana.Client
-	lock   *serverlock.ServerLockService
+	cfg      *setting.Cfg
+	log      log.Logger
+	features featuremgmt.FeatureToggles
+	store    db.DB
+	client   zanzana.Client
+	lock     *serverlock.ServerLockService
 	// reconcilers are migrations that tries to reconcile the state of grafana db to zanzana store.
 	// These are run periodically to try to maintain a consistent state.
 	reconcilers []resourceReconciler
@@ -38,13 +39,14 @@ type ZanzanaReconciler struct {
 	globalReconcilers []resourceReconciler
 }
 
-func NewZanzanaReconciler(cfg *setting.Cfg, client zanzana.Client, store db.DB, lock *serverlock.ServerLockService, folderService folder.Service) *ZanzanaReconciler {
+func ProvideZanzanaReconciler(cfg *setting.Cfg, features featuremgmt.FeatureToggles, client zanzana.Client, store db.DB, lock *serverlock.ServerLockService, folderService folder.Service) *ZanzanaReconciler {
 	zanzanaReconciler := &ZanzanaReconciler{
-		cfg:    cfg,
-		log:    log.New("zanzana.reconciler"),
-		client: client,
-		lock:   lock,
-		store:  store,
+		cfg:      cfg,
+		log:      log.New("zanzana.reconciler"),
+		features: features,
+		client:   client,
+		lock:     lock,
+		store:    store,
 		reconcilers: []resourceReconciler{
 			newResourceReconciler(
 				"team memberships",
@@ -119,6 +121,14 @@ func NewZanzanaReconciler(cfg *setting.Cfg, client zanzana.Client, store db.DB,
 	return zanzanaReconciler
 }
 
+// Run implements registry.BackgroundService
+func (r *ZanzanaReconciler) Run(ctx context.Context) error {
+	if r.features.IsEnabledGlobally(featuremgmt.FlagZanzana) {
+		return r.Reconcile(ctx)
+	}
+	return nil
+}
+
 // Reconcile schedules as job that will run and reconcile resources between
 // legacy access control and zanzana.
 func (r *ZanzanaReconciler) Reconcile(ctx context.Context) error {
@@ -137,13 +147,6 @@ func (r *ZanzanaReconciler) Reconcile(ctx context.Context) error {
 	}
 }
 
-// ReconcileSync runs reconciliation and returns. Useful for tests to perform
-// reconciliation in a synchronous way.
-func (r *ZanzanaReconciler) ReconcileSync(ctx context.Context) error {
-	r.reconcile(ctx)
-	return nil
-}
-
 func (r *ZanzanaReconciler) reconcile(ctx context.Context) {
 	runGlobal := func(ctx context.Context) {
 		for _, reconciler := range r.globalReconcilers {

pkg/services/accesscontrol/ossaccesscontrol/testutil/testutil.go

@@ -10,7 +10,6 @@ import (
 	"github.com/grafana/grafana/pkg/services/accesscontrol/ossaccesscontrol"
 	"github.com/grafana/grafana/pkg/services/accesscontrol/permreg"
 	"github.com/grafana/grafana/pkg/services/accesscontrol/resourcepermissions"
-	"github.com/grafana/grafana/pkg/services/authz/zanzana"
 	"github.com/grafana/grafana/pkg/services/dashboards/database"
 	"github.com/grafana/grafana/pkg/services/featuremgmt"
 	"github.com/grafana/grafana/pkg/services/folder/folderimpl"
@@ -52,7 +51,7 @@ func ProvideFolderPermissions(
 
 	acSvc := acimpl.ProvideOSSService(
 		cfg, acdb.ProvideService(sqlStore), actionSets, localcache.ProvideService(),
-		features, tracing.InitializeTracerForTest(), zanzana.NewNoopClient(), sqlStore, permreg.ProvidePermissionRegistry(),
+		features, tracing.InitializeTracerForTest(), sqlStore, permreg.ProvidePermissionRegistry(),
 		nil, fService,
 	)
 

pkg/services/serviceaccounts/extsvcaccounts/service_test.go

@@ -54,7 +54,7 @@ func setupTestEnv(t *testing.T) *TestEnv {
 		enabled: true,
 		acSvc: acimpl.ProvideOSSService(
 			cfg, env.AcStore, &resourcepermissions.FakeActionSetSvc{},
-			localcache.New(0, 0), fmgt, tracing.InitializeTracerForTest(), nil, nil,
+			localcache.New(0, 0), fmgt, tracing.InitializeTracerForTest(), nil,
 			permreg.ProvidePermissionRegistry(), nil, nil,
 		),
 		defaultOrgID: autoAssignOrgID,