RBAC caching

pkg/accesscontrol/access_set.go

@@ -7,6 +7,7 @@ import (
 )
 
 type AccessSet struct {
+	ID  string
 	set map[key]resourceAccessSet
 }
 

pkg/accesscontrol/access_store.go

@@ -1,7 +1,17 @@
 package accesscontrol
 
 import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"sort"
+	"sync"
+	"time"
+
 	v1 "github.com/rancher/wrangler-api/pkg/generated/controllers/rbac/v1"
+	"github.com/rancher/wrangler/pkg/kv"
+	k8srbac "k8s.io/api/rbac/v1"
+	"k8s.io/apimachinery/pkg/util/cache"
 	"k8s.io/apiserver/pkg/authentication/user"
 )
 
@@ -12,19 +22,99 @@ type AccessSetLookup interface {
 type AccessStore struct {
 	users         *policyRuleIndex
 	groups        *policyRuleIndex
+	roleRevisions sync.Map
+	cache         *cache.LRUExpireCache
 }
 
-func NewAccessStore(rbac v1.Interface) *AccessStore {
-	return &AccessStore{
+type roleKey struct {
+	namespace string
+	name      string
+}
+
+func NewAccessStore(ctx context.Context, cacheResults bool, rbac v1.Interface) *AccessStore {
+	as := &AccessStore{
 		users:  newPolicyRuleIndex(true, rbac),
 		groups: newPolicyRuleIndex(false, rbac),
 	}
+	rbac.Role().OnChange(ctx, "role-revision-indexer", as.onRoleChanged)
+	rbac.ClusterRole().OnChange(ctx, "role-revision-indexer", as.onClusterRoleChanged)
+	if cacheResults {
+		as.cache = cache.NewLRUExpireCache(1000)
+	}
+	return as
+}
+
+func (l *AccessStore) onClusterRoleChanged(key string, cr *k8srbac.ClusterRole) (role *k8srbac.ClusterRole, err error) {
+	if cr == nil {
+		l.roleRevisions.Delete(roleKey{
+			name: key,
+		})
+	} else {
+		l.roleRevisions.Store(roleKey{
+			name: key,
+		}, cr.ResourceVersion)
+	}
+	return cr, nil
+}
+
+func (l *AccessStore) onRoleChanged(key string, cr *k8srbac.Role) (role *k8srbac.Role, err error) {
+	if cr == nil {
+		namespace, name := kv.Split(key, "/")
+		l.roleRevisions.Delete(roleKey{
+			name:      name,
+			namespace: namespace,
+		})
+	} else {
+		l.roleRevisions.Store(roleKey{
+			name:      cr.Name,
+			namespace: cr.Namespace,
+		}, cr.ResourceVersion)
+	}
+	return cr, nil
 }
 
 func (l *AccessStore) AccessFor(user user.Info) *AccessSet {
+	var cacheKey string
+	if l.cache != nil {
+		cacheKey = l.CacheKey(user)
+		val, ok := l.cache.Get(cacheKey)
+		if ok {
+			as, _ := val.(*AccessSet)
+			return as
+		}
+	}
+
 	result := l.users.get(user.GetName())
 	for _, group := range user.GetGroups() {
 		result.Merge(l.groups.get(group))
 	}
+
+	if l.cache != nil {
+		result.ID = cacheKey
+		l.cache.Add(cacheKey, result, 24*time.Hour)
+	}
+
 	return result
 }
+
+func (l *AccessStore) CacheKey(user user.Info) string {
+	roles := map[roleKey]struct{}{}
+	l.users.addRolesToMap(roles, user.GetName())
+	for _, group := range user.GetGroups() {
+		l.groups.addRolesToMap(roles, group)
+	}
+
+	revs := make([]string, 0, len(roles))
+	for roleKey := range roles {
+		val, _ := l.roleRevisions.Load(roleKey)
+		rev, _ := val.(string)
+		revs = append(revs, roleKey.namespace+"/"+roleKey.name+":"+rev)
+	}
+
+	sort.Strings(revs)
+	d := sha256.New()
+	for _, rev := range revs {
+		d.Write([]byte(rev))
+	}
+	return hex.EncodeToString(d.Sum(nil))
+}

pkg/accesscontrol/policy_rule_index.go

@@ -44,6 +44,15 @@ func newPolicyRuleIndex(user bool, rbac v1.Interface) *policyRuleIndex {
 
 func (p *policyRuleIndex) clusterRoleBindingBySubjectIndexer(crb *rbacv1.ClusterRoleBinding) (result []string, err error) {
 	for _, subject := range crb.Subjects {
+		if subject.APIGroup == rbacGroup && subject.Kind == p.kind && crb.RoleRef.Kind == "ClusterRole" {
+			result = append(result, subject.Name)
+		}
+	}
+	return
+}
+
+func (p *policyRuleIndex) roleBindingBySubject(rb *rbacv1.RoleBinding) (result []string, err error) {
+	for _, subject := range rb.Subjects {
 		if subject.APIGroup == rbacGroup && subject.Kind == p.kind {
 			result = append(result, subject.Name)
 		}
@@ -51,13 +60,26 @@ func (p *policyRuleIndex) clusterRoleBindingBySubjectIndexer(crb *rbacv1.Cluster
 	return
 }
 
-func (p *policyRuleIndex) roleBindingBySubject(crb *rbacv1.RoleBinding) (result []string, err error) {
-	for _, subject := range crb.Subjects {
-		if subject.APIGroup == rbacGroup && subject.Kind == p.kind {
-			result = append(result, subject.Name)
+func (p *policyRuleIndex) addRolesToMap(roles map[roleKey]struct{}, subjectName string) {
+	for _, crb := range p.getClusterRoleBindings(subjectName) {
+		roles[roleKey{
+			name: crb.RoleRef.Name,
+		}] = struct{}{}
+	}
+
+	for _, rb := range p.getRoleBindings(subjectName) {
+		switch rb.RoleRef.Kind {
+		case "Role":
+			roles[roleKey{
+				name:      rb.RoleRef.Name,
+				namespace: rb.Namespace,
+			}] = struct{}{}
+		case "ClusterRole":
+			roles[roleKey{
+				name: rb.RoleRef.Name,
+			}] = struct{}{}
 		}
 	}
-	return
 }
 
 func (p *policyRuleIndex) get(subjectName string) *AccessSet {

pkg/schema/collection.go

@@ -3,15 +3,15 @@ package schema
 import (
 	"context"
 	"strings"
+	"sync"
 
 	"github.com/rancher/steve/pkg/accesscontrol"
 	"github.com/rancher/steve/pkg/attributes"
-	"github.com/rancher/steve/pkg/schema/table"
 	"github.com/rancher/steve/pkg/schemaserver/types"
-	"github.com/rancher/wrangler/pkg/data"
 	"github.com/rancher/wrangler/pkg/name"
-	"github.com/rancher/wrangler/pkg/schemas"
+	"github.com/sirupsen/logrus"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/util/cache"
 	"k8s.io/apiserver/pkg/authentication/user"
 )
 
@@ -28,7 +28,11 @@ type Collection struct {
 	templates  map[string]*Template
 	byGVR      map[schema.GroupVersionResource]string
 	byGVK      map[schema.GroupVersionKind]string
+	cache      *cache.LRUExpireCache
+	lock       sync.RWMutex
 
+	ctx     context.Context
+	running map[string]func()
 	as      accesscontrol.AccessSetLookup
 }
 
@@ -41,19 +45,19 @@ type Template struct {
 	Store        types.Store
 	Start        func(ctx context.Context) error
 	StoreFactory func(types.Store) types.Store
-	Mapper          schemas.Mapper
-	Columns         []table.Column
-	ComputedColumns func(data.Object)
 }
 
-func NewCollection(baseSchema *types.APISchemas, access accesscontrol.AccessSetLookup) *Collection {
+func NewCollection(ctx context.Context, baseSchema *types.APISchemas, access accesscontrol.AccessSetLookup) *Collection {
 	return &Collection{
 		baseSchema: baseSchema,
 		schemas:    map[string]*types.APISchema{},
 		templates:  map[string]*Template{},
 		byGVR:      map[schema.GroupVersionResource]string{},
 		byGVK:      map[schema.GroupVersionKind]string{},
+		cache:      cache.NewLRUExpireCache(1000),
+		ctx:        ctx,
 		as:         access,
+		running:    map[string]func(){},
 	}
 }
 
@@ -70,18 +74,58 @@ func (c *Collection) Reset(schemas map[string]*types.APISchema) {
 		if gvk.Kind != "" {
 			byGVK[gvk] = s.ID
 		}
+
+		c.applyTemplates(s)
 	}
 
+	c.lock.Lock()
+	c.startStopTemplate(schemas)
 	c.schemas = schemas
 	c.byGVR = byGVR
 	c.byGVK = byGVK
+	for _, k := range c.cache.Keys() {
+		c.cache.Remove(k)
+	}
+	c.lock.Unlock()
+}
+
+func (c *Collection) startStopTemplate(schemas map[string]*types.APISchema) {
+	for id := range schemas {
+		if _, ok := c.running[id]; ok {
+			continue
+		}
+		template := c.templates[id]
+		if template == nil || template.Start == nil {
+			continue
+		}
+
+		subCtx, cancel := context.WithCancel(c.ctx)
+		if err := template.Start(subCtx); err != nil {
+			logrus.Errorf("failed to start schema template: %s", id)
+			continue
+		}
+		c.running[id] = cancel
+	}
+
+	for id, cancel := range c.running {
+		if _, ok := schemas[id]; !ok {
+			cancel()
+			delete(c.running, id)
+		}
+	}
 }
 
 func (c *Collection) Schema(id string) *types.APISchema {
+	c.lock.RLock()
+	defer c.lock.RUnlock()
+
 	return c.schemas[id]
 }
 
 func (c *Collection) IDs() (result []string) {
+	c.lock.RLock()
+	defer c.lock.RUnlock()
+
 	seen := map[string]bool{}
 	for _, id := range c.byGVR {
 		if seen[id] {
@@ -94,6 +138,9 @@ func (c *Collection) IDs() (result []string) {
 }
 
 func (c *Collection) ByGVR(gvr schema.GroupVersionResource) string {
+	c.lock.RLock()
+	defer c.lock.RUnlock()
+
 	id, ok := c.byGVR[gvr]
 	if ok {
 		return id
@@ -107,14 +154,16 @@ func (c *Collection) ByGVR(gvr schema.GroupVersionResource) string {
 }
 
 func (c *Collection) ByGVK(gvk schema.GroupVersionKind) string {
+	c.lock.RLock()
+	defer c.lock.RUnlock()
+
 	return c.byGVK[gvk]
 }
 
-func (c *Collection) TemplateForSchemaID(id string) *Template {
-	return c.templates[id]
-}
-
 func (c *Collection) AddTemplate(template *Template) {
+	c.lock.RLock()
+	defer c.lock.RUnlock()
+
 	if template.Kind != "" {
 		c.templates[template.Group+"/"+template.Kind] = template
 	}

pkg/schema/factory.go

@@ -3,13 +3,12 @@ package schema
 import (
 	"fmt"
 	"net/http"
+	"time"
 
 	"github.com/rancher/steve/pkg/accesscontrol"
 	"github.com/rancher/steve/pkg/attributes"
-	"github.com/rancher/steve/pkg/schema/table"
 	"github.com/rancher/steve/pkg/schemaserver/builtin"
 	"github.com/rancher/steve/pkg/schemaserver/types"
-	"github.com/rancher/wrangler/pkg/schemas"
 	"k8s.io/apiserver/pkg/authentication/user"
 )
 
@@ -18,19 +17,31 @@ func newSchemas() (*types.APISchemas, error) {
 	if err := apiSchemas.AddSchemas(builtin.Schemas); err != nil {
 		return nil, err
 	}
-	apiSchemas.InternalSchemas.DefaultMapper = func() schemas.Mapper {
-		return newDefaultMapper()
-	}
 
 	return apiSchemas, nil
 }
 
 func (c *Collection) Schemas(user user.Info) (*types.APISchemas, error) {
 	access := c.as.AccessFor(user)
-	return c.schemasForSubject(access)
+	val, ok := c.cache.Get(access.ID)
+	if ok {
+		schemas, _ := val.(*types.APISchemas)
+		return schemas, nil
+	}
+
+	schemas, err := c.schemasForSubject(access)
+	if err != nil {
+		return nil, err
+	}
+
+	c.cache.Add(access.ID, schemas, 24*time.Hour)
+	return schemas, nil
 }
 
 func (c *Collection) schemasForSubject(access *accesscontrol.AccessSet) (*types.APISchemas, error) {
+	c.lock.RLock()
+	defer c.lock.RUnlock()
+
 	result, err := newSchemas()
 	if err != nil {
 		return nil, err
@@ -81,8 +92,6 @@ func (c *Collection) schemasForSubject(access *accesscontrol.AccessSet) (*types.
 			s.CollectionMethods = append(s.CollectionMethods, http.MethodPost)
 		}
 
-		c.applyTemplates(result, s)
-
 		if err := result.AddSchema(*s); err != nil {
 			return nil, err
 		}
@@ -91,7 +100,10 @@ func (c *Collection) schemasForSubject(access *accesscontrol.AccessSet) (*types.
 	return result, nil
 }
 
-func (c *Collection) applyTemplates(schemas *types.APISchemas, schema *types.APISchema) {
+func (c *Collection) applyTemplates(schema *types.APISchema) {
+	c.lock.RLock()
+	defer c.lock.RUnlock()
+
 	templates := []*Template{
 		c.templates[schema.ID],
 		c.templates[fmt.Sprintf("%s/%s", attributes.Group(schema), attributes.Kind(schema))],
@@ -102,9 +114,6 @@ func (c *Collection) applyTemplates(schemas *types.APISchemas, schema *types.API
 		if t == nil {
 			continue
 		}
-		if t.Mapper != nil {
-			schemas.InternalSchemas.AddMapper(schema.ID, t.Mapper)
-		}
 		if schema.Formatter == nil {
 			schema.Formatter = t.Formatter
 		}
@@ -118,8 +127,5 @@ func (c *Collection) applyTemplates(schemas *types.APISchemas, schema *types.API
 		if t.Customize != nil {
 			t.Customize(schema)
 		}
-		if len(t.Columns) > 0 {
-			schemas.InternalSchemas.AddMapper(schema.ID, table.NewColumns(t.ComputedColumns, t.Columns...))
-		}
 	}
 }