grafana/pkg/services/accesscontrol/dualwrite/reconciler.go

package dualwrite

import (
	"context"
	"strconv"
	"time"

	"github.com/grafana/authlib/claims"
	"go.opentelemetry.io/otel"

	"github.com/grafana/grafana/pkg/infra/db"
	"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/setting"
)

var tracer = otel.Tracer("github.com/grafana/grafana/pkg/accesscontrol/migrator")

// ZanzanaReconciler is a component to reconcile RBAC permissions to zanzana.
// 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
	// 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
}

func NewZanzanaReconciler(cfg *setting.Cfg, client zanzana.Client, store db.DB, lock *serverlock.ServerLockService) *ZanzanaReconciler {
	return &ZanzanaReconciler{
		cfg:    cfg,
		log:    log.New("zanzana.reconciler"),
		client: client,
		lock:   lock,
		store:  store,
		reconcilers: []resourceReconciler{
			newResourceReconciler(
				"team memberships",
				teamMembershipCollector(store),
				zanzanaCollector([]string{zanzana.RelationTeamMember, zanzana.RelationTeamAdmin}),
				client,
			),
			newResourceReconciler(
				"folder tree",
				folderTreeCollector(store),
				zanzanaCollector([]string{zanzana.RelationParent}),
				client,
			),
			newResourceReconciler(
				"managed folder permissions",
				managedPermissionsCollector(store, zanzana.KindFolders),
				zanzanaCollector(zanzana.FolderRelations),
				client,
			),
			newResourceReconciler(
				"managed dashboard permissions",
				managedPermissionsCollector(store, zanzana.KindDashboards),
				zanzanaCollector(zanzana.ResourceRelations),
				client,
			),
			newResourceReconciler(
				"basic role permissions",
				basicRolePermissionsCollector(store),
				zanzanaCollector(zanzana.FolderRelations),
				client,
			),
			newResourceReconciler(
				"basic role bindings",
				basicRoleBindingsCollector(store),
				zanzanaCollector([]string{zanzana.RelationAssignee}),
				client,
			),
		},
	}
}

// Reconcile schedules as job that will run and reconcile resources between
// legacy access control and zanzana.
func (r *ZanzanaReconciler) Reconcile(ctx context.Context) error {
	r.reconcile(ctx)

	// FIXME:
	// 1. We should be a bit graceful about reconciliations so we are not hammering dbs
	// 2. We should be able to configure reconciliation interval
	ticker := time.NewTicker(1 * time.Hour)
	for {
		select {
		case <-ticker.C:
			r.reconcile(ctx)
		case <-ctx.Done():
			return ctx.Err()
		}
	}
}

func (r *ZanzanaReconciler) reconcile(ctx context.Context) {
	run := func(ctx context.Context, namespace string) {
		now := time.Now()
		for _, reconciler := range r.reconcilers {
			if err := reconciler.reconcile(ctx, namespace); err != nil {
				r.log.Warn("Failed to perform reconciliation for resource", "err", err)
			}
		}
		r.log.Debug("Finished reconciliation", "elapsed", time.Since(now))
	}

	var namespaces []string
	if r.cfg.StackID != "" {
		id, err := strconv.ParseInt(r.cfg.StackID, 10, 64)
		if err != nil {
			r.log.Error("cannot perform reconciliation, malformed stack id", "id", r.cfg.StackID, "err", err)
			return
		}

		namespaces = []string{claims.CloudNamespaceFormatter(id)}
	} else {
		ids, err := r.getOrgs(ctx)
		if err != nil {
			r.log.Error("cannot perform reconciliation, failed to fetch orgs", "err", err)
			return
		}

		for _, id := range ids {
			namespaces = append(namespaces, claims.OrgNamespaceFormatter(id))
		}
	}

	if r.lock == nil {
		for _, ns := range namespaces {
			run(ctx, ns)
		}
		return
	}

	// We ignore the error for now
	_ = r.lock.LockExecuteAndRelease(ctx, "zanzana-reconciliation", 10*time.Hour, func(ctx context.Context) {
		for _, ns := range namespaces {
			run(ctx, ns)
		}
	})
}

func (r *ZanzanaReconciler) getOrgs(ctx context.Context) ([]int64, error) {
	orgs := make([]int64, 0)
	err := r.store.WithDbSession(ctx, func(sess *db.Session) error {
		q := "SELECT id FROM org"
		if err := sess.SQL(q).Find(&orgs); err != nil {
			return err
		}
		return nil
	})
	if err != nil {
		return nil, err
	}
	return orgs, nil
}