public/grafana
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Zanzana: Use separate store for each org (#96015)

Browse Source 
* Move server init into server package

* map store name to id

* refactor model loading

* pass namespace into reconcilers and collectors

* refactor

* Extend authz server with Read and Write methods

* use new read/write in reconciler

* implement server side read and write

* Sync permissions for every org

* handle namespace in check and list

* split read and write

* provide conditions

* Fix client implementation

* fix nil conditions

* remove unused client code

* use lock for store access

* move type translators to common package

* fix folder collector

* fix store creation

* remove unused AuthorizationModelId

* fix server tests

* fix linter
This commit is contained in:
Alexander Zobnin
2024-11-08 14:54:36 +01:00
committed by GitHub
parent 86bc087257
commit 910ec7e7dc
22 changed files with 1497 additions and 396 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/services/accesscontrol/dualwrite/collectors.go
+17 -12
View File
@@ -3,13 +3,16 @@ package dualwrite
import (
"context"
openfgav1 "github.com/openfga/api/proto/openfga/v1"
"github.com/grafana/grafana/pkg/infra/db"
"github.com/grafana/grafana/pkg/services/authz/zanzana"
openfgav1 "github.com/openfga/api/proto/openfga/v1"
"github.com/grafana/grafana/pkg/services/authz/zanzana/common"
authzextv1 "github.com/grafana/grafana/pkg/services/authz/zanzana/proto/v1"
)
func teamMembershipCollector(store db.DB) legacyTupleCollector {
return func(ctx context.Context) (map[string]map[string]*openfgav1.TupleKey, error) {
return func(ctx context.Context, orgId int64) (map[string]map[string]*openfgav1.TupleKey, error) {
query := `
SELECT t.uid as team_uid, u.uid as user_uid, tm.permission
FROM team_member tm
@@ -60,7 +63,7 @@ func teamMembershipCollector(store db.DB) legacyTupleCollector {
// folderTreeCollector collects folder tree structure and writes it as relation tuples
func folderTreeCollector(store db.DB) legacyTupleCollector {
return func(ctx context.Context) (map[string]map[string]*openfgav1.TupleKey, error) {
return func(ctx context.Context, orgId int64) (map[string]map[string]*openfgav1.TupleKey, error) {
ctx, span := tracer.Start(ctx, "accesscontrol.migrator.folderTreeCollector")
defer span.End()
@@ -91,9 +94,9 @@ func folderTreeCollector(store db.DB) legacyTupleCollector {
}
tuple = &openfgav1.TupleKey{
Object: zanzana.NewTupleEntry("folder2", f.FolderUID, ""),
Object: zanzana.NewTupleEntry(common.TypeFolder, f.FolderUID, ""),
Relation: zanzana.RelationParent,
User: zanzana.NewTupleEntry("folder2", f.ParentUID, ""),
User: zanzana.NewTupleEntry(common.TypeFolder, f.ParentUID, ""),
}
if tuples[tuple.Object] == nil {
@@ -111,7 +114,7 @@ func folderTreeCollector(store db.DB) legacyTupleCollector {
// It will only store actions that are supported by our schema. Managed permissions can
// be directly mapped to user/team/role without having to write an intermediate role.
func managedPermissionsCollector(store db.DB, kind string) legacyTupleCollector {
return func(ctx context.Context) (map[string]map[string]*openfgav1.TupleKey, error) {
return func(ctx context.Context, orgId int64) (map[string]map[string]*openfgav1.TupleKey, error) {
query := `
SELECT u.uid as user_uid, t.uid as team_uid, p.action, p.kind, p.identifier, r.org_id
FROM permission p
@@ -194,11 +197,12 @@ func tupleStringWithoutCondition(tuple *openfgav1.TupleKey) string {
}
func zanzanaCollector(relations []string) zanzanaTupleCollector {
return func(ctx context.Context, client zanzana.Client, object string) (map[string]*openfgav1.TupleKey, error) {
return func(ctx context.Context, client zanzana.Client, object string, namespace string) (map[string]*openfgav1.TupleKey, error) {
// list will use continuation token to collect all tuples for object and relation
list := func(relation string) ([]*openfgav1.Tuple, error) {
first, err := client.Read(ctx, &openfgav1.ReadRequest{
TupleKey: &openfgav1.ReadRequestTupleKey{
first, err := client.Read(ctx, &authzextv1.ReadRequest{
Namespace: namespace,
TupleKey: &authzextv1.ReadRequestTupleKey{
Object: object,
Relation: relation,
},
@@ -211,8 +215,9 @@ func zanzanaCollector(relations []string) zanzanaTupleCollector {
c := first.ContinuationToken
for c != "" {
res, err := client.Read(ctx, &openfgav1.ReadRequest{
TupleKey: &openfgav1.ReadRequestTupleKey{
res, err := client.Read(ctx, &authzextv1.ReadRequest{
Namespace: namespace,
TupleKey: &authzextv1.ReadRequestTupleKey{
Object: object,
Relation: relation,
},
@@ -225,7 +230,7 @@ func zanzanaCollector(relations []string) zanzanaTupleCollector {
first.Tuples = append(first.Tuples, res.Tuples...)
}
return first.Tuples, nil
return common.ToOpenFGATuples(first.Tuples), nil
}
out := make(map[string]*openfgav1.TupleKey)
pkg/services/accesscontrol/dualwrite/reconciler.go
+40 -8
View File
@@ -4,6 +4,8 @@ import (
"context"
"time"
"github.com/grafana/authlib/claims"
openfgav1 "github.com/openfga/api/proto/openfga/v1"
"go.opentelemetry.io/otel"
"github.com/grafana/grafana/pkg/infra/db"
@@ -12,7 +14,11 @@ import (
"github.com/grafana/grafana/pkg/services/authz/zanzana"
)
var tracer = otel.Tracer("github.com/grafana/grafana/pkg/accesscontrol/reconciler")
var tracer = otel.Tracer("github.com/grafana/grafana/pkg/accesscontrol/migrator")
// A TupleCollector is responsible to build and store [openfgav1.TupleKey] into provided tuple map.
// They key used should be a unique group key for the collector so we can skip over an already synced group.
type TupleCollector func(ctx context.Context, namespace string, tuples map[string][]*openfgav1.TupleKey) error
// ZanzanaReconciler is a component to reconcile RBAC permissions to zanzana.
// We should rewrite the migration after we have "migrated" all possible actions
@@ -20,6 +26,7 @@ var tracer = otel.Tracer("github.com/grafana/grafana/pkg/accesscontrol/reconcile
type ZanzanaReconciler struct {
lock *serverlock.ServerLockService
log log.Logger
store db.DB
client zanzana.Client
// 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.
@@ -31,6 +38,7 @@ func NewZanzanaReconciler(client zanzana.Client, store db.DB, lock *serverlock.S
client: client,
lock: lock,
log: log.New("zanzana.reconciler"),
store: store,
reconcilers: []resourceReconciler{
newResourceReconciler(
"team memberships",
@@ -93,23 +101,47 @@ func (r *ZanzanaReconciler) Reconcile(ctx context.Context) error {
}
func (r *ZanzanaReconciler) reconcile(ctx context.Context) {
run := func(ctx context.Context) {
run := func(ctx context.Context, namespace string) {
now := time.Now()
for _, reconciler := range r.reconcilers {
if err := reconciler.reconcile(ctx); err != nil {
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))
}
if r.lock == nil {
run(ctx)
orgIds, err := r.getOrgs(ctx)
if err != nil {
return
}
// We ignore the error for now
_ = r.lock.LockExecuteAndRelease(ctx, "zanzana-reconciliation", 10*time.Hour, func(ctx context.Context) {
run(ctx)
for _, orgId := range orgIds {
ns := claims.OrgNamespaceFormatter(orgId)
if r.lock == nil {
run(ctx, ns)
return
}
// We ignore the error for now
_ = r.lock.LockExecuteAndRelease(ctx, "zanzana-reconciliation", 10*time.Hour, func(ctx context.Context) {
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
}
pkg/services/accesscontrol/dualwrite/resource_reconciler.go
+19 -9
View File
@@ -4,16 +4,19 @@ import (
"context"
"fmt"
"github.com/grafana/authlib/claims"
openfgav1 "github.com/openfga/api/proto/openfga/v1"
"github.com/grafana/grafana/pkg/services/authz/zanzana"
"github.com/grafana/grafana/pkg/services/authz/zanzana/common"
authzextv1 "github.com/grafana/grafana/pkg/services/authz/zanzana/proto/v1"
)
// legacyTupleCollector collects tuples groupd by object and tupleKey
type legacyTupleCollector func(ctx context.Context) (map[string]map[string]*openfgav1.TupleKey, error)
type legacyTupleCollector func(ctx context.Context, orgId int64) (map[string]map[string]*openfgav1.TupleKey, error)
// zanzanaTupleCollector collects tuples from zanzana for given object
type zanzanaTupleCollector func(ctx context.Context, client zanzana.Client, object string) (map[string]*openfgav1.TupleKey, error)
type zanzanaTupleCollector func(ctx context.Context, client zanzana.Client, object string, namespace string) (map[string]*openfgav1.TupleKey, error)
type resourceReconciler struct {
name string
@@ -26,9 +29,14 @@ func newResourceReconciler(name string, legacy legacyTupleCollector, zanzana zan
return resourceReconciler{name, legacy, zanzana, client}
}
func (r resourceReconciler) reconcile(ctx context.Context) error {
func (r resourceReconciler) reconcile(ctx context.Context, namespace string) error {
info, err := claims.ParseNamespace(namespace)
if err != nil {
return err
}
// 1. Fetch grafana resources stored in grafana db.
res, err := r.legacy(ctx)
res, err := r.legacy(ctx, info.OrgID)
if err != nil {
return fmt.Errorf("failed to collect legacy tuples for %s: %w", r.name, err)
}
@@ -41,7 +49,7 @@ func (r resourceReconciler) reconcile(ctx context.Context) error {
for object, tuples := range res {
// 2. Fetch all tuples for given object.
// Due to limitations in open fga api we need to collect tuples per object
zanzanaTuples, err := r.zanzana(ctx, r.client, object)
zanzanaTuples, err := r.zanzana(ctx, r.client, object, namespace)
if err != nil {
return fmt.Errorf("failed to collect zanzanaa tuples for %s: %w", r.name, err)
}
@@ -85,8 +93,9 @@ func (r resourceReconciler) reconcile(ctx context.Context) error {
if len(deletes) > 0 {
err := batch(deletes, 100, func(items []*openfgav1.TupleKeyWithoutCondition) error {
return r.client.Write(ctx, &openfgav1.WriteRequest{
Deletes: &openfgav1.WriteRequestDeletes{TupleKeys: items},
return r.client.Write(ctx, &authzextv1.WriteRequest{
Namespace: namespace,
Deletes: &authzextv1.WriteRequestDeletes{TupleKeys: common.ToAuthzExtTupleKeysWithoutCondition(items)},
})
})
@@ -97,8 +106,9 @@ func (r resourceReconciler) reconcile(ctx context.Context) error {
if len(writes) > 0 {
err := batch(writes, 100, func(items []*openfgav1.TupleKey) error {
return r.client.Write(ctx, &openfgav1.WriteRequest{
Writes: &openfgav1.WriteRequestWrites{TupleKeys: items},
return r.client.Write(ctx, &authzextv1.WriteRequest{
Namespace: namespace,
Writes: &authzextv1.WriteRequestWrites{TupleKeys: common.ToAuthzExtTupleKeys(items)},
})
})