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

Zanzana: Use cache for both streamed and non-stream version of list objects (#98882)

Browse Source 
* Add prefix constants and use string builders / string concatinations

* Use cache for both streamed and non-stream versions of list objects

* Remove unused constants
This commit is contained in:
Karl Persson
2025-01-14 16:00:59 +01:00
committed by GitHub
parent 7151ea6abc
commit ce0d986673
4 changed files with 113 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/services/authz/zanzana/common/tuple.go
+20 -8
View File
@@ -1,7 +1,7 @@
package common
import (
"fmt"
"strings"
openfgav1 "github.com/openfga/api/proto/openfga/v1"
"google.golang.org/protobuf/types/known/structpb"
@@ -28,6 +28,12 @@ const (
TypeGroupResouce string = "group_resource"
)
const (
TypeFolderPrefix string = TypeFolder + ":"
TypeResourcePrefix string = TypeResource + ":"
TypeGroupResoucePrefix string = TypeGroupResouce + ":"
)
const (
RelationTeamMember string = "member"
RelationTeamAdmin string = "admin"
@@ -140,31 +146,37 @@ func isValidRelation(relation string, valid []string) bool {
}
func FolderResourceRelation(relation string) string {
return fmt.Sprintf("%s_%s", TypeResource, relation)
return TypeResource + "_" + relation
}
func NewTypedIdent(typ string, name string) string {
return fmt.Sprintf("%s:%s", typ, name)
return typ + ":" + name
}
func NewResourceIdent(group, resource, subresource, name string) string {
return fmt.Sprintf("%s:%s/%s", TypeResource, FormatGroupResource(group, resource, subresource), name)
return TypeResourcePrefix + FormatGroupResource(group, resource, subresource) + "/" + name
}
func NewFolderIdent(name string) string {
return fmt.Sprintf("%s:%s", TypeFolder, name)
return TypeFolderPrefix + name
}
func NewGroupResourceIdent(group, resource, subresource string) string {
return fmt.Sprintf("%s:%s", TypeGroupResouce, FormatGroupResource(group, resource, subresource))
return TypeGroupResoucePrefix + FormatGroupResource(group, resource, subresource)
}
func FormatGroupResource(group, resource, subresource string) string {
b := strings.Builder{}
b.WriteString(group)
b.WriteRune('/')
b.WriteString(resource)
if subresource != "" {
return fmt.Sprintf("%s/%s/%s", group, resource, subresource)
b.WriteRune('/')
b.WriteString(subresource)
}
return fmt.Sprintf("%s/%s", group, resource)
return b.String()
}
func NewResourceTuple(subject, relation, group, resource, subresource, name string) *openfgav1.TupleKey {
pkg/services/authz/zanzana/server/server.go
+1 -7
View File
@@ -20,13 +20,7 @@ import (
"github.com/grafana/grafana/pkg/setting"
)
const (
resourceType = "resource"
namespaceType = "namespace"
folderTypePrefix = "folder:"
cacheCleanInterval = 2 * time.Minute
)
const cacheCleanInterval = 2 * time.Minute
var _ authzv1.AuthzServiceServer = (*Server)(nil)
var _ authzextv1.AuthzExtentionServiceServer = (*Server)(nil)
pkg/services/authz/zanzana/server/server_list.go
+92 -13
View File
@@ -2,7 +2,10 @@ package server
import (
"context"
"fmt"
"encoding/base64"
"errors"
"hash/fnv"
"io"
"strings"
authzv1 "github.com/grafana/authlib/authz/proto/v1"
@@ -48,13 +51,6 @@ func (s *Server) List(ctx context.Context, r *authzv1.ListRequest) (*authzv1.Lis
return s.listTyped(ctx, r.GetSubject(), relation, resource, contextuals, store)
}
func (s *Server) listObjects(ctx context.Context, req *openfgav1.ListObjectsRequest) (*openfgav1.ListObjectsResponse, error) {
if s.cfg.UseStreamedListObjects {
return s.streamedListObjects(ctx, req)
}
return s.openfga.ListObjects(ctx, req)
}
func (s *Server) listTyped(ctx context.Context, subject, relation string, resource common.ResourceInfo, contextuals *openfgav1.ContextualTupleKeys, store *storeInfo) (*authzv1.ListResponse, error) {
if !resource.IsValidRelation(relation) {
return &authzv1.ListResponse{}, nil
@@ -125,20 +121,103 @@ func (s *Server) listGeneric(ctx context.Context, subject, relation string, reso
return &authzv1.ListResponse{
Folders: folderObject(folders),
Items: directObjects(resource.GroupResource(), objects),
Items: genericObjects(resource.GroupResource(), objects),
}, nil
}
func (s *Server) listObjects(ctx context.Context, req *openfgav1.ListObjectsRequest) (*openfgav1.ListObjectsResponse, error) {
fn := s.openfga.ListObjects
if s.cfg.UseStreamedListObjects {
fn = s.streamedListObjects
}
if s.cfg.CheckQueryCache {
return s.listObjectCached(ctx, req, fn)
}
return fn(ctx, req)
}
type listFn func(ctx context.Context, req *openfgav1.ListObjectsRequest) (*openfgav1.ListObjectsResponse, error)
func (s *Server) listObjectCached(ctx context.Context, req *openfgav1.ListObjectsRequest, fn listFn) (*openfgav1.ListObjectsResponse, error) {
ctx, span := tracer.Start(ctx, "server.listObjectCached")
defer span.End()
key, err := getRequestHash(req)
if err != nil {
return nil, err
}
if res, ok := s.cache.Get(key); ok {
return res.(*openfgav1.ListObjectsResponse), nil
}
res, err := fn(ctx, req)
if err != nil {
return nil, err
}
s.cache.Set(key, res, 0)
return res, nil
}
func (s *Server) streamedListObjects(ctx context.Context, req *openfgav1.ListObjectsRequest) (*openfgav1.ListObjectsResponse, error) {
ctx, span := tracer.Start(ctx, "server.streamedListObjects")
defer span.End()
r := &openfgav1.StreamedListObjectsRequest{
StoreId: req.GetStoreId(),
AuthorizationModelId: req.GetAuthorizationModelId(),
Type: req.GetType(),
Relation: req.GetRelation(),
User: req.GetUser(),
Context: req.GetContext(),
ContextualTuples: req.ContextualTuples,
}
stream, err := s.openfgaClient.StreamedListObjects(ctx, r)
if err != nil {
return nil, err
}
var objects []string
for {
res, err := stream.Recv()
if err != nil {
if errors.Is(err, io.EOF) {
break
}
return nil, err
}
objects = append(objects, res.GetObject())
}
return &openfgav1.ListObjectsResponse{
Objects: objects,
}, nil
}
func getRequestHash(req *openfgav1.ListObjectsRequest) (string, error) {
hash := fnv.New64a()
_, err := hash.Write([]byte(req.String()))
if err != nil {
return "", err
}
return base64.StdEncoding.EncodeToString(hash.Sum(nil)), nil
}
func typedObjects(typ string, objects []string) []string {
prefix := fmt.Sprintf("%s:", typ)
prefix := typ + ":"
for i := range objects {
objects[i] = strings.TrimPrefix(objects[i], prefix)
}
return objects
}
func directObjects(gr string, objects []string) []string {
prefix := fmt.Sprintf("%s:%s/", resourceType, gr)
func genericObjects(gr string, objects []string) []string {
prefix := common.TypeResourcePrefix + gr + "/"
for i := range objects {
objects[i] = strings.TrimPrefix(objects[i], prefix)
}
@@ -147,7 +226,7 @@ func directObjects(gr string, objects []string) []string {
func folderObject(objects []string) []string {
for i := range objects {
objects[i] = strings.TrimPrefix(objects[i], folderTypePrefix)
objects[i] = strings.TrimPrefix(objects[i], common.TypeFolderPrefix)
}
return objects
}
pkg/services/authz/zanzana/server/server_list_streamed.go
-85
View File
@@ -1,85 +0,0 @@
package server
import (
"context"
"encoding/base64"
"errors"
"hash/fnv"
"io"
openfgav1 "github.com/openfga/api/proto/openfga/v1"
)
func (s *Server) streamedListObjects(ctx context.Context, req *openfgav1.ListObjectsRequest) (*openfgav1.ListObjectsResponse, error) {
if !s.cfg.CheckQueryCache {
return s.listObjectsWithStream(ctx, req)
}
return s.listObjectsWithStreamCached(ctx, req)
}
func (s *Server) listObjectsWithStreamCached(ctx context.Context, req *openfgav1.ListObjectsRequest) (*openfgav1.ListObjectsResponse, error) {
ctx, span := tracer.Start(ctx, "server.listObjectsWithStreamCached")
defer span.End()
reqHash, err := getRequestHash(req)
if err != nil {
return nil, err
}
if res, ok := s.cache.Get(reqHash); ok {
return res.(*openfgav1.ListObjectsResponse), nil
}
res, err := s.listObjectsWithStream(ctx, req)
if err != nil {
return nil, err
}
s.cache.Set(reqHash, res, 0)
return res, nil
}
func (s *Server) listObjectsWithStream(ctx context.Context, req *openfgav1.ListObjectsRequest) (*openfgav1.ListObjectsResponse, error) {
ctx, span := tracer.Start(ctx, "server.listObjectsWithStream")
defer span.End()
r := &openfgav1.StreamedListObjectsRequest{
StoreId: req.GetStoreId(),
AuthorizationModelId: req.GetAuthorizationModelId(),
Type: req.GetType(),
Relation: req.GetRelation(),
User: req.GetUser(),
Context: req.GetContext(),
ContextualTuples: req.ContextualTuples,
}
stream, err := s.openfgaClient.StreamedListObjects(ctx, r)
if err != nil {
return nil, err
}
var objects []string
for {
res, err := stream.Recv()
if err != nil {
if errors.Is(err, io.EOF) {
break
}
return nil, err
}
objects = append(objects, res.GetObject())
}
return &openfgav1.ListObjectsResponse{
Objects: objects,
}, nil
}
func getRequestHash(req *openfgav1.ListObjectsRequest) (string, error) {
hash := fnv.New64a()
_, err := hash.Write([]byte(req.String()))
if err != nil {
return "", err
}
return base64.StdEncoding.EncodeToString(hash.Sum(nil)), nil
}